Макет SQLAlchemy 2.0 вставляет данные

Я пытаюсь протестировать репозиторий 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)
.flush() возможно, но я не думаю, что этот подход сработает, т.е. Издеваться над всей сессией
Ian Wilson 01.06.2023 23:41

Некоторые люди используют SQLite в памяти, в то время как другие люди откатывают каждую транзакцию для каждого теста.

Ian Wilson 01.06.2023 23:41

@IanWilson, как я издеваюсь, я понимаю, что SQLAlchemy не должен запускаться и данные не должны быть вставлены. Разве я не прав?

Claudio Shigueo Watanabe 02.06.2023 03:55

Как вы определяете свое Session в src.infra.db_models.db_base? Вы используете sessionmaker()?

Ian Wilson 02.06.2023 09:42

Да. Вот как я это делаю: Session = scoped_session(sessionmaker(bind=engine)) Я думаю, что моя ошибка здесь заключалась в том, что я должен указать профессии_db_model_mock на модуль репозитория следующим образом: .profession_postgresql_repository.Pro‌​fessionsDBModel') Но я думаю, что мне все же нужно установить, что должны возвращать профессии_db_model_mock

Claudio Shigueo Watanabe 02.06.2023 15:41
Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
0
5
51
2
Перейти к ответу Данный вопрос помечен как решенный

Ответы 2

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

Другие вопросы по теме