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