test_musehub_api_contracts.py
python
| 1 | """Deep API contract tests for core MuseHub endpoints. |
| 2 | |
| 3 | This file addresses the shallow-assertion gap: many existing tests only |
| 4 | assert on status codes. Here we verify complete response bodies, field |
| 5 | types, and envelope structure for the most critical endpoints — repos, |
| 6 | commits, branches, issues, and explore. |
| 7 | |
| 8 | Uses ``tests.factories`` for clean, declarative data setup. |
| 9 | All tests are module-level async functions (not class-based) to ensure |
| 10 | pytest-asyncio fixture injection works correctly. |
| 11 | """ |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import pytest |
| 15 | from httpx import AsyncClient |
| 16 | from sqlalchemy.ext.asyncio import AsyncSession |
| 17 | |
| 18 | from tests.factories import create_repo, create_branch, create_commit |
| 19 | |
| 20 | |
| 21 | # --------------------------------------------------------------------------- |
| 22 | # Repo CRUD response contracts |
| 23 | # --------------------------------------------------------------------------- |
| 24 | |
| 25 | @pytest.mark.anyio |
| 26 | async def test_create_repo_response_shape( |
| 27 | client: AsyncClient, |
| 28 | auth_headers: dict[str, str], |
| 29 | db_session: AsyncSession, |
| 30 | ) -> None: |
| 31 | """POST /repos returns all required fields with correct types.""" |
| 32 | resp = await client.post( |
| 33 | "/api/v1/musehub/repos", |
| 34 | json={"name": "contract-test", "owner": "tester", "visibility": "public"}, |
| 35 | headers=auth_headers, |
| 36 | ) |
| 37 | assert resp.status_code == 201 |
| 38 | body = resp.json() |
| 39 | |
| 40 | for key in ("repoId", "name", "owner", "slug", "visibility", "ownerUserId", |
| 41 | "cloneUrl", "createdAt"): |
| 42 | assert key in body, f"Missing field: {key}" |
| 43 | assert isinstance(body[key], str), f"Field {key} should be a string" |
| 44 | |
| 45 | assert body["name"] == "contract-test" |
| 46 | assert body["owner"] == "tester" |
| 47 | assert body["visibility"] == "public" |
| 48 | assert body["slug"] == "contract-test" |
| 49 | assert isinstance(body["tags"], list) |
| 50 | |
| 51 | |
| 52 | @pytest.mark.anyio |
| 53 | async def test_get_repo_response_shape( |
| 54 | client: AsyncClient, |
| 55 | auth_headers: dict[str, str], |
| 56 | db_session: AsyncSession, |
| 57 | ) -> None: |
| 58 | """GET /repos/{id} returns all expected fields.""" |
| 59 | create = await client.post( |
| 60 | "/api/v1/musehub/repos", |
| 61 | json={"name": "get-shape-test", "owner": "tester"}, |
| 62 | headers=auth_headers, |
| 63 | ) |
| 64 | repo_id = create.json()["repoId"] |
| 65 | |
| 66 | resp = await client.get(f"/api/v1/musehub/repos/{repo_id}", headers=auth_headers) |
| 67 | assert resp.status_code == 200 |
| 68 | body = resp.json() |
| 69 | |
| 70 | assert body["repoId"] == repo_id |
| 71 | assert body["name"] == "get-shape-test" |
| 72 | assert isinstance(body["tags"], list) |
| 73 | assert isinstance(body["createdAt"], str) |
| 74 | |
| 75 | |
| 76 | @pytest.mark.anyio |
| 77 | async def test_update_repo_settings_returns_updated_fields( |
| 78 | client: AsyncClient, |
| 79 | auth_headers: dict[str, str], |
| 80 | db_session: AsyncSession, |
| 81 | ) -> None: |
| 82 | """PATCH /repos/{id}/settings returns the updated repo fields.""" |
| 83 | create = await client.post( |
| 84 | "/api/v1/musehub/repos", |
| 85 | json={"name": "patch-test-repo", "owner": "patcher"}, |
| 86 | headers=auth_headers, |
| 87 | ) |
| 88 | repo_id = create.json()["repoId"] |
| 89 | |
| 90 | patch_resp = await client.patch( |
| 91 | f"/api/v1/musehub/repos/{repo_id}/settings", |
| 92 | json={"description": "Updated description", "visibility": "public"}, |
| 93 | headers=auth_headers, |
| 94 | ) |
| 95 | assert patch_resp.status_code == 200 |
| 96 | body = patch_resp.json() |
| 97 | assert body["description"] == "Updated description" |
| 98 | assert body["visibility"] == "public" |
| 99 | assert "name" in body # RepoSettingsResponse fields |
| 100 | |
| 101 | |
| 102 | # --------------------------------------------------------------------------- |
| 103 | # Branch response contracts |
| 104 | # --------------------------------------------------------------------------- |
| 105 | |
| 106 | @pytest.mark.anyio |
| 107 | async def test_list_branches_envelope( |
| 108 | client: AsyncClient, |
| 109 | auth_headers: dict[str, str], |
| 110 | db_session: AsyncSession, |
| 111 | ) -> None: |
| 112 | """GET /repos/{id}/branches returns a 'branches' list with name fields.""" |
| 113 | repo = await create_repo(db_session, owner="brancher", slug="branch-contract") |
| 114 | await create_branch(db_session, repo_id=str(repo.repo_id), name="main") |
| 115 | await create_branch(db_session, repo_id=str(repo.repo_id), name="feature-x") |
| 116 | |
| 117 | resp = await client.get( |
| 118 | f"/api/v1/musehub/repos/{repo.repo_id}/branches", |
| 119 | headers=auth_headers, |
| 120 | ) |
| 121 | assert resp.status_code == 200 |
| 122 | body = resp.json() |
| 123 | |
| 124 | assert "branches" in body |
| 125 | assert isinstance(body["branches"], list) |
| 126 | assert len(body["branches"]) == 2 |
| 127 | for branch in body["branches"]: |
| 128 | assert "name" in branch |
| 129 | assert isinstance(branch["name"], str) |
| 130 | |
| 131 | |
| 132 | @pytest.mark.anyio |
| 133 | async def test_branch_names_are_correct( |
| 134 | client: AsyncClient, |
| 135 | auth_headers: dict[str, str], |
| 136 | db_session: AsyncSession, |
| 137 | ) -> None: |
| 138 | """Branch names returned by the API match what was inserted.""" |
| 139 | repo = await create_repo(db_session, owner="brancher2", slug="branch-names") |
| 140 | await create_branch(db_session, repo_id=str(repo.repo_id), name="develop") |
| 141 | |
| 142 | resp = await client.get( |
| 143 | f"/api/v1/musehub/repos/{repo.repo_id}/branches", |
| 144 | headers=auth_headers, |
| 145 | ) |
| 146 | names = [b["name"] for b in resp.json()["branches"]] |
| 147 | assert "develop" in names |
| 148 | |
| 149 | |
| 150 | # --------------------------------------------------------------------------- |
| 151 | # Commit response contracts |
| 152 | # --------------------------------------------------------------------------- |
| 153 | |
| 154 | @pytest.mark.anyio |
| 155 | async def test_list_commits_envelope( |
| 156 | client: AsyncClient, |
| 157 | auth_headers: dict[str, str], |
| 158 | db_session: AsyncSession, |
| 159 | ) -> None: |
| 160 | """GET /repos/{id}/commits returns a 'commits' list envelope.""" |
| 161 | repo = await create_repo(db_session, owner="committer", slug="commit-contract") |
| 162 | await create_commit(db_session, str(repo.repo_id), message="init: first commit") |
| 163 | await create_commit(db_session, str(repo.repo_id), message="feat: second commit") |
| 164 | |
| 165 | resp = await client.get( |
| 166 | f"/api/v1/musehub/repos/{repo.repo_id}/commits", |
| 167 | headers=auth_headers, |
| 168 | ) |
| 169 | assert resp.status_code == 200 |
| 170 | body = resp.json() |
| 171 | |
| 172 | assert "commits" in body |
| 173 | assert isinstance(body["commits"], list) |
| 174 | assert len(body["commits"]) == 2 |
| 175 | |
| 176 | |
| 177 | @pytest.mark.anyio |
| 178 | async def test_commit_fields( |
| 179 | client: AsyncClient, |
| 180 | auth_headers: dict[str, str], |
| 181 | db_session: AsyncSession, |
| 182 | ) -> None: |
| 183 | """Each commit object has message, author, branch, commitId, timestamp, parentIds.""" |
| 184 | repo = await create_repo(db_session, owner="committer2", slug="commit-fields") |
| 185 | await create_commit( |
| 186 | db_session, str(repo.repo_id), |
| 187 | message="feat: piano track", |
| 188 | author="mozart", |
| 189 | branch="main", |
| 190 | ) |
| 191 | |
| 192 | resp = await client.get( |
| 193 | f"/api/v1/musehub/repos/{repo.repo_id}/commits", |
| 194 | headers=auth_headers, |
| 195 | ) |
| 196 | commits = resp.json()["commits"] |
| 197 | assert len(commits) == 1 |
| 198 | c = commits[0] |
| 199 | |
| 200 | assert c["message"] == "feat: piano track" |
| 201 | assert c["author"] == "mozart" |
| 202 | assert c["branch"] == "main" |
| 203 | assert "commitId" in c |
| 204 | assert "timestamp" in c |
| 205 | assert isinstance(c["parentIds"], list) |
| 206 | |
| 207 | |
| 208 | # --------------------------------------------------------------------------- |
| 209 | # Issue response contracts |
| 210 | # --------------------------------------------------------------------------- |
| 211 | |
| 212 | @pytest.mark.anyio |
| 213 | async def test_create_issue_response_shape( |
| 214 | client: AsyncClient, |
| 215 | auth_headers: dict[str, str], |
| 216 | db_session: AsyncSession, |
| 217 | ) -> None: |
| 218 | """POST /repos/{id}/issues returns title, body, status, number, createdAt.""" |
| 219 | create_repo_resp = await client.post( |
| 220 | "/api/v1/musehub/repos", |
| 221 | json={"name": "issue-contract-repo", "owner": "issuer"}, |
| 222 | headers=auth_headers, |
| 223 | ) |
| 224 | repo_id = create_repo_resp.json()["repoId"] |
| 225 | |
| 226 | resp = await client.post( |
| 227 | f"/api/v1/musehub/repos/{repo_id}/issues", |
| 228 | json={"title": "Bug: tempo drift", "body": "The tempo drifts by 3 BPM"}, |
| 229 | headers=auth_headers, |
| 230 | ) |
| 231 | assert resp.status_code == 201 |
| 232 | body = resp.json() |
| 233 | |
| 234 | assert body["title"] == "Bug: tempo drift" |
| 235 | assert body["body"] == "The tempo drifts by 3 BPM" |
| 236 | assert body["state"] == "open" |
| 237 | assert "number" in body |
| 238 | assert isinstance(body["number"], int) |
| 239 | assert "createdAt" in body |
| 240 | |
| 241 | |
| 242 | @pytest.mark.anyio |
| 243 | async def test_list_issues_returns_open_issues( |
| 244 | client: AsyncClient, |
| 245 | auth_headers: dict[str, str], |
| 246 | db_session: AsyncSession, |
| 247 | ) -> None: |
| 248 | """GET /repos/{id}/issues returns issues envelope with status=open.""" |
| 249 | create_repo_resp = await client.post( |
| 250 | "/api/v1/musehub/repos", |
| 251 | json={"name": "issue-list-contract", "owner": "issuer2"}, |
| 252 | headers=auth_headers, |
| 253 | ) |
| 254 | repo_id = create_repo_resp.json()["repoId"] |
| 255 | |
| 256 | for i in range(3): |
| 257 | await client.post( |
| 258 | f"/api/v1/musehub/repos/{repo_id}/issues", |
| 259 | json={"title": f"Issue {i}"}, |
| 260 | headers=auth_headers, |
| 261 | ) |
| 262 | |
| 263 | resp = await client.get( |
| 264 | f"/api/v1/musehub/repos/{repo_id}/issues", |
| 265 | headers=auth_headers, |
| 266 | ) |
| 267 | assert resp.status_code == 200 |
| 268 | body = resp.json() |
| 269 | |
| 270 | assert "issues" in body |
| 271 | assert len(body["issues"]) == 3 |
| 272 | for issue in body["issues"]: |
| 273 | assert issue["state"] == "open" |
| 274 | assert "title" in issue |
| 275 | assert "number" in issue |
| 276 | |
| 277 | |
| 278 | @pytest.mark.anyio |
| 279 | async def test_close_issue_changes_status( |
| 280 | client: AsyncClient, |
| 281 | auth_headers: dict[str, str], |
| 282 | db_session: AsyncSession, |
| 283 | ) -> None: |
| 284 | """POST /repos/{id}/issues/{n}/close sets status to 'closed'.""" |
| 285 | create_repo_resp = await client.post( |
| 286 | "/api/v1/musehub/repos", |
| 287 | json={"name": "close-issue-contract", "owner": "closer"}, |
| 288 | headers=auth_headers, |
| 289 | ) |
| 290 | repo_id = create_repo_resp.json()["repoId"] |
| 291 | |
| 292 | issue_resp = await client.post( |
| 293 | f"/api/v1/musehub/repos/{repo_id}/issues", |
| 294 | json={"title": "Close me"}, |
| 295 | headers=auth_headers, |
| 296 | ) |
| 297 | number = issue_resp.json()["number"] |
| 298 | |
| 299 | close_resp = await client.post( |
| 300 | f"/api/v1/musehub/repos/{repo_id}/issues/{number}/close", |
| 301 | headers=auth_headers, |
| 302 | ) |
| 303 | assert close_resp.status_code == 200 |
| 304 | assert close_resp.json()["state"] == "closed" |
| 305 | |
| 306 | |
| 307 | # --------------------------------------------------------------------------- |
| 308 | # Explore / discover |
| 309 | # --------------------------------------------------------------------------- |
| 310 | |
| 311 | @pytest.mark.anyio |
| 312 | async def test_explore_returns_public_repos( |
| 313 | client: AsyncClient, |
| 314 | auth_headers: dict[str, str], |
| 315 | db_session: AsyncSession, |
| 316 | ) -> None: |
| 317 | """GET /repos/explore returns public repos and excludes private ones.""" |
| 318 | await client.post( |
| 319 | "/api/v1/musehub/repos", |
| 320 | json={"name": "explore-public", "owner": "explorer", "visibility": "public"}, |
| 321 | headers=auth_headers, |
| 322 | ) |
| 323 | await client.post( |
| 324 | "/api/v1/musehub/repos", |
| 325 | json={"name": "explore-private", "owner": "explorer", "visibility": "private"}, |
| 326 | headers=auth_headers, |
| 327 | ) |
| 328 | |
| 329 | resp = await client.get("/api/v1/musehub/discover/repos") |
| 330 | assert resp.status_code == 200 |
| 331 | body = resp.json() |
| 332 | repos = body if isinstance(body, list) else body.get("repos", body.get("items", [])) |
| 333 | |
| 334 | slugs = [r.get("slug", "") for r in repos] |
| 335 | assert "explore-public" in slugs |
| 336 | assert "explore-private" not in slugs |