gabriel / musehub public
test_context_section23.py python
1,058 lines 36.4 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Section 23 — Agent Context API: 7-layer test suite.
2
3 Covers musehub/services/musehub_context.py and musehub/models/musehub_context.py.
4 The 15 existing tests in test_musehub_context.py cover E2E + integration basics;
5 this suite adds unit, stress, data-integrity, security, and performance layers.
6
7 Layer map
8 ---------
9 1. Unit — pure functions, constants, Pydantic models
10 2. Integration — service functions against real PostgreSQL DB
11 3. E2E — HTTP client against the full app
12 4. Stress — large datasets, concurrent requests
13 5. Data Integrity — ordering, filtering, exclusion rules
14 6. Security — auth enforcement, private repo visibility
15 7. Performance — timing budgets
16 """
17 from __future__ import annotations
18
19 import asyncio
20 import time
21 import uuid
22 from datetime import datetime, timezone
23
24 import pytest
25 from httpx import AsyncClient
26 from sqlalchemy.ext.asyncio import AsyncSession
27
28 from musehub.muse_contracts.json_types import StrDict
29 from musehub.db.musehub_models import (
30 MusehubBranch,
31 MusehubCommit,
32 MusehubIssue,
33 MusehubProposal,
34 MusehubRepo,
35 )
36 from musehub.models.musehub_context import (
37 ActiveProposalContext,
38 AgentContextResponse,
39 AnalysisSummaryContext,
40 ContextDepth,
41 ContextFormat,
42 HistoryEntryContext,
43 MusicalStateContext,
44 OpenIssueContext,
45 )
46 from musehub.services.musehub_context import (
47 _HISTORY_LIMIT,
48 _INCLUDE_ISSUE_BODY,
49 _INCLUDE_PROPOSAL_BODY,
50 _extract_tracks_from_snapshot,
51 _generate_suggestions,
52 _get_latest_commit,
53 _get_open_issues,
54 _get_open_proposals,
55 _resolve_ref_to_commit,
56 _utc_iso,
57 build_agent_context,
58 )
59
60
61 # ---------------------------------------------------------------------------
62 # DB helpers
63 # ---------------------------------------------------------------------------
64
65
66 def _uid() -> str:
67 return str(uuid.uuid4())
68
69
70 async def _db_repo(session: AsyncSession, *, visibility: str = "private") -> str:
71 slug = f"test-repo-{_uid()[:8]}"
72 repo = MusehubRepo(
73 repo_id=_uid(),
74 name=slug,
75 slug=slug,
76 owner="testuser",
77 owner_user_id="testuser",
78 visibility=visibility,
79 )
80 session.add(repo)
81 await session.flush()
82 return repo.repo_id
83
84
85 async def _db_commit(
86 session: AsyncSession,
87 repo_id: str,
88 *,
89 branch: str = "main",
90 message: str = "add groove",
91 ts: datetime | None = None,
92 parent_id: str | None = None,
93 ) -> str:
94 commit_id = _uid().replace("-", "")
95 c = MusehubCommit(
96 commit_id=commit_id,
97 repo_id=repo_id,
98 branch=branch,
99 parent_ids=[parent_id] if parent_id else [],
100 message=message,
101 author="agent",
102 timestamp=ts or datetime.now(timezone.utc),
103 )
104 session.add(c)
105 await session.flush()
106 return commit_id
107
108
109 async def _db_branch(session: AsyncSession, repo_id: str, name: str, head: str) -> None:
110 session.add(MusehubBranch(repo_id=repo_id, name=name, head_commit_id=head))
111 await session.flush()
112
113
114 async def _db_issue(
115 session: AsyncSession,
116 repo_id: str,
117 *,
118 number: int = 1,
119 title: str = "fix harmony",
120 body: str = "needs fixing",
121 state: str = "open",
122 labels: list[str] | None = None,
123 ) -> str:
124 issue = MusehubIssue(
125 repo_id=repo_id,
126 number=number,
127 title=title,
128 body=body,
129 state=state,
130 labels=labels or [],
131 )
132 session.add(issue)
133 await session.flush()
134 return issue.issue_id
135
136
137 async def _db_proposal_ctx(
138 session: AsyncSession,
139 repo_id: str,
140 *,
141 proposal_number: int = 1,
142 title: str = "add tritone sub",
143 body: str = "see description",
144 state: str = "open",
145 from_branch: str = "feat/x",
146 to_branch: str = "main",
147 ) -> str:
148 proposal = MusehubProposal(
149 repo_id=repo_id,
150 proposal_number=proposal_number,
151 title=title,
152 body=body,
153 state=state,
154 from_branch=from_branch,
155 to_branch=to_branch,
156 )
157 session.add(proposal)
158 await session.flush()
159 return proposal.proposal_id
160
161
162 async def _api_repo(
163 client: AsyncClient,
164 auth_headers: StrDict,
165 *,
166 name: str | None = None,
167 visibility: str = "private",
168 ) -> str:
169 name = name or f"repo-{_uid()[:8]}"
170 r = await client.post(
171 "/api/repos",
172 json={"name": name, "owner": "testuser", "visibility": visibility},
173 headers=auth_headers,
174 )
175 assert r.status_code == 201, r.text
176 return r.json()["repoId"]
177
178
179 # ===========================================================================
180 # Layer 1 — Unit
181 # ===========================================================================
182
183
184 class TestUnitUtcIso:
185 def test_naive_datetime_gets_utc(self) -> None:
186 dt = datetime(2026, 1, 15, 12, 0, 0)
187 result = _utc_iso(dt)
188 assert "+00:00" in result or "Z" in result.upper() or "UTC" in result
189 assert "2026-01-15" in result
190
191 def test_aware_datetime_preserved(self) -> None:
192 dt = datetime(2026, 6, 1, 0, 0, 0, tzinfo=timezone.utc)
193 result = _utc_iso(dt)
194 assert "2026-06-01" in result
195
196 def test_returns_string(self) -> None:
197 assert isinstance(_utc_iso(datetime.now(timezone.utc)), str)
198
199 def test_iso_format_parseable(self) -> None:
200 dt = datetime(2025, 3, 14, 9, 26, 53, tzinfo=timezone.utc)
201 result = _utc_iso(dt)
202 parsed = datetime.fromisoformat(result)
203 assert parsed.year == 2025
204 assert parsed.month == 3
205 assert parsed.day == 14
206
207
208 class TestUnitExtractTracks:
209 def test_none_snapshot_returns_empty(self) -> None:
210 assert _extract_tracks_from_snapshot(None) == []
211
212 def test_any_object_returns_empty(self) -> None:
213 # stub returns [] regardless — just verifies the contract
214 assert _extract_tracks_from_snapshot(object()) == []
215
216 def test_returns_list(self) -> None:
217 result = _extract_tracks_from_snapshot(None)
218 assert isinstance(result, list)
219
220
221 class TestUnitHistoryLimit:
222 def test_brief_is_three(self) -> None:
223 assert _HISTORY_LIMIT[ContextDepth.brief] == 3
224
225 def test_standard_is_ten(self) -> None:
226 assert _HISTORY_LIMIT[ContextDepth.standard] == 10
227
228 def test_verbose_is_fifty(self) -> None:
229 assert _HISTORY_LIMIT[ContextDepth.verbose] == 50
230
231 def test_all_depths_covered(self) -> None:
232 for depth in ContextDepth:
233 assert depth in _HISTORY_LIMIT
234
235
236 class TestUnitIncludeFlags:
237 def test_proposal_body_brief_false(self) -> None:
238 assert _INCLUDE_PROPOSAL_BODY[ContextDepth.brief] is False
239
240 def test_proposal_body_standard_true(self) -> None:
241 assert _INCLUDE_PROPOSAL_BODY[ContextDepth.standard] is True
242
243 def test_proposal_body_verbose_true(self) -> None:
244 assert _INCLUDE_PROPOSAL_BODY[ContextDepth.verbose] is True
245
246 def test_issue_body_brief_false(self) -> None:
247 assert _INCLUDE_ISSUE_BODY[ContextDepth.brief] is False
248
249 def test_issue_body_standard_false(self) -> None:
250 assert _INCLUDE_ISSUE_BODY[ContextDepth.standard] is False
251
252 def test_issue_body_verbose_true(self) -> None:
253 assert _INCLUDE_ISSUE_BODY[ContextDepth.verbose] is True
254
255
256 class TestUnitGenerateSuggestions:
257 def _empty_state(self) -> MusicalStateContext:
258 return MusicalStateContext(active_tracks=[])
259
260 def _state_with_tracks(self) -> MusicalStateContext:
261 return MusicalStateContext(active_tracks=["drums", "bass"])
262
263 def _issue(self, n: int = 1) -> OpenIssueContext:
264 return OpenIssueContext(
265 issue_id=_uid(), number=n, title=f"issue {n}", labels=[], body=""
266 )
267
268 def _proposal_ctx(self) -> ActiveProposalContext:
269 return ActiveProposalContext(
270 proposal_id=_uid(),
271 title="add swing feel",
272 from_branch="feat/swing",
273 to_branch="main",
274 state="open",
275 body="",
276 )
277
278 def test_no_tracks_generates_suggestion(self) -> None:
279 s = _generate_suggestions(self._empty_state(), [], [], ContextDepth.standard)
280 assert len(s) >= 1
281 assert any("No files" in x for x in s)
282
283 def test_with_tracks_no_no_files_suggestion(self) -> None:
284 s = _generate_suggestions(
285 self._state_with_tracks(), [], [], ContextDepth.standard
286 )
287 assert not any("No files" in x for x in s)
288
289 def test_open_issue_generates_suggestion(self) -> None:
290 s = _generate_suggestions(
291 self._state_with_tracks(), [self._issue(5)], [], ContextDepth.standard
292 )
293 assert any("#5" in x for x in s)
294
295 def test_open_pr_generates_suggestion(self) -> None:
296 s = _generate_suggestions(
297 self._state_with_tracks(), [], [self._proposal_ctx()], ContextDepth.standard
298 )
299 assert any("add swing feel" in x for x in s)
300
301 def test_brief_caps_at_two(self) -> None:
302 # force 3 suggestions: no tracks + issue + proposal
303 s = _generate_suggestions(
304 self._empty_state(), [self._issue()], [self._proposal_ctx()], ContextDepth.brief
305 )
306 assert len(s) <= 2
307
308 def test_standard_caps_at_four(self) -> None:
309 issues = [self._issue(i) for i in range(1, 5)]
310 proposals_ctx = [self._proposal_ctx()]
311 # empty state + 4 issues + 1 proposal = 6 raw suggestions; capped at 4
312 s = _generate_suggestions(self._empty_state(), issues, proposals_ctx, ContextDepth.standard)
313 assert len(s) <= 4
314
315 def test_verbose_uncapped(self) -> None:
316 issues = [self._issue(i) for i in range(1, 5)]
317 proposals_ctx = [self._proposal_ctx()]
318 s = _generate_suggestions(self._empty_state(), issues, proposals_ctx, ContextDepth.verbose)
319 # 1 (no tracks) + 1 (first issue) + 1 (first proposal) = 3 — all returned
320 assert len(s) == 3
321
322 def test_returns_strings(self) -> None:
323 s = _generate_suggestions(self._empty_state(), [], [], ContextDepth.brief)
324 assert all(isinstance(x, str) for x in s)
325
326 def test_deterministic(self) -> None:
327 state = self._empty_state()
328 issues = [self._issue()]
329 proposals_ctx = [self._proposal_ctx()]
330 s1 = _generate_suggestions(state, issues, proposals_ctx, ContextDepth.standard)
331 s2 = _generate_suggestions(state, issues, proposals_ctx, ContextDepth.standard)
332 assert s1 == s2
333
334
335 class TestUnitModels:
336 def test_context_depth_values(self) -> None:
337 assert ContextDepth.brief == "brief"
338 assert ContextDepth.standard == "standard"
339 assert ContextDepth.verbose == "verbose"
340
341 def test_context_format_values(self) -> None:
342 assert ContextFormat.json == "json"
343 assert ContextFormat.yaml == "yaml"
344
345 def test_musical_state_default_empty_tracks(self) -> None:
346 m = MusicalStateContext()
347 assert m.active_tracks == []
348
349 def test_history_entry_context_fields(self) -> None:
350 h = HistoryEntryContext(
351 commit_id="abc123",
352 message="add bass",
353 author="agent",
354 timestamp="2026-01-01T00:00:00+00:00",
355 )
356 assert h.commit_id == "abc123"
357 assert h.active_tracks == []
358
359 def test_analysis_all_none_by_default(self) -> None:
360 a = AnalysisSummaryContext()
361 assert a.key_finding is None
362 assert a.chord_progression is None
363 assert a.groove_score is None
364 assert a.emotion is None
365 assert a.harmonic_tension is None
366 assert a.melodic_contour is None
367
368 def test_open_issue_context_defaults(self) -> None:
369 i = OpenIssueContext(issue_id=_uid(), number=1, title="fix")
370 assert i.labels == []
371 assert i.body == ""
372
373 def test_active_pr_context_defaults(self) -> None:
374 p = ActiveProposalContext(
375 proposal_id=_uid(),
376 title="Proposal",
377 from_branch="a",
378 to_branch="b",
379 state="open",
380 )
381 assert p.body == ""
382
383 def test_agent_context_response_camel_fields(self) -> None:
384 resp = AgentContextResponse(
385 repo_id="r1",
386 ref="main",
387 depth="standard",
388 musical_state=MusicalStateContext(),
389 analysis=AnalysisSummaryContext(),
390 )
391 d = resp.model_dump(by_alias=True)
392 assert "repoId" in d
393 assert "musicalState" in d
394 assert "activeProposals" in d
395 assert "openIssues" in d
396
397
398 # ===========================================================================
399 # Layer 2 — Integration
400 # ===========================================================================
401
402
403 class TestIntegrationResolveRef:
404 @pytest.mark.anyio
405 async def test_resolve_branch_name(self, db_session: AsyncSession) -> None:
406 repo_id = await _db_repo(db_session)
407 commit_id = await _db_commit(db_session, repo_id)
408 await _db_branch(db_session, repo_id, "main", commit_id)
409 await db_session.flush()
410
411 result = await _resolve_ref_to_commit(db_session, repo_id, "main")
412 assert result is not None
413 assert result.commit_id == commit_id
414
415 @pytest.mark.anyio
416 async def test_resolve_commit_id_directly(self, db_session: AsyncSession) -> None:
417 repo_id = await _db_repo(db_session)
418 commit_id = await _db_commit(db_session, repo_id)
419 await db_session.flush()
420
421 result = await _resolve_ref_to_commit(db_session, repo_id, commit_id)
422 assert result is not None
423 assert result.commit_id == commit_id
424
425 @pytest.mark.anyio
426 async def test_resolve_nonexistent_returns_none(self, db_session: AsyncSession) -> None:
427 repo_id = await _db_repo(db_session)
428 await db_session.flush()
429
430 result = await _resolve_ref_to_commit(db_session, repo_id, "nonexistent-ref")
431 assert result is None
432
433 @pytest.mark.anyio
434 async def test_branch_takes_priority_over_commit_id(self, db_session: AsyncSession) -> None:
435 """If a branch name happens to equal a commit ID substring, branch wins."""
436 repo_id = await _db_repo(db_session)
437 commit_id = await _db_commit(db_session, repo_id)
438 branch_commit_id = await _db_commit(db_session, repo_id, message="branch head")
439 await _db_branch(db_session, repo_id, "main", branch_commit_id)
440 await db_session.flush()
441
442 # Resolving "main" returns the branch head, not commit_id
443 result = await _resolve_ref_to_commit(db_session, repo_id, "main")
444 assert result is not None
445 assert result.commit_id == branch_commit_id
446
447
448 class TestIntegrationGetLatestCommit:
449 @pytest.mark.anyio
450 async def test_returns_most_recent(self, db_session: AsyncSession) -> None:
451 repo_id = await _db_repo(db_session)
452 ts_old = datetime(2026, 1, 1, tzinfo=timezone.utc)
453 ts_new = datetime(2026, 6, 1, tzinfo=timezone.utc)
454 await _db_commit(db_session, repo_id, ts=ts_old, message="old")
455 new_id = await _db_commit(db_session, repo_id, ts=ts_new, message="new")
456 await db_session.flush()
457
458 result = await _get_latest_commit(db_session, repo_id)
459 assert result is not None
460 assert result.commit_id == new_id
461
462 @pytest.mark.anyio
463 async def test_no_commits_returns_none(self, db_session: AsyncSession) -> None:
464 repo_id = await _db_repo(db_session)
465 await db_session.flush()
466
467 result = await _get_latest_commit(db_session, repo_id)
468 assert result is None
469
470
471 class TestIntegrationGetOpenProposals:
472 @pytest.mark.anyio
473 async def test_include_body_true(self, db_session: AsyncSession) -> None:
474 repo_id = await _db_repo(db_session)
475 await _db_proposal_ctx(db_session, repo_id, body="detailed body text")
476 await db_session.flush()
477
478 results = await _get_open_proposals(db_session, repo_id, include_body=True)
479 assert len(results) == 1
480 assert results[0].body == "detailed body text"
481
482 @pytest.mark.anyio
483 async def test_include_body_false(self, db_session: AsyncSession) -> None:
484 repo_id = await _db_repo(db_session)
485 await _db_proposal_ctx(db_session, repo_id, body="detailed body text")
486 await db_session.flush()
487
488 results = await _get_open_proposals(db_session, repo_id, include_body=False)
489 assert len(results) == 1
490 assert results[0].body == ""
491
492 @pytest.mark.anyio
493 async def test_closed_prs_excluded(self, db_session: AsyncSession) -> None:
494 repo_id = await _db_repo(db_session)
495 await _db_proposal_ctx(db_session, repo_id, state="closed")
496 await db_session.flush()
497
498 results = await _get_open_proposals(db_session, repo_id, include_body=False)
499 assert results == []
500
501
502 class TestIntegrationGetOpenIssues:
503 @pytest.mark.anyio
504 async def test_include_body_verbose(self, db_session: AsyncSession) -> None:
505 repo_id = await _db_repo(db_session)
506 await _db_issue(db_session, repo_id, body="full body text")
507 await db_session.flush()
508
509 results = await _get_open_issues(db_session, repo_id, include_body=True)
510 assert len(results) == 1
511 assert results[0].body == "full body text"
512
513 @pytest.mark.anyio
514 async def test_include_body_false_empty_string(self, db_session: AsyncSession) -> None:
515 repo_id = await _db_repo(db_session)
516 await _db_issue(db_session, repo_id, body="full body text")
517 await db_session.flush()
518
519 results = await _get_open_issues(db_session, repo_id, include_body=False)
520 assert results[0].body == ""
521
522 @pytest.mark.anyio
523 async def test_closed_issues_excluded(self, db_session: AsyncSession) -> None:
524 repo_id = await _db_repo(db_session)
525 await _db_issue(db_session, repo_id, state="closed")
526 await db_session.flush()
527
528 results = await _get_open_issues(db_session, repo_id, include_body=False)
529 assert results == []
530
531 @pytest.mark.anyio
532 async def test_ordered_by_number(self, db_session: AsyncSession) -> None:
533 repo_id = await _db_repo(db_session)
534 await _db_issue(db_session, repo_id, number=5, title="five")
535 await _db_issue(db_session, repo_id, number=2, title="two")
536 await _db_issue(db_session, repo_id, number=8, title="eight")
537 await db_session.flush()
538
539 results = await _get_open_issues(db_session, repo_id, include_body=False)
540 assert [r.number for r in results] == [2, 5, 8]
541
542
543 class TestIntegrationBuildAgentContext:
544 @pytest.mark.anyio
545 async def test_repo_not_found_returns_none(self, db_session: AsyncSession) -> None:
546 result = await build_agent_context(
547 db_session, repo_id="nonexistent-repo", ref="main"
548 )
549 assert result is None
550
551 @pytest.mark.anyio
552 async def test_no_commits_returns_none(self, db_session: AsyncSession) -> None:
553 repo_id = await _db_repo(db_session)
554 await db_session.flush()
555
556 result = await build_agent_context(db_session, repo_id=repo_id, ref="HEAD")
557 assert result is None
558
559 @pytest.mark.anyio
560 async def test_head_resolves_to_latest(self, db_session: AsyncSession) -> None:
561 repo_id = await _db_repo(db_session)
562 ts_old = datetime(2026, 1, 1, tzinfo=timezone.utc)
563 ts_new = datetime(2026, 6, 1, tzinfo=timezone.utc)
564 await _db_commit(db_session, repo_id, ts=ts_old, branch="main", message="old")
565 new_id = await _db_commit(
566 db_session, repo_id, ts=ts_new, branch="main", message="new"
567 )
568 await _db_branch(db_session, repo_id, "main", new_id)
569 await db_session.flush()
570
571 result = await build_agent_context(db_session, repo_id=repo_id, ref="HEAD")
572 assert result is not None
573 assert result.repo_id == repo_id
574 # History excludes the head commit; head is the new one
575 history_ids = [h.commit_id for h in result.history]
576 assert new_id not in history_ids
577
578 @pytest.mark.anyio
579 async def test_branch_ref_resolution(self, db_session: AsyncSession) -> None:
580 repo_id = await _db_repo(db_session)
581 commit_id = await _db_commit(db_session, repo_id, branch="feature")
582 await _db_branch(db_session, repo_id, "feature", commit_id)
583 await db_session.flush()
584
585 result = await build_agent_context(
586 db_session, repo_id=repo_id, ref="feature"
587 )
588 assert result is not None
589 assert result.ref == "feature"
590
591 @pytest.mark.anyio
592 async def test_brief_depth_history_limit(self, db_session: AsyncSession) -> None:
593 repo_id = await _db_repo(db_session)
594 for i in range(8):
595 ts = datetime(2026, 1, i + 1, tzinfo=timezone.utc)
596 await _db_commit(
597 db_session, repo_id, ts=ts, branch="main", message=f"commit {i}"
598 )
599 await db_session.flush()
600
601 result = await build_agent_context(
602 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.brief
603 )
604 assert result is not None
605 assert len(result.history) <= 3
606
607 @pytest.mark.anyio
608 async def test_verbose_depth_issue_body_included(
609 self, db_session: AsyncSession
610 ) -> None:
611 repo_id = await _db_repo(db_session)
612 await _db_commit(db_session, repo_id)
613 await _db_issue(db_session, repo_id, body="verbose body")
614 await db_session.flush()
615
616 result = await build_agent_context(
617 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
618 )
619 assert result is not None
620 assert len(result.open_issues) == 1
621 assert result.open_issues[0].body == "verbose body"
622
623 @pytest.mark.anyio
624 async def test_standard_depth_issue_body_empty(
625 self, db_session: AsyncSession
626 ) -> None:
627 repo_id = await _db_repo(db_session)
628 await _db_commit(db_session, repo_id)
629 await _db_issue(db_session, repo_id, body="hidden")
630 await db_session.flush()
631
632 result = await build_agent_context(
633 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.standard
634 )
635 assert result is not None
636 assert result.open_issues[0].body == ""
637
638
639 # ===========================================================================
640 # Layer 3 — E2E
641 # ===========================================================================
642
643
644 class TestE2EContextEndpoint:
645 @pytest.mark.anyio
646 async def test_200_with_all_sections(
647 self,
648 client: AsyncClient,
649 auth_headers: StrDict,
650 db_session: AsyncSession,
651 ) -> None:
652 repo_id = await _api_repo(client, auth_headers)
653 await _db_commit(db_session, repo_id)
654 await db_session.commit()
655
656 r = await client.get(f"/api/repos/{repo_id}/context", headers=auth_headers)
657 assert r.status_code == 200
658 body = r.json()
659 for key in ("repoId", "ref", "depth", "musicalState", "history", "analysis", "activeProposals", "openIssues", "suggestions"):
660 assert key in body
661
662 @pytest.mark.anyio
663 async def test_depth_brief_param(
664 self,
665 client: AsyncClient,
666 auth_headers: StrDict,
667 db_session: AsyncSession,
668 ) -> None:
669 repo_id = await _api_repo(client, auth_headers)
670 for i in range(6):
671 await _db_commit(db_session, repo_id, message=f"c{i}")
672 await db_session.commit()
673
674 r = await client.get(
675 f"/api/repos/{repo_id}/context?depth=brief", headers=auth_headers
676 )
677 assert r.status_code == 200
678 body = r.json()
679 assert body["depth"] == "brief"
680 assert len(body["history"]) <= 3
681
682 @pytest.mark.anyio
683 async def test_depth_verbose_param(
684 self,
685 client: AsyncClient,
686 auth_headers: StrDict,
687 db_session: AsyncSession,
688 ) -> None:
689 repo_id = await _api_repo(client, auth_headers)
690 await _db_commit(db_session, repo_id)
691 await _db_issue(db_session, repo_id, body="full body verbose")
692 await db_session.commit()
693
694 r = await client.get(
695 f"/api/repos/{repo_id}/context?depth=verbose", headers=auth_headers
696 )
697 assert r.status_code == 200
698 body = r.json()
699 assert body["depth"] == "verbose"
700 assert body["openIssues"][0]["body"] == "full body verbose"
701
702 @pytest.mark.anyio
703 async def test_invalid_depth_422(
704 self,
705 client: AsyncClient,
706 auth_headers: StrDict,
707 ) -> None:
708 r = await client.get(
709 "/api/repos/any-id/context?depth=ultra", headers=auth_headers
710 )
711 assert r.status_code == 422
712
713 @pytest.mark.anyio
714 async def test_unknown_repo_404(
715 self,
716 client: AsyncClient,
717 auth_headers: StrDict,
718 ) -> None:
719 r = await client.get(
720 "/api/repos/no-such-repo/context", headers=auth_headers
721 )
722 assert r.status_code == 404
723
724 @pytest.mark.anyio
725 async def test_nonexistent_ref_404(
726 self,
727 client: AsyncClient,
728 auth_headers: StrDict,
729 db_session: AsyncSession,
730 ) -> None:
731 repo_id = await _api_repo(client, auth_headers)
732 await db_session.commit()
733
734 r = await client.get(
735 f"/api/repos/{repo_id}/context?ref=no-such-branch", headers=auth_headers
736 )
737 assert r.status_code == 404
738
739 @pytest.mark.anyio
740 async def test_yaml_format_returns_yaml_content_type(
741 self,
742 client: AsyncClient,
743 auth_headers: StrDict,
744 db_session: AsyncSession,
745 ) -> None:
746 import yaml
747
748 repo_id = await _api_repo(client, auth_headers)
749 await _db_commit(db_session, repo_id)
750 await db_session.commit()
751
752 r = await client.get(
753 f"/api/repos/{repo_id}/context?format=yaml", headers=auth_headers
754 )
755 assert r.status_code == 200
756 assert "yaml" in r.headers.get("content-type", "")
757 parsed = yaml.safe_load(r.text)
758 assert isinstance(parsed, dict)
759 assert "repoId" in parsed
760
761
762 # ===========================================================================
763 # Layer 4 — Stress
764 # ===========================================================================
765
766
767 class TestStress:
768 @pytest.mark.anyio
769 async def test_verbose_depth_50_commit_history(
770 self,
771 db_session: AsyncSession,
772 ) -> None:
773 """build_agent_context handles 60 commits; verbose history capped at 50."""
774 repo_id = await _db_repo(db_session)
775 for i in range(60):
776 ts = datetime(2026, 1, 1, 0, i, 0, tzinfo=timezone.utc)
777 await _db_commit(db_session, repo_id, ts=ts, message=f"commit {i}")
778 await db_session.flush()
779
780 result = await build_agent_context(
781 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
782 )
783 assert result is not None
784 assert len(result.history) <= 50
785
786 @pytest.mark.anyio
787 async def test_concurrent_context_builds(
788 self,
789 db_session: AsyncSession,
790 ) -> None:
791 """5 concurrent build_agent_context calls on the same repo all succeed."""
792 repo_id = await _db_repo(db_session)
793 for i in range(5):
794 await _db_commit(db_session, repo_id, message=f"c{i}")
795 await db_session.flush()
796
797 results = await asyncio.gather(
798 *[
799 build_agent_context(
800 db_session, repo_id=repo_id, ref="HEAD"
801 )
802 for _ in range(5)
803 ]
804 )
805 assert all(r is not None for r in results)
806
807 @pytest.mark.anyio
808 async def test_many_open_issues_all_returned_verbose(
809 self,
810 db_session: AsyncSession,
811 ) -> None:
812 repo_id = await _db_repo(db_session)
813 await _db_commit(db_session, repo_id)
814 for i in range(20):
815 await _db_issue(db_session, repo_id, number=i + 1, title=f"issue {i}")
816 await db_session.flush()
817
818 result = await build_agent_context(
819 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
820 )
821 assert result is not None
822 assert len(result.open_issues) == 20
823
824
825 # ===========================================================================
826 # Layer 5 — Data Integrity
827 # ===========================================================================
828
829
830 class TestDataIntegrity:
831 @pytest.mark.anyio
832 async def test_history_newest_first(self, db_session: AsyncSession) -> None:
833 repo_id = await _db_repo(db_session)
834 for i in range(5):
835 ts = datetime(2026, 1, i + 1, tzinfo=timezone.utc)
836 await _db_commit(db_session, repo_id, ts=ts, message=f"c{i}")
837 await db_session.flush()
838
839 result = await build_agent_context(
840 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
841 )
842 assert result is not None
843 timestamps = [h.timestamp for h in result.history]
844 assert timestamps == sorted(timestamps, reverse=True)
845
846 @pytest.mark.anyio
847 async def test_head_commit_excluded_from_history(
848 self, db_session: AsyncSession
849 ) -> None:
850 repo_id = await _db_repo(db_session)
851 ts_old = datetime(2026, 1, 1, tzinfo=timezone.utc)
852 ts_new = datetime(2026, 6, 1, tzinfo=timezone.utc)
853 await _db_commit(db_session, repo_id, ts=ts_old, message="old")
854 new_id = await _db_commit(db_session, repo_id, ts=ts_new, message="new")
855 await db_session.flush()
856
857 # ref=HEAD resolves to new_id; it must NOT appear in history
858 result = await build_agent_context(
859 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
860 )
861 assert result is not None
862 history_ids = [h.commit_id for h in result.history]
863 assert new_id not in history_ids
864
865 @pytest.mark.anyio
866 async def test_closed_proposals_not_in_active_proposals(
867 self, db_session: AsyncSession
868 ) -> None:
869 repo_id = await _db_repo(db_session)
870 await _db_commit(db_session, repo_id)
871 await _db_proposal_ctx(db_session, repo_id, proposal_number=1, state="closed")
872 await _db_proposal_ctx(db_session, repo_id, proposal_number=2, state="merged", title="merged")
873 await db_session.flush()
874
875 result = await build_agent_context(
876 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
877 )
878 assert result is not None
879 assert result.active_proposals == []
880
881 @pytest.mark.anyio
882 async def test_closed_issues_not_in_open_issues(
883 self, db_session: AsyncSession
884 ) -> None:
885 repo_id = await _db_repo(db_session)
886 await _db_commit(db_session, repo_id)
887 await _db_issue(db_session, repo_id, state="closed")
888 await db_session.flush()
889
890 result = await build_agent_context(
891 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
892 )
893 assert result is not None
894 assert result.open_issues == []
895
896 @pytest.mark.anyio
897 async def test_proposal_body_empty_at_brief_depth(
898 self, db_session: AsyncSession
899 ) -> None:
900 repo_id = await _db_repo(db_session)
901 await _db_commit(db_session, repo_id)
902 await _db_proposal_ctx(db_session, repo_id, body="secret details")
903 await db_session.flush()
904
905 result = await build_agent_context(
906 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.brief
907 )
908 assert result is not None
909 assert result.active_proposals[0].body == ""
910
911 @pytest.mark.anyio
912 async def test_analysis_fields_all_none(self, db_session: AsyncSession) -> None:
913 repo_id = await _db_repo(db_session)
914 await _db_commit(db_session, repo_id)
915 await db_session.flush()
916
917 result = await build_agent_context(
918 db_session, repo_id=repo_id, ref="HEAD"
919 )
920 assert result is not None
921 a = result.analysis
922 assert a.key_finding is None
923 assert a.chord_progression is None
924 assert a.groove_score is None
925 assert a.emotion is None
926
927 @pytest.mark.anyio
928 async def test_repo_id_echoed_in_response(self, db_session: AsyncSession) -> None:
929 repo_id = await _db_repo(db_session)
930 await _db_commit(db_session, repo_id)
931 await db_session.flush()
932
933 result = await build_agent_context(db_session, repo_id=repo_id, ref="HEAD")
934 assert result is not None
935 assert result.repo_id == repo_id
936
937
938 # ===========================================================================
939 # Layer 6 — Security
940 # ===========================================================================
941
942
943 class TestSecurity:
944 @pytest.mark.anyio
945 async def test_private_repo_requires_auth(
946 self,
947 client: AsyncClient,
948 db_session: AsyncSession,
949 ) -> None:
950 """Context endpoint returns 403/401/404 for private repos without token."""
951 # Create repo and commit directly in DB (no auth_headers to avoid fixture override)
952 repo_id = await _db_repo(db_session, visibility="private")
953 await _db_commit(db_session, repo_id)
954 await db_session.commit()
955
956 r = await client.get(f"/api/repos/{repo_id}/context")
957 # private repo without auth → 403 or 401 (implementation may 404 for privacy)
958 assert r.status_code in (401, 403, 404)
959
960 @pytest.mark.anyio
961 async def test_public_repo_context_accessible_without_auth(
962 self,
963 client: AsyncClient,
964 db_session: AsyncSession,
965 ) -> None:
966 """Public repo context is readable without authentication."""
967 repo_id = await _db_repo(db_session, visibility="public")
968 await _db_commit(db_session, repo_id)
969 await db_session.commit()
970
971 r = await client.get(f"/api/repos/{repo_id}/context")
972 assert r.status_code == 200
973
974 @pytest.mark.anyio
975 async def test_sql_injection_in_ref_param_safe(
976 self,
977 client: AsyncClient,
978 auth_headers: StrDict,
979 db_session: AsyncSession,
980 ) -> None:
981 """SQL injection in ?ref param is handled safely (returns 404, not 500)."""
982 repo_id = await _api_repo(client, auth_headers)
983 await db_session.commit()
984
985 malicious_ref = "'; DROP TABLE musehub_commits; --"
986 r = await client.get(
987 f"/api/repos/{repo_id}/context",
988 params={"ref": malicious_ref},
989 headers=auth_headers,
990 )
991 assert r.status_code in (404, 422)
992
993 @pytest.mark.anyio
994 async def test_xss_in_ref_not_echoed_as_html(
995 self,
996 client: AsyncClient,
997 auth_headers: StrDict,
998 db_session: AsyncSession,
999 ) -> None:
1000 """XSS attempt in ?ref is not reflected as raw HTML in a 200 response."""
1001 repo_id = await _api_repo(client, auth_headers)
1002 await db_session.commit()
1003
1004 r = await client.get(
1005 f"/api/repos/{repo_id}/context",
1006 params={"ref": "<script>alert(1)</script>"},
1007 headers=auth_headers,
1008 )
1009 # Either rejected (404/422) or if echoed, must be JSON-escaped
1010 if r.status_code == 200:
1011 assert "<script>" not in r.text
1012 else:
1013 assert r.status_code in (404, 422)
1014
1015
1016 # ===========================================================================
1017 # Layer 7 — Performance
1018 # ===========================================================================
1019
1020
1021 class TestPerformance:
1022 @pytest.mark.anyio
1023 async def test_build_context_under_200ms(self, db_session: AsyncSession) -> None:
1024 """build_agent_context for 20 commits completes in under 200ms."""
1025 repo_id = await _db_repo(db_session)
1026 for i in range(20):
1027 ts = datetime(2026, 1, 1, 0, i, 0, tzinfo=timezone.utc)
1028 await _db_commit(db_session, repo_id, ts=ts, message=f"commit {i}")
1029 await db_session.flush()
1030
1031 start = time.perf_counter()
1032 result = await build_agent_context(
1033 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.standard
1034 )
1035 elapsed = time.perf_counter() - start
1036
1037 assert result is not None
1038 assert elapsed < 0.2, f"build_agent_context took {elapsed:.3f}s, expected <0.2s"
1039
1040 @pytest.mark.anyio
1041 async def test_verbose_context_50_commits_under_500ms(
1042 self, db_session: AsyncSession
1043 ) -> None:
1044 """Verbose depth with 55 commits (50 history + head) completes under 500ms."""
1045 repo_id = await _db_repo(db_session)
1046 for i in range(55):
1047 ts = datetime(2026, 1, 1, 0, 0, i, tzinfo=timezone.utc)
1048 await _db_commit(db_session, repo_id, ts=ts, message=f"commit {i}")
1049 await db_session.flush()
1050
1051 start = time.perf_counter()
1052 result = await build_agent_context(
1053 db_session, repo_id=repo_id, ref="HEAD", depth=ContextDepth.verbose
1054 )
1055 elapsed = time.perf_counter() - start
1056
1057 assert result is not None
1058 assert elapsed < 0.5, f"verbose build_agent_context took {elapsed:.3f}s"
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago