为什么我的单元测试失败并显示错误 'str' object has no attribute 'cursor'?

Why is my unittest failing with the error 'str' object has no attribute 'cursor'?

我有代码:

def get_address_query(trader_id, connection):
    cursor = connection.cursor()
    sql_address_data = "SELECT * FROM address where id = '%s' AND type = 'registered'"
    cursor.execute(sql_address_data, trader_id)
    address_query = cursor.fetchall()
    if address_query is None:
        raise Exception("address details do not exist")
    else:
        return address_query

我写了下面的测试:

    @mock.patch('lambda_function.pymysql')
    def test_get_address_query(self, mock_database_connection):
        mock_record = (('test',),('first',),('last',))
        mock_database_connection.cursor.return_value.fetchall.return_value = mock_record
        result = lambda_function.get_address_query('test', 'connection')
        self.assertEqual(result, mock_record)

但我遇到以下错误:

id = 'test', connection = 'connection'

    def get_address_query(id, connection):
>       cursor = connection.cursor()
E       AttributeError: 'str' object has no attribute 'cursor'

../lambda_function.py:92: AttributeError

我知道这个错误是由于我没有正确地模拟 connection.cursor.fetchall 但我不确定如何正确地做到这一点。任何帮助将不胜感激。

您将字符串 'connection' 作为第二个参数传递给了 get_address_query()

lambda_function.get_address_query('test', 'connection')

然后在 get_address_query() 中,它使用第二个参数 connection(它是一个字符串)调用方法 .cursor()

def get_address_query(trader_id, connection):
    cursor = connection.cursor()
    ...

所以这就像调用 'connection'.cursor() 显然会失败,因为 str 没有这样的方法。

相反,您需要做的是将模拟对象 mock_database_connection 传递给 connection 参数。这样,所有对 connection 的调用(例如 connection.cursor() )都将使用配置的模拟对象。

以下是您可能希望用作参考的最少测试代码:

import unittest
from unittest import mock


def get_address_query(trader_id, connection):
    cursor = connection.cursor()
    sql_address_data = "SELECT * FROM address where id = '%s' AND type = 'registered'"
    cursor.execute(sql_address_data, trader_id)
    address_query = cursor.fetchall()
    if address_query is None:
        raise Exception("address details do not exist")
    else:
        return address_query


class TestSample(unittest.TestCase):
    def test_get_address_query(self):
        mock_database_connection = mock.MagicMock()

        mock_record = (('test',),('first',),('last',))
        mock_database_connection.cursor.return_value.fetchall.return_value = mock_record

        result = get_address_query('test', mock_database_connection)

        self.assertEqual(result, (mock_record))


if __name__ == '__main__':
    unittest.main()

输出:

$ python3 test_src.py 
.
----------------------------------------------------------------------
Ran 1 test in 0.002s

OK
  • 现在,cursor.fetchall() 的响应与我们在模拟对象 mock_database_connection 中配置的一样。