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