在 contextManager 下使用 unittest 模拟 MysqlConnection
Mock MysqlConnection with unittest under a contextManager
我正在尝试测试一种在 with 块中检索 mysql 连接的方法,但是当我尝试从中获取模拟游标时,游标的执行结果也不正确嘲笑。
这里是获取连接的方法
def has_pending_task(self, host, port):
with self.get_mysql_connection(host, port) as conn:
cursor = conn.cursor()
table = self.config.get_table()
cursor.execute(
'select count(*) from {} where status in ("PENDING", "RUNNING") and host=%s'.format(
table), host)
count = cursor.fetchone()
cursor.close()
return count > 0
@contextmanager
def get_mysql_connection(self, host, port):
db_credentials = self.config.get_credentials()
schema = self.config.get_schema()
if not db_credentials:
logging.error(
"There was an error getting db info for cluster {}. Please review the config".format(self.cluster))
return None
conn = None
try:
conn = MySQLConnection(user=db_credentials[0], password=db_credentials[1],
host=host, port=port, database=schema)
except Error as error:
logging.error("Couldn't create a mysql connection: {}".format(error.msg))
return conn
测试看起来像这样:
def test_has_no_pending_tasks(self):
with mock.patch("my_package.my_class.Config") as config_mock:
config = config_mock.return_value
config_mock.get_instance.return_value = config
config.get_table.return_value = "table"
with mock.patch("mysql.connector.connection.MySQLConnection") as mysql_mock:
mysql = mysql_mock.return_value
with mock.patch('package.MyClass.get_mysql_connection') as conn_mock:
conn_mock.return_value = mysql
cursor = mysql_mock.cursor.return_value
cursor.fetchone.return_value = 0
Orchestrator("host", 3306).has_pending_tasks("host", 3306)
conn_mock.assert_called_once_with("host", 3306)
问题是因为它没有正确地模拟事物,测试函数中的 return 何时出现,count 是 MagicMock 的一个实例并使测试崩溃.
知道我遗漏了什么吗?
谢谢!
根据 的说法,问题出在上下文管理器上。像
一样模拟 get_mysql_connection() 方法
onn_mock.return_value.__enter__.return_value = mysql
它完美地完成了技巧
我正在尝试测试一种在 with 块中检索 mysql 连接的方法,但是当我尝试从中获取模拟游标时,游标的执行结果也不正确嘲笑。
这里是获取连接的方法
def has_pending_task(self, host, port):
with self.get_mysql_connection(host, port) as conn:
cursor = conn.cursor()
table = self.config.get_table()
cursor.execute(
'select count(*) from {} where status in ("PENDING", "RUNNING") and host=%s'.format(
table), host)
count = cursor.fetchone()
cursor.close()
return count > 0
@contextmanager
def get_mysql_connection(self, host, port):
db_credentials = self.config.get_credentials()
schema = self.config.get_schema()
if not db_credentials:
logging.error(
"There was an error getting db info for cluster {}. Please review the config".format(self.cluster))
return None
conn = None
try:
conn = MySQLConnection(user=db_credentials[0], password=db_credentials[1],
host=host, port=port, database=schema)
except Error as error:
logging.error("Couldn't create a mysql connection: {}".format(error.msg))
return conn
测试看起来像这样:
def test_has_no_pending_tasks(self):
with mock.patch("my_package.my_class.Config") as config_mock:
config = config_mock.return_value
config_mock.get_instance.return_value = config
config.get_table.return_value = "table"
with mock.patch("mysql.connector.connection.MySQLConnection") as mysql_mock:
mysql = mysql_mock.return_value
with mock.patch('package.MyClass.get_mysql_connection') as conn_mock:
conn_mock.return_value = mysql
cursor = mysql_mock.cursor.return_value
cursor.fetchone.return_value = 0
Orchestrator("host", 3306).has_pending_tasks("host", 3306)
conn_mock.assert_called_once_with("host", 3306)
问题是因为它没有正确地模拟事物,测试函数中的 return 何时出现,count 是 MagicMock 的一个实例并使测试崩溃.
知道我遗漏了什么吗?
谢谢!
根据
onn_mock.return_value.__enter__.return_value = mysql
它完美地完成了技巧