Python FetchMany 循环的单元测试
Python Unit Test for FetchMany loop
我想对从 SQL 服务器数据库获取结果的一段代码进行单元测试,然后使用 fetchmany 进行循环。我没有使用 fetchall,因为我预计结果集会很大。
但是,模拟它会导致无限循环,因为我认为我无法正确设置 return 值。
所以,这是我要测试的代码
with(MsSqlIntegratedHook(mssql_conn_id="detbds")).get_conn() as conndet:
with conndet.cursor() as cursordet:
while True:
current_data = cursorswh.fetchmany(100)
if not current_data:
break
cursordet.executemany("insert into input.Events (EventTime, EventTypeID, EventPayload) values (getdate() ,1, ?)", current_data)
conndet.commit()
conndet.commit()
然而,由于行
,像下面这样的模拟导致它是无限的
current_data = cursorswh.fetchmany(100)
这是我试图模拟的方式:
cursor_mock = mock_MsSqlIntegratedHook.return_value.__enter__.return_value
cursor_mock.execute.return_value.fetchmany.return_value=[("123","SH", "1", "AUD", "100", "100", "100")]
也试过
mock_MsSqlIntegratedHook.get_conn.__enter__.cursor.__enter__.fetchmany.return_value=[("123","SH", "1", "AUD", "100", "100", "100")]
但是因为 current_data 是模拟对象,“if not current_data”永远不会 returns false。
我认为模拟没有正常进行,因为 fetchmany return 是模拟对象而不是我在模拟的 return_value 中指定的值。
我该如何处理?
如果您需要我的任何解释或澄清,请开枪。
对此很陌生。
所以,我终于想通了。 S.O post 谈到了这一点,Martijn Pieters 的回答谈到了使用“with”子句时的“enter”方法。
所以,最终的解决方案是
class MoveTest(unittest.TestCase):
@mock.patch('Module_positionmove.MsSqlIntegratedHook')
def test_method(self, mock_MsSqlIntegratedHook):
mock_con_cm = mock_MsSqlIntegratedHook.return_value
mock_con = mock_con_cm.get_conn.return_value.__enter__.return_value #because of using the with block, we get a __enter__
mock_con.cursor.return_value.__enter__.return_value.fetchmany.side_effect = [("123","SH", "1", "AUD", "100", "100", "100"), None]
livepositionswith52weeklowInstruments (("1","2"))
mock_con.cursor.return_value.__enter__.return_value.fetchmany.assert_called()
第二个成员作为None的列表的副作用和分配确保循环退出并调用“break”。
我想对从 SQL 服务器数据库获取结果的一段代码进行单元测试,然后使用 fetchmany 进行循环。我没有使用 fetchall,因为我预计结果集会很大。
但是,模拟它会导致无限循环,因为我认为我无法正确设置 return 值。
所以,这是我要测试的代码
with(MsSqlIntegratedHook(mssql_conn_id="detbds")).get_conn() as conndet:
with conndet.cursor() as cursordet:
while True:
current_data = cursorswh.fetchmany(100)
if not current_data:
break
cursordet.executemany("insert into input.Events (EventTime, EventTypeID, EventPayload) values (getdate() ,1, ?)", current_data)
conndet.commit()
conndet.commit()
然而,由于行
,像下面这样的模拟导致它是无限的current_data = cursorswh.fetchmany(100)
这是我试图模拟的方式:
cursor_mock = mock_MsSqlIntegratedHook.return_value.__enter__.return_value
cursor_mock.execute.return_value.fetchmany.return_value=[("123","SH", "1", "AUD", "100", "100", "100")]
也试过
mock_MsSqlIntegratedHook.get_conn.__enter__.cursor.__enter__.fetchmany.return_value=[("123","SH", "1", "AUD", "100", "100", "100")]
但是因为 current_data 是模拟对象,“if not current_data”永远不会 returns false。
我认为模拟没有正常进行,因为 fetchmany return 是模拟对象而不是我在模拟的 return_value 中指定的值。
我该如何处理?
如果您需要我的任何解释或澄清,请开枪。
对此很陌生。
所以,我终于想通了。
所以,最终的解决方案是
class MoveTest(unittest.TestCase):
@mock.patch('Module_positionmove.MsSqlIntegratedHook')
def test_method(self, mock_MsSqlIntegratedHook):
mock_con_cm = mock_MsSqlIntegratedHook.return_value
mock_con = mock_con_cm.get_conn.return_value.__enter__.return_value #because of using the with block, we get a __enter__
mock_con.cursor.return_value.__enter__.return_value.fetchmany.side_effect = [("123","SH", "1", "AUD", "100", "100", "100"), None]
livepositionswith52weeklowInstruments (("1","2"))
mock_con.cursor.return_value.__enter__.return_value.fetchmany.assert_called()
第二个成员作为None的列表的副作用和分配确保循环退出并调用“break”。