1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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
101
102
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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
|
"""Unit tests verifying to_model() operation flow for service.update()
This test suite validates GitHub issue #555 fix - ensuring that service.update()
calls to_model(data, "update") for ALL data types (dict, Pydantic, msgspec, attrs, model).
Before the fix, dict/Pydantic/msgspec/attrs data bypassed to_model() entirely.
"""
from __future__ import annotations
from typing import Any, Optional
from unittest.mock import AsyncMock, MagicMock
import pytest
from advanced_alchemy.repository import SQLAlchemyAsyncRepository, SQLAlchemySyncRepository
from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService, SQLAlchemySyncRepositoryService
from advanced_alchemy.service.typing import ATTRS_INSTALLED, MSGSPEC_INSTALLED, PYDANTIC_INSTALLED, ModelDictT
# Use real SQLAlchemy models from fixtures instead of mock
# Import from test fixtures which have proper SQLAlchemy declarative models
from tests.fixtures.uuid.models import UUIDAuthor as MockModel
pytestmark = [pytest.mark.unit]
class MockRepository(SQLAlchemyAsyncRepository[MockModel]):
"""Mock repository for testing."""
model_type = MockModel
def __init__(self) -> None:
# Don't call super().__init__ to avoid needing session
self.model_type = MockModel
self.id_attribute = "id"
class MockSyncRepository(SQLAlchemySyncRepository[MockModel]):
"""Mock sync repository for testing."""
model_type = MockModel
def __init__(self) -> None:
# Don't call super().__init__ to avoid needing session
self.model_type = MockModel
self.id_attribute = "id"
class TrackingService(SQLAlchemyAsyncRepositoryService[MockModel, MockRepository]):
"""Service that tracks to_model() calls for testing."""
repository_type = MockRepository
def __init__(self) -> None:
# Create mock repository
self._repository = MockRepository()
# Mock model with proper SQLAlchemy attributes
mock_model = MockModel()
mock_model.id = "existing-id" # type: ignore[assignment]
mock_model.name = "existing"
mock_model.dob = None # type: ignore[assignment]
self._repository.get = AsyncMock(return_value=mock_model) # type: ignore[method-assign]
self._repository.update = AsyncMock(side_effect=lambda data, **kwargs: data) # type: ignore[method-assign]
# Track method calls
self.to_model_calls: list[tuple[Any, Optional[str]]] = []
self.to_model_on_update_calls: list[Any] = []
@property
def repository(self) -> MockRepository:
"""Return mock repository."""
return self._repository
async def to_model(
self,
data: ModelDictT[MockModel],
operation: Optional[str] = None,
) -> MockModel:
"""Track to_model calls."""
self.to_model_calls.append((data, operation))
return await super().to_model(data, operation)
async def to_model_on_update(self, data: ModelDictT[MockModel]) -> ModelDictT[MockModel]:
"""Track to_model_on_update calls."""
self.to_model_on_update_calls.append(data)
return await super().to_model_on_update(data)
class TrackingSyncService(SQLAlchemySyncRepositoryService[MockModel, MockSyncRepository]):
"""Sync service that tracks to_model() calls for testing."""
repository_type = MockSyncRepository
def __init__(self) -> None:
# Create mock repository
self._repository = MockSyncRepository()
# Mock model with proper SQLAlchemy attributes
mock_model = MockModel()
mock_model.id = "existing-id" # type: ignore[assignment]
mock_model.name = "existing"
mock_model.dob = None # type: ignore[assignment]
self._repository.get = MagicMock(return_value=mock_model) # type: ignore[method-assign]
self._repository.update = MagicMock(side_effect=lambda data, **kwargs: data) # type: ignore[method-assign]
# Track method calls
self.to_model_calls: list[tuple[Any, Optional[str]]] = []
self.to_model_on_update_calls: list[Any] = []
@property
def repository(self) -> MockSyncRepository:
"""Return mock repository."""
return self._repository
def to_model(
self,
data: ModelDictT[MockModel],
operation: Optional[str] = None,
) -> MockModel:
"""Track to_model calls."""
self.to_model_calls.append((data, operation))
return super().to_model(data, operation)
def to_model_on_update(self, data: ModelDictT[MockModel]) -> ModelDictT[MockModel]:
"""Track to_model_on_update calls."""
self.to_model_on_update_calls.append(data)
return super().to_model_on_update(data)
# Tests for async service
@pytest.mark.asyncio
async def test_update_dict_calls_to_model_with_operation() -> None:
"""Test that update() with dict data calls to_model(data, 'update')."""
service = TrackingService()
# Update with dict data and item_id
await service.update({"name": "Updated Name"}, item_id="test-id")
# Verify to_model was called with operation="update"
assert len(service.to_model_calls) == 1
data, operation = service.to_model_calls[0]
assert operation == "update"
assert data == {"name": "Updated Name"}
# Verify to_model_on_update was also called (via operation_map)
assert len(service.to_model_on_update_calls) == 1
@pytest.mark.asyncio
async def test_update_model_instance_calls_to_model_with_operation() -> None:
"""Test that update() with model instance calls to_model(data, 'update')."""
service = TrackingService()
# Update with model instance
model = MockModel()
model.id = "test-id" # type: ignore[assignment]
model.name = "Updated Name"
await service.update(model)
# Verify to_model was called with operation="update"
assert len(service.to_model_calls) == 1
data, operation = service.to_model_calls[0]
assert operation == "update"
assert data is model
@pytest.mark.skipif(not PYDANTIC_INSTALLED, reason="Pydantic not installed")
@pytest.mark.asyncio
async def test_update_pydantic_calls_to_model_with_operation() -> None:
"""Test that update() with Pydantic data calls to_model(data, 'update')."""
from pydantic import BaseModel
class AuthorSchema(BaseModel):
name: str
service = TrackingService()
# Update with Pydantic model
schema = AuthorSchema(name="Updated Name")
await service.update(schema, item_id="test-id")
# Verify to_model was called with operation="update"
assert len(service.to_model_calls) == 1
data, operation = service.to_model_calls[0]
assert operation == "update"
assert isinstance(data, AuthorSchema)
@pytest.mark.skipif(not MSGSPEC_INSTALLED, reason="msgspec not installed")
@pytest.mark.asyncio
async def test_update_msgspec_calls_to_model_with_operation() -> None:
"""Test that update() with msgspec data calls to_model(data, 'update')."""
import msgspec
class AuthorStruct(msgspec.Struct):
name: str
service = TrackingService()
# Update with msgspec struct
struct = AuthorStruct(name="Updated Name")
await service.update(struct, item_id="test-id")
# Verify to_model was called with operation="update"
assert len(service.to_model_calls) == 1
data, operation = service.to_model_calls[0]
assert operation == "update"
assert isinstance(data, AuthorStruct)
@pytest.mark.skipif(not ATTRS_INSTALLED, reason="attrs not installed")
@pytest.mark.asyncio
async def test_update_attrs_calls_to_model_with_operation() -> None:
"""Test that update() with attrs data calls to_model(data, 'update')."""
from attrs import define
@define
class AuthorAttrs:
name: str
service = TrackingService()
# Update with attrs instance
attrs_obj = AuthorAttrs(name="Updated Name")
await service.update(attrs_obj, item_id="test-id")
# Verify to_model was called with operation="update"
assert len(service.to_model_calls) == 1
data, operation = service.to_model_calls[0]
assert operation == "update"
assert isinstance(data, AuthorAttrs)
@pytest.mark.asyncio
async def test_update_operation_map_routes_to_to_model_on_update() -> None:
"""Test that to_model() with operation='update' routes to to_model_on_update()."""
service = TrackingService()
# Update with dict data
await service.update({"name": "updated"}, item_id="test-id")
# Verify both methods were called
assert len(service.to_model_calls) == 1
assert len(service.to_model_on_update_calls) == 1
# Verify operation_map routing worked
_, operation = service.to_model_calls[0]
assert operation == "update"
# Tests for sync service
def test_sync_update_dict_calls_to_model_with_operation() -> None:
"""Test that sync update() with dict data calls to_model(data, 'update')."""
service = TrackingSyncService()
# Update with dict data and item_id
service.update({"name": "Updated Name"}, item_id="test-id")
# Verify to_model was called with operation="update"
assert len(service.to_model_calls) == 1
data, operation = service.to_model_calls[0]
assert operation == "update"
assert data == {"name": "Updated Name"}
# Verify to_model_on_update was also called (via operation_map)
assert len(service.to_model_on_update_calls) == 1
def test_sync_update_model_instance_calls_to_model_with_operation() -> None:
"""Test that sync update() with model instance calls to_model(data, 'update')."""
service = TrackingSyncService()
# Update with model instance
model = MockModel()
model.id = "test-id" # type: ignore[assignment]
model.name = "Updated Name"
service.update(model)
# Verify to_model was called with operation="update"
assert len(service.to_model_calls) == 1
data, operation = service.to_model_calls[0]
assert operation == "update"
assert data is model
# Tests for backward compatibility
@pytest.mark.asyncio
async def test_backward_compat_to_model_on_update_only() -> None:
"""Test backward compatibility - service with ONLY to_model_on_update() override."""
class LegacyService(SQLAlchemyAsyncRepositoryService[MockModel, MockRepository]):
repository_type = MockRepository
def __init__(self) -> None:
self._repository = MockRepository()
mock_model = MockModel()
mock_model.id = "test-id" # type: ignore[assignment]
mock_model.name = "Old Name"
self._repository.get = AsyncMock(return_value=mock_model) # type: ignore[method-assign]
self._repository.update = AsyncMock(side_effect=lambda data, **kwargs: data) # type: ignore[method-assign]
self.update_hook_called = False
@property
def repository(self) -> MockRepository:
return self._repository
async def to_model_on_update(self, data: ModelDictT[MockModel]) -> ModelDictT[MockModel]:
"""Legacy pattern - only override to_model_on_update."""
self.update_hook_called = True
return await super().to_model_on_update(data)
service = LegacyService()
await service.update({"name": "Updated Name"}, item_id="test-id")
# Verify to_model_on_update was called (backward compatible)
assert service.update_hook_called
# Real-world pattern tests using SlugBook fixtures are in integration tests
# The SlugBookAsyncService and SlugBookSyncService in tests/fixtures/uuid/services.py
# demonstrate the exact pattern this fix enables:
# - Custom to_model() that checks operation == "update"
# - Regenerates slug when title changes during update
# - This pattern would have been broken before the fix for dict/Pydantic/msgspec/attrs data
# Edge case tests
@pytest.mark.asyncio
async def test_update_without_item_id_uses_model_id() -> None:
"""Test update without item_id uses ID from model instance."""
service = TrackingService()
# Update with model that has ID
model = MockModel()
model.id = "model-id" # type: ignore[assignment]
model.name = "Updated Name"
await service.update(model)
# Should work - uses model's ID
assert len(service.to_model_calls) == 1
@pytest.mark.asyncio
async def test_update_preserves_existing_instance_attributes() -> None:
"""Test that update with item_id preserves existing instance attributes."""
service = TrackingService()
# Update only name field
result = await service.update({"name": "New Name"}, item_id="test-id")
# Existing ID should be preserved from existing instance
assert result.name == "New Name"
assert result.id == "existing-id" # From mock repository's get()
def test_sync_update_preserves_existing_instance_attributes() -> None:
"""Test that sync update with item_id preserves existing instance attributes."""
service = TrackingSyncService()
# Update only name field
result = service.update({"name": "New Name"}, item_id="test-id")
# Existing ID should be preserved from existing instance
assert result.name == "New Name"
assert result.id == "existing-id" # From mock repository's get()
|
