Я построил небольшую MySQL ORM в python, используя библиотеку mysql.connector
, которую я использую для выполнения запросов.
Имя файла: mysql_db
и хранится в utils
Mysql ORM:
import mysql.connector
from config import settings
class PythonMySqlOp:
def __init__(self, host, user, passwd, database):
self.host = host
self.user = user
self.passwd = passwd
self.database = database
def create_mysql_connector(self):
mysql_connector = mysql.connector.connect(
host=self.host,
user=self.user,
passwd=self.passwd,
database=self.database
)
return mysql_connector
def execute_query(self, query):
try:
mysql_connector = self.create_mysql_connector()
cursor = mysql_connector.cursor(buffered=True)
print('INFO: Executing: {}'.format(query))
cursor.execute(query)
query_results = []
if not cursor:
return query_results
for query_result in cursor:
query_results.append(query_result)
cursor.close()
# print('Query results: {}'.format(query_results))
return query_results
except Exception as e:
raise Exception(e)
Это код, который требуетсяиспользовать это. Он называется batch_sql_cassandra_miner.py
и хранится в batch
from utils import mysql_db
from constants import mysql_constants
class BatchSqlCassandraMiner:
def __init__(self):
self.mysql_db = mysql_db.PythonMysqlOp(
mysql_constants.HOSTNAME,
mysql_constants.USER,
mysql_constants.PASSWORD,
mysql_constants.DATABASE
)
def miner(self):
'''a bunch of code here to build up the query'''
query = ('sql query')
try:
results = self.mysql_db.execute_query(
query
)
return results
except Exception as e:
raise Exception(e)
Мне нужно проверить этот метод miner
. Вот что я придумал.
from batch import batch_sql_cassandra_miner
from unittest import TestCase
from unittest.mock import patch
from constants import mysql_constants
class TestBatchSqlCassandraMiner(unittest.TestCase):
@patch('batch.batch_sql_cassandra_miner.BatchSqlCassandraMiner')
@patch('utils.mysql_db.PythonMysqlOp')
def setUp(self, mocked_mysql_op, mocked_batch_sql_cass_miner):
self.batch_sql_cass_miner = batch_sql_cassandra_miner.BatchSqlCassandraMiner()
``` Mocks ```
self.mocked_mysql_op = mocked_mysql_op
self.mocked_batch_sql_cass_miner = mocked_batch_sql_cass_miner
@patch('utils.mysql_db.PythonMysqlOp')
def test_constructor(self, mocked_mysql_db):
sql_cass_miner = batch_sql_cassandra_miner.BatchSqlCassandraMiner()
mocked_mysql_db.assert_called_with(
mysql_constants.HOSTNAME,
mysql_constants.USER,
mysql_constants.PASSWORD,
mysql_constants.DATABASE
)
def test_miner_query_execution(self):
pass
if __name__ == "__main__":
unittest.main()
До этого момента тест проходит. Пересмотренная переменная MySQL называется.
(myvirtualenv) user@hostname:~/workspace$ python -m unittest discover -s 'tests/integration/batch_test/' -p 'test_*.py'
INFO * Creating logs Directory: /logs
INFO * /logs: Directory already exists
..
----------------------------------------------------------------------
Ran 2 tests in 0.020s
OK
Итак, я проделал то же самое для test_miner_query_execution
from batch import batch_sql_cassandra_miner
from unittest import TestCase
from unittest.mock import patch
from constants import mysql_constants
class TestBatchSqlCassandraMiner(unittest.TestCase):
@patch('batch.batch_sql_cassandra_miner.BatchSqlCassandraMiner')
@patch('utils.mysql_db.PythonMysqlOp')
def setUp(self, mocked_mysql_op, mocked_batch_sql_cass_miner):
self.batch_sql_cass_miner = batch_sql_cassandra_miner.BatchSqlCassandraMiner()
``` Mocked Classes ```
self.mocked_mysql_op = mocked_mysql_op
self.mocked_batch_sql_cass_miner = mocked_batch_sql_cass_miner
@patch('utils.mysql_db.PythonMysqlOp')
def test_constructor(self, mocked_mysql_db):
sql_cass_miner = batch_sql_cassandra_miner.BatchSqlCassandraMiner()
mocked_mysql_db.assert_called_with(
mysql_constants.HOSTNAME,
mysql_constants.USER,
mysql_constants.PASSWORD,
mysql_constants.DATABASE
)
@patch('utils.mysql_db.PythonMysqlOp')
def test_miner_query_execution(self, mocked_mysql_db):
self.batch_sql_cass_miner.miner()
mocked_mysql_db.execute_query.assert_called_with()
if __name__ == "__main__":
unittest.main()
Сбой теста, при котором макет не вызывается
INFO * Creating logs Directory: /home/spark/alt_balaji/logs
INFO * /home/spark/alt_balaji/logs: Directory already exists
..F
======================================================================
FAIL: test_get_exp_records (test_batch_sql_cassandra_miner.TestBatchSqlCassandraDump)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/usr/lib/python3.5/unittest/mock.py", line 1157, in patched
return func(*args, **keywargs)
File "/home/spark/alt_balaji/tests/integration/batch_test/test_batch_sql_cassandra_miner.py", line 54, in test_get_exp_records
mocked_mysql_db.execute_query.assert_called_with()
File "/usr/lib/python3.5/unittest/mock.py", line 783, in assert_called_with
raise AssertionError('Expected call: %s\nNot called' % (expected,))
AssertionError: Expected call: execute_query()
Not called
----------------------------------------------------------------------
Ran 3 tests in 0.029s
FAILED (failures=1)
Итак, я подумал о создании фиктивного объекта. Затем я изменил метод test_miner_query_execution
следующим образом:
def test_miner_query_execution(self):
mocked_batch_sql_cass_miner_obj = self.mocked_batch_sql_cass_miner()
# print(vars(mocked_batch_sql_cass_miner_obj))
'''
By printing this object I noticed that, this mock did not have the mysql_db instance variable under `_mock_children`, where everythingelse was located.
So, I created the variable manually
'''
mocked_batch_sql_cass_miner_obj.mysql_db = self.mocked_mysql_db()
# print(vars(mocked_batch_sql_cass_miner_obj))
'''
This time, `mysql_db` was registered as a `NonCallableMagicMock` and still not under `_mock_children`
'''
#print(vars(mocked_batch_sql_cass_miner_obj.mysql_db))
'''
The mysql_db variable did indeed have an execute_query method
'''
mocked_batch_sql_cass_miner_obj.miner()
mocked_batch_sql_cass_miner_obj.mysql_db.execute_query.assert_called_with()
Нет результатов:
AssertionError: Expected call: execute_query()
Not called
----------------------------------------------------------------------
Ran 3 tests in 0.028s
Я просмотрел несколько ссылок на StackOverflow, но не нашел решения. Как мне посмеяться над поведением self.mysqldb
, или мне вообще стоит над ним издеваться?