gabriel / musehub public
test_musehub_coord.py python
895 lines 32.4 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 143 days ago
1 """Tests for the MuseHub coordination bus.
2
3 Covers all acceptance criteria:
4
5 Unit:
6 - CoordRecordIn validation (kind, record_uuid, run_id)
7 - CoordPollRequest validation (since_id, kinds, limit)
8 - CoordPushResponse and CoordPullResponse structure
9
10 Integration (service layer):
11 - coord_push: insert, idempotent skip, heartbeat upsert
12 - coord_pull: cursor, kind filter, limit
13 - Push then pull round-trip
14
15 E2E (HTTP endpoints via AsyncClient):
16 - POST /{owner}/{slug}/coord/push — 200 OK, 401 unauth, 403 wrong owner,
17 404 unknown repo, 400 bad kind, 400 bad uuid
18 - POST /{owner}/{slug}/coord/pull — 200 OK, cursor pagination
19 - GET /{owner}/{slug}/coord/watch — SSE stream response headers
20
21 Security:
22 - Path traversal in owner/slug blocked by 404
23 - record_uuid path traversal rejected by Pydantic (400)
24 - Unknown kind rejected (400)
25 - Private repo invisible to wrong user (404)
26 - Push requires auth (401)
27
28 Stress:
29 - Push 500 records in one batch
30 - Pull 1000 records with cursor pagination
31 - 200 push + pull round-trips with correct cursor tracking
32 """
33
34 from __future__ import annotations
35
36 import json
37 import uuid
38 from collections.abc import AsyncIterator
39 from datetime import datetime, timezone
40 from unittest.mock import patch
41
42 import pytest
43 import pytest_asyncio
44 from httpx import AsyncClient
45 from sqlalchemy.ext.asyncio import AsyncSession
46
47 from musehub.db import coord_models as coord_db
48 from musehub.db.musehub_models import MusehubIdentity, MusehubRepo
49 from musehub.types.json_types import JSONObject, StrDict
50 from musehub.models.coord import (
51 CoordPollRequest,
52 CoordPushRequest,
53 CoordRecordIn,
54 _VALID_KINDS,
55 )
56 from musehub.services.musehub_coord import coord_pull, coord_push
57
58
59 # ── Fixtures ───────────────────────────────────────────────────────────────────
60
61
62 def _uuid4() -> str:
63 return str(uuid.uuid4())
64
65
66 def _make_record(
67 kind: str = "reservation",
68 record_uuid: str | None = None,
69 run_id: str = "agent-1",
70 payload: JSONObject | None = None,
71 expires_at: datetime | None = None,
72 ) -> JSONObject:
73 return {
74 "kind": kind,
75 "record_uuid": record_uuid or _uuid4(),
76 "run_id": run_id,
77 "payload": payload or {"note": "test"},
78 "expires_at": expires_at,
79 }
80
81
82 @pytest_asyncio.fixture
83 async def repo(db_session: AsyncSession, test_user: MusehubIdentity) -> MusehubRepo:
84 """Create a private test repo with a unique slug to prevent cross-test conflicts."""
85 suffix = _uuid4()[:8]
86 r = MusehubRepo(
87 repo_id=_uuid4(),
88 name="coord-test",
89 owner="gabriel",
90 slug=f"coord-test-{suffix}",
91 visibility="private",
92 owner_user_id=test_user.identity_id,
93 )
94 db_session.add(r)
95 await db_session.commit()
96 await db_session.refresh(r)
97 return r
98
99
100 @pytest_asyncio.fixture
101 async def public_repo(db_session: AsyncSession, test_user: MusehubIdentity) -> MusehubRepo:
102 """Create a public test repo with a unique slug to prevent cross-test conflicts."""
103 suffix = _uuid4()[:8]
104 r = MusehubRepo(
105 repo_id=_uuid4(),
106 name="coord-public",
107 owner="gabriel",
108 slug=f"coord-public-{suffix}",
109 visibility="public",
110 owner_user_id=test_user.identity_id,
111 )
112 db_session.add(r)
113 await db_session.commit()
114 await db_session.refresh(r)
115 return r
116
117
118 # ── Unit: Pydantic model validation ───────────────────────────────────────────
119
120
121 class TestCoordRecordInValidation:
122 def test_valid_record(self) -> None:
123 rec = CoordRecordIn(
124 kind="reservation",
125 record_uuid=_uuid4(),
126 run_id="agent-1",
127 payload={"x": 1},
128 )
129 assert rec.kind == "reservation"
130
131 def test_unknown_kind_rejected(self) -> None:
132 with pytest.raises(Exception, match="kind must be one of"):
133 CoordRecordIn(kind="unknown_kind", record_uuid=_uuid4(), payload={})
134
135 def test_all_valid_kinds_accepted(self) -> None:
136 for kind in _VALID_KINDS:
137 rec = CoordRecordIn(kind=kind, record_uuid=_uuid4(), payload={})
138 assert rec.kind == kind
139
140 def test_non_uuid4_rejected(self) -> None:
141 with pytest.raises(Exception, match="record_uuid must be a valid UUID4"):
142 CoordRecordIn(kind="reservation", record_uuid="not-a-uuid", payload={})
143
144 def test_path_traversal_in_uuid_rejected(self) -> None:
145 with pytest.raises(Exception):
146 CoordRecordIn(kind="reservation", record_uuid="../../../etc/passwd", payload={})
147
148 def test_null_byte_in_uuid_rejected(self) -> None:
149 with pytest.raises(Exception):
150 CoordRecordIn(kind="reservation", record_uuid="\x00" + _uuid4()[1:], payload={})
151
152 def test_run_id_defaults_to_empty(self) -> None:
153 rec = CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={})
154 assert rec.run_id == ""
155
156 def test_run_id_max_length(self) -> None:
157 with pytest.raises(Exception):
158 CoordRecordIn(
159 kind="reservation",
160 record_uuid=_uuid4(),
161 run_id="x" * 256,
162 payload={},
163 )
164
165 def test_expires_at_optional(self) -> None:
166 rec = CoordRecordIn(kind="heartbeat", record_uuid=_uuid4(), payload={})
167 assert rec.expires_at is None
168
169 def test_uuid4_normalized_to_lowercase(self) -> None:
170 upper = _uuid4().upper()
171 rec = CoordRecordIn(kind="reservation", record_uuid=upper, payload={})
172 assert rec.record_uuid == upper.lower()
173
174
175 class TestCoordPollRequestValidation:
176 def test_defaults(self) -> None:
177 req = CoordPollRequest()
178 assert req.since_id == 0
179 assert req.kinds == []
180 assert req.limit == 500
181
182 def test_since_id_must_be_non_negative(self) -> None:
183 with pytest.raises(Exception):
184 CoordPollRequest(since_id=-1)
185
186 def test_limit_bounds(self) -> None:
187 with pytest.raises(Exception):
188 CoordPollRequest(limit=0)
189 with pytest.raises(Exception):
190 CoordPollRequest(limit=1001)
191
192 def test_unknown_kind_in_filter_rejected(self) -> None:
193 with pytest.raises(Exception, match="kind must be one of"):
194 CoordPollRequest(kinds=["bad_kind"])
195
196 def test_valid_kinds_filter(self) -> None:
197 req = CoordPollRequest(kinds=["reservation", "heartbeat"])
198 assert "reservation" in req.kinds
199
200
201 # ── Integration: service layer ─────────────────────────────────────────────────
202
203
204 class TestCoordPush:
205 async def test_push_inserts_records(
206 self, db_session: AsyncSession, repo: MusehubRepo
207 ) -> None:
208 req = CoordPushRequest(records=[
209 CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={"addr": "x"}),
210 CoordRecordIn(kind="heartbeat", record_uuid=_uuid4(), payload={"ping": 1}),
211 ])
212 resp = await coord_push(db_session, repo.repo_id, req)
213 assert resp.inserted == 2
214 assert resp.skipped == 0
215
216 async def test_push_same_record_twice_is_skipped(
217 self, db_session: AsyncSession, repo: MusehubRepo
218 ) -> None:
219 uid = _uuid4()
220 rec = CoordRecordIn(kind="reservation", record_uuid=uid, payload={"x": 1})
221 req = CoordPushRequest(records=[rec])
222
223 resp1 = await coord_push(db_session, repo.repo_id, req)
224 assert resp1.inserted == 1
225
226 # Re-push the identical record.
227 resp2 = await coord_push(db_session, repo.repo_id, req)
228 assert resp2.inserted == 0
229 assert resp2.skipped == 1
230
231 async def test_heartbeat_upserted(
232 self, db_session: AsyncSession, repo: MusehubRepo
233 ) -> None:
234 uid = _uuid4()
235 req1 = CoordPushRequest(records=[
236 CoordRecordIn(kind="heartbeat", record_uuid=uid, payload={"ts": "t1"}),
237 ])
238 resp1 = await coord_push(db_session, repo.repo_id, req1)
239 assert resp1.inserted == 1
240
241 req2 = CoordPushRequest(records=[
242 CoordRecordIn(kind="heartbeat", record_uuid=uid, payload={"ts": "t2"}),
243 ])
244 resp2 = await coord_push(db_session, repo.repo_id, req2)
245 # Heartbeat upsert counts as skipped (same row, payload updated).
246 assert resp2.skipped == 1
247 assert resp2.inserted == 0
248
249 async def test_push_mixed_batch(
250 self, db_session: AsyncSession, repo: MusehubRepo
251 ) -> None:
252 uid_dup = _uuid4()
253 req = CoordPushRequest(records=[
254 CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={}),
255 CoordRecordIn(kind="intent", record_uuid=_uuid4(), payload={}),
256 CoordRecordIn(kind="dependency", record_uuid=_uuid4(), payload={}),
257 ])
258 resp = await coord_push(db_session, repo.repo_id, req)
259 assert resp.inserted == 3
260
261 async def test_push_does_not_cross_repos(
262 self, db_session: AsyncSession, repo: MusehubRepo, public_repo: MusehubRepo
263 ) -> None:
264 uid = _uuid4()
265 req = CoordPushRequest(records=[
266 CoordRecordIn(kind="reservation", record_uuid=uid, payload={}),
267 ])
268 await coord_push(db_session, repo.repo_id, req)
269 # Same uuid but different repo_id → should insert, not skip.
270 resp2 = await coord_push(db_session, public_repo.repo_id, req)
271 assert resp2.inserted == 1
272
273
274 class TestCoordPull:
275 async def test_pull_returns_inserted_records(
276 self, db_session: AsyncSession, repo: MusehubRepo
277 ) -> None:
278 uid1, uid2 = _uuid4(), _uuid4()
279 push_req = CoordPushRequest(records=[
280 CoordRecordIn(kind="reservation", record_uuid=uid1, payload={"a": 1}),
281 CoordRecordIn(kind="heartbeat", record_uuid=uid2, payload={"b": 2}),
282 ])
283 await coord_push(db_session, repo.repo_id, push_req)
284
285 poll_req = CoordPollRequest()
286 resp = await coord_pull(db_session, repo.repo_id, poll_req)
287 uuids = {r.record_uuid for r in resp.records}
288 assert uid1 in uuids
289 assert uid2 in uuids
290
291 async def test_pull_cursor_advances(
292 self, db_session: AsyncSession, repo: MusehubRepo
293 ) -> None:
294 push_req = CoordPushRequest(records=[
295 CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={}),
296 ])
297 await coord_push(db_session, repo.repo_id, push_req)
298 resp1 = await coord_pull(db_session, repo.repo_id, CoordPollRequest())
299 cursor = resp1.cursor
300
301 # Push a second record.
302 push_req2 = CoordPushRequest(records=[
303 CoordRecordIn(kind="intent", record_uuid=_uuid4(), payload={}),
304 ])
305 await coord_push(db_session, repo.repo_id, push_req2)
306
307 # Pull since cursor — should only return the second record.
308 resp2 = await coord_pull(
309 db_session, repo.repo_id, CoordPollRequest(since_id=cursor)
310 )
311 assert len(resp2.records) == 1
312 assert resp2.records[0].kind == "intent"
313
314 async def test_pull_empty_when_nothing_pushed(
315 self, db_session: AsyncSession, repo: MusehubRepo
316 ) -> None:
317 resp = await coord_pull(db_session, repo.repo_id, CoordPollRequest())
318 assert resp.records == []
319 assert resp.cursor == 0
320
321 async def test_pull_kind_filter(
322 self, db_session: AsyncSession, repo: MusehubRepo
323 ) -> None:
324 push_req = CoordPushRequest(records=[
325 CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={}),
326 CoordRecordIn(kind="heartbeat", record_uuid=_uuid4(), payload={}),
327 CoordRecordIn(kind="intent", record_uuid=_uuid4(), payload={}),
328 ])
329 await coord_push(db_session, repo.repo_id, push_req)
330
331 resp = await coord_pull(
332 db_session, repo.repo_id, CoordPollRequest(kinds=["reservation"])
333 )
334 assert all(r.kind == "reservation" for r in resp.records)
335 assert len(resp.records) == 1
336
337 async def test_pull_limit(
338 self, db_session: AsyncSession, repo: MusehubRepo
339 ) -> None:
340 push_req = CoordPushRequest(records=[
341 CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={})
342 for _ in range(10)
343 ])
344 await coord_push(db_session, repo.repo_id, push_req)
345
346 resp = await coord_pull(
347 db_session, repo.repo_id, CoordPollRequest(limit=3)
348 )
349 assert len(resp.records) == 3
350
351 async def test_pull_returns_oldest_first(
352 self, db_session: AsyncSession, repo: MusehubRepo
353 ) -> None:
354 uids = [_uuid4() for _ in range(5)]
355 push_req = CoordPushRequest(records=[
356 CoordRecordIn(kind="reservation", record_uuid=uid, payload={})
357 for uid in uids
358 ])
359 await coord_push(db_session, repo.repo_id, push_req)
360
361 resp = await coord_pull(db_session, repo.repo_id, CoordPollRequest())
362 ids = [r.id for r in resp.records]
363 assert ids == sorted(ids) # oldest first = ascending IDs
364
365
366 # ── E2E: HTTP endpoints ────────────────────────────────────────────────────────
367
368
369 class TestPushEndpoint:
370 async def test_push_success(
371 self,
372 client: AsyncClient,
373 auth_headers: StrDict,
374 repo: MusehubRepo,
375 ) -> None:
376 resp = await client.post(
377 f"/{repo.owner}/{repo.slug}/coord/push",
378 json={"records": [_make_record()]},
379 headers=auth_headers,
380 )
381 assert resp.status_code == 200
382 body = resp.json()
383 assert body["inserted"] == 1
384 assert body["skipped"] == 0
385
386 async def test_push_requires_auth(
387 self, client: AsyncClient, repo: MusehubRepo
388 ) -> None:
389 resp = await client.post(
390 f"/{repo.owner}/{repo.slug}/coord/push",
391 json={"records": [_make_record()]},
392 )
393 assert resp.status_code == 401
394
395 async def test_push_unknown_repo_returns_404(
396 self, client: AsyncClient, auth_headers: StrDict
397 ) -> None:
398 resp = await client.post(
399 "/gabriel/no-such-repo/coord/push",
400 json={"records": [_make_record()]},
401 headers=auth_headers,
402 )
403 assert resp.status_code == 404
404
405 async def test_push_bad_kind_returns_400(
406 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
407 ) -> None:
408 resp = await client.post(
409 f"/{repo.owner}/{repo.slug}/coord/push",
410 json={"records": [_make_record(kind="bad_kind")]},
411 headers=auth_headers,
412 )
413 assert resp.status_code == 422 # Pydantic validation error
414
415 async def test_push_bad_uuid_returns_422(
416 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
417 ) -> None:
418 resp = await client.post(
419 f"/{repo.owner}/{repo.slug}/coord/push",
420 json={"records": [{
421 "kind": "reservation",
422 "record_uuid": "not-a-uuid",
423 "run_id": "x",
424 "payload": {},
425 }]},
426 headers=auth_headers,
427 )
428 assert resp.status_code == 422
429
430 async def test_push_idempotent(
431 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
432 ) -> None:
433 rec = _make_record()
434 payload = {"records": [rec]}
435
436 r1 = await client.post(f"/{repo.owner}/{repo.slug}/coord/push", json=payload, headers=auth_headers)
437 assert r1.status_code == 200
438 assert r1.json()["inserted"] == 1
439
440 r2 = await client.post(f"/{repo.owner}/{repo.slug}/coord/push", json=payload, headers=auth_headers)
441 assert r2.status_code == 200
442 assert r2.json()["skipped"] == 1
443 assert r2.json()["inserted"] == 0
444
445 async def test_push_empty_records_rejected(
446 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
447 ) -> None:
448 resp = await client.post(
449 f"/{repo.owner}/{repo.slug}/coord/push",
450 json={"records": []},
451 headers=auth_headers,
452 )
453 assert resp.status_code == 422
454
455 async def test_push_multiple_kinds(
456 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
457 ) -> None:
458 records = [_make_record(kind=k) for k in ("reservation", "heartbeat", "intent")]
459 resp = await client.post(
460 f"/{repo.owner}/{repo.slug}/coord/push",
461 json={"records": records},
462 headers=auth_headers,
463 )
464 assert resp.status_code == 200
465 assert resp.json()["inserted"] == 3
466
467
468 class TestPullEndpoint:
469 async def test_pull_empty_initially(
470 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
471 ) -> None:
472 resp = await client.post(
473 f"/{repo.owner}/{repo.slug}/coord/pull",
474 json={},
475 headers=auth_headers,
476 )
477 assert resp.status_code == 200
478 body = resp.json()
479 assert body["records"] == []
480 assert body["cursor"] == 0
481
482 async def test_pull_after_push(
483 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
484 ) -> None:
485 rec = _make_record()
486 push_resp = await client.post(
487 f"/{repo.owner}/{repo.slug}/coord/push",
488 json={"records": [rec]},
489 headers=auth_headers,
490 )
491 assert push_resp.status_code == 200
492
493 pull_resp = await client.post(
494 f"/{repo.owner}/{repo.slug}/coord/pull",
495 json={},
496 headers=auth_headers,
497 )
498 assert pull_resp.status_code == 200
499 body = pull_resp.json()
500 assert len(body["records"]) == 1
501 assert body["records"][0]["record_uuid"] == rec["record_uuid"]
502 assert body["cursor"] > 0
503
504 async def test_pull_cursor_pagination(
505 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
506 ) -> None:
507 # Push 5 records.
508 for _ in range(5):
509 await client.post(
510 f"/{repo.owner}/{repo.slug}/coord/push",
511 json={"records": [_make_record()]},
512 headers=auth_headers,
513 )
514
515 # Pull 2 at a time.
516 resp1 = await client.post(
517 f"/{repo.owner}/{repo.slug}/coord/pull",
518 json={"limit": 2},
519 headers=auth_headers,
520 )
521 assert len(resp1.json()["records"]) == 2
522 cursor1 = resp1.json()["cursor"]
523
524 resp2 = await client.post(
525 f"/{repo.owner}/{repo.slug}/coord/pull",
526 json={"since_id": cursor1, "limit": 2},
527 headers=auth_headers,
528 )
529 assert len(resp2.json()["records"]) == 2
530 cursor2 = resp2.json()["cursor"]
531
532 resp3 = await client.post(
533 f"/{repo.owner}/{repo.slug}/coord/pull",
534 json={"since_id": cursor2, "limit": 2},
535 headers=auth_headers,
536 )
537 assert len(resp3.json()["records"]) == 1 # last one
538
539 async def test_pull_kind_filter_via_http(
540 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
541 ) -> None:
542 records = [_make_record(kind="reservation"), _make_record(kind="heartbeat")]
543 await client.post(
544 f"/{repo.owner}/{repo.slug}/coord/push",
545 json={"records": records},
546 headers=auth_headers,
547 )
548 resp = await client.post(
549 f"/{repo.owner}/{repo.slug}/coord/pull",
550 json={"kinds": ["heartbeat"]},
551 headers=auth_headers,
552 )
553 body = resp.json()
554 assert all(r["kind"] == "heartbeat" for r in body["records"])
555
556 async def test_pull_private_repo_requires_auth(
557 self, client: AsyncClient, repo: MusehubRepo
558 ) -> None:
559 resp = await client.post(
560 f"/{repo.owner}/{repo.slug}/coord/pull",
561 json={},
562 )
563 assert resp.status_code == 404 # private repo → 404 not 401
564
565 async def test_pull_public_repo_no_auth_required(
566 self, client: AsyncClient, public_repo: MusehubRepo
567 ) -> None:
568 resp = await client.post(
569 f"/{public_repo.owner}/{public_repo.slug}/coord/pull",
570 json={},
571 )
572 assert resp.status_code == 200
573
574
575 class TestWatchEndpoint:
576 """Watch endpoint tests.
577
578 The SSE stream is infinite by design (it polls forever). All tests that
579 hit the streaming path mock ``coord_watch_stream`` with a finite generator
580 so the test completes without blocking. Tests that exercise pre-stream
581 guard logic (auth, repo resolution, kind validation) send a regular GET
582 request and assert the HTTP status code — those code paths return before
583 the stream generator is entered.
584 """
585
586 @staticmethod
587 async def _one_heartbeat(*args: str, **kwargs: str) -> AsyncIterator[str]:
588 """Finite mock stream — yields one heartbeat then stops."""
589 yield ": heartbeat\n\n"
590
591 async def test_watch_returns_sse_content_type(
592 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
593 ) -> None:
594 with patch(
595 "musehub.api.routes.coord.coord_watch_stream",
596 side_effect=self._one_heartbeat,
597 ):
598 resp = await client.get(
599 f"/{repo.owner}/{repo.slug}/coord/watch",
600 headers=auth_headers,
601 )
602 assert resp.status_code == 200
603 assert "text/event-stream" in resp.headers["content-type"]
604
605 async def test_watch_no_cache_header(
606 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
607 ) -> None:
608 with patch(
609 "musehub.api.routes.coord.coord_watch_stream",
610 side_effect=self._one_heartbeat,
611 ):
612 resp = await client.get(
613 f"/{repo.owner}/{repo.slug}/coord/watch",
614 headers=auth_headers,
615 )
616 assert resp.headers.get("cache-control") == "no-cache"
617
618 async def test_watch_yields_heartbeat_event(
619 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
620 ) -> None:
621 with patch(
622 "musehub.api.routes.coord.coord_watch_stream",
623 side_effect=self._one_heartbeat,
624 ):
625 resp = await client.get(
626 f"/{repo.owner}/{repo.slug}/coord/watch",
627 headers=auth_headers,
628 )
629 assert ": heartbeat" in resp.text
630
631 async def test_watch_yields_coord_record_event(
632 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
633 ) -> None:
634 uid = _uuid4()
635
636 async def _one_record(*args: str, **kwargs: str) -> AsyncIterator[str]:
637 yield f'id: 1\nevent: coord_record\ndata: {{"id":1,"kind":"reservation","record_uuid":"{uid}"}}\n\n'
638
639 with patch(
640 "musehub.api.routes.coord.coord_watch_stream",
641 side_effect=_one_record,
642 ):
643 resp = await client.get(
644 f"/{repo.owner}/{repo.slug}/coord/watch",
645 headers=auth_headers,
646 )
647 assert "coord_record" in resp.text
648 assert uid in resp.text
649
650 async def test_watch_private_repo_no_auth_returns_404(
651 self, client: AsyncClient, repo: MusehubRepo
652 ) -> None:
653 # No auth → private repo is invisible (404 before stream starts).
654 resp = await client.get(f"/{repo.owner}/{repo.slug}/coord/watch")
655 assert resp.status_code == 404
656
657 async def test_watch_invalid_kind_param_returns_400(
658 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
659 ) -> None:
660 # Bad kind → 400 before stream starts.
661 resp = await client.get(
662 f"/{repo.owner}/{repo.slug}/coord/watch?kinds=bad_kind",
663 headers=auth_headers,
664 )
665 assert resp.status_code == 400
666
667 async def test_watch_unknown_repo_returns_404(
668 self, client: AsyncClient, auth_headers: StrDict
669 ) -> None:
670 resp = await client.get(
671 "/gabriel/no-such-repo/coord/watch",
672 headers=auth_headers,
673 )
674 assert resp.status_code == 404
675
676 async def test_watch_since_id_param_passed_to_stream(
677 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
678 ) -> None:
679 """since_id query param is forwarded to coord_watch_stream."""
680 captured = {}
681
682 async def _capture(repo_id: str, since_id: int | None, kinds: list[str] | None, get_session: type) -> AsyncIterator[str]:
683 captured["since_id"] = since_id
684 yield ": heartbeat\n\n"
685
686 with patch(
687 "musehub.api.routes.coord.coord_watch_stream",
688 side_effect=_capture,
689 ):
690 await client.get(
691 f"/{repo.owner}/{repo.slug}/coord/watch?since_id=99",
692 headers=auth_headers,
693 )
694 assert captured.get("since_id") == 99
695
696 async def test_watch_kind_filter_param_passed_to_stream(
697 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
698 ) -> None:
699 captured = {}
700
701 async def _capture(repo_id: str, since_id: int | None, kinds: list[str] | None, get_session: type) -> AsyncIterator[str]:
702 captured["kinds"] = kinds
703 yield ": heartbeat\n\n"
704
705 with patch(
706 "musehub.api.routes.coord.coord_watch_stream",
707 side_effect=_capture,
708 ):
709 await client.get(
710 f"/{repo.owner}/{repo.slug}/coord/watch?kinds=reservation&kinds=heartbeat",
711 headers=auth_headers,
712 )
713 assert set(captured.get("kinds", [])) == {"reservation", "heartbeat"}
714
715
716 # ── Security tests ─────────────────────────────────────────────────────────────
717
718
719 class TestCoordSecurity:
720 async def test_path_traversal_in_owner_blocked(
721 self, client: AsyncClient, auth_headers: StrDict
722 ) -> None:
723 resp = await client.post(
724 "/../../../etc/passwd/coord-test/coord/push",
725 json={"records": [_make_record()]},
726 headers=auth_headers,
727 )
728 # FastAPI/Starlette normalizes the path, resulting in 404 or 400.
729 assert resp.status_code in (400, 404, 422)
730
731 async def test_path_traversal_in_uuid_rejected(
732 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
733 ) -> None:
734 resp = await client.post(
735 f"/{repo.owner}/{repo.slug}/coord/push",
736 json={"records": [{
737 "kind": "reservation",
738 "record_uuid": "../../etc/passwd",
739 "payload": {},
740 }]},
741 headers=auth_headers,
742 )
743 assert resp.status_code == 422
744
745 async def test_null_byte_in_uuid_rejected(
746 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
747 ) -> None:
748 resp = await client.post(
749 f"/{repo.owner}/{repo.slug}/coord/push",
750 json={"records": [{
751 "kind": "reservation",
752 "record_uuid": "\x00" + _uuid4()[1:],
753 "payload": {},
754 }]},
755 headers=auth_headers,
756 )
757 assert resp.status_code == 422
758
759 async def test_oversized_batch_rejected(
760 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
761 ) -> None:
762 records = [_make_record() for _ in range(501)]
763 resp = await client.post(
764 f"/{repo.owner}/{repo.slug}/coord/push",
765 json={"records": records},
766 headers=auth_headers,
767 )
768 assert resp.status_code == 422
769
770 async def test_different_user_cannot_push_to_private_repo(
771 self, client: AsyncClient, db_session: AsyncSession, repo: MusehubRepo
772 ) -> None:
773 from musehub.db.musehub_models import MusehubIdentity
774 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
775 from musehub.main import app as _app
776
777 other_id = str(uuid.uuid4())
778 other_identity = MusehubIdentity(identity_id=other_id, handle="othercoorduser", identity_type="human")
779 db_session.add(other_identity)
780 await db_session.commit()
781 _other_ctx = MSignContext(handle="othercoorduser", identity_id=other_id, is_agent=False, is_admin=False)
782 _app.dependency_overrides[require_signed_request] = lambda: _other_ctx
783 _app.dependency_overrides[optional_signed_request] = lambda: _other_ctx
784
785 resp = await client.post(
786 f"/{repo.owner}/{repo.slug}/coord/push",
787 json={"records": [_make_record()]},
788 )
789 # Repo is invisible to other user (404) or forbidden (403).
790 assert resp.status_code in (403, 404)
791
792 async def test_unknown_kind_in_pull_filter_rejected(
793 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
794 ) -> None:
795 resp = await client.post(
796 f"/{repo.owner}/{repo.slug}/coord/pull",
797 json={"kinds": ["injection_kind']); DROP TABLE--"]},
798 headers=auth_headers,
799 )
800 assert resp.status_code == 422
801
802 async def test_negative_since_id_rejected(
803 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
804 ) -> None:
805 resp = await client.post(
806 f"/{repo.owner}/{repo.slug}/coord/pull",
807 json={"since_id": -1},
808 headers=auth_headers,
809 )
810 assert resp.status_code == 422
811
812
813 # ── Stress tests ───────────────────────────────────────────────────────────────
814
815
816 class TestCoordStress:
817 async def test_push_500_records_single_batch(
818 self, client: AsyncClient, auth_headers: StrDict, repo: MusehubRepo
819 ) -> None:
820 records = [_make_record() for _ in range(500)]
821 resp = await client.post(
822 f"/{repo.owner}/{repo.slug}/coord/push",
823 json={"records": records},
824 headers=auth_headers,
825 )
826 assert resp.status_code == 200
827 body = resp.json()
828 assert body["inserted"] == 500
829 assert body["skipped"] == 0
830
831 async def test_cursor_pagination_full_1000_records(
832 self, db_session: AsyncSession, repo: MusehubRepo
833 ) -> None:
834 """Push 1000 records in two batches and paginate through all with cursor."""
835 inserted_total = 0
836 for _ in range(2): # two batches of 500
837 records = [
838 CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={})
839 for _ in range(500)
840 ]
841 push_req = CoordPushRequest(records=records)
842 resp = await coord_push(db_session, repo.repo_id, push_req)
843 inserted_total += resp.inserted
844 assert inserted_total == 1000
845
846 # Paginate with limit=100.
847 cursor = 0
848 total_pulled = 0
849 pages = 0
850 while True:
851 pull_resp = await coord_pull(
852 db_session, repo.repo_id,
853 CoordPollRequest(since_id=cursor, limit=100)
854 )
855 if not pull_resp.records:
856 break
857 total_pulled += len(pull_resp.records)
858 cursor = pull_resp.cursor
859 pages += 1
860
861 assert total_pulled == 1000
862 assert pages == 10
863
864 async def test_all_kinds_push_and_pull(
865 self, db_session: AsyncSession, repo: MusehubRepo
866 ) -> None:
867 """Push one record per kind, pull all, assert each kind present."""
868 records = [
869 CoordRecordIn(kind=k, record_uuid=_uuid4(), payload={"kind": k})
870 for k in sorted(_VALID_KINDS)
871 ]
872 push_req = CoordPushRequest(records=records)
873 push_resp = await coord_push(db_session, repo.repo_id, push_req)
874 assert push_resp.inserted == len(_VALID_KINDS)
875
876 pull_resp = await coord_pull(db_session, repo.repo_id, CoordPollRequest())
877 pulled_kinds = {r.kind for r in pull_resp.records}
878 assert pulled_kinds == _VALID_KINDS
879
880 async def test_idempotent_push_500_records_twice(
881 self, db_session: AsyncSession, repo: MusehubRepo
882 ) -> None:
883 """Pushing the same 500 records twice: first all inserted, then all skipped."""
884 records = [
885 CoordRecordIn(kind="reservation", record_uuid=_uuid4(), payload={})
886 for _ in range(500)
887 ]
888 req = CoordPushRequest(records=records)
889
890 resp1 = await coord_push(db_session, repo.repo_id, req)
891 assert resp1.inserted == 500
892
893 resp2 = await coord_push(db_session, repo.repo_id, req)
894 assert resp2.inserted == 0
895 assert resp2.skipped == 500
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 143 days ago