Я пытаюсь протестировать репозиторий SQLAlchemy 2.0 и получаю сообщение об ошибке:
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) повторяющееся значение ключа нарушает уникальное ограничение «professions_name_key»
Итак, хотя я издеваюсь над тестом, он вставляет данные в базу данных. Что сделать, чтобы тест не вставлял данные в базу?
Я использую pytest-mock.
Вот модель SQLAlchemy
# File src.infra.db_models.profession_db_model.py
import uuid
from sqlalchemy import Column, String
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.dialects.postgresql import UUID
from src.infra.db_models.db_base import Base
class ProfessionsDBModel(Base):
""" Defines the professions database model.
"""
__tablename__ = "professions"
profession_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(80), nullable=False, unique=True)
description: Mapped[str] = mapped_column(String(200), nullable=False)
Вот репозиторий:
# File src.infra.repositories.profession_postgresql_repository.py
from typing import Dict, Optional
import copy
import uuid
from src.domain.entities.profession import Profession
from src.interactor.interfaces.repositories.profession_repository \
import ProfessionRepositoryInterface
from src.domain.value_objects import ProfessionId
from src.infra.db_models.db_base import Session
from src.infra.db_models.profession_db_model import ProfessionsDBModel
class ProfessionPostgresqlRepository(ProfessionRepositoryInterface):
""" Postgresql Repository for Profession
"""
def __init__(self) -> None:
self._data: Dict[ProfessionId, Profession] = {}
def __db_to_entity(self, db_row: ProfessionsDBModel) -> Optional[Profession]:
return Profession(
profession_id=db_row.profession_id,
name=db_row.name,
description=db_row.description
)
def create(self, name: str, description: str) -> Optional[Profession]:
session = Session()
profession_id=uuid.uuid4()
profession = ProfessionsDBModel(
profession_id=profession_id,
name=name,
description=description
)
session.add(profession)
session.commit()
session.refresh(profession)
if profession is not None:
return self.__db_to_entity(profession)
return None
Вот тест:
import uuid
import pytest
from src.infra.db_models.db_base import Session
from src.domain.entities.profession import Profession
from src.infra.db_models.profession_db_model import ProfessionsDBModel
from .profession_postgresql_repository import ProfessionPostgresqlRepository
from unittest.mock import patch
def test_profession_postgresql_repository(mocker, fixture_profession_developer):
mocker.patch(
'uuid.uuid4',
return_value=fixture_profession_developer["profession_id"]
)
professions_db_model_mock = mocker.patch(
'src.infra.db_models.profession_db_model.ProfessionsDBModel')
session_add_mock = mocker.patch.object(
Session,
"add"
)
session_commit_mock = mocker.patch.object(
Session,
"commit"
)
session_refresh_mock = mocker.patch.object(
Session,
"refresh"
)
repository = ProfessionPostgresqlRepository()
repository.create(
fixture_profession_developer["name"],
fixture_profession_developer["description"]
)
assert session_add_mock.add.call_once_with(professions_db_model_mock)
assert session_commit_mock.commit.call_once_with()
assert session_refresh_mock.refresh.call_once_with(professions_db_model_mock)
Некоторые люди используют SQLite в памяти, в то время как другие люди откатывают каждую транзакцию для каждого теста.
@IanWilson, как я издеваюсь, я понимаю, что SQLAlchemy не должен запускаться и данные не должны быть вставлены. Разве я не прав?
Как вы определяете свое Session
в src.infra.db_models.db_base
? Вы используете sessionmaker()
?
Да. Вот как я это делаю: Session = scoped_session(sessionmaker(bind=engine)) Я думаю, что моя ошибка здесь заключалась в том, что я должен указать профессии_db_model_mock на модуль репозитория следующим образом: .profession_postgresql_repository.ProfessionsDBModel') Но я думаю, что мне все же нужно установить, что должны возвращать профессии_db_model_mock
Мне удалось найти решение, передающее сеанс в качестве параметра классу репозитория. Я использовал MagicMock, чтобы издеваться над сеансом.
""" Module for ProfessionPostgresqlRepository
"""
from typing import Dict, Optional
import copy
import uuid
from src.domain.entities.profession import Profession
from src.interactor.interfaces.repositories.profession_repository \
import ProfessionRepositoryInterface
from src.domain.value_objects import ProfessionId
from src.infra.db_models.db_base import Session
from src.infra.db_models.profession_db_model import ProfessionsDBModel
class ProfessionPostgresqlRepository(ProfessionRepositoryInterface):
""" Postgresql Repository for Profession
"""
def __init__(self, session) -> None:
self.__session = session
self.__data: Dict[ProfessionId, Profession] = {}
def __db_to_entity(self, db_row: ProfessionsDBModel) -> Optional[Profession]:
return Profession(
profession_id=db_row.profession_id,
name=db_row.name,
description=db_row.description
)
def create(self, name: str, description: str) -> Optional[Profession]:
profession_id=uuid.uuid4()
profession = ProfessionsDBModel(
profession_id=profession_id,
name=name,
description=description
)
self.__session.add(profession)
self.__session.commit()
self.__session.refresh(profession)
if profession is not None:
return self.__db_to_entity(profession)
return None
Вот тест:
import uuid
import pytest
from src.infra.db_models.db_base import Session
from src.domain.entities.profession import Profession
from src.infra.db_models.profession_db_model import ProfessionsDBModel
from .profession_postgresql_repository import ProfessionPostgresqlRepository
from unittest.mock import patch
def test_profession_postgresql_repository(mocker, fixture_profession_developer):
session = mocker.MagicMock()
mocker.patch(
'src.infra.repositories.profession_postgresql_repository.uuid.uuid4',
return_value=fixture_profession_developer["profession_id"]
)
profession = ProfessionsDBModel(
profession_id=fixture_profession_developer["profession_id"],
name=fixture_profession_developer["name"],
description=fixture_profession_developer["description"]
)
professions_db_model_mock = mocker.patch(
'src.infra.repositories.profession_postgresql_repository.ProfessionsDBModel')
professions_db_model_mock.return_value = profession
session_add_mock = mocker.patch.object(
Session,
"add"
)
session_commit_mock = mocker.patch.object(
Session,
"commit"
)
session_refresh_mock = mocker.patch.object(
Session,
"refresh"
)
repository = ProfessionPostgresqlRepository(session)
repository.create(
fixture_profession_developer["name"],
fixture_profession_developer["description"]
)
assert session_add_mock.add.call_once_with(professions_db_model_mock)
assert session_commit_mock.commit.call_once_with()
assert session_refresh_mock.refresh.call_once_with(professions_db_model_mock)
Этому решению не нужно передавать сеанс в качестве параметра. Решение состояло в том, чтобы издеваться над сеансом, а не над его методами отдельно.
Как преимущество, тест стал более кратким!
import uuid
import pytest
from src.domain.entities.profession import Profession
from src.infra.db_models.profession_db_model import ProfessionsDBModel
from .profession_postgresql_repository import ProfessionPostgresqlRepository
def test_profession_postgresql_repository(
mocker,
fixture_profession_developer
):
mocker.patch(
'uuid.uuid4',
return_value=fixture_profession_developer["profession_id"]
)
professions_db_model_mock = mocker.patch(
'src.infra.repositories.profession_postgresql_repository.\
ProfessionsDBModel')
session_mock = mocker.patch(
'src.infra.repositories.profession_postgresql_repository.Session')
professions_db_model = ProfessionsDBModel(
profession_id = fixture_profession_developer["profession_id"],
name = fixture_profession_developer["name"],
description = fixture_profession_developer["description"]
)
professions_db_model_mock.return_value = professions_db_model
repository = ProfessionPostgresqlRepository()
result = repository.create(
fixture_profession_developer["name"],
fixture_profession_developer["description"]
)
profession = Profession(
professions_db_model_mock.return_value.profession_id,
professions_db_model_mock.return_value.name,
professions_db_model_mock.return_value.description
)
session_mock.add.assert_called_once_with(professions_db_model_mock())
session_mock.commit.assert_called_once_with()
session_mock.refresh.assert_called_once_with(professions_db_model_mock())
assert result == profession
.flush()
возможно, но я не думаю, что этот подход сработает, т.е. Издеваться над всей сессией