How to mock cx_Oracle cursor in Python

user_10374013 0
Viewed 3394

I have two below functions in a class and I need to mock the database connection and cursor results. cx_Oracle.connect successfully patched it works as expected. But the cursor, callfunc, fetch_all doesn't get mock. Any idea what would be the correct syntax ? 

class dbconnect:

    def db_connect(self, connection_details):
        connection = cx_Oracle.connect(user_name,pwd,<connection_string>)
        return connection

    def execute_function(self, sqlFunction, args):
        cursor = self.connection.cursor()
        res=cursor.callfunc(sqlFunction, cursor.var(cx_Oracle.CURSOR), args)
        results = {'headers' : [x[0] for x in res.description],
            'data': res.fetchall()
        }
        cursor.close()
        return results

My Mocking code

with mock.patch('dbconnect.cx_Oracle.connect') as mockOracle:

   result_set = {}
   mockOracle.cursor.callfunc.fetch_all = result_set
python-unittest cx-oracle patch mocking python
2 Answers

You don't actually have to "mock" up the connection and cursor at all! You can simply subclass the connection and cursor yourself, like this:

class MyConnection(cx_Oracle.Connection):

    def cursor(self, scrollable=False):
        return MyCursor(self, scrollable)

class MyCursor(cx_Oracle.Cursor):

    def execute(self, sql, args):
        result = super(MyConnection, self).execute(sql, args)
        print("My mock execute...")
        return result

I'm not sure if that was what you were intending or if you knew about this possibility. With this you can add new functions and you can override or extend existing ones.

user_25103470

I had a similar issue so I patched the cx_Oracle.connect() method with a class that I created, so it both works within a context manager and returns the mock cursor. The cursor should work with both fetchmany and fetchall

import cx_Oracle

class MockCursor:
    def __init__(self, *args, **kwargs):
        self.fetched = False
        self.rows = None
        self.description = [("id",), ("name",)]

    def callfunc(self, *args):
        return self

    @staticmethod
    def var(*args):
        pass

    @staticmethod
    def execute(query):
        pass

    @staticmethod
    def close():
        pass

    def fetchmany(self, *args):
        if not self.fetched:
            self.rows = self.fetch_rows()
            self.fetched = True

        return next(self.rows)

    @staticmethod
    def fetchall():
        return [
            (1, "TEST A"),
            (2, "TEST B"),
            (3, "TEST C"),
        ]


    def fetch_rows(self):
        yield self.fetchall()
        yield None


class MockConnect:
    def __enter__(self, *args, **kwargs):
        return self

    def __exit__(self, type, value, traceback):
        pass

    @staticmethod
    def cursor():
        return MockCursor()

    @staticmethod
    def close():
        pass
 

@mock.patch.object(cx_Oracle, "connect")
def test_foo(mock_connect):
    mock_connect.return_value = MockConnect()
user_111480810
Related