"""Section 10 — Issues & Milestones: 7-layer test suite. Covers: - musehub/services/musehub_issues.py (all 19 service functions) - musehub/api/routes/musehub/issues.py (14 endpoints) - musehub/api/routes/musehub/milestones.py (5 endpoints) - musehub/api/routes/musehub/labels.py (label CRUD + assignment) - musehub/mcp/write_tools/issues.py (3 agent tools) - musehub/db/musehub_models.py (MusehubIssue, MusehubIssueComment, MusehubMilestone, MusehubIssueMilestone) Layers: 1. Unit — pure service functions, no HTTP 2. Integration — service calls against real test DB, no HTTP 3. End-to-End — full HTTP via AsyncClient 4. Stress — 100-issue repos, bulk label ops, deep comment threads 5. Data Integrity — sequential numbering, soft-delete, constraint enforcement 6. Security — auth enforcement, cross-repo isolation, input limits 7. Performance — list under 200ms, pagination correct, N+1 not catastrophic """ from __future__ import annotations import time import uuid from datetime import datetime, timedelta, timezone import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from musehub.db import musehub_models as db from musehub.models.musehub import IssueResponse, MilestoneResponse from musehub.services import musehub_issues from tests.factories import create_repo from musehub.muse_contracts.json_types import JSONObject, StrDict # ── Helpers ─────────────────────────────────────────────────────────────────── def _uid() -> str: return str(uuid.uuid4()) def _now() -> datetime: return datetime.now(tz=timezone.utc) async def _repo(session: AsyncSession, slug: str) -> db.MusehubRepo: return await create_repo(session, slug=slug) async def _issue( session: AsyncSession, repo_id: str, *, title: str = "Test issue", body: str = "", labels: list[str] | None = None, author: str = "tester", ) -> IssueResponse: issue = await musehub_issues.create_issue( session, repo_id=repo_id, title=title, body=body, labels=labels or [], author=author, ) await session.commit() return issue async def _milestone( session: AsyncSession, repo_id: str, *, title: str = "v1.0", description: str = "", due_on: datetime | None = None, ) -> MilestoneResponse: ms = await musehub_issues.create_milestone( session, repo_id=repo_id, title=title, description=description, due_on=due_on, ) await session.commit() return ms # HTTP helpers async def _api_repo(client: AsyncClient, auth_headers: StrDict, name: str = "repo") -> str: r = await client.post( "/api/repos", json={"name": name, "owner": "testuser"}, headers=auth_headers ) assert r.status_code == 201 return r.json()["repoId"] async def _api_issue( client: AsyncClient, auth_headers: StrDict, repo_id: str, title: str = "Issue", **kwargs: str | list[str], ) -> JSONObject: r = await client.post( f"/api/repos/{repo_id}/issues", json={"title": title, "body": kwargs.get("body", ""), "labels": kwargs.get("labels", [])}, headers=auth_headers, ) assert r.status_code == 201 return r.json() # =========================================================================== # Layer 1 — Unit tests (service layer, no HTTP) # =========================================================================== class TestUnitIssueCreate: @pytest.mark.anyio async def test_creates_in_open_state(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-create-1") issue = await _issue(db_session, repo.repo_id, title="My issue") assert issue.state == "open" @pytest.mark.anyio async def test_sequential_numbers(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-seq") i1 = await _issue(db_session, repo.repo_id, title="First") i2 = await _issue(db_session, repo.repo_id, title="Second") i3 = await _issue(db_session, repo.repo_id, title="Third") assert [i1.number, i2.number, i3.number] == [1, 2, 3] @pytest.mark.anyio async def test_numbers_independent_per_repo(self, db_session: AsyncSession) -> None: r1 = await _repo(db_session, "u-seq-r1") r2 = await _repo(db_session, "u-seq-r2") i1 = await _issue(db_session, r1.repo_id, title="R1 issue") i2 = await _issue(db_session, r2.repo_id, title="R2 issue") assert i1.number == 1 assert i2.number == 1 # independent sequence @pytest.mark.anyio async def test_stores_author(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-author") issue = await _issue(db_session, repo.repo_id, author="alice") assert issue.author == "alice" @pytest.mark.anyio async def test_stores_labels(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-labels") issue = await _issue(db_session, repo.repo_id, labels=["bug", "p1"]) assert set(issue.labels) == {"bug", "p1"} class TestUnitIssueStateTransitions: @pytest.mark.anyio async def test_close_issue(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-close") issue = await _issue(db_session, repo.repo_id) closed = await musehub_issues.close_issue(db_session, repo.repo_id, issue.number) assert closed is not None assert closed.state == "closed" @pytest.mark.anyio async def test_reopen_issue(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-reopen") issue = await _issue(db_session, repo.repo_id) await musehub_issues.close_issue(db_session, repo.repo_id, issue.number) reopened = await musehub_issues.reopen_issue(db_session, repo.repo_id, issue.number) assert reopened is not None assert reopened.state == "open" @pytest.mark.anyio async def test_close_nonexistent_returns_none(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-close-miss") result = await musehub_issues.close_issue(db_session, repo.repo_id, 9999) assert result is None @pytest.mark.anyio async def test_reopen_idempotent(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-reopen-idem") issue = await _issue(db_session, repo.repo_id) r1 = await musehub_issues.reopen_issue(db_session, repo.repo_id, issue.number) r2 = await musehub_issues.reopen_issue(db_session, repo.repo_id, issue.number) assert r1.state == "open" assert r2.state == "open" class TestUnitIssueUpdate: @pytest.mark.anyio async def test_update_title(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-upd-title") issue = await _issue(db_session, repo.repo_id, title="Old") updated = await musehub_issues.update_issue( db_session, repo.repo_id, issue.number, title="New" ) assert updated is not None assert updated.title == "New" @pytest.mark.anyio async def test_update_body_only(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-upd-body") issue = await _issue(db_session, repo.repo_id, title="Keep") updated = await musehub_issues.update_issue( db_session, repo.repo_id, issue.number, body="New body" ) assert updated.title == "Keep" assert updated.body == "New body" @pytest.mark.anyio async def test_update_labels_replaces_all(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-upd-labels") issue = await _issue(db_session, repo.repo_id, labels=["a", "b"]) updated = await musehub_issues.update_issue( db_session, repo.repo_id, issue.number, labels=["c"] ) assert updated.labels == ["c"] @pytest.mark.anyio async def test_update_nonexistent_returns_none(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-upd-miss") result = await musehub_issues.update_issue( db_session, repo.repo_id, 9999, title="X" ) assert result is None class TestUnitLabelOps: @pytest.mark.anyio async def test_assign_labels_replaces(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-lbl-replace") issue = await _issue(db_session, repo.repo_id, labels=["x", "y"]) result = await musehub_issues.assign_labels( db_session, repo.repo_id, issue.number, labels=["z"] ) assert result.labels == ["z"] @pytest.mark.anyio async def test_remove_label_single(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-lbl-rm") issue = await _issue(db_session, repo.repo_id, labels=["bug", "p1"]) result = await musehub_issues.remove_label( db_session, repo.repo_id, issue.number, label="bug" ) assert "bug" not in result.labels assert "p1" in result.labels @pytest.mark.anyio async def test_remove_label_idempotent(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-lbl-idem") issue = await _issue(db_session, repo.repo_id, labels=["bug"]) r1 = await musehub_issues.remove_label( db_session, repo.repo_id, issue.number, label="bug" ) r2 = await musehub_issues.remove_label( db_session, repo.repo_id, issue.number, label="bug" ) assert r1.labels == [] assert r2.labels == [] class TestUnitComments: @pytest.mark.anyio async def test_create_and_list_comment(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-comment-1") issue = await _issue(db_session, repo.repo_id) comment = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Hello", author="alice", ) await db_session.commit() listed = await musehub_issues.list_comments(db_session, issue.issue_id) assert listed.total == 1 assert listed.comments[0].body == "Hello" @pytest.mark.anyio async def test_soft_delete_hides_comment(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-comment-del") issue = await _issue(db_session, repo.repo_id) comment = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Soon deleted", author="bob", ) await db_session.commit() await musehub_issues.delete_comment(db_session, comment.comment_id, issue.issue_id) await db_session.commit() listed = await musehub_issues.list_comments(db_session, issue.issue_id) assert listed.total == 0 @pytest.mark.anyio async def test_include_deleted_exposes_soft_deleted(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-comment-incl") issue = await _issue(db_session, repo.repo_id) comment = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Deleted body", author="bob", ) await db_session.commit() await musehub_issues.delete_comment(db_session, comment.comment_id, issue.issue_id) await db_session.commit() listed = await musehub_issues.list_comments(db_session, issue.issue_id, include_deleted=True) assert listed.total == 1 assert listed.comments[0].is_deleted is True @pytest.mark.anyio async def test_threaded_reply_validates_parent(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-thread") issue = await _issue(db_session, repo.repo_id) parent = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Parent", author="alice", ) await db_session.commit() reply = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Reply", author="bob", parent_id=parent.comment_id, ) await db_session.commit() assert reply.parent_id == parent.comment_id @pytest.mark.anyio async def test_invalid_parent_raises_value_error(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-thread-invalid") issue = await _issue(db_session, repo.repo_id) with pytest.raises(ValueError, match="not found"): await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Orphan", author="eve", parent_id=_uid(), ) class TestUnitMilestones: @pytest.mark.anyio async def test_create_milestone(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-ms-create") ms = await _milestone(db_session, repo.repo_id, title="Sprint 1") assert ms.title == "Sprint 1" assert ms.state == "open" assert ms.number == 1 @pytest.mark.anyio async def test_milestone_numbers_sequential(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-ms-seq") m1 = await _milestone(db_session, repo.repo_id, title="M1") m2 = await _milestone(db_session, repo.repo_id, title="M2") assert m1.number == 1 assert m2.number == 2 @pytest.mark.anyio async def test_set_issue_milestone(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-ms-assign") issue = await _issue(db_session, repo.repo_id) ms = await _milestone(db_session, repo.repo_id, title="V1") result = await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=ms.milestone_id ) await db_session.commit() assert result.milestone_id == ms.milestone_id assert result.milestone_title == "V1" @pytest.mark.anyio async def test_clear_issue_milestone(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-ms-clear") issue = await _issue(db_session, repo.repo_id) ms = await _milestone(db_session, repo.repo_id, title="V1") await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=ms.milestone_id ) await db_session.commit() result = await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=None ) await db_session.commit() assert result.milestone_id is None @pytest.mark.anyio async def test_cross_repo_milestone_raises(self, db_session: AsyncSession) -> None: r1 = await _repo(db_session, "u-ms-xrepo-1") r2 = await _repo(db_session, "u-ms-xrepo-2") issue = await _issue(db_session, r1.repo_id) ms = await _milestone(db_session, r2.repo_id, title="Other") with pytest.raises(ValueError): await musehub_issues.set_issue_milestone( db_session, r1.repo_id, issue.number, milestone_id=ms.milestone_id ) @pytest.mark.anyio async def test_list_milestones_open_only_by_default(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-ms-state") await _milestone(db_session, repo.repo_id, title="Open") closed_ms = await _milestone(db_session, repo.repo_id, title="Closed") # Manually close it row = await db_session.get(db.MusehubMilestone, closed_ms.milestone_id) row.state = "closed" await db_session.commit() result = await musehub_issues.list_milestones(db_session, repo.repo_id, state="open") assert all(m.state == "open" for m in result.milestones) assert len(result.milestones) == 1 @pytest.mark.anyio async def test_completeness_sort(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "u-ms-complete") m1 = await _milestone(db_session, repo.repo_id, title="Empty") m2 = await _milestone(db_session, repo.repo_id, title="Full") # Assign an issue to m2 and close it issue = await _issue(db_session, repo.repo_id) await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=m2.milestone_id ) await musehub_issues.close_issue(db_session, repo.repo_id, issue.number) await db_session.commit() result = await musehub_issues.list_milestones( db_session, repo.repo_id, state="all", sort="completeness" ) # "Full" (100% closed) should come before "Empty" (0%) assert result.milestones[0].title == "Full" assert result.milestones[1].title == "Empty" # =========================================================================== # Layer 2 — Integration tests (service + real DB, no HTTP) # =========================================================================== class TestIntegrationCommentCount: @pytest.mark.anyio async def test_comment_count_in_issue_response(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-cc-1") issue = await _issue(db_session, repo.repo_id) for i in range(3): await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body=f"Comment {i}", author="u", ) await db_session.commit() fetched = await musehub_issues.get_issue(db_session, repo.repo_id, issue.number) assert fetched.comment_count == 3 @pytest.mark.anyio async def test_soft_deleted_comments_not_counted(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-cc-2") issue = await _issue(db_session, repo.repo_id) c1 = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Keep", author="u", ) c2 = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Delete me", author="u", ) await db_session.commit() await musehub_issues.delete_comment(db_session, c2.comment_id, issue.issue_id) await db_session.commit() fetched = await musehub_issues.get_issue(db_session, repo.repo_id, issue.number) assert fetched.comment_count == 1 class TestIntegrationListFilters: @pytest.mark.anyio async def test_list_open_default(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-lf-1") i1 = await _issue(db_session, repo.repo_id, title="Open") i2 = await _issue(db_session, repo.repo_id, title="Closed") await musehub_issues.close_issue(db_session, repo.repo_id, i2.number) await db_session.commit() results = await musehub_issues.list_issues(db_session, repo.repo_id, state="open") assert all(r.state == "open" for r in results) assert len(results) == 1 @pytest.mark.anyio async def test_list_all_state(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-lf-2") for i in range(3): await _issue(db_session, repo.repo_id, title=f"Issue {i}") await musehub_issues.close_issue(db_session, repo.repo_id, 1) await db_session.commit() results = await musehub_issues.list_issues(db_session, repo.repo_id, state="all") assert len(results) == 3 @pytest.mark.anyio async def test_list_by_label(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-lf-3") await _issue(db_session, repo.repo_id, labels=["bug"]) await _issue(db_session, repo.repo_id, labels=["enhancement"]) await _issue(db_session, repo.repo_id, labels=["bug", "p1"]) results = await musehub_issues.list_issues(db_session, repo.repo_id, label="bug") assert len(results) == 2 assert all("bug" in r.labels for r in results) @pytest.mark.anyio async def test_list_by_milestone(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-lf-4") ms = await _milestone(db_session, repo.repo_id, title="Sprint") i1 = await _issue(db_session, repo.repo_id, title="In milestone") i2 = await _issue(db_session, repo.repo_id, title="No milestone") await musehub_issues.set_issue_milestone( db_session, repo.repo_id, i1.number, milestone_id=ms.milestone_id ) await db_session.commit() results = await musehub_issues.list_issues( db_session, repo.repo_id, milestone_id=ms.milestone_id ) assert len(results) == 1 assert results[0].number == i1.number @pytest.mark.anyio async def test_list_ordered_by_number(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-lf-5") for i in range(5): await _issue(db_session, repo.repo_id, title=f"Issue {i}") results = await musehub_issues.list_issues(db_session, repo.repo_id, state="all") numbers = [r.number for r in results] assert numbers == sorted(numbers) class TestIntegrationMilestoneProgress: @pytest.mark.anyio async def test_open_and_closed_counts(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "int-mp-1") ms = await _milestone(db_session, repo.repo_id, title="Sprint") for i in range(4): issue = await _issue(db_session, repo.repo_id, title=f"Task {i}") await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=ms.milestone_id ) # Close 2 of them await musehub_issues.close_issue(db_session, repo.repo_id, 1) await musehub_issues.close_issue(db_session, repo.repo_id, 2) await db_session.commit() fetched = await musehub_issues.get_milestone(db_session, repo.repo_id, ms.number) assert fetched.open_issues == 2 assert fetched.closed_issues == 2 @pytest.mark.anyio async def test_delete_milestone_unlinks_issues(self, db_session: AsyncSession) -> None: from musehub.api.routes.musehub import milestones as ms_routes from sqlalchemy import update repo = await _repo(db_session, "int-mp-del") ms = await _milestone(db_session, repo.repo_id, title="Doomed") issue = await _issue(db_session, repo.repo_id) await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=ms.milestone_id ) await db_session.commit() # Unlink issues then delete (mirrors route handler logic) await db_session.execute( update(db.MusehubIssue) .where(db.MusehubIssue.milestone_id == ms.milestone_id) .values(milestone_id=None) ) ms_row = await db_session.get(db.MusehubMilestone, ms.milestone_id) await db_session.delete(ms_row) await db_session.commit() # Issue should now have no milestone result = await musehub_issues.get_issue(db_session, repo.repo_id, issue.number) assert result.milestone_id is None # =========================================================================== # Layer 3 — End-to-End tests (full HTTP via AsyncClient) # =========================================================================== class TestE2EIssues: @pytest.mark.anyio async def test_create_issue_201( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-create") r = await client.post( f"/api/repos/{repo_id}/issues", json={"title": "First issue", "body": "Details here", "labels": ["bug"]}, headers=auth_headers, ) assert r.status_code == 201 data = r.json() assert data["state"] == "open" assert data["number"] == 1 assert data["title"] == "First issue" @pytest.mark.anyio async def test_get_issue_by_number( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-get") issue = await _api_issue(client, auth_headers, repo_id, title="Fetchable") r = await client.get(f"/api/repos/{repo_id}/issues/{issue['number']}") assert r.status_code == 200 assert r.json()["title"] == "Fetchable" @pytest.mark.anyio async def test_get_nonexistent_issue_404( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-get-miss") r = await client.get(f"/api/repos/{repo_id}/issues/9999") assert r.status_code == 404 @pytest.mark.anyio async def test_list_issues_default_open( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-list") issue = await _api_issue(client, auth_headers, repo_id, title="Open") await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/close", headers=auth_headers, ) await _api_issue(client, auth_headers, repo_id, title="Still open") r = await client.get(f"/api/repos/{repo_id}/issues") assert r.status_code == 200 data = r.json() assert all(i["state"] == "open" for i in data["issues"]) @pytest.mark.anyio async def test_close_issue( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-close") issue = await _api_issue(client, auth_headers, repo_id) r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/close", headers=auth_headers, ) assert r.status_code == 200 assert r.json()["state"] == "closed" @pytest.mark.anyio async def test_reopen_issue( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-reopen") issue = await _api_issue(client, auth_headers, repo_id) await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/close", headers=auth_headers, ) r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/reopen", headers=auth_headers, ) assert r.status_code == 200 assert r.json()["state"] == "open" @pytest.mark.anyio async def test_update_issue_patch( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-patch") issue = await _api_issue(client, auth_headers, repo_id, title="Old title") r = await client.patch( f"/api/repos/{repo_id}/issues/{issue['number']}", json={"title": "New title", "labels": ["enhancement"]}, headers=auth_headers, ) assert r.status_code == 200 data = r.json() assert data["title"] == "New title" assert "enhancement" in data["labels"] @pytest.mark.anyio async def test_assign_issue( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-assign") issue = await _api_issue(client, auth_headers, repo_id) r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/assign", json={"assignee": "dev-team-lead"}, headers=auth_headers, ) assert r.status_code == 200 assert r.json()["assignee"] == "dev-team-lead" @pytest.mark.anyio async def test_unassign_issue( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-unassign") issue = await _api_issue(client, auth_headers, repo_id) await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/assign", json={"assignee": "somebody"}, headers=auth_headers, ) r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/assign", json={"assignee": None}, headers=auth_headers, ) assert r.json()["assignee"] is None class TestE2EComments: @pytest.mark.anyio async def test_create_comment_201( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-c-create") issue = await _api_issue(client, auth_headers, repo_id) r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/comments", json={"body": "Looks good to me"}, headers=auth_headers, ) assert r.status_code == 201 data = r.json() assert data["comments"][0]["body"] == "Looks good to me" @pytest.mark.anyio async def test_list_comments( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-c-list") issue = await _api_issue(client, auth_headers, repo_id) for i in range(3): await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/comments", json={"body": f"Comment {i}"}, headers=auth_headers, ) r = await client.get( f"/api/repos/{repo_id}/issues/{issue['number']}/comments" ) assert r.status_code == 200 assert r.json()["total"] == 3 @pytest.mark.anyio async def test_delete_comment_204( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-c-del") issue = await _api_issue(client, auth_headers, repo_id) post_r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/comments", json={"body": "Delete me"}, headers=auth_headers, ) comment_id = post_r.json()["comments"][0]["commentId"] r = await client.delete( f"/api/repos/{repo_id}/issues/{issue['number']}/comments/{comment_id}", headers=auth_headers, ) assert r.status_code == 204 # Not visible in subsequent list listed = await client.get( f"/api/repos/{repo_id}/issues/{issue['number']}/comments" ) assert listed.json()["total"] == 0 class TestE2EMilestones: @pytest.mark.anyio async def test_create_milestone_201( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-ms-create") r = await client.post( f"/api/repos/{repo_id}/milestones", json={"title": "Sprint 1", "description": "First sprint"}, headers=auth_headers, ) assert r.status_code == 201 data = r.json() assert data["title"] == "Sprint 1" assert data["number"] == 1 assert data["state"] == "open" @pytest.mark.anyio async def test_get_milestone_by_number( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-ms-get") await client.post( f"/api/repos/{repo_id}/milestones", json={"title": "M1"}, headers=auth_headers, ) r = await client.get(f"/api/repos/{repo_id}/milestones/1") assert r.status_code == 200 assert r.json()["title"] == "M1" @pytest.mark.anyio async def test_update_milestone( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-ms-upd") await client.post( f"/api/repos/{repo_id}/milestones", json={"title": "Old"}, headers=auth_headers, ) r = await client.patch( f"/api/repos/{repo_id}/milestones/1", json={"title": "New", "state": "closed"}, headers=auth_headers, ) assert r.status_code == 200 data = r.json() assert data["title"] == "New" assert data["state"] == "closed" @pytest.mark.anyio async def test_update_milestone_clear_due_on( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-ms-due") due = (_now() + timedelta(days=7)).isoformat() await client.post( f"/api/repos/{repo_id}/milestones", json={"title": "Due soon", "dueOn": due}, headers=auth_headers, ) r = await client.patch( f"/api/repos/{repo_id}/milestones/1", json={"dueOn": None}, headers=auth_headers, ) assert r.status_code == 200 assert r.json()["dueOn"] is None @pytest.mark.anyio async def test_delete_milestone_204( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-ms-del") await client.post( f"/api/repos/{repo_id}/milestones", json={"title": "Gone"}, headers=auth_headers, ) r = await client.delete( f"/api/repos/{repo_id}/milestones/1", headers=auth_headers ) assert r.status_code == 204 r2 = await client.get(f"/api/repos/{repo_id}/milestones/1") assert r2.status_code == 404 @pytest.mark.anyio async def test_set_issue_milestone_via_http( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-ms-link") issue = await _api_issue(client, auth_headers, repo_id) ms_r = await client.post( f"/api/repos/{repo_id}/milestones", json={"title": "Sprint"}, headers=auth_headers, ) ms_id = ms_r.json()["milestoneId"] r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/milestone", params={"milestone_id": ms_id}, headers=auth_headers, ) assert r.status_code == 200 assert r.json()["milestoneId"] == ms_id @pytest.mark.anyio async def test_remove_issue_milestone_via_http( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "e2e-ms-unlink") issue = await _api_issue(client, auth_headers, repo_id) ms_r = await client.post( f"/api/repos/{repo_id}/milestones", json={"title": "Sprint"}, headers=auth_headers ) ms_id = ms_r.json()["milestoneId"] await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/milestone", params={"milestone_id": ms_id}, headers=auth_headers, ) r = await client.delete( f"/api/repos/{repo_id}/issues/{issue['number']}/milestone", headers=auth_headers, ) assert r.status_code == 200 assert r.json()["milestoneId"] is None class TestE2EMCPTools: @pytest.mark.anyio async def test_execute_create_issue(self, db_session: AsyncSession) -> None: from musehub.mcp.write_tools.issues import execute_create_issue from musehub.db.database import AsyncSessionLocal repo = await _repo(db_session, "mcp-create") await db_session.commit() result = await execute_create_issue( repo_id=repo.repo_id, title="Agent-filed issue", body="From MCP", actor="agent-1", ) assert result.ok is True assert result.data["title"] == "Agent-filed issue" assert result.data["number"] == 1 @pytest.mark.anyio async def test_execute_create_issue_missing_repo(self, db_session: AsyncSession) -> None: from musehub.mcp.write_tools.issues import execute_create_issue result = await execute_create_issue( repo_id=_uid(), title="Won't work", actor="agent" ) assert result.ok is False assert result.error_code == "repo_not_found" @pytest.mark.anyio async def test_execute_update_issue_close(self, db_session: AsyncSession) -> None: from musehub.mcp.write_tools.issues import execute_update_issue repo = await _repo(db_session, "mcp-update") issue = await _issue(db_session, repo.repo_id, title="Open issue") result = await execute_update_issue( repo_id=repo.repo_id, issue_number=issue.number, state="closed" ) assert result.ok is True assert result.data["state"] == "closed" @pytest.mark.anyio async def test_execute_create_comment(self, db_session: AsyncSession) -> None: from musehub.mcp.write_tools.issues import execute_create_issue_comment repo = await _repo(db_session, "mcp-comment") issue = await _issue(db_session, repo.repo_id) await db_session.commit() result = await execute_create_issue_comment( repo_id=repo.repo_id, issue_number=issue.number, body="Agent comment", actor="agent-2", ) assert result.ok is True assert result.data["body"] == "Agent comment" # =========================================================================== # Layer 4 — Stress tests # =========================================================================== class TestStress: @pytest.mark.anyio async def test_100_issues_in_one_repo(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "stress-100") for i in range(100): await musehub_issues.create_issue( db_session, repo_id=repo.repo_id, title=f"Issue {i:03d}", body="", labels=[], ) await db_session.commit() results = await musehub_issues.list_issues(db_session, repo.repo_id, state="all") assert len(results) == 100 assert results[-1].number == 100 @pytest.mark.anyio async def test_bulk_label_replacement(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "stress-labels") issues = [] for i in range(20): issues.append(await _issue(db_session, repo.repo_id, labels=["old"])) await db_session.commit() for issue in issues: await musehub_issues.assign_labels( db_session, repo.repo_id, issue.number, labels=["new"] ) await db_session.commit() results = await musehub_issues.list_issues(db_session, repo.repo_id, state="all") assert all(r.labels == ["new"] for r in results) @pytest.mark.anyio async def test_deep_comment_thread(self, db_session: AsyncSession) -> None: """10-level deep reply chain must be created without error.""" repo = await _repo(db_session, "stress-thread") issue = await _issue(db_session, repo.repo_id) parent_id = None for depth in range(10): comment = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body=f"Depth {depth}", author="u", parent_id=parent_id, ) await db_session.commit() parent_id = comment.comment_id listed = await musehub_issues.list_comments(db_session, issue.issue_id) assert listed.total == 10 @pytest.mark.anyio async def test_milestone_with_50_issues(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "stress-ms-50") ms = await _milestone(db_session, repo.repo_id, title="Big Sprint") for i in range(50): issue = await musehub_issues.create_issue( db_session, repo_id=repo.repo_id, title=f"Task {i}", body="", labels=[], ) await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=ms.milestone_id ) # Close half for n in range(1, 26): await musehub_issues.close_issue(db_session, repo.repo_id, n) await db_session.commit() fetched = await musehub_issues.get_milestone(db_session, repo.repo_id, ms.number) assert fetched.open_issues == 25 assert fetched.closed_issues == 25 # =========================================================================== # Layer 5 — Data Integrity tests # =========================================================================== class TestDataIntegrity: @pytest.mark.anyio async def test_issue_number_survives_deletions(self, db_session: AsyncSession) -> None: """Numbers must not be reused when issues are closed or deleted.""" repo = await _repo(db_session, "di-number-gap") i1 = await _issue(db_session, repo.repo_id, title="First") i2 = await _issue(db_session, repo.repo_id, title="Second") i3 = await _issue(db_session, repo.repo_id, title="Third") # Close the middle one — next issue must still be #4 await musehub_issues.close_issue(db_session, repo.repo_id, i2.number) i4 = await _issue(db_session, repo.repo_id, title="Fourth") assert i4.number == 4 @pytest.mark.anyio async def test_update_does_not_change_number(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "di-update-num") issue = await _issue(db_session, repo.repo_id) original_number = issue.number updated = await musehub_issues.update_issue( db_session, repo.repo_id, issue.number, title="Changed title" ) assert updated.number == original_number @pytest.mark.anyio async def test_cross_repo_issue_isolation(self, db_session: AsyncSession) -> None: r1 = await _repo(db_session, "di-iso-1") r2 = await _repo(db_session, "di-iso-2") await _issue(db_session, r1.repo_id, title="In R1") result = await musehub_issues.get_issue(db_session, r2.repo_id, 1) assert result is None @pytest.mark.anyio async def test_assign_issue_does_not_affect_other_issues( self, db_session: AsyncSession ) -> None: repo = await _repo(db_session, "di-assign-iso") i1 = await _issue(db_session, repo.repo_id) i2 = await _issue(db_session, repo.repo_id) await musehub_issues.assign_issue( db_session, repo.repo_id, i1.number, assignee="alice" ) await db_session.commit() fetched_i2 = await musehub_issues.get_issue(db_session, repo.repo_id, i2.number) assert fetched_i2.assignee is None @pytest.mark.anyio async def test_soft_delete_preserves_comment_in_db( self, db_session: AsyncSession ) -> None: repo = await _repo(db_session, "di-softdel") issue = await _issue(db_session, repo.repo_id) comment = await musehub_issues.create_comment( db_session, issue_id=issue.issue_id, repo_id=repo.repo_id, body="Still in DB", author="u", ) await db_session.commit() await musehub_issues.delete_comment(db_session, comment.comment_id, issue.issue_id) await db_session.commit() # Raw DB row still exists with is_deleted=True row = await db_session.get(db.MusehubIssueComment, comment.comment_id) assert row is not None assert row.is_deleted is True assert row.body == "Still in DB" @pytest.mark.anyio async def test_milestone_state_closed_persists(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "di-ms-closed") ms = await _milestone(db_session, repo.repo_id, title="Done sprint") row = await db_session.get(db.MusehubMilestone, ms.milestone_id) row.state = "closed" await db_session.commit() fetched = await musehub_issues.get_milestone(db_session, repo.repo_id, ms.number) assert fetched.state == "closed" @pytest.mark.anyio async def test_delete_comment_returns_false_for_wrong_issue( self, db_session: AsyncSession ) -> None: repo = await _repo(db_session, "di-del-wrong") i1 = await _issue(db_session, repo.repo_id) i2 = await _issue(db_session, repo.repo_id) comment = await musehub_issues.create_comment( db_session, issue_id=i1.issue_id, repo_id=repo.repo_id, body="On issue 1", author="u", ) await db_session.commit() # Passing wrong issue_id should return False (comment not found on i2) result = await musehub_issues.delete_comment( db_session, comment.comment_id, i2.issue_id ) assert result is False # =========================================================================== # Layer 6 — Security tests # =========================================================================== class TestSecurity: @pytest.mark.anyio async def test_create_issue_requires_auth( self, client: AsyncClient ) -> None: r = await client.post( "/api/repos/some-repo-id/issues", json={"title": "No auth"}, ) assert r.status_code in (401, 403, 422) @pytest.mark.anyio async def test_close_requires_auth( self, client: AsyncClient, db_session: AsyncSession ) -> None: # Set up via DB directly (not HTTP auth) so auth override isn't active repo = await _repo(db_session, "sec-close-auth") issue = await _issue(db_session, repo.repo_id, title="Issue") await db_session.commit() r = await client.post( f"/api/repos/{repo.repo_id}/issues/{issue.number}/close" ) assert r.status_code in (401, 403) @pytest.mark.anyio async def test_comment_create_requires_auth( self, client: AsyncClient, db_session: AsyncSession ) -> None: repo = await _repo(db_session, "sec-comment-auth") issue = await _issue(db_session, repo.repo_id, title="Issue") await db_session.commit() r = await client.post( f"/api/repos/{repo.repo_id}/issues/{issue.number}/comments", json={"body": "Unauthenticated"}, ) assert r.status_code in (401, 403) @pytest.mark.anyio async def test_milestone_create_requires_auth( self, client: AsyncClient, db_session: AsyncSession ) -> None: repo = await _repo(db_session, "sec-ms-auth") await db_session.commit() r = await client.post( f"/api/repos/{repo.repo_id}/milestones", json={"title": "No auth milestone"}, ) assert r.status_code in (401, 403) @pytest.mark.anyio async def test_cross_repo_issue_not_accessible( self, client: AsyncClient, auth_headers: StrDict ) -> None: r1 = await _api_repo(client, auth_headers, "sec-xrepo-1") r2 = await _api_repo(client, auth_headers, "sec-xrepo-2") await _api_issue(client, auth_headers, r1, title="Private issue in R1") # Issue #1 of r2 doesn't exist r = await client.get(f"/api/repos/{r2}/issues/1") assert r.status_code == 404 @pytest.mark.anyio async def test_title_max_length_enforced( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "sec-title-len") r = await client.post( f"/api/repos/{repo_id}/issues", json={"title": "x" * 501, "body": ""}, headers=auth_headers, ) assert r.status_code == 422 @pytest.mark.anyio async def test_comment_body_max_length_enforced( self, client: AsyncClient, auth_headers: StrDict ) -> None: repo_id = await _api_repo(client, auth_headers, "sec-body-len") issue = await _api_issue(client, auth_headers, repo_id) r = await client.post( f"/api/repos/{repo_id}/issues/{issue['number']}/comments", json={"body": "x" * 10_001}, headers=auth_headers, ) assert r.status_code == 422 @pytest.mark.anyio async def test_read_endpoints_accessible_without_auth( self, client: AsyncClient, auth_headers: StrDict ) -> None: """Public repos: read endpoints must not require auth.""" repo_id = await _api_repo(client, auth_headers, "sec-public-read") await _api_issue(client, auth_headers, repo_id) # Read without any auth headers r = await client.get(f"/api/repos/{repo_id}/issues") assert r.status_code == 200 # =========================================================================== # Layer 7 — Performance tests # =========================================================================== class TestPerformance: @pytest.mark.anyio async def test_list_50_issues_under_200ms(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "perf-list-50") for i in range(50): await musehub_issues.create_issue( db_session, repo_id=repo.repo_id, title=f"Issue {i}", body="", labels=[], ) await db_session.commit() t0 = time.perf_counter() results = await musehub_issues.list_issues(db_session, repo.repo_id, state="all") elapsed_ms = (time.perf_counter() - t0) * 1000 assert len(results) == 50 assert elapsed_ms < 200, f"list_issues(50) took {elapsed_ms:.1f}ms" @pytest.mark.anyio async def test_list_milestones_with_counts_under_300ms( self, db_session: AsyncSession ) -> None: repo = await _repo(db_session, "perf-ms-counts") for i in range(10): ms = await musehub_issues.create_milestone( db_session, repo_id=repo.repo_id, title=f"Sprint {i}", ) for j in range(5): issue = await musehub_issues.create_issue( db_session, repo_id=repo.repo_id, title=f"Task {i}-{j}", body="", labels=[], ) await musehub_issues.set_issue_milestone( db_session, repo.repo_id, issue.number, milestone_id=ms.milestone_id ) await db_session.commit() t0 = time.perf_counter() result = await musehub_issues.list_milestones( db_session, repo.repo_id, state="open" ) elapsed_ms = (time.perf_counter() - t0) * 1000 assert len(result.milestones) == 10 assert elapsed_ms < 300, f"list_milestones(10, 5 issues each) took {elapsed_ms:.1f}ms" @pytest.mark.anyio async def test_create_issue_under_50ms(self, db_session: AsyncSession) -> None: repo = await _repo(db_session, "perf-create") t0 = time.perf_counter() await musehub_issues.create_issue( db_session, repo_id=repo.repo_id, title="Perf issue", body="", labels=[], ) elapsed_ms = (time.perf_counter() - t0) * 1000 assert elapsed_ms < 50, f"create_issue took {elapsed_ms:.1f}ms" @pytest.mark.anyio async def test_pagination_total_consistent( self, client: AsyncClient, auth_headers: StrDict ) -> None: """total in IssueListResponse must match actual record count.""" repo_id = await _api_repo(client, auth_headers, "perf-pagination") for i in range(15): await _api_issue(client, auth_headers, repo_id, title=f"Issue {i}") r1 = await client.get( f"/api/repos/{repo_id}/issues", params={"page": 1, "per_page": 10} ) r2 = await client.get( f"/api/repos/{repo_id}/issues", params={"page": 2, "per_page": 10} ) total = r1.json()["total"] assert total == 15 assert len(r1.json()["issues"]) == 10 assert len(r2.json()["issues"]) == 5