Пересмешивать переменную экземпляра в Python - PullRequest
0 голосов
/ 15 октября 2019

Я построил небольшую 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, или мне вообще стоит над ним издеваться?

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...