Сценарии (часть 3)
Git tag: Step09
Наша реализация ответов и запросов, наконец, завершена. И теперь мы можем реализовать последнюю версию нашего сценария. Сценарий корректно возвращает объект ResponseSuccess, но до сих пор не проверяет корректность входящего запроса.
Давайте изменим тест в файле tests/use_cases/test_storageroom_list_use_case.py и добавим ещё 2 теста. Полученный набор тестов (после фикстуры domain_storagerooms) выглядит следующим образом:
import pytest
from unittest import mock
from rentomatic.domain.storageroom import StorageRoom
from rentomatic.shared import response_object as res
from rentomatic.use_cases import request_objects as req
from rentomatic.use_cases import storageroom_use_cases as uc
@pytest.fixture
def domain_storagerooms():
[…]
def test_storageroom_list_without_parameters(domain_storagerooms):
repo = mock.Mock()
repo.list.return_value = domain_storagerooms
storageroom_list_use_case = uc.StorageRoomListUseCase(repo)
request_object = req.StorageRoomListRequestObject.from_dict({})
response_object = storageroom_list_use_case.execute(request_object)
assert bool(response_object) is True
repo.list.assert_called_with(filters=None)
assert response_object.value == domain_storagerooms
Этот тест отличается от предыдущего тем, что теперь метод assert_called_with() вызывается с параметром filters=None. В строках с импортом появились response_objects и request_objects. Фикстура Domain_storagerooms была исключена из кода для лаконичности.
def test_storageroom_list_with_filters(domain_storagerooms):
repo = mock.Mock()
repo.list.return_value = domain_storagerooms
storageroom_list_use_case = uc.StorageRoomListUseCase(repo)
qry_filters = {'a': 5}
request_object = req.StorageRoomListRequestObject.from_dict({'filters': qry_filters})
response_object = storageroom_list_use_case.execute(request_object)
assert bool(response_object) is True
repo.list.assert_called_with(filters=qry_filters)
assert response_object.value == domain_storagerooms
Этот тест проверяет, что при вызове хранилища используется значение ключа filters, который используется и для создания запросов.
def test_storageroom_list_handles_generic_error():
repo = mock.Mock()
repo.list.side_effect = Exception('Just an error message')
storageroom_list_use_case = uc.StorageRoomListUseCase(repo)
request_object = req.StorageRoomListRequestObject.from_dict({})
response_object = storageroom_list_use_case.execute(request_object)
assert bool(response_object) is False
assert response_object.value == {
'type': res.ResponseFailure.SYSTEM_ERROR,
'message': "Exception: Just an error message"
}
def test_storageroom_list_handles_bad_request():
repo = mock.Mock()
storageroom_list_use_case = uc.StorageRoomListUseCase(repo)
request_object = req.StorageRoomListRequestObject.from_dict({'filters': 5})
response_object = storageroom_list_use_case.execute(request_object)
assert bool(response_object) is False
assert response_object.value == {
'type': res.ResponseFailure.PARAMETERS_ERROR,
'message': "filters: Is not iterable"
}
Эти последние два теста проверяют поведение сценария при возникновении исключения в хранилище или при неверно сформированном запросе.
Изменим файл rentomatic/use_cases/storageroom_use_cases.py так, чтобы он содержал новую реализацию сценария, позволяющую тестам проходить успешно.
from rentomatic.shared import response_object as res
class StorageRoomListUseCase(object):
def __init__(self, repo):
self.repo = repo
def execute(self, request_object):
if not request_object:
return res.ResponseFailure.build_from_invalid_request_object(request_object)
try:
storage_rooms = self.repo.list(filters=request_object.filters)
return res.ResponseSuccess(storage_rooms)
except Exception as exc:
return res.ResponseFailure.build_system_error(
"{}: {}".format(exc.__class__.__name__, "{}".format(exc)))
Как видите, метод execute() проверяет корректность запроса, в противном случае возвращает ResponseFailure, созданный на основе того же объекта запроса. Вот теперь бизнес-логика реализована, вызывая хранилище и возвращая успешный ответ. Если же в этой фазе что-то пойдет не так, перехватывается исключение и возвращается сформированный нужным образом ResponseFailure.
Антракт: рефакторинг
Git tag: Step10
Чистая архитектура — это не фреймворк. Не стоит ждать от него столько же, сколько от таких продуктов, как Django, который предоставляет модели, ORM, всевозможные структуры и библиотеки. Тем не менее, некоторые классы могут быть выделены из нашего кода и представлены в виде библиотеки, так, что мы можем повторно использовать этот код. В этом разделе я познакомлю вас с рефакторингом уже существующего кода, при помощи которого мы изолирует основные фичи для запросов, ответов и сценариев.
Мы уже выделили объект ответа. Так как test_valid_request_object_cannot_be_used проверяет общее поведение, а не связь между моделью StorageRoom и сценариями, то мы можем переместить его из tests/use_cases/test_storageroom_list_request_objects.py в tests/shared/test_response_object.py.
Затем мы можем переместить классы InvalidRequestObject и ValidRequestObject из rentomatic/use_cases/request_objects.py в rentomatic/shared/request_object.py, внося необходимые изменения в классе StorageRoomListRequestObject, который теперь наследуется от внешнего класса.
А вот класс сценария подвергается значительным изменениям. Класс UseCase тестируем кодом файла tests/shared/test_use_case.py:
from unittest import mock
from rentomatic.shared import request_object as req, response_object as res
from rentomatic.shared import use_case as uc
def test_use_case_cannot_process_valid_requests():
valid_request_object = mock.MagicMock()
valid_request_object.__bool__.return_value = True
use_case = uc.UseCase()
response = use_case.execute(valid_request_object)
assert not response
assert response.type == res.ResponseFailure.SYSTEM_ERROR
assert response.message == \
'NotImplementedError: process_request() not implemented by UseCase class'
Этот тест проверяет, что класс UseCase нельзя использовать для обработки входящих запросов.
def test_use_case_can_process_invalid_requests_and_returns_response_failure():
invalid_request_object = req.InvalidRequestObject()
invalid_request_object.add_error('someparam', 'somemessage')
use_case = uc.UseCase()
response = use_case.execute(invalid_request_object)
assert not response
assert response.type == res.ResponseFailure.PARAMETERS_ERROR
assert response.message == 'someparam: somemessage'
Тест выполняет сценарий с неправильным запросом и проверяет ответ на корректность. Так как запрос неправильный, типом ответа является PARAMETERS_ERROR, тем самым говоря о наличии проблемы в параметрах запроса.
def test_use_case_can_manage_generic_exception_from_process_request():
use_case = uc.UseCase()
class TestException(Exception):
pass
use_case.process_request = mock.Mock()
use_case.process_request.side_effect = TestException('somemessage')
response = use_case.execute(mock.Mock)
assert not response
assert response.type == res.ResponseFailure.SYSTEM_ERROR
assert response.message == 'TestException: somemessage'
Этот тест заставляет сценарий вызывать исключение. Этот тип ошибки классифицируется как SYSTEM_ERROR, который представляет собой общее название для исключения, которое не связано с параметрами запроса или фактическими сущностями.
Как видно из последнего теста, идея состоит в том, чтобы предоставить метод execute() класса UseCase и вызвать метод process_request(), который определён в каждом дочернем классе, являющимся нашим сценарием.
Файл rentomatic/shared/use_case.py содержит следующий код для успешного прохождения теста:
from rentomatic.shared import response_object as res
class UseCase(object):
def execute(self, request_object):
if not request_object:
return res.ResponseFailure.build_from_invalid_request_object(request_object)
try:
return self.process_request(request_object)
except Exception as exc:
return res.ResponseFailure.build_system_error(
"{}: {}".format(exc.__class__.__name__, "{}".format(exc)))
def process_request(self, request_object):
raise NotImplementedError(
"process_request() not implemented by UseCase class")
И вот теперь файл rentomatic/use_cases/storageroom_use_cases.py содержит следующий код:
from rentomatic.shared import use_case as uc
from rentomatic.shared import response_object as res
class StorageRoomListUseCase(uc.UseCase):
def __init__(self, repo):
self.repo = repo
def process_request(self, request_object):
domain_storageroom = self.repo.list(filters=request_object.filters)
return res.ResponseSuccess(domain_storageroom)
Слой хранилища
Git tag: Step11
Слой хранилища — это тот слой, в котором выполняется система хранения данных. Как вы видели, когда мы реализовали сценарий, мы получаем доступ к хранилищу данных через API, в данном случае через метод хранилища list(). Уровень абстракции, предоставляемой уровнем хранилища, выше, чем уровень, предоставляемый ORM или таким инструментом, как SQLAlchemy. Слой хранилища обеспечивает только необходимые для приложения конечные точки с интерфейсом, который адаптирован для конкретных бизнес-задач и целей приложения.
Для ясности, говоря в терминах конкретных технологий, SQLAlchemy является прекрасным инструментом для абстрактного доступа к базе данных SQL. Внутренняя реализация хранилища слоя может использовать его для доступа к базе данных PostgreSQL. Но внешний API слоя — не то же самое, что даёт SQLAlchemy. API — это (как правило, уменьшенное) множество функций, которые вызывают сценарии для получения данных. В самом деле внутренняя реализация API может использовать сырые запросы SQL. Хранилище даже может не быть основанным на базе данных. Мы можем иметь слой хранилища, который извлекает данные из службы REST или делает удаленные вызовы процедур через RabbitMQ.
Очень важной особенностью слоя хранилища является то, что он всегда возвращает модели предметной области, и это совпадает с тем, что обычно делают ORM-фреймворки.
В этой статье я не буду разворачивать реальную базу данных. Я переложу эту задачу на будущую статью, в которой должно хватить места для реализации двух различных решений, и которая сможет показать, как API хранилища может скрыть фактическую реализацию.
Вместо этого я собираюсь создать очень простую систему хранения памяти с некоторыми заранее заданными данными. Думаю, для демонстрации концепции хранилища этого будет достаточно.
Первое, что нужно сделать, это написать несколько тестов, которые описывают публичный API хранилища. Файл tests/repository/test_memrepo.py содержит тесты для данной задачи.
Сначала мы добавим некоторые данные, которые мы будем использовать в тестах. Импортируем доменную модель для проверки корректного типа результата вызова API:
import pytest
from rentomatic.shared.domain_model import DomainModel
from rentomatic.repository import memrepo
storageroom1 = {
'code': 'f853578c-fc0f-4e65-81b8-566c5dffa35a',
'size': 215,
'price': 39,
'longitude': '-0.09998975',
'latitude': '51.75436293',
}
storageroom2 = {
'code': 'fe2c3195-aeff-487a-a08f-e0bdc0ec6e9a',
'size': 405,
'price': 66,
'longitude': '0.18228006',
'latitude': '51.74640997',
}
storageroom3 = {
'code': '913694c6-435a-4366-ba0d-da5334a611b2',
'size': 56,
'price': 60,
'longitude': '0.27891577',
'latitude': '51.45994069',
}
storageroom4 = {
'code': 'eed76e77-55c1-41ce-985d-ca49bf6c0585',
'size': 93,
'price': 48,
'longitude': '0.33894476',
'latitude': '51.39916678',
}
@pytest.fixture
def storagerooms():
return [storageroom1, storageroom2, storageroom3, storageroom4]
Поскольку объект хранилища будет возвращать модели предметной области, нам нужна вспомогательная функция для проверки правильности результатов. Следующая функция проверяет длину двух списков и гарантирует, что все возвращенные элементы модели из предметной области, и сравнивает коды. Обратите внимание, что мы можем использовать встроенную функцию isinstance(), поскольку DomainModel является абстрактным базовым классом и наши модели зарегистрированы (см rentomatic/domian/storagerooms.py)
def _check_results(domain_models_list, data_list):
assert len(domain_models_list) == len(data_list)
assert all([isinstance(dm, DomainModel) for dm in domain_models_list])
assert set([dm.code for dm in domain_models_list]) == set([d['code'] for d in data_list])
Нам нужно иметь возможность инициализировать хранилище со списком словарей, и чтобы вызов метода list() без какого-либо параметра возвращал тот же список записей.
def test_repository_list_without_parameters(storagerooms):
repo = memrepo.MemRepo(storagerooms)
assert repo.list() == storagerooms
Метод list() должен принимать параметр filters, являющийся словарём. Ключи словаря должны быть в виде <attribute>__<operator>, похожий на синтаксис, используемый в Django ORM. Таким образом, чтобы выразить, что цена должна быть меньше 65, можно записать filters={'price__lt': 60}.
Нужно проверить несколько условий возникновения ошибки: неизвестный ключ должен вызвать исключение KeyError, а использование неверного оператора должно вызвать исключение ValueError.
def test_repository_list_with_filters_unknown_key(storagerooms):
repo = memrepo.MemRepo(storagerooms)
with pytest.raises(KeyError):
repo.list(filters={'name': 'aname'})
def test_repository_list_with_filters_unknown_operator(storagerooms):
repo = memrepo.MemRepo(storagerooms)
with pytest.raises(ValueError):
repo.list(filters={'price__in': [20, 30]})
Давайте затем проверим, что механизм фильтрации работает. Мы хотим, чтобы оператором по умолчанию был __eq, что значит, что если мы не передаём какой-либо оператор, должен выполняться оператор равенства.
def test_repository_list_with_filters_price(storagerooms):
repo = memrepo.MemRepo(storagerooms)
_check_results(repo.list(filters={'price': 60}), [storageroom3])
_check_results(repo.list(filters={'price__eq': 60}), [storageroom3])
_check_results(repo.list(filters={'price__lt': 60}), [storageroom1, storageroom4])
_check_results(repo.list(filters={'price__gt': 60}), [storageroom2])
def test_repository_list_with_filters_size(storagerooms):
repo = memrepo.MemRepo(storagerooms)
_check_results(repo.list(filters={'size': 93}), [storageroom4])
_check_results(repo.list(filters={'size__eq': 93}), [storageroom4])
_check_results(repo.list(filters={'size__lt': 60}), [storageroom3])
_check_results(repo.list(filters={'size__gt': 400}), [storageroom2])
def test_repository_list_with_filters_code(storagerooms):
repo = memrepo.MemRepo(storagerooms)
_check_results(
repo.list(filters={'code': '913694c6-435a-4366-ba0d-da5334a611b2'}), [storageroom3])
Реализация класса MemRepo довольно проста, и я не буду погружаться в нее построчно.
from rentomatic.domain import storageroom as sr
class MemRepo:
def __init__(self, entries=None):
self._entries = []
if entries:
self._entries.extend(entries)
def _check(self, element, key, value):
if '__' not in key:
key = key + '__eq'
key, operator = key.split('__')
if operator not in ['eq', 'lt', 'gt']:
raise ValueError('Operator {} is not supported'.format(operator))
operator = '__{}__'.format(operator)
return getattr(element[key], operator)(value)
def list(self, filters=None):
if not filters:
return self._entries
result = []
result.extend(self._entries)
for key, value in filters.items():
result = [e for e in result if self._check(e, key, value)]
return [sr.StorageRoom.from_dict(r) for r in result]
Продолжение следует...
Комментарии (0)