gabriel / musehub public
test_derived_agent_provisioner.py python
386 lines 13.9 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 20 days ago
1 """TDD tests for derived-agent auto-provisioning.
2
3 When a profile page is visited for an agent that only exists as an `agent_id`
4 string in commit metadata (never formally registered), we:
5 1. Compute a deterministic genesis identity_id from the handle.
6 2. Upsert a real `musehub_identities` row (idempotent).
7 3. Return the real identity_id so the Spectral Sigil route can serve the SVG.
8
9 Tests are RED-first. Each assertion drives one concrete implementation decision.
10
11 New symbols:
12 musehub.core.genesis.compute_derived_agent_id
13 musehub.services.derived_agent_provisioner.ensure_agent_identity
14 musehub.services.derived_agent_provisioner.provision_if_derived
15 """
16 from __future__ import annotations
17
18 import re
19 from datetime import datetime, timezone
20 from unittest.mock import AsyncMock, MagicMock, patch
21
22 import pytest
23 import pytest_asyncio
24 from httpx import ASGITransport, AsyncClient
25 from sqlalchemy import select
26 from sqlalchemy.ext.asyncio import AsyncSession
27
28 from muse.core.types import blob_id, long_id
29 from musehub.main import app
30
31 # ---------------------------------------------------------------------------
32 # Helpers
33 # ---------------------------------------------------------------------------
34
35 _SONNET_HANDLE = "claude-sonnet-4-6"
36 _OPUS_HANDLE = "claude-opus-4-7"
37 _CUSTOM_HANDLE = "my-custom-agent-42"
38
39 _KNOWN_FIRST_SEEN = datetime(2026, 1, 31, 12, 0, 0, tzinfo=timezone.utc)
40
41
42 def _expected_id(handle: str) -> str:
43 """Restate the genesis formula so tests catch any deviation in implementation."""
44 return blob_id(f"agent_handle\x00{handle}".encode())
45
46
47 # ---------------------------------------------------------------------------
48 # Module imports — must exist
49 # ---------------------------------------------------------------------------
50
51
52 def test_genesis_module_exports_compute_derived_agent_id() -> None:
53 from musehub.core.genesis import compute_derived_agent_id # noqa: F401
54
55 assert callable(compute_derived_agent_id)
56
57
58 def test_provisioner_module_importable() -> None:
59 from musehub.services import derived_agent_provisioner # noqa: F401
60
61
62 def test_ensure_agent_identity_importable() -> None:
63 from musehub.services.derived_agent_provisioner import ensure_agent_identity # noqa: F401
64
65 assert callable(ensure_agent_identity)
66
67
68 def test_provision_if_derived_importable() -> None:
69 from musehub.services.derived_agent_provisioner import provision_if_derived # noqa: F401
70
71 assert callable(provision_if_derived)
72
73
74 # ---------------------------------------------------------------------------
75 # compute_derived_agent_id — deterministic genesis formula
76 # ---------------------------------------------------------------------------
77
78
79 def test_compute_derived_agent_id_returns_sha256_prefixed() -> None:
80 from musehub.core.genesis import compute_derived_agent_id
81
82 result = compute_derived_agent_id(_SONNET_HANDLE)
83 assert result.startswith("sha256:")
84
85
86 def test_compute_derived_agent_id_hex_part_is_64_chars() -> None:
87 from musehub.core.genesis import compute_derived_agent_id
88
89 result = compute_derived_agent_id(_SONNET_HANDLE)
90 hex_part = result[len("sha256:"):]
91 assert len(hex_part) == 64
92 assert re.fullmatch(r"[0-9a-f]{64}", hex_part)
93
94
95 def test_compute_derived_agent_id_is_deterministic() -> None:
96 from musehub.core.genesis import compute_derived_agent_id
97
98 assert compute_derived_agent_id(_SONNET_HANDLE) == compute_derived_agent_id(_SONNET_HANDLE)
99
100
101 def test_compute_derived_agent_id_differs_by_handle() -> None:
102 from musehub.core.genesis import compute_derived_agent_id
103
104 assert compute_derived_agent_id(_SONNET_HANDLE) != compute_derived_agent_id(_OPUS_HANDLE)
105 assert compute_derived_agent_id(_OPUS_HANDLE) != compute_derived_agent_id(_CUSTOM_HANDLE)
106
107
108 def test_compute_derived_agent_id_matches_known_formula() -> None:
109 """Nail the exact formula so any future refactor breaks loudly."""
110 from musehub.core.genesis import compute_derived_agent_id
111
112 assert compute_derived_agent_id(_SONNET_HANDLE) == _expected_id(_SONNET_HANDLE)
113 assert compute_derived_agent_id(_OPUS_HANDLE) == _expected_id(_OPUS_HANDLE)
114 assert compute_derived_agent_id(_CUSTOM_HANDLE) == _expected_id(_CUSTOM_HANDLE)
115
116
117 # ---------------------------------------------------------------------------
118 # ensure_agent_identity — upsert into musehub_identities
119 # ---------------------------------------------------------------------------
120
121
122 @pytest.mark.asyncio
123 async def test_ensure_agent_identity_creates_row(db_session: AsyncSession) -> None:
124 from musehub.db.musehub_identity_models import MusehubIdentity
125 from musehub.services.derived_agent_provisioner import ensure_agent_identity
126
127 result = await ensure_agent_identity(
128 db_session,
129 handle=_SONNET_HANDLE,
130 agent_model="claude-sonnet-4-6",
131 first_seen_at=_KNOWN_FIRST_SEEN,
132 )
133 await db_session.commit()
134
135 row = (await db_session.execute(
136 select(MusehubIdentity).where(MusehubIdentity.handle == _SONNET_HANDLE)
137 )).scalar_one_or_none()
138
139 assert row is not None
140 assert result.handle == _SONNET_HANDLE
141
142
143 @pytest.mark.asyncio
144 async def test_ensure_agent_identity_sets_correct_identity_id(db_session: AsyncSession) -> None:
145 from musehub.services.derived_agent_provisioner import ensure_agent_identity
146
147 result = await ensure_agent_identity(
148 db_session,
149 handle=_SONNET_HANDLE,
150 agent_model="claude-sonnet-4-6",
151 first_seen_at=_KNOWN_FIRST_SEEN,
152 )
153
154 assert result.identity_id == _expected_id(_SONNET_HANDLE)
155
156
157 @pytest.mark.asyncio
158 async def test_ensure_agent_identity_sets_identity_type_agent(db_session: AsyncSession) -> None:
159 from musehub.services.derived_agent_provisioner import ensure_agent_identity
160
161 result = await ensure_agent_identity(
162 db_session,
163 handle=_SONNET_HANDLE,
164 agent_model="claude-sonnet-4-6",
165 first_seen_at=_KNOWN_FIRST_SEEN,
166 )
167
168 assert result.identity_type == "agent"
169
170
171 @pytest.mark.asyncio
172 async def test_ensure_agent_identity_stores_agent_model(db_session: AsyncSession) -> None:
173 from musehub.services.derived_agent_provisioner import ensure_agent_identity
174
175 result = await ensure_agent_identity(
176 db_session,
177 handle=_SONNET_HANDLE,
178 agent_model="claude-sonnet-4-6",
179 first_seen_at=_KNOWN_FIRST_SEEN,
180 )
181
182 assert result.agent_model == "claude-sonnet-4-6"
183
184
185 @pytest.mark.asyncio
186 async def test_ensure_agent_identity_model_none_is_accepted(db_session: AsyncSession) -> None:
187 from musehub.services.derived_agent_provisioner import ensure_agent_identity
188
189 result = await ensure_agent_identity(
190 db_session,
191 handle=_CUSTOM_HANDLE,
192 agent_model=None,
193 first_seen_at=None,
194 )
195
196 assert result.identity_type == "agent"
197 assert result.agent_model is None
198
199
200 @pytest.mark.asyncio
201 async def test_ensure_agent_identity_is_idempotent(db_session: AsyncSession) -> None:
202 """Calling twice with the same handle must not raise and must not duplicate."""
203 from musehub.db.musehub_identity_models import MusehubIdentity
204 from musehub.services.derived_agent_provisioner import ensure_agent_identity
205
206 await ensure_agent_identity(db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN)
207 await db_session.commit()
208
209 await ensure_agent_identity(db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN)
210 await db_session.commit()
211
212 rows = (await db_session.execute(
213 select(MusehubIdentity).where(MusehubIdentity.handle == _SONNET_HANDLE)
214 )).scalars().all()
215
216 assert len(rows) == 1
217
218
219 @pytest.mark.asyncio
220 async def test_ensure_agent_identity_returns_existing_row_unchanged(db_session: AsyncSession) -> None:
221 """If the row already exists, return it without touching model or timestamps."""
222 from musehub.db.musehub_identity_models import MusehubIdentity
223 from musehub.services.derived_agent_provisioner import ensure_agent_identity
224
225 first = await ensure_agent_identity(db_session, _SONNET_HANDLE, "v1-model", _KNOWN_FIRST_SEEN)
226 await db_session.commit()
227 original_id = first.identity_id
228
229 second = await ensure_agent_identity(db_session, _SONNET_HANDLE, "v2-model-different", None)
230
231 assert second.identity_id == original_id
232
233
234 @pytest.mark.asyncio
235 async def test_ensure_agent_identity_different_handles_create_separate_rows(db_session: AsyncSession) -> None:
236 from musehub.db.musehub_identity_models import MusehubIdentity
237 from musehub.services.derived_agent_provisioner import ensure_agent_identity
238
239 await ensure_agent_identity(db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN)
240 await ensure_agent_identity(db_session, _OPUS_HANDLE, "claude-opus-4-7", _KNOWN_FIRST_SEEN)
241 await db_session.commit()
242
243 rows = (await db_session.execute(
244 select(MusehubIdentity).where(
245 MusehubIdentity.handle.in_([_SONNET_HANDLE, _OPUS_HANDLE])
246 )
247 )).scalars().all()
248
249 assert len(rows) == 2
250 ids = {r.identity_id for r in rows}
251 assert len(ids) == 2 # distinct genesis IDs
252
253
254 # ---------------------------------------------------------------------------
255 # provision_if_derived — the _resolve_identity integration hook
256 # ---------------------------------------------------------------------------
257
258
259 @pytest.mark.asyncio
260 async def test_provision_if_derived_provisions_derived_agent(db_session: AsyncSession) -> None:
261 """Given a derived identity dict, provision_if_derived upserts and updates user_id."""
262 from musehub.db.musehub_identity_models import MusehubIdentity
263 from musehub.services.derived_agent_provisioner import provision_if_derived
264
265 derived_identity = {
266 "handle": _SONNET_HANDLE,
267 "type": "agent",
268 "user_id": None,
269 "agent_model": "claude-sonnet-4-6",
270 "is_derived": True,
271 "member_since": _KNOWN_FIRST_SEEN,
272 }
273
274 updated = await provision_if_derived(db_session, derived_identity)
275 await db_session.commit()
276
277 assert updated["user_id"] == _expected_id(_SONNET_HANDLE)
278 assert updated["is_derived"] is False
279
280 row = (await db_session.execute(
281 select(MusehubIdentity).where(MusehubIdentity.handle == _SONNET_HANDLE)
282 )).scalar_one_or_none()
283 assert row is not None
284
285
286 @pytest.mark.asyncio
287 async def test_provision_if_derived_skips_non_agent(db_session: AsyncSession) -> None:
288 """Humans and orgs are not auto-provisioned."""
289 from musehub.services.derived_agent_provisioner import provision_if_derived
290
291 human_identity = {
292 "handle": "gabriel",
293 "type": "human",
294 "user_id": None,
295 "is_derived": True,
296 "member_since": _KNOWN_FIRST_SEEN,
297 }
298
299 updated = await provision_if_derived(db_session, human_identity)
300
301 assert updated["user_id"] is None
302 assert updated["is_derived"] is True
303
304
305 @pytest.mark.asyncio
306 async def test_provision_if_derived_skips_already_registered(db_session: AsyncSession) -> None:
307 """If user_id is already set (registered agent), don't touch it."""
308 from musehub.services.derived_agent_provisioner import provision_if_derived
309
310 existing_id = long_id("a" * 64)
311 registered = {
312 "handle": _SONNET_HANDLE,
313 "type": "agent",
314 "user_id": existing_id,
315 "is_derived": False,
316 "member_since": _KNOWN_FIRST_SEEN,
317 }
318
319 updated = await provision_if_derived(db_session, registered)
320
321 assert updated["user_id"] == existing_id
322
323
324 # ---------------------------------------------------------------------------
325 # Avatar route — sigil is served after provisioning
326 # ---------------------------------------------------------------------------
327
328
329 @pytest.mark.asyncio
330 async def test_avatar_route_serves_sigil_for_provisioned_agent(db_session: AsyncSession) -> None:
331 """After ensure_agent_identity, the avatar route returns 200 image/svg+xml."""
332 from musehub.services.derived_agent_provisioner import ensure_agent_identity
333
334 identity = await ensure_agent_identity(
335 db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN
336 )
337 await db_session.commit()
338
339 hex_part = identity.identity_id[len("sha256:"):]
340
341 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
342 response = await client.get(f"/avatars/sha256/{hex_part}.svg")
343
344 assert response.status_code == 200
345 assert "image/svg+xml" in response.headers["content-type"]
346
347
348 @pytest.mark.asyncio
349 async def test_avatar_route_sigil_reflects_agent_archetype(db_session: AsyncSession) -> None:
350 """The generated SVG must use the agent archetype (polygon/path for hex shape)."""
351 import xml.etree.ElementTree as ET
352 from musehub.services.derived_agent_provisioner import ensure_agent_identity
353
354 identity = await ensure_agent_identity(
355 db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN
356 )
357 await db_session.commit()
358
359 hex_part = identity.identity_id[len("sha256:"):]
360
361 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
362 response = await client.get(f"/avatars/sha256/{hex_part}.svg")
363
364 root = ET.fromstring(response.content)
365 paths = root.findall(".//{http://www.w3.org/2000/svg}path") or root.findall(".//path")
366 polygons = root.findall(".//{http://www.w3.org/2000/svg}polygon") or root.findall(".//polygon")
367 assert len(paths) + len(polygons) >= 1, "agent sigil must contain path or polygon"
368
369
370 @pytest.mark.asyncio
371 async def test_avatar_sigil_is_deterministic_for_provisioned_agent(db_session: AsyncSession) -> None:
372 """Two requests for the same provisioned agent return identical SVG bytes."""
373 from musehub.services.derived_agent_provisioner import ensure_agent_identity
374
375 identity = await ensure_agent_identity(
376 db_session, _SONNET_HANDLE, "claude-sonnet-4-6", _KNOWN_FIRST_SEEN
377 )
378 await db_session.commit()
379
380 hex_part = identity.identity_id[len("sha256:"):]
381
382 async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
383 r1 = await client.get(f"/avatars/sha256/{hex_part}.svg")
384 r2 = await client.get(f"/avatars/sha256/{hex_part}.svg")
385
386 assert r1.content == r2.content
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 20 days ago