pytest - 如何模拟 sqlalchemy.engine.cursor.LegacyCursorResult？

Question

我有以下类连接到我的项目中的 MSSQL Server 实例，

"""MSSql Connection class."""

import logging
import logging.config
import sqlalchemy

class MSSQLConnection:

   _connection = None
   _engine = None

def __init__(self, host, database, username, password):
    connection_string = self.build_connection_string(
        host, database, username, password)
    self._engine = sqlalchemy.create_engine(
        connection_string, fast_executemany=True,
        isolation_level="READ COMMITTED")

def connect(self):
    self._connection = self._engine.connect()

def get_result_tuple(self, table):
    logger = logging.getLogger()
    metadata = sqlalchemy.MetaData()
    db_table = sqlalchemy.Table(
        table, metadata, autoload=True, autoload_with=self._engine)
    query = sqlalchemy.select([db_table])
    transaction_id = self.get_transaction_id()
    result_proxy = self._connection.execute(query)
    return transaction_id, result_proxy

def get_transaction_id(self):
    """Get the transaction id."""
    connection = self._connection
    sql_query = sqlalchemy.text("SELECT CURRENT_TRANSACTION_ID()")
    result = connection.execute(sql_query)
    row = result.fetchone()
    return row[0]

def build_connection_string(self, host, database, username, password):
    connection_string = ('mssql+pyodbc://' + username + ':'
            + password + '@' + host + '/' + database
            + '?driver=ODBC+Driver+17+for+SQL+Server')
    return connection_string

我想模拟该方法

get_result_tuple

返回“transaction_id”和 sqlalchemy.engine.cursor.LegacyCursorResult 的实例。

如何模拟 sqlalchemy.engine.cursor.LegacyCursorResult 并在 ResultProxy 对象上返回一些虚拟数据？

调用者有以下代码，

mssql_connection = MSSQLConnection(
      host, database, username, password)
mssql_connection.connect()
result_tuple = mssql_connection.get_result_tuple(table)
transaction_id, result_proxy = result_tuple
logger.info(f'Transaction id = {transaction_id}')
current_date = date.today().strftime("%Y_%m_%d")
while True:
    partial_results = result_proxy.fetchmany(rows_fetch_limit)
    results_count = len(partial_results)
    if (partial_results == [] or results_count == 0):
        return
    else:
     // other logic

请指教。