Skip to content

castlecraft_engineer.testing.repository

castlecraft_engineer.testing.repository

BaseAsyncModelRepositoryTest

Bases: Generic[TSQLModel, AsyncRepo]

Base class for testing AsyncModelRepository implementations.

Provides pytest fixtures and helper methods for testing asynchronous repository operations. Subclasses should define repository_class and model_class.

Source code in src/castlecraft_engineer/testing/repository.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class BaseAsyncModelRepositoryTest(Generic[TSQLModel, AsyncRepo]):
    """
    Base class for testing AsyncModelRepository implementations.

    Provides pytest fixtures and helper methods for testing asynchronous
    repository operations. Subclasses should define `repository_class`
    and `model_class`.
    """

    repository_class: Optional[Type[AsyncRepo]] = None
    model_class: Optional[Type[TSQLModel]] = None

    @pytest.fixture
    def mock_async_session(self) -> AsyncMock:
        """Provides an AsyncMock instance simulating an AsyncSession."""
        mock = AsyncMock(spec=AsyncSession)
        # Mock the execute chain for get_all
        mock_execute_result = AsyncMock()
        mock_scalars_result = AsyncMock()
        mock_scalars_result.all.return_value = []  # Default to empty list

        mock_execute_result.scalars.return_value = mock_scalars_result
        mock.execute.return_value = mock_execute_result
        return mock

    @pytest.fixture
    def sample_model_id(self) -> Any:
        """Provides a sample ID for the model.
        Can be overridden if the PK is not UUID."""
        return uuid.uuid4()

    @pytest.fixture
    def sample_model_instance(self, sample_model_id: Any) -> TSQLModel:
        if not self.model_class:
            pytest.skip("model_class not set")
        try:
            return self.model_class(id=sample_model_id)  # type: ignore
        except TypeError:
            return MagicMock(spec=self.model_class, id=sample_model_id)

    @pytest.fixture
    def repository_instance(self) -> AsyncRepo:
        if not self.repository_class:
            pytest.skip("repository_class not set")
        if not self.model_class:
            pytest.skip("model_class not set for repository test")
        return self.repository_class(model=self.model_class)

    def setup_session_get_mock_async(
        self,
        mock_async_session: AsyncMock,
        return_value: Optional[TSQLModel],
    ):
        """Configure mock async_session.get to return a specific value."""
        mock_async_session.get.return_value = return_value

    def setup_session_execute_mock_async(
        self,
        mock_async_session: AsyncMock,
        return_value: List[TSQLModel],
    ):
        """Configure mock async_session.execute().scalars().all() chain."""
        mock_async_session.execute.return_value.scalars.return_value.all.return_value = (
            return_value
        )

    def assert_session_add_called(
        self, mock_async_session: AsyncMock, expected_model: TSQLModel
    ):
        # session.add is synchronous even in AsyncSession
        mock_async_session.add.assert_called_once_with(expected_model)

    def assert_session_commit_awaited(self, mock_async_session: AsyncMock):
        mock_async_session.commit.assert_awaited_once()

    def assert_session_refresh_awaited(
        self, mock_async_session: AsyncMock, expected_model: TSQLModel
    ):
        mock_async_session.refresh.assert_awaited_once_with(expected_model)

    def assert_session_delete_awaited(
        self, mock_async_session: AsyncMock, expected_model: TSQLModel
    ):
        # session.delete is synchronous, but the repo method is async
        # and might await other things or be part of an async flow.
        # The actual call to session.delete itself is not awaited.
        # However, the repository's delete_by_id method will await session.commit()
        mock_async_session.delete.assert_called_once_with(expected_model)

mock_async_session()

Provides an AsyncMock instance simulating an AsyncSession.

Source code in src/castlecraft_engineer/testing/repository.py 
115
116
117
118
119
120
121
122
123
124
125
126
@pytest.fixture
def mock_async_session(self) -> AsyncMock:
    """Provides an AsyncMock instance simulating an AsyncSession."""
    mock = AsyncMock(spec=AsyncSession)
    # Mock the execute chain for get_all
    mock_execute_result = AsyncMock()
    mock_scalars_result = AsyncMock()
    mock_scalars_result.all.return_value = []  # Default to empty list

    mock_execute_result.scalars.return_value = mock_scalars_result
    mock.execute.return_value = mock_execute_result
    return mock

sample_model_id()

Provides a sample ID for the model. Can be overridden if the PK is not UUID.

Source code in src/castlecraft_engineer/testing/repository.py 
128
129
130
131
132
@pytest.fixture
def sample_model_id(self) -> Any:
    """Provides a sample ID for the model.
    Can be overridden if the PK is not UUID."""
    return uuid.uuid4()

setup_session_execute_mock_async(mock_async_session, return_value)

Configure mock async_session.execute().scalars().all() chain.

Source code in src/castlecraft_engineer/testing/repository.py 
159
160
161
162
163
164
165
166
167
def setup_session_execute_mock_async(
    self,
    mock_async_session: AsyncMock,
    return_value: List[TSQLModel],
):
    """Configure mock async_session.execute().scalars().all() chain."""
    mock_async_session.execute.return_value.scalars.return_value.all.return_value = (
        return_value
    )

setup_session_get_mock_async(mock_async_session, return_value)

Configure mock async_session.get to return a specific value.

Source code in src/castlecraft_engineer/testing/repository.py 
151
152
153
154
155
156
157
def setup_session_get_mock_async(
    self,
    mock_async_session: AsyncMock,
    return_value: Optional[TSQLModel],
):
    """Configure mock async_session.get to return a specific value."""
    mock_async_session.get.return_value = return_value

BaseModelRepositoryTest

Bases: Generic[TSQLModel, SyncRepo]

Base class for testing ModelRepository implementations.

Provides pytest fixtures and helper methods for testing synchronous repository operations. Subclasses should define repository_class and model_class.

Source code in src/castlecraft_engineer/testing/repository.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class BaseModelRepositoryTest(Generic[TSQLModel, SyncRepo]):
    """
    Base class for testing ModelRepository implementations.

    Provides pytest fixtures and helper methods for testing synchronous
    repository operations. Subclasses should define `repository_class`
    and `model_class`.
    """

    repository_class: Optional[Type[SyncRepo]] = None
    model_class: Optional[Type[TSQLModel]] = None

    @pytest.fixture
    def mock_session(self) -> MagicMock:
        """Provides a MagicMock instance simulating a Session."""
        mock = MagicMock(spec=Session)
        # Mock the execute chain for get_all
        mock.execute.return_value.scalars.return_value.all.return_value = []
        return mock

    @pytest.fixture
    def sample_model_id(self) -> Any:
        """Provides a sample ID for the model.
        Can be overridden if the PK is not UUID."""
        return uuid.uuid4()

    @pytest.fixture
    def sample_model_instance(self, sample_model_id: Any) -> TSQLModel:
        if not self.model_class:
            pytest.skip("model_class not set")
        # Assumes model can be instantiated with its ID or relevant fields
        # Adjust if your model's __init__ is different
        try:
            return self.model_class(id=sample_model_id)  # type: ignore
        except TypeError:
            # Fallback for models without 'id' in __init__ or complex init
            return MagicMock(spec=self.model_class, id=sample_model_id)

    @pytest.fixture
    def repository_instance(self) -> SyncRepo:
        if not self.repository_class:
            pytest.skip("repository_class not set")
        if not self.model_class:
            pytest.skip("model_class not set for repository test")
        return self.repository_class(model=self.model_class)

    def setup_session_get_mock(
        self,
        mock_session: MagicMock,
        return_value: Optional[TSQLModel],
    ):
        """Configure mock session.get to return a specific value."""
        mock_session.get.return_value = return_value

    def setup_session_execute_mock(
        self,
        mock_session: MagicMock,
        return_value: List[TSQLModel],
    ):
        """Configure mock session.execute().scalars().all() chain."""
        mock_session.execute.return_value.scalars.return_value.all.return_value = (
            return_value
        )

    def assert_session_add_called(
        self, mock_session: MagicMock, expected_model: TSQLModel
    ):
        mock_session.add.assert_called_once_with(expected_model)

    def assert_session_commit_called(self, mock_session: MagicMock):
        mock_session.commit.assert_called_once()

    def assert_session_refresh_called(
        self, mock_session: MagicMock, expected_model: TSQLModel
    ):
        mock_session.refresh.assert_called_once_with(expected_model)

    def assert_session_delete_called(
        self, mock_session: MagicMock, expected_model: TSQLModel
    ):
        mock_session.delete.assert_called_once_with(expected_model)

mock_session()

Provides a MagicMock instance simulating a Session.

Source code in src/castlecraft_engineer/testing/repository.py 
32
33
34
35
36
37
38
@pytest.fixture
def mock_session(self) -> MagicMock:
    """Provides a MagicMock instance simulating a Session."""
    mock = MagicMock(spec=Session)
    # Mock the execute chain for get_all
    mock.execute.return_value.scalars.return_value.all.return_value = []
    return mock

sample_model_id()

Provides a sample ID for the model. Can be overridden if the PK is not UUID.

Source code in src/castlecraft_engineer/testing/repository.py 
40
41
42
43
44
@pytest.fixture
def sample_model_id(self) -> Any:
    """Provides a sample ID for the model.
    Can be overridden if the PK is not UUID."""
    return uuid.uuid4()

setup_session_execute_mock(mock_session, return_value)

Configure mock session.execute().scalars().all() chain.

Source code in src/castlecraft_engineer/testing/repository.py 
74
75
76
77
78
79
80
81
82
def setup_session_execute_mock(
    self,
    mock_session: MagicMock,
    return_value: List[TSQLModel],
):
    """Configure mock session.execute().scalars().all() chain."""
    mock_session.execute.return_value.scalars.return_value.all.return_value = (
        return_value
    )

setup_session_get_mock(mock_session, return_value)

Configure mock session.get to return a specific value.

Source code in src/castlecraft_engineer/testing/repository.py 
66
67
68
69
70
71
72
def setup_session_get_mock(
    self,
    mock_session: MagicMock,
    return_value: Optional[TSQLModel],
):
    """Configure mock session.get to return a specific value."""
    mock_session.get.return_value = return_value