从MySQLdb python库模拟callproc方法时是否可以模拟某些MySQL过程存在

问题描述

我有以下单元测试，需要测试烧瓶路径。

@unittest.mock.patch('flask_login.utils._get_user')
@unittest.mock.patch('flask_sqlalchemy.sqlAlchemy')
def test_attendances_management(self,sqlAlchemy,current_user):

    user = Magicmock()

    sqlalchemy = sqlAlchemy()
    sqlalchemy.engine.return_value = Magicmock()

    sqlalchemy.engine.raw_connection.return_value = Magicmock()

    sqlalchemy.engine.raw_connection.cursor.return_value = Magicmock()

    sqlalchemy.engine.raw_connection.cursor.callproc("get_project_members",[3,9,2020]).return_value = ((1,3,4),)

    current_user.return_value = user
    current_user.return_value.role.name = 'Administrator'
    current_user.return_value.id = 1

    response = self.client.get('attendances/management/3/2020/11')
    self.assertEqual(response.status_code,200)

一切正常（模拟数据库引擎，raw_connection和游标），但是我有一个问题，即get_project_members不存在。我需要模拟flask-sqlalchemy或MysqLdb库的哪一部分，以便伪造某些过程实际上存在，因为我使用flask-testing来创建例如表ExampleTable。 table .create(db.session.bind），并且我不知道如何创建存储过程或如何使用模拟来模拟该过程。

异常日志：

    File "C\:my_app\app\attendances\views.py",line 169,in management
    cursor.callproc("get_project_members",[project_id,month,year])
  File "c:\my_app\flaskvenv\lib\site-packages\MysqLdb\cursors.py",line 365,in callproc
    self._query(q)
  File "c:\my_app\flaskvenv\lib\site-packages\MysqLdb\cursors.py",line 411,in _query
    rowcount = self._do_query(q)
  File "c:\my_app\flaskvenv\lib\site-packages\MysqLdb\cursors.py",line 374,in _do_query
    db.query(q)
  File "c:\my_app\flaskvenv\lib\site-packages\MysqLdb\connections.py",line 277,in query
    _MysqL.connection.query(self,query)
_MysqL_exceptions.OperationalError: (1305,'PROCEDURE my_app_test_db.get_project_members does not exist')

解决方法

我刚刚使用cursor.execute方法解决了这个问题，该方法使用原始查询来创建MySQL过程。另外，我已经将db作为SQLAlchemy的实例，因此不需要模拟已经定义的连接。

    @unittest.mock.patch('flask_login.utils._get_user')
    def test_attendances_management(self,current_user):

        user = MagicMock()
        q = "CREATE PROCEDURE `get_project_members` (IN `project_id` INT,IN `month` INT,IN `year` INT) BEGIN END"
        connection = db.engine.raw_connection()
        cursor = connection.cursor()
        cursor.execute(q)

        current_user.return_value = user
        current_user.return_value.role.name = 'Administrator'
        current_user.return_value.id = 1

        response = self.client.get('attendances/management/3/2020/11')
        self.assertEqual(response.status_code,200)

类似的答案： Creating Stored Procedures with SQLAlchemy