Python Unit Test for FetchMany loop

Question

I would like to unit test a piece of code where I am fetching results from a SQL server database and then loop through using fetchmany . I am not using fetchall as I expect the resultset to be huge.

However, mocking it is causing an infinite loop as I think I am unable to set the return value properly.

So, this is the piece of code I would like to test

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()

However a mock like the below is causing it to be infinite because of the line

current_data = cursorswh.fetchmany(100)

Here is how I am trying to mock:

cursor_mock = mock_MsSqlIntegratedHook.return_value.__enter__.return_value
cursor_mock.execute.return_value.fetchmany.return_value=[("123","SH", "1", "AUD", "100", "100", "100")]

also tried

mock_MsSqlIntegratedHook.get_conn.__enter__.cursor.__enter__.fetchmany.return_value=[("123","SH", "1", "AUD", "100", "100", "100")]

But because current_data being a mock object, " if not current_data" never returns false.

I think the mocking isn't happening properly, because the fetchmany returns a Mock object instead of the value that I specify in the return_value of the mock.

How should I approach this?

If you need any explanations or clarifications from me, just shoot.

Quite new to this.

Answer 1

So, I finally figured it out. This S.O post touched upon this, Martijn Pieters's answer touched upon the "enter" method when "with" clauses are used.

So, the finally solution was

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()

The side effect and the assigning of the list with the 2nd member as None ensures that the loop is exited with the "break" being called.