I am trying to test a SQLAlchemy 2.0 repository and I am getting the error:
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "professions_name_key"
So, although I am mocking the test, it inserts data into the database. What should I do to the test not insert data into the database?
I am using pytest-mock.
Here is the SQLAlchemy model
# 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)
Here is the repository:
# 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
Here is the test:
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)
This solution don't need to pass the session as parameter. The solution was to mock the Session and not its methods separately.
As an advantage, the test is more concise now!
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