gabriel / musehub public
test_coordination.py python
1,109 lines 44.7 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 144 days ago
1 """Section 7 — Coordination (muse coord): 7-layer test suite.
2
3 Covers:
4 - musehub/api/routes/coord.py (push_coord, pull_coord, watch_coord HTTP handlers,
5 _resolve_repo, _assert_readable, _assert_writable)
6 - musehub/services/musehub_coord.py (coord_push, coord_pull, coord_watch_stream,
7 _row_to_out, write-once semantics, heartbeat upsert)
8 - musehub/services/musehub_coord_server.py (materialize_coord_record, list_reservations,
9 conflict_check, extend_reservation,
10 list_tasks, claim_task, complete_task, fail_task)
11 - musehub/models/coord.py (CoordRecordIn validators, CoordPushRequest,
12 CoordPollRequest, _validate_uuid4)
13 - musehub/db/coord_models.py (MusehubCoordRecord, MusehubCoordReservation,
14 MusehubCoordTask)
15
16 Layers:
17 1. Unit — model validators, pure helpers, no DB
18 2. Integration — real DB (PostgreSQL), service-layer calls, no HTTP
19 3. End-to-End — full HTTP via AsyncClient, real DB
20 4. Stress — 500-record push, 100 tasks, cursor pagination
21 5. Data Integrity — write-once, heartbeat upsert, constraint enforcement,
22 task lifecycle state machine
23 6. Security — auth guards, ownership enforcement, private-repo 404,
24 invalid kind rejection
25 7. Performance — latency budgets for push/pull/materialize
26 """
27 from __future__ import annotations
28
29 import asyncio
30 import time
31 import uuid
32 from datetime import datetime, timedelta, timezone
33
34 import pytest
35 import pytest_asyncio
36 from httpx import AsyncClient
37 from sqlalchemy.ext.asyncio import AsyncSession
38
39 from musehub.core.genesis import compute_identity_id
40 from musehub.types.json_types import JSONObject, StrDict
41 from musehub.models.coord import (
42 CoordPollRequest,
43 CoordPushRequest,
44 CoordRecordIn,
45 _VALID_KINDS,
46 )
47 from muse.core.types import fake_id
48 from tests.factories import create_repo
49
50 # ---------------------------------------------------------------------------
51 # Local helpers
52 # ---------------------------------------------------------------------------
53
54 def _now() -> datetime:
55 return datetime.now(tz=timezone.utc)
56
57
58 def _uuid() -> str:
59 return str(uuid.uuid4())
60
61
62 def _record(
63 kind: str = "intent",
64 record_uuid: str | None = None,
65 run_id: str = "agent-1",
66 payload: JSONObject | None = None,
67 expires_at: datetime | None = None,
68 ) -> CoordRecordIn:
69 return CoordRecordIn(
70 kind=kind,
71 record_uuid=record_uuid or _uuid(),
72 run_id=run_id,
73 payload=payload or {"action": kind, "data": "test"},
74 expires_at=expires_at,
75 )
76
77
78 def _push_body(*records: CoordRecordIn) -> JSONObject:
79 return {"records": [r.model_dump(mode="json") for r in records]}
80
81
82 def _pull_body(since_id: int = 0, kinds: list[str] | None = None, limit: int = 500) -> JSONObject:
83 return {"since_id": since_id, "kinds": kinds or [], "limit": limit}
84
85
86 # ===========================================================================
87 # Layer 1 — Unit tests (model validators, pure helpers)
88 # ===========================================================================
89
90 class TestCoordRecordInValidators:
91 def test_valid_kinds_accepted(self) -> None:
92 for kind in _VALID_KINDS:
93 r = _record(kind=kind)
94 assert r.kind == kind
95
96 def test_invalid_kind_raises(self) -> None:
97 import pytest
98 with pytest.raises(Exception):
99 _record(kind="garbage")
100
101 def test_record_uuid_must_be_uuid4(self) -> None:
102 with pytest.raises(Exception):
103 _record(record_uuid="not-a-uuid")
104
105 def test_record_uuid_normalized_lowercase(self) -> None:
106 uid = str(uuid.uuid4()).upper()
107 r = _record(record_uuid=uid)
108 assert r.record_uuid == uid.lower()
109
110 def test_run_id_empty_string_allowed(self) -> None:
111 r = _record(run_id="")
112 assert r.run_id == ""
113
114 def test_expires_at_optional(self) -> None:
115 r = _record()
116 assert r.expires_at is None
117
118 def test_expires_at_accepted(self) -> None:
119 exp = _now() + timedelta(seconds=300)
120 r = _record(expires_at=exp)
121 assert r.expires_at is not None
122
123
124 class TestCoordPushRequestValidators:
125 def test_empty_records_rejected(self) -> None:
126 with pytest.raises(Exception):
127 CoordPushRequest(records=[])
128
129 def test_max_500_records_accepted(self) -> None:
130 records = [_record() for _ in range(500)]
131 req = CoordPushRequest(records=records)
132 assert len(req.records) == 500
133
134 def test_501_records_rejected(self) -> None:
135 with pytest.raises(Exception):
136 CoordPushRequest(records=[_record() for _ in range(501)])
137
138 def test_single_record_accepted(self) -> None:
139 req = CoordPushRequest(records=[_record()])
140 assert len(req.records) == 1
141
142
143 class TestCoordPollRequestValidators:
144 def test_default_values(self) -> None:
145 req = CoordPollRequest()
146 assert req.since_id == 0
147 assert req.kinds == []
148 assert req.limit == 500
149
150 def test_since_id_must_be_non_negative(self) -> None:
151 with pytest.raises(Exception):
152 CoordPollRequest(since_id=-1)
153
154 def test_invalid_kind_in_pull_rejected(self) -> None:
155 with pytest.raises(Exception):
156 CoordPollRequest(kinds=["garbage"])
157
158 def test_valid_kinds_filter_accepted(self) -> None:
159 req = CoordPollRequest(kinds=["reservation", "task"])
160 assert set(req.kinds) == {"reservation", "task"}
161
162 def test_limit_range(self) -> None:
163 assert CoordPollRequest(limit=1).limit == 1
164 assert CoordPollRequest(limit=1000).limit == 1000
165 with pytest.raises(Exception):
166 CoordPollRequest(limit=0)
167 with pytest.raises(Exception):
168 CoordPollRequest(limit=1001)
169
170
171 class TestValidKinds:
172 def test_all_expected_kinds_present(self) -> None:
173 expected = {"reservation", "intent", "release", "heartbeat",
174 "dependency", "task", "claim"}
175 assert expected == _VALID_KINDS
176
177
178 # ===========================================================================
179 # Layer 2 — Integration tests (real DB, service layer, no HTTP)
180 # ===========================================================================
181
182 class TestCoordPushIntegration:
183 @pytest.mark.asyncio
184 async def test_push_inserts_records(self, db_session: AsyncSession) -> None:
185 from musehub.services.musehub_coord import coord_push
186
187 repo = await create_repo(db_session, slug="push-insert")
188 req = CoordPushRequest(records=[_record("intent"), _record("dependency")])
189 resp = await coord_push(db_session, repo.repo_id, req)
190 assert resp.inserted == 2
191 assert resp.skipped == 0
192
193 @pytest.mark.asyncio
194 async def test_push_write_once_skips_duplicate(self, db_session: AsyncSession) -> None:
195 from musehub.services.musehub_coord import coord_push
196
197 repo = await create_repo(db_session, slug="push-writeonce")
198 rec = _record("intent")
199 req = CoordPushRequest(records=[rec])
200
201 resp1 = await coord_push(db_session, repo.repo_id, req)
202 assert resp1.inserted == 1
203
204 resp2 = await coord_push(db_session, repo.repo_id, req)
205 assert resp2.skipped == 1
206 assert resp2.inserted == 0
207
208 @pytest.mark.asyncio
209 async def test_push_heartbeat_upserts_payload(self, db_session: AsyncSession) -> None:
210 from musehub.services.musehub_coord import coord_push, coord_pull
211
212 repo = await create_repo(db_session, slug="push-hb-upsert")
213 uid = _uuid()
214 req1 = CoordPushRequest(records=[_record("heartbeat", record_uuid=uid,
215 payload={"tick": 1})])
216 req2 = CoordPushRequest(records=[_record("heartbeat", record_uuid=uid,
217 payload={"tick": 2})])
218
219 r1 = await coord_push(db_session, repo.repo_id, req1)
220 assert r1.inserted == 1
221
222 r2 = await coord_push(db_session, repo.repo_id, req2)
223 # Re-push of same heartbeat → upsert, counted as skipped (no new row)
224 assert r2.skipped == 1
225
226 # Payload should be updated
227 pull_resp = await coord_pull(db_session, repo.repo_id,
228 CoordPollRequest(kinds=["heartbeat"]))
229 assert len(pull_resp.records) == 1
230 assert pull_resp.records[0].payload["tick"] == 2
231
232 @pytest.mark.asyncio
233 async def test_push_all_valid_kinds(self, db_session: AsyncSession) -> None:
234 from musehub.services.musehub_coord import coord_push
235
236 repo = await create_repo(db_session, slug="push-all-kinds")
237 records = [_record(k) for k in _VALID_KINDS]
238 req = CoordPushRequest(records=records)
239 resp = await coord_push(db_session, repo.repo_id, req)
240 assert resp.inserted == len(_VALID_KINDS)
241
242 @pytest.mark.asyncio
243 async def test_push_materializes_reservation(self, db_session: AsyncSession) -> None:
244 from musehub.services.musehub_coord import coord_push
245 from musehub.services.musehub_coord_server import list_reservations
246
247 repo = await create_repo(db_session, slug="push-materialize-res")
248 rec_uuid = _uuid()
249 res_id = fake_id(rec_uuid)
250 payload = {
251 "reservation_id": res_id,
252 "run_id": "agent-42",
253 "addresses": ["src/main.py::process"],
254 "ttl_s": 300,
255 }
256 req = CoordPushRequest(records=[_record("reservation", record_uuid=rec_uuid, payload=payload)])
257 await coord_push(db_session, repo.repo_id, req)
258
259 reservations = await list_reservations(db_session, repo.repo_id)
260 assert len(reservations) == 1
261 assert reservations[0].symbol_address == "src/main.py::process"
262 assert reservations[0].agent_id == "agent-42"
263
264 @pytest.mark.asyncio
265 async def test_push_materializes_task(self, db_session: AsyncSession) -> None:
266 from musehub.services.musehub_coord import coord_push
267 from musehub.services.musehub_coord_server import list_tasks
268
269 repo = await create_repo(db_session, slug="push-materialize-task")
270 rec_uuid = _uuid()
271 task_id = fake_id(rec_uuid)
272 payload = {
273 "task_id": task_id,
274 "queue": "ci",
275 "priority": 10,
276 "created_by": "dispatcher",
277 }
278 req = CoordPushRequest(records=[_record("task", record_uuid=rec_uuid, payload=payload)])
279 await coord_push(db_session, repo.repo_id, req)
280
281 tasks = await list_tasks(db_session, repo.repo_id)
282 assert len(tasks) == 1
283 assert tasks[0].task_id == task_id
284 assert tasks[0].queue == "ci"
285 assert tasks[0].priority == 10
286 assert tasks[0].status == "pending"
287
288
289 class TestCoordPullIntegration:
290 @pytest.mark.asyncio
291 async def test_pull_empty_returns_cursor_zero(self, db_session: AsyncSession) -> None:
292 from musehub.services.musehub_coord import coord_pull
293
294 repo = await create_repo(db_session, slug="pull-empty")
295 resp = await coord_pull(db_session, repo.repo_id, CoordPollRequest())
296 assert resp.records == []
297 assert resp.cursor == 0
298
299 @pytest.mark.asyncio
300 async def test_pull_returns_all_pushed_records(self, db_session: AsyncSession) -> None:
301 from musehub.services.musehub_coord import coord_push, coord_pull
302
303 repo = await create_repo(db_session, slug="pull-all")
304 req = CoordPushRequest(records=[_record("intent"), _record("dependency"),
305 _record("heartbeat")])
306 await coord_push(db_session, repo.repo_id, req)
307
308 resp = await coord_pull(db_session, repo.repo_id, CoordPollRequest())
309 assert len(resp.records) == 3
310 assert resp.cursor == resp.records[-1].id
311
312 @pytest.mark.asyncio
313 async def test_pull_since_id_cursor_pagination(self, db_session: AsyncSession) -> None:
314 from musehub.services.musehub_coord import coord_push, coord_pull
315
316 repo = await create_repo(db_session, slug="pull-cursor")
317 for _ in range(5):
318 await coord_push(db_session, repo.repo_id,
319 CoordPushRequest(records=[_record("intent")]))
320
321 # Fetch first 3
322 resp1 = await coord_pull(db_session, repo.repo_id,
323 CoordPollRequest(limit=3))
324 assert len(resp1.records) == 3
325 cursor = resp1.cursor
326
327 # Fetch next 2 using cursor
328 resp2 = await coord_pull(db_session, repo.repo_id,
329 CoordPollRequest(since_id=cursor))
330 assert len(resp2.records) == 2
331 # IDs must be strictly greater than cursor
332 assert all(r.id > cursor for r in resp2.records)
333
334 @pytest.mark.asyncio
335 async def test_pull_kinds_filter(self, db_session: AsyncSession) -> None:
336 from musehub.services.musehub_coord import coord_push, coord_pull
337
338 repo = await create_repo(db_session, slug="pull-kinds-filter")
339 await coord_push(db_session, repo.repo_id,
340 CoordPushRequest(records=[_record("intent"), _record("heartbeat"),
341 _record("dependency")]))
342
343 resp = await coord_pull(db_session, repo.repo_id,
344 CoordPollRequest(kinds=["intent"]))
345 assert len(resp.records) == 1
346 assert resp.records[0].kind == "intent"
347
348 @pytest.mark.asyncio
349 async def test_pull_ordered_oldest_first(self, db_session: AsyncSession) -> None:
350 from musehub.services.musehub_coord import coord_push, coord_pull
351
352 repo = await create_repo(db_session, slug="pull-ordered")
353 for _ in range(3):
354 await coord_push(db_session, repo.repo_id,
355 CoordPushRequest(records=[_record("intent")]))
356
357 resp = await coord_pull(db_session, repo.repo_id, CoordPollRequest())
358 ids = [r.id for r in resp.records]
359 assert ids == sorted(ids)
360
361 @pytest.mark.asyncio
362 async def test_pull_limit_respected(self, db_session: AsyncSession) -> None:
363 from musehub.services.musehub_coord import coord_push, coord_pull
364
365 repo = await create_repo(db_session, slug="pull-limit")
366 for _ in range(10):
367 await coord_push(db_session, repo.repo_id,
368 CoordPushRequest(records=[_record("intent")]))
369
370 resp = await coord_pull(db_session, repo.repo_id,
371 CoordPollRequest(limit=4))
372 assert len(resp.records) == 4
373
374
375 class TestCoordServerIntegration:
376 @pytest.mark.asyncio
377 async def test_conflict_check_no_reservations(self, db_session: AsyncSession) -> None:
378 from musehub.services.musehub_coord_server import conflict_check
379
380 repo = await create_repo(db_session, slug="conflict-empty")
381 result = await conflict_check(db_session, repo.repo_id, ["a.py::Fn"])
382 assert result == []
383
384 @pytest.mark.asyncio
385 async def test_conflict_check_finds_active_reservation(
386 self, db_session: AsyncSession
387 ) -> None:
388 from musehub.services.musehub_coord import coord_push
389 from musehub.services.musehub_coord_server import conflict_check
390
391 repo = await create_repo(db_session, slug="conflict-found")
392 rec_uuid = _uuid()
393 res_id = fake_id(rec_uuid)
394 exp = _now() + timedelta(seconds=300)
395 payload = {
396 "reservation_id": res_id,
397 "run_id": "worker-1",
398 "addresses": ["a.py::MyFn"],
399 "ttl_s": 300,
400 "expires_at": exp.isoformat(),
401 }
402 await coord_push(db_session, repo.repo_id,
403 CoordPushRequest(records=[_record("reservation", record_uuid=rec_uuid,
404 payload=payload,
405 expires_at=exp)]))
406
407 conflicts = await conflict_check(db_session, repo.repo_id, ["a.py::MyFn"])
408 assert len(conflicts) == 1
409 assert conflicts[0]["symbol_address"] == "a.py::MyFn"
410
411 @pytest.mark.asyncio
412 async def test_conflict_check_ignores_expired_reservation(
413 self, db_session: AsyncSession
414 ) -> None:
415 from musehub.services.musehub_coord import coord_push
416 from musehub.services.musehub_coord_server import conflict_check
417 from musehub.db import coord_models as _cm
418
419 repo = await create_repo(db_session, slug="conflict-expired")
420 # Insert reservation directly with past expires_at
421 past = _now() - timedelta(seconds=10)
422 row = _cm.MusehubCoordReservation(
423 reservation_id=fake_id("expired-reservation"),
424 repo_id=repo.repo_id,
425 symbol_address="a.py::OldFn",
426 agent_id="old-agent",
427 ttl_s=10,
428 created_at=_now() - timedelta(seconds=20),
429 expires_at=past,
430 )
431 db_session.add(row)
432 await db_session.commit()
433
434 conflicts = await conflict_check(db_session, repo.repo_id, ["a.py::OldFn"])
435 assert conflicts == []
436
437 @pytest.mark.asyncio
438 async def test_extend_reservation(self, db_session: AsyncSession) -> None:
439 from musehub.services.musehub_coord import coord_push
440 from musehub.services.musehub_coord_server import extend_reservation, list_reservations
441
442 repo = await create_repo(db_session, slug="extend-reservation")
443 rec_uuid = _uuid()
444 res_id = fake_id(rec_uuid)
445 exp = _now() + timedelta(seconds=60)
446 payload = {
447 "reservation_id": res_id,
448 "run_id": "agent-ext",
449 "addresses": ["b.py::Fn"],
450 "ttl_s": 60,
451 "expires_at": exp.isoformat(),
452 }
453 await coord_push(db_session, repo.repo_id,
454 CoordPushRequest(records=[_record("reservation", record_uuid=rec_uuid,
455 payload=payload, expires_at=exp)]))
456
457 res_before = await list_reservations(db_session, repo.repo_id)
458 old_exp = res_before[0].expires_at
459
460 updated = await extend_reservation(db_session, repo.repo_id, res_id, extend_by_s=600)
461 assert updated is not None
462 # New expiry must be later than original
463 new_exp = updated.expires_at
464 if old_exp.tzinfo is None:
465 old_exp = old_exp.replace(tzinfo=timezone.utc)
466 if new_exp.tzinfo is None:
467 new_exp = new_exp.replace(tzinfo=timezone.utc)
468 assert new_exp > old_exp
469
470 @pytest.mark.asyncio
471 async def test_task_lifecycle_claim_complete(self, db_session: AsyncSession) -> None:
472 from musehub.services.musehub_coord import coord_push
473 from musehub.services.musehub_coord_server import claim_task, complete_task
474
475 repo = await create_repo(db_session, slug="task-lifecycle")
476 rec_uuid = _uuid()
477 task_id = fake_id(rec_uuid)
478 payload = {"task_id": task_id, "queue": "default", "priority": 50,
479 "created_by": "dispatcher"}
480 await coord_push(db_session, repo.repo_id,
481 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
482 payload=payload)]))
483
484 claimed = await claim_task(db_session, repo.repo_id, task_id, "worker-1")
485 assert claimed is not None
486 assert claimed.status == "claimed"
487 assert claimed.claimed_by == "worker-1"
488
489 completed = await complete_task(db_session, repo.repo_id, task_id, "worker-1",
490 result={"output": "done"})
491 assert completed is not None
492 assert completed.status == "completed"
493 assert completed.payload.get("result") == {"output": "done"}
494
495 @pytest.mark.asyncio
496 async def test_task_lifecycle_claim_fail(self, db_session: AsyncSession) -> None:
497 from musehub.services.musehub_coord import coord_push
498 from musehub.services.musehub_coord_server import claim_task, fail_task
499
500 repo = await create_repo(db_session, slug="task-fail")
501 rec_uuid = _uuid()
502 task_id = fake_id(rec_uuid)
503 payload = {"task_id": task_id, "queue": "default", "priority": 50}
504 await coord_push(db_session, repo.repo_id,
505 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
506 payload=payload)]))
507
508 await claim_task(db_session, repo.repo_id, task_id, "worker-2")
509 failed = await fail_task(db_session, repo.repo_id, task_id, "worker-2",
510 reason="OOM")
511 assert failed is not None
512 assert failed.status == "failed"
513 assert failed.payload.get("failure_reason") == "OOM"
514
515 @pytest.mark.asyncio
516 async def test_claim_already_claimed_task_returns_none(
517 self, db_session: AsyncSession
518 ) -> None:
519 from musehub.services.musehub_coord import coord_push
520 from musehub.services.musehub_coord_server import claim_task
521
522 repo = await create_repo(db_session, slug="double-claim")
523 rec_uuid = _uuid()
524 task_id = fake_id(rec_uuid)
525 payload = {"task_id": task_id, "queue": "default"}
526 await coord_push(db_session, repo.repo_id,
527 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
528 payload=payload)]))
529
530 r1 = await claim_task(db_session, repo.repo_id, task_id, "worker-A")
531 assert r1 is not None
532 r2 = await claim_task(db_session, repo.repo_id, task_id, "worker-B")
533 assert r2 is None # Already claimed by worker-A
534
535 @pytest.mark.asyncio
536 async def test_list_tasks_filter_by_status(self, db_session: AsyncSession) -> None:
537 from musehub.services.musehub_coord import coord_push
538 from musehub.services.musehub_coord_server import claim_task, list_tasks
539
540 repo = await create_repo(db_session, slug="list-tasks-status")
541 tids = []
542 for i in range(3):
543 rec_uuid = _uuid()
544 tid = fake_id(rec_uuid)
545 tids.append(tid)
546 await coord_push(db_session, repo.repo_id,
547 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
548 payload={"task_id": tid})]))
549 if i == 0:
550 await claim_task(db_session, repo.repo_id, tid, "worker-X")
551
552 pending = await list_tasks(db_session, repo.repo_id, status="pending")
553 claimed = await list_tasks(db_session, repo.repo_id, status="claimed")
554 assert len(pending) == 2
555 assert len(claimed) == 1
556
557
558 # ===========================================================================
559 # Layer 3 — End-to-End tests (full HTTP via AsyncClient, real DB)
560 # ===========================================================================
561
562 class TestCoordEndToEnd:
563 @pytest.mark.asyncio
564 async def test_push_404_unknown_repo(
565 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
566 ) -> None:
567 resp = await client.post(
568 "/ghost-owner/ghost-repo/coord/push",
569 json=_push_body(_record()),
570 headers=auth_headers,
571 )
572 assert resp.status_code == 404
573
574 @pytest.mark.asyncio
575 async def test_push_requires_auth(
576 self, client: AsyncClient, db_session: AsyncSession
577 ) -> None:
578 repo = await create_repo(db_session, slug="push-noauth")
579 await db_session.commit()
580 resp = await client.post(
581 f"/{repo.owner}/{repo.slug}/coord/push",
582 json=_push_body(_record()),
583 )
584 assert resp.status_code == 401
585
586 @pytest.mark.asyncio
587 async def test_push_403_non_owner(
588 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
589 ) -> None:
590 # auth_headers gives identity_id = _TEST_IDENTITY_ID; create repo with different owner_user_id
591 repo = await create_repo(db_session, slug="push-nonowner", owner_user_id=compute_identity_id(b"other-user"))
592 await db_session.commit()
593 resp = await client.post(
594 f"/{repo.owner}/{repo.slug}/coord/push",
595 json=_push_body(_record()),
596 headers=auth_headers,
597 )
598 assert resp.status_code == 403
599
600 @pytest.mark.asyncio
601 async def test_push_success_returns_counts(
602 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
603 ) -> None:
604 from tests.conftest import _TEST_IDENTITY_ID
605 repo = await create_repo(db_session, slug="push-e2e-ok",
606 owner_user_id=_TEST_IDENTITY_ID)
607 await db_session.commit()
608 resp = await client.post(
609 f"/{repo.owner}/{repo.slug}/coord/push",
610 json=_push_body(_record("intent"), _record("dependency")),
611 headers=auth_headers,
612 )
613 assert resp.status_code == 200
614 data = resp.json()
615 assert data["inserted"] == 2
616 assert data["skipped"] == 0
617
618 @pytest.mark.asyncio
619 async def test_pull_public_repo_no_auth(
620 self, client: AsyncClient, db_session: AsyncSession
621 ) -> None:
622 repo = await create_repo(db_session, slug="pull-e2e-pub", visibility="public")
623 await db_session.commit()
624 resp = await client.post(
625 f"/{repo.owner}/{repo.slug}/coord/pull",
626 json=_pull_body(),
627 )
628 assert resp.status_code == 200
629 data = resp.json()
630 assert "records" in data
631 assert "cursor" in data
632
633 @pytest.mark.asyncio
634 async def test_pull_private_repo_404_no_auth(
635 self, client: AsyncClient, db_session: AsyncSession
636 ) -> None:
637 repo = await create_repo(db_session, slug="pull-e2e-priv", visibility="private")
638 await db_session.commit()
639 resp = await client.post(
640 f"/{repo.owner}/{repo.slug}/coord/pull",
641 json=_pull_body(),
642 )
643 assert resp.status_code == 404
644
645 @pytest.mark.asyncio
646 async def test_pull_returns_pushed_records(
647 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
648 ) -> None:
649 from tests.conftest import _TEST_IDENTITY_ID
650 repo = await create_repo(db_session, slug="pull-e2e-round",
651 owner_user_id=_TEST_IDENTITY_ID, visibility="public")
652 await db_session.commit()
653
654 push_resp = await client.post(
655 f"/{repo.owner}/{repo.slug}/coord/push",
656 json=_push_body(_record("intent"), _record("heartbeat")),
657 headers=auth_headers,
658 )
659 assert push_resp.status_code == 200
660
661 pull_resp = await client.post(
662 f"/{repo.owner}/{repo.slug}/coord/pull",
663 json=_pull_body(),
664 )
665 assert pull_resp.status_code == 200
666 data = pull_resp.json()
667 assert len(data["records"]) == 2
668
669 @pytest.mark.asyncio
670 async def test_pull_kinds_filter_via_http(
671 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
672 ) -> None:
673 from tests.conftest import _TEST_IDENTITY_ID
674 repo = await create_repo(db_session, slug="pull-e2e-filter",
675 owner_user_id=_TEST_IDENTITY_ID, visibility="public")
676 await db_session.commit()
677
678 await client.post(
679 f"/{repo.owner}/{repo.slug}/coord/push",
680 json=_push_body(_record("intent"), _record("dependency"), _record("heartbeat")),
681 headers=auth_headers,
682 )
683
684 pull_resp = await client.post(
685 f"/{repo.owner}/{repo.slug}/coord/pull",
686 json=_pull_body(kinds=["heartbeat"]),
687 )
688 assert pull_resp.status_code == 200
689 records = pull_resp.json()["records"]
690 assert len(records) == 1
691 assert records[0]["kind"] == "heartbeat"
692
693 @pytest.mark.asyncio
694 async def test_watch_invalid_kind_400(
695 self, client: AsyncClient, db_session: AsyncSession
696 ) -> None:
697 repo = await create_repo(db_session, slug="watch-invalid-kind", visibility="public")
698 await db_session.commit()
699 resp = await client.get(
700 f"/{repo.owner}/{repo.slug}/coord/watch",
701 params={"kinds": "garbage"},
702 )
703 assert resp.status_code == 400
704
705 @pytest.mark.asyncio
706 async def test_watch_404_unknown_repo(
707 self, client: AsyncClient, db_session: AsyncSession
708 ) -> None:
709 resp = await client.get("/ghost/norepo/coord/watch")
710 assert resp.status_code == 404
711
712
713 # ===========================================================================
714 # Layer 4 — Stress tests
715 # ===========================================================================
716
717 class TestStress:
718 @pytest.mark.asyncio
719 async def test_push_500_records_single_call(self, db_session: AsyncSession) -> None:
720 from musehub.services.musehub_coord import coord_push, coord_pull
721
722 repo = await create_repo(db_session, slug="stress-push-500")
723 records = [_record("intent") for _ in range(500)]
724 req = CoordPushRequest(records=records)
725 resp = await coord_push(db_session, repo.repo_id, req)
726 assert resp.inserted == 500
727 assert resp.skipped == 0
728
729 # All 500 must be pullable
730 pull = await coord_pull(db_session, repo.repo_id,
731 CoordPollRequest(limit=1000))
732 assert len(pull.records) == 500
733
734 @pytest.mark.asyncio
735 async def test_cursor_pagination_through_500_records(
736 self, db_session: AsyncSession
737 ) -> None:
738 from musehub.services.musehub_coord import coord_push, coord_pull
739
740 repo = await create_repo(db_session, slug="stress-cursor-500")
741 records = [_record("dependency") for _ in range(500)]
742 await coord_push(db_session, repo.repo_id, CoordPushRequest(records=records))
743
744 cursor = 0
745 fetched = 0
746 pages = 0
747 while True:
748 page = await coord_pull(db_session, repo.repo_id,
749 CoordPollRequest(since_id=cursor, limit=100))
750 if not page.records:
751 break
752 fetched += len(page.records)
753 cursor = page.cursor
754 pages += 1
755 assert fetched == 500
756 assert pages == 5
757
758 @pytest.mark.asyncio
759 async def test_task_queue_100_tasks(self, db_session: AsyncSession) -> None:
760 from musehub.services.musehub_coord import coord_push
761 from musehub.services.musehub_coord_server import list_tasks, claim_task
762
763 repo = await create_repo(db_session, slug="stress-100-tasks")
764 for _ in range(100):
765 rec_uuid = _uuid()
766 tid = fake_id(rec_uuid)
767 payload = {"task_id": tid, "queue": "batch", "priority": 50}
768 await coord_push(db_session, repo.repo_id,
769 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
770 payload=payload)]))
771
772 tasks = await list_tasks(db_session, repo.repo_id, queue="batch", limit=100)
773 assert len(tasks) == 100
774
775 # Claim first 10
776 claimed_count = 0
777 for task in tasks[:10]:
778 result = await claim_task(db_session, repo.repo_id, task.task_id, "batch-worker")
779 if result is not None:
780 claimed_count += 1
781 assert claimed_count == 10
782
783 @pytest.mark.asyncio
784 async def test_conflict_check_100_reserved_symbols(
785 self, db_session: AsyncSession
786 ) -> None:
787 from musehub.services.musehub_coord import coord_push
788 from musehub.services.musehub_coord_server import conflict_check
789 from musehub.db import coord_models as _cm
790
791 repo = await create_repo(db_session, slug="stress-conflict-100")
792 # Insert 100 reservations directly
793 exp = _now() + timedelta(seconds=300)
794 for i in range(100):
795 row = _cm.MusehubCoordReservation(
796 reservation_id=fake_id(f"stress-res-{i}"),
797 repo_id=repo.repo_id,
798 symbol_address=f"module/file_{i}.py::Fn{i}",
799 agent_id=f"agent-{i}",
800 ttl_s=300,
801 created_at=_now(),
802 expires_at=exp,
803 )
804 db_session.add(row)
805 await db_session.commit()
806
807 # Check the last 50 — all should conflict
808 addresses = [f"module/file_{i}.py::Fn{i}" for i in range(50, 100)]
809 conflicts = await conflict_check(db_session, repo.repo_id, addresses)
810 assert len(conflicts) == 50
811
812
813 # ===========================================================================
814 # Layer 5 — Data Integrity tests
815 # ===========================================================================
816
817 class TestDataIntegrity:
818 @pytest.mark.asyncio
819 async def test_write_once_constraint_enforced(self, db_session: AsyncSession) -> None:
820 """The UniqueConstraint on (repo_id, kind, record_uuid) must hold."""
821 from musehub.services.musehub_coord import coord_push
822
823 repo = await create_repo(db_session, slug="di-unique-constraint")
824 uid = _uuid()
825 rec = _record("intent", record_uuid=uid)
826
827 r1 = await coord_push(db_session, repo.repo_id, CoordPushRequest(records=[rec]))
828 r2 = await coord_push(db_session, repo.repo_id, CoordPushRequest(records=[rec]))
829 # First → inserted, second → skipped (not error)
830 assert r1.inserted == 1
831 assert r2.skipped == 1
832
833 @pytest.mark.asyncio
834 async def test_heartbeat_upsert_does_not_create_new_row(
835 self, db_session: AsyncSession
836 ) -> None:
837 from musehub.services.musehub_coord import coord_push, coord_pull
838
839 repo = await create_repo(db_session, slug="di-hb-no-dup")
840 uid = _uuid()
841 for i in range(5):
842 rec = _record("heartbeat", record_uuid=uid, payload={"tick": i})
843 await coord_push(db_session, repo.repo_id, CoordPushRequest(records=[rec]))
844
845 resp = await coord_pull(db_session, repo.repo_id,
846 CoordPollRequest(kinds=["heartbeat"]))
847 # Only 1 row despite 5 pushes
848 assert len(resp.records) == 1
849 assert resp.records[0].payload["tick"] == 4
850
851 @pytest.mark.asyncio
852 async def test_coord_record_fields_complete(self, db_session: AsyncSession) -> None:
853 from musehub.services.musehub_coord import coord_push, coord_pull
854
855 repo = await create_repo(db_session, slug="di-record-fields")
856 uid = _uuid()
857 exp = _now() + timedelta(seconds=120)
858 rec = _record("dependency", record_uuid=uid, run_id="run-99",
859 payload={"dep": "x"}, expires_at=exp)
860 await coord_push(db_session, repo.repo_id, CoordPushRequest(records=[rec]))
861
862 resp = await coord_pull(db_session, repo.repo_id,
863 CoordPollRequest(kinds=["dependency"]))
864 r = resp.records[0]
865 assert r.kind == "dependency"
866 assert r.record_uuid == uid
867 assert r.run_id == "run-99"
868 assert r.payload == {"dep": "x"}
869 assert r.repo_id == repo.repo_id
870 assert r.created_at is not None
871
872 @pytest.mark.asyncio
873 async def test_task_depends_on_preserved(self, db_session: AsyncSession) -> None:
874 from musehub.services.musehub_coord import coord_push
875 from musehub.services.musehub_coord_server import list_tasks
876
877 repo = await create_repo(db_session, slug="di-depends-on")
878 dep_a = fake_id("dep-a")
879 dep_b = fake_id("dep-b")
880 rec_uuid = _uuid()
881 task_id = fake_id(rec_uuid)
882 payload = {"task_id": task_id, "queue": "default", "depends_on": [dep_a, dep_b]}
883 await coord_push(db_session, repo.repo_id,
884 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
885 payload=payload)]))
886 tasks = await list_tasks(db_session, repo.repo_id)
887 assert tasks[0].depends_on == [dep_a, dep_b]
888
889 @pytest.mark.asyncio
890 async def test_release_marks_reservation_released(self, db_session: AsyncSession) -> None:
891 from musehub.services.musehub_coord import coord_push
892 from musehub.services.musehub_coord_server import list_reservations
893 from musehub.db import coord_models as _cm
894
895 repo = await create_repo(db_session, slug="di-release")
896 rec_uuid = _uuid()
897 res_id = fake_id(rec_uuid)
898 exp = _now() + timedelta(seconds=300)
899 res_payload = {
900 "reservation_id": res_id,
901 "run_id": "agent-r",
902 "addresses": ["c.py::Fn"],
903 "ttl_s": 300,
904 "expires_at": exp.isoformat(),
905 }
906 await coord_push(db_session, repo.repo_id,
907 CoordPushRequest(records=[_record("reservation", record_uuid=rec_uuid,
908 payload=res_payload,
909 expires_at=exp)]))
910
911 # Confirm reservation exists
912 active = await list_reservations(db_session, repo.repo_id)
913 assert len(active) == 1
914
915 # Push a release record
916 rel_id = _uuid()
917 rel_payload = {"reservation_id": res_id}
918 await coord_push(db_session, repo.repo_id,
919 CoordPushRequest(records=[_record("release", record_uuid=rel_id,
920 payload=rel_payload)]))
921
922 # Reservation should now be gone from active list
923 active_after = await list_reservations(db_session, repo.repo_id)
924 assert len(active_after) == 0
925
926
927 # ===========================================================================
928 # Layer 6 — Security tests
929 # ===========================================================================
930
931 class TestSecurity:
932 @pytest.mark.asyncio
933 async def test_push_requires_authentication(
934 self, client: AsyncClient, db_session: AsyncSession
935 ) -> None:
936 repo = await create_repo(db_session, slug="sec-push-noauth", visibility="public")
937 await db_session.commit()
938 resp = await client.post(
939 f"/{repo.owner}/{repo.slug}/coord/push",
940 json=_push_body(_record()),
941 )
942 assert resp.status_code == 401
943
944 @pytest.mark.asyncio
945 async def test_push_403_for_non_owner_authenticated(
946 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
947 ) -> None:
948 repo = await create_repo(db_session, slug="sec-push-nonowner",
949 owner_user_id=compute_identity_id(b"other-owner"))
950 await db_session.commit()
951 resp = await client.post(
952 f"/{repo.owner}/{repo.slug}/coord/push",
953 json=_push_body(_record()),
954 headers=auth_headers,
955 )
956 assert resp.status_code == 403
957
958 @pytest.mark.asyncio
959 async def test_private_repo_pull_returns_404_unauthenticated(
960 self, client: AsyncClient, db_session: AsyncSession
961 ) -> None:
962 repo = await create_repo(db_session, slug="sec-priv-pull", visibility="private")
963 await db_session.commit()
964 resp = await client.post(
965 f"/{repo.owner}/{repo.slug}/coord/pull",
966 json=_pull_body(),
967 )
968 assert resp.status_code == 404
969
970 @pytest.mark.asyncio
971 async def test_push_invalid_kind_rejected(
972 self, client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
973 ) -> None:
974 from tests.conftest import _TEST_IDENTITY_ID
975 repo = await create_repo(db_session, slug="sec-invalid-kind",
976 owner_user_id=_TEST_IDENTITY_ID)
977 await db_session.commit()
978 bad_payload = {
979 "records": [{
980 "kind": "INJECT_SQL",
981 "record_uuid": str(uuid.uuid4()),
982 "run_id": "",
983 "payload": {},
984 }]
985 }
986 resp = await client.post(
987 f"/{repo.owner}/{repo.slug}/coord/push",
988 json=bad_payload,
989 headers=auth_headers,
990 )
991 assert resp.status_code == 422
992
993 @pytest.mark.asyncio
994 async def test_watch_invalid_kind_query_param_400(
995 self, client: AsyncClient, db_session: AsyncSession
996 ) -> None:
997 repo = await create_repo(db_session, slug="sec-watch-kind", visibility="public")
998 await db_session.commit()
999 resp = await client.get(
1000 f"/{repo.owner}/{repo.slug}/coord/watch",
1001 params={"kinds": "evil_kind"},
1002 )
1003 assert resp.status_code == 400
1004
1005 @pytest.mark.asyncio
1006 async def test_complete_task_wrong_agent_rejected(
1007 self, db_session: AsyncSession
1008 ) -> None:
1009 from musehub.services.musehub_coord import coord_push
1010 from musehub.services.musehub_coord_server import claim_task, complete_task
1011
1012 repo = await create_repo(db_session, slug="sec-complete-wrong-agent")
1013 rec_uuid = _uuid()
1014 task_id = fake_id(rec_uuid)
1015 payload = {"task_id": task_id, "queue": "default"}
1016 await coord_push(db_session, repo.repo_id,
1017 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
1018 payload=payload)]))
1019
1020 await claim_task(db_session, repo.repo_id, task_id, "worker-A")
1021 result = await complete_task(db_session, repo.repo_id, task_id, "worker-B")
1022 # worker-B did not claim it — must return None
1023 assert result is None
1024
1025
1026 # ===========================================================================
1027 # Layer 7 — Performance tests
1028 # ===========================================================================
1029
1030 class TestPerformance:
1031 @pytest.mark.asyncio
1032 async def test_push_100_records_under_500ms(self, db_session: AsyncSession) -> None:
1033 from musehub.services.musehub_coord import coord_push
1034
1035 repo = await create_repo(db_session, slug="perf-push-100")
1036 records = [_record("intent") for _ in range(100)]
1037 req = CoordPushRequest(records=records)
1038
1039 t0 = time.perf_counter()
1040 resp = await coord_push(db_session, repo.repo_id, req)
1041 elapsed_ms = (time.perf_counter() - t0) * 1000
1042
1043 assert resp.inserted == 100
1044 assert elapsed_ms < 500, f"push 100 records took {elapsed_ms:.1f}ms"
1045
1046 @pytest.mark.asyncio
1047 async def test_pull_500_records_under_200ms(self, db_session: AsyncSession) -> None:
1048 from musehub.services.musehub_coord import coord_push, coord_pull
1049
1050 repo = await create_repo(db_session, slug="perf-pull-500")
1051 records = [_record("dependency") for _ in range(500)]
1052 await coord_push(db_session, repo.repo_id, CoordPushRequest(records=records))
1053
1054 t0 = time.perf_counter()
1055 resp = await coord_pull(db_session, repo.repo_id, CoordPollRequest(limit=1000))
1056 elapsed_ms = (time.perf_counter() - t0) * 1000
1057
1058 assert len(resp.records) == 500
1059 assert elapsed_ms < 200, f"pull 500 records took {elapsed_ms:.1f}ms"
1060
1061 @pytest.mark.asyncio
1062 async def test_conflict_check_50_addresses_under_100ms(
1063 self, db_session: AsyncSession
1064 ) -> None:
1065 from musehub.services.musehub_coord_server import conflict_check
1066 from musehub.db import coord_models as _cm
1067
1068 repo = await create_repo(db_session, slug="perf-conflict")
1069 exp = _now() + timedelta(seconds=300)
1070 for i in range(50):
1071 db_session.add(_cm.MusehubCoordReservation(
1072 reservation_id=fake_id(f"perf-res-{i}"),
1073 repo_id=repo.repo_id,
1074 symbol_address=f"pkg/file_{i}.py::Fn{i}",
1075 agent_id="agent",
1076 ttl_s=300,
1077 created_at=_now(),
1078 expires_at=exp,
1079 ))
1080 await db_session.commit()
1081
1082 addresses = [f"pkg/file_{i}.py::Fn{i}" for i in range(50)]
1083 t0 = time.perf_counter()
1084 conflicts = await conflict_check(db_session, repo.repo_id, addresses)
1085 elapsed_ms = (time.perf_counter() - t0) * 1000
1086
1087 assert len(conflicts) == 50
1088 assert elapsed_ms < 100, f"conflict_check 50 addresses took {elapsed_ms:.1f}ms"
1089
1090 @pytest.mark.asyncio
1091 async def test_task_queue_list_100_under_100ms(self, db_session: AsyncSession) -> None:
1092 from musehub.services.musehub_coord import coord_push
1093 from musehub.services.musehub_coord_server import list_tasks
1094
1095 repo = await create_repo(db_session, slug="perf-tasklist-100")
1096 for _ in range(100):
1097 rec_uuid = _uuid()
1098 tid = fake_id(rec_uuid)
1099 await coord_push(db_session, repo.repo_id,
1100 CoordPushRequest(records=[_record("task", record_uuid=rec_uuid,
1101 payload={"task_id": tid,
1102 "queue": "perf"})]))
1103
1104 t0 = time.perf_counter()
1105 tasks = await list_tasks(db_session, repo.repo_id, limit=100)
1106 elapsed_ms = (time.perf_counter() - t0) * 1000
1107
1108 assert len(tasks) == 100
1109 assert elapsed_ms < 100, f"list_tasks 100 took {elapsed_ms:.1f}ms"
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 144 days ago