python насмешливый sqlalchemy соединение - PullRequest
Новая
       51

python насмешливый sqlalchemy соединение

1 голос
/ 18 февраля 2020

У меня есть простая функция, которая подключается к БД и извлекает некоторые данные. 

db.py

from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool


def _create_engine(app):
    impac_engine = create_engine(
        app['DB'],
        poolclass=NullPool  # this setting enables NOT to use Pooling, preventing from timeout issues.
    )
    return impac_engine


def get_all_pos(app):
    engine = _create_engine(app)
    qry = """SELECT DISTINCT id, name FROM p_t ORDER BY name ASC"""
    try:
        cursor = engine.execute(qry)
        rows = cursor.fetchall()
        return rows
    except Exception as re:
        raise re

Я пытаюсь написать несколько тестовых случаев, высмеивая это соединение -

tests.py 

import unittest
from db import get_all_pos
from unittest.mock import patch
from unittest.mock import Mock


class TestPosition(unittest.TestCase):

    @patch('db.sqlalchemy')
    def test_get_all_pos(self, mock_sqlalchemy):
        mock_sqlalchemy.create_engine = Mock()
        get_all_pos({'DB': 'test'})




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

Когда я запускаю вышеуказанный файл python tests.py, я получаю следующую ошибку - 

   "Could not parse rfc1738 URL from string '%s'" % name
sqlalchemy.exc.ArgumentError: Could not parse rfc1738 URL from string 'test'

Не должно ли mock_sqlalchemy.create_engine = Mock () дать мне макет объекта и обход проверки URL.

1 Ответ

0 голосов
/ 18 февраля 2020

Другим вариантом было бы издеваться над вашей _create_engine функцией. Поскольку это модульный тест, и мы хотим протестировать get_all_pos, нам не нужно полагаться на поведение _create_engine, поэтому мы можем просто исправить это так. 

import unittest
import db
from unittest.mock import patch


class TestPosition(unittest.TestCase):

    @patch.object(db, '_create_engine')
    def test_get_all_pos(self, mock_sqlalchemy):
        args = {'DB': 'test'}
        db.get_all_pos(args)
        mock_sqlalchemy.assert_called_once()
        mock_sqlalchemy.assert_called_with({'DB': 'test'})


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

Если вы хотите Чтобы проверить определенные результаты, вам нужно будет правильно установить все соответствующие атрибуты. Я бы рекомендовал не связывать его в один вызов, чтобы он был более читабельным, как показано ниже. 

import unittest
import db
from unittest.mock import patch
from unittest.mock import Mock


class Cursor:
    def __init__(self, vals):
        self.vals = vals

    def fetchall(self):
        return self.vals


class TestPosition(unittest.TestCase):

    @patch.object(db, '_create_engine')
    def test_get_all_pos(self, mock_sqlalchemy):
        to_test = [1, 2, 3]

        mock_cursor = Mock()
        cursor_attrs = {'fetchall.return_value': to_test}
        mock_cursor.configure_mock(**cursor_attrs)

        mock_execute = Mock()
        engine_attrs = {'execute.return_value': mock_cursor}
        mock_execute.configure_mock(**engine_attrs)

        mock_sqlalchemy.return_value = mock_execute

        args = {'DB': 'test'}
        rows = db.get_all_pos(args)

        mock_sqlalchemy.assert_called_once()
        mock_sqlalchemy.assert_called_with({'DB': 'test'})
        self.assertEqual(to_test, rows)
...