gabriel / musehub public
test_labels_section28.py python
1,027 lines 36.3 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Section 28 — Labels: 7-layer test suite.
2
3 Covers: musehub/api/routes/musehub/labels.py, musehub/db/musehub_label_models.py
4
5 Layers:
6 1. Unit — pure-function tests, no DB, no HTTP
7 2. Integration — real DB session, service-level calls
8 3. End-to-End — full HTTP stack via AsyncClient
9 4. Stress — concurrency, bulk operations
10 5. Data Integrity— constraint enforcement, rollback
11 6. Security — auth bypass, privilege escalation
12 7. Performance — latency budgets, query efficiency
13 """
14 from __future__ import annotations
15
16 import time
17 import uuid
18 from typing import AsyncGenerator
19
20 import pytest
21 from httpx import AsyncClient
22 from sqlalchemy import text
23 from sqlalchemy.ext.asyncio import AsyncSession
24
25 from musehub.muse_contracts.json_types import StrDict
26 from musehub.api.routes.musehub.labels import (
27 DEFAULT_LABELS,
28 LabelCreate,
29 LabelListResponse,
30 LabelResponse,
31 LabelUpdate,
32 AssignLabelsRequest,
33 _get_label_or_404,
34 seed_default_labels,
35 )
36 from musehub.db.musehub_label_models import MusehubIssueLabel, MusehubLabel, MusehubProposalLabel
37 from musehub.db.musehub_models import MusehubIssue, MusehubProposal, MusehubRepo
38
39
40 # ── helpers ───────────────────────────────────────────────────────────────────
41
42
43 def _uid() -> str:
44 return str(uuid.uuid4())
45
46
47 async def _db_repo(session: AsyncSession, *, visibility: str = "public") -> MusehubRepo:
48 slug = f"label-repo-{_uid()[:8]}"
49 repo = MusehubRepo(
50 repo_id=_uid(),
51 name=slug,
52 slug=slug,
53 owner="testuser",
54 owner_user_id="testuser",
55 visibility=visibility,
56 )
57 session.add(repo)
58 await session.flush()
59 return repo
60
61
62 async def _db_label(
63 session: AsyncSession,
64 repo_id: str,
65 *,
66 name: str | None = None,
67 color: str = "#aabbcc",
68 ) -> MusehubLabel:
69 label = MusehubLabel(
70 id=_uid(),
71 repo_id=repo_id,
72 name=name or f"label-{_uid()[:8]}",
73 color=color,
74 description="test label",
75 )
76 session.add(label)
77 await session.flush()
78 return label
79
80
81 async def _db_issue(session: AsyncSession, repo_id: str, *, number: int = 1) -> MusehubIssue:
82 issue = MusehubIssue(
83 issue_id=_uid(),
84 repo_id=repo_id,
85 number=number,
86 title="Test issue",
87 body="body",
88 state="open",
89 labels=[],
90 author="testuser",
91 )
92 session.add(issue)
93 await session.flush()
94 return issue
95
96
97 async def _db_proposal(
98 session: AsyncSession, repo_id: str, *, proposal_number: int = 1
99 ) -> MusehubProposal:
100 proposal = MusehubProposal(
101 proposal_id=_uid(),
102 repo_id=repo_id,
103 proposal_number=proposal_number,
104 title="Test Proposal",
105 body="",
106 state="open",
107 from_branch="feature",
108 to_branch="dev",
109 author="testuser",
110 )
111 session.add(proposal)
112 await session.flush()
113 return proposal
114
115
116 # ═══════════════════════════════════════════════════════════════════════════════
117 # Layer 1 — Unit
118 # ═══════════════════════════════════════════════════════════════════════════════
119
120
121 class TestUnitLabelModels:
122 """Pure model/schema validation — no DB, no HTTP."""
123
124 def test_default_labels_not_empty(self) -> None:
125 assert len(DEFAULT_LABELS) > 0
126
127 def test_default_labels_have_required_fields(self) -> None:
128 for entry in DEFAULT_LABELS:
129 assert "name" in entry
130 assert "color" in entry
131 assert entry["color"].startswith("#")
132 assert len(entry["color"]) == 7 # #rrggbb
133
134 def test_default_labels_names_unique(self) -> None:
135 names = [e["name"] for e in DEFAULT_LABELS]
136 assert len(names) == len(set(names))
137
138 def test_label_create_valid(self) -> None:
139 lc = LabelCreate(name="bug", color="#d73a4a")
140 assert lc.name == "bug"
141 assert lc.color == "#d73a4a"
142 assert lc.description is None
143
144 def test_label_create_with_description(self) -> None:
145 lc = LabelCreate(name="bug", color="#d73a4a", description="It breaks")
146 assert lc.description == "It breaks"
147
148 def test_label_update_all_optional(self) -> None:
149 lu = LabelUpdate()
150 assert lu.name is None
151 assert lu.color is None
152 assert lu.description is None
153
154 def test_label_update_partial(self) -> None:
155 lu = LabelUpdate(color="#ffffff")
156 assert lu.color == "#ffffff"
157 assert lu.name is None
158
159 def test_label_response_round_trip(self) -> None:
160 lr = LabelResponse(
161 label_id="abc",
162 repo_id="repo1",
163 name="bug",
164 color="#d73a4a",
165 description=None,
166 )
167 assert lr.label_id == "abc"
168 assert lr.repo_id == "repo1"
169 assert lr.description is None
170
171 def test_label_list_response(self) -> None:
172 items = [
173 LabelResponse(label_id=_uid(), repo_id="r", name="a", color="#111111"),
174 LabelResponse(label_id=_uid(), repo_id="r", name="b", color="#222222"),
175 ]
176 llr = LabelListResponse(items=items, total=2)
177 assert llr.total == 2
178 assert len(llr.items) == 2
179
180 def test_assign_labels_request_requires_min_one(self) -> None:
181 with pytest.raises(Exception):
182 AssignLabelsRequest(label_ids=[])
183
184 def test_assign_labels_request_valid(self) -> None:
185 req = AssignLabelsRequest(label_ids=["abc", "def"])
186 assert req.label_ids == ["abc", "def"]
187
188
189 # ═══════════════════════════════════════════════════════════════════════════════
190 # Layer 2 — Integration
191 # ═══════════════════════════════════════════════════════════════════════════════
192
193
194 class TestIntegrationLabelDB:
195 """Real DB session, service-layer functions."""
196
197 @pytest.mark.anyio
198 async def test_seed_default_labels_inserts_all(self, db_session: AsyncSession) -> None:
199 repo = await _db_repo(db_session)
200 await db_session.commit()
201 await seed_default_labels(db_session, repo.repo_id)
202 await db_session.commit()
203
204 result = await db_session.execute(
205 text("SELECT COUNT(*) FROM musehub_labels WHERE repo_id = :rid"),
206 {"rid": repo.repo_id},
207 )
208 count = result.scalar_one()
209 assert count == len(DEFAULT_LABELS)
210
211 @pytest.mark.anyio
212 async def test_seed_default_labels_idempotent(self, db_session: AsyncSession) -> None:
213 repo = await _db_repo(db_session)
214 await db_session.commit()
215 await seed_default_labels(db_session, repo.repo_id)
216 await db_session.commit()
217 await seed_default_labels(db_session, repo.repo_id)
218 await db_session.commit()
219
220 result = await db_session.execute(
221 text("SELECT COUNT(*) FROM musehub_labels WHERE repo_id = :rid"),
222 {"rid": repo.repo_id},
223 )
224 assert result.scalar_one() == len(DEFAULT_LABELS)
225
226 @pytest.mark.anyio
227 async def test_get_label_or_404_found(self, db_session: AsyncSession) -> None:
228 repo = await _db_repo(db_session)
229 label = await _db_label(db_session, repo.repo_id, name="found-label")
230 await db_session.commit()
231
232 result = await _get_label_or_404(db_session, repo.repo_id, label.id)
233 assert result.name == "found-label"
234
235 @pytest.mark.anyio
236 async def test_get_label_or_404_missing(self, db_session: AsyncSession) -> None:
237 from fastapi import HTTPException
238
239 repo = await _db_repo(db_session)
240 await db_session.commit()
241
242 with pytest.raises(HTTPException) as exc_info:
243 await _get_label_or_404(db_session, repo.repo_id, "nonexistent-id")
244 assert exc_info.value.status_code == 404
245
246 @pytest.mark.anyio
247 async def test_label_unique_constraint_within_repo(self, db_session: AsyncSession) -> None:
248 from sqlalchemy.exc import IntegrityError
249
250 repo = await _db_repo(db_session)
251 await _db_label(db_session, repo.repo_id, name="duplicate")
252 await db_session.flush()
253
254 label2 = MusehubLabel(
255 id=_uid(), repo_id=repo.repo_id, name="duplicate", color="#000000"
256 )
257 db_session.add(label2)
258 with pytest.raises(IntegrityError):
259 await db_session.flush()
260
261 @pytest.mark.anyio
262 async def test_same_name_different_repos_allowed(self, db_session: AsyncSession) -> None:
263 repo1 = await _db_repo(db_session)
264 repo2 = await _db_repo(db_session)
265 await _db_label(db_session, repo1.repo_id, name="shared-name")
266 await _db_label(db_session, repo2.repo_id, name="shared-name")
267 await db_session.flush() # no error expected
268
269 @pytest.mark.anyio
270 async def test_issue_label_assignment_db(self, db_session: AsyncSession) -> None:
271 repo = await _db_repo(db_session)
272 label = await _db_label(db_session, repo.repo_id)
273 issue = await _db_issue(db_session, repo.repo_id)
274 await db_session.commit()
275
276 il = MusehubIssueLabel(issue_id=issue.issue_id, label_id=label.id)
277 db_session.add(il)
278 await db_session.flush()
279
280 result = await db_session.execute(
281 text("SELECT COUNT(*) FROM musehub_issue_labels WHERE issue_id = :iid"),
282 {"iid": issue.issue_id},
283 )
284 assert result.scalar_one() == 1
285
286 @pytest.mark.anyio
287 async def test_proposal_label_assignment_db(self, db_session: AsyncSession) -> None:
288 repo = await _db_repo(db_session)
289 label = await _db_label(db_session, repo.repo_id)
290 proposal = await _db_proposal(db_session, repo.repo_id)
291 await db_session.commit()
292
293 prl = MusehubProposalLabel(proposal_id=proposal.proposal_id, label_id=label.id)
294 db_session.add(prl)
295 await db_session.flush()
296
297 result = await db_session.execute(
298 text("SELECT COUNT(*) FROM musehub_proposal_labels WHERE proposal_id = :pid"),
299 {"pid": proposal.proposal_id},
300 )
301 assert result.scalar_one() == 1
302
303
304 # ═══════════════════════════════════════════════════════════════════════════════
305 # Layer 3 — End-to-End
306 # ═══════════════════════════════════════════════════════════════════════════════
307
308
309 class TestE2ELabels:
310 """Full HTTP stack via AsyncClient."""
311
312 @pytest.mark.anyio
313 async def test_list_labels_empty(
314 self, client: AsyncClient, db_session: AsyncSession
315 ) -> None:
316 repo = await _db_repo(db_session)
317 await db_session.commit()
318
319 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
320 assert resp.status_code == 200
321 data = resp.json()
322 assert data["total"] == 0
323 assert data["items"] == []
324
325 @pytest.mark.anyio
326 async def test_list_labels_with_data(
327 self, client: AsyncClient, db_session: AsyncSession
328 ) -> None:
329 repo = await _db_repo(db_session)
330 await _db_label(db_session, repo.repo_id, name="alpha")
331 await _db_label(db_session, repo.repo_id, name="beta")
332 await db_session.commit()
333
334 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
335 assert resp.status_code == 200
336 data = resp.json()
337 assert data["total"] == 2
338 names = [i["name"] for i in data["items"]]
339 assert "alpha" in names
340 assert "beta" in names
341
342 @pytest.mark.anyio
343 async def test_list_labels_sorted_alphabetically(
344 self, client: AsyncClient, db_session: AsyncSession
345 ) -> None:
346 repo = await _db_repo(db_session)
347 await _db_label(db_session, repo.repo_id, name="zzz")
348 await _db_label(db_session, repo.repo_id, name="aaa")
349 await db_session.commit()
350
351 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
352 names = [i["name"] for i in resp.json()["items"]]
353 assert names == sorted(names)
354
355 @pytest.mark.anyio
356 async def test_list_labels_repo_not_found(self, client: AsyncClient) -> None:
357 resp = await client.get("/api/repos/nonexistent-repo/labels")
358 assert resp.status_code == 404
359
360 @pytest.mark.anyio
361 async def test_create_label_success(
362 self,
363 client: AsyncClient,
364 db_session: AsyncSession,
365 auth_headers: StrDict,
366 ) -> None:
367 repo = await _db_repo(db_session)
368 await db_session.commit()
369
370 resp = await client.post(
371 f"/api/repos/{repo.repo_id}/labels",
372 json={"name": "enhancement", "color": "#a2eeef", "description": "New feature"},
373 headers=auth_headers,
374 )
375 assert resp.status_code == 201
376 data = resp.json()
377 assert data["name"] == "enhancement"
378 assert data["color"] == "#a2eeef"
379 assert data["label_id"] is not None
380
381 @pytest.mark.anyio
382 async def test_create_label_duplicate_name_409(
383 self,
384 client: AsyncClient,
385 db_session: AsyncSession,
386 auth_headers: StrDict,
387 ) -> None:
388 repo = await _db_repo(db_session)
389 await _db_label(db_session, repo.repo_id, name="existing")
390 await db_session.commit()
391
392 resp = await client.post(
393 f"/api/repos/{repo.repo_id}/labels",
394 json={"name": "existing", "color": "#ffffff"},
395 headers=auth_headers,
396 )
397 assert resp.status_code == 409
398
399 @pytest.mark.anyio
400 async def test_update_label_name(
401 self,
402 client: AsyncClient,
403 db_session: AsyncSession,
404 auth_headers: StrDict,
405 ) -> None:
406 repo = await _db_repo(db_session)
407 label = await _db_label(db_session, repo.repo_id, name="old-name")
408 await db_session.commit()
409
410 resp = await client.patch(
411 f"/api/repos/{repo.repo_id}/labels/{label.id}",
412 json={"name": "new-name"},
413 headers=auth_headers,
414 )
415 assert resp.status_code == 200
416 assert resp.json()["name"] == "new-name"
417
418 @pytest.mark.anyio
419 async def test_update_label_color(
420 self,
421 client: AsyncClient,
422 db_session: AsyncSession,
423 auth_headers: StrDict,
424 ) -> None:
425 repo = await _db_repo(db_session)
426 label = await _db_label(db_session, repo.repo_id, name="colored", color="#000000")
427 await db_session.commit()
428
429 resp = await client.patch(
430 f"/api/repos/{repo.repo_id}/labels/{label.id}",
431 json={"color": "#ffffff"},
432 headers=auth_headers,
433 )
434 assert resp.status_code == 200
435 assert resp.json()["color"] == "#ffffff"
436
437 @pytest.mark.anyio
438 async def test_update_label_name_conflict_409(
439 self,
440 client: AsyncClient,
441 db_session: AsyncSession,
442 auth_headers: StrDict,
443 ) -> None:
444 repo = await _db_repo(db_session)
445 await _db_label(db_session, repo.repo_id, name="taken")
446 label = await _db_label(db_session, repo.repo_id, name="mine")
447 await db_session.commit()
448
449 resp = await client.patch(
450 f"/api/repos/{repo.repo_id}/labels/{label.id}",
451 json={"name": "taken"},
452 headers=auth_headers,
453 )
454 assert resp.status_code == 409
455
456 @pytest.mark.anyio
457 async def test_delete_label_success(
458 self,
459 client: AsyncClient,
460 db_session: AsyncSession,
461 auth_headers: StrDict,
462 ) -> None:
463 repo = await _db_repo(db_session)
464 label = await _db_label(db_session, repo.repo_id, name="to-delete")
465 await db_session.commit()
466
467 resp = await client.delete(
468 f"/api/repos/{repo.repo_id}/labels/{label.id}",
469 headers=auth_headers,
470 )
471 assert resp.status_code == 204
472
473 @pytest.mark.anyio
474 async def test_delete_label_not_found(
475 self,
476 client: AsyncClient,
477 db_session: AsyncSession,
478 auth_headers: StrDict,
479 ) -> None:
480 repo = await _db_repo(db_session)
481 await db_session.commit()
482
483 resp = await client.delete(
484 f"/api/repos/{repo.repo_id}/labels/nonexistent-id",
485 headers=auth_headers,
486 )
487 assert resp.status_code == 404
488
489 @pytest.mark.anyio
490 async def test_list_labels_after_create(
491 self,
492 client: AsyncClient,
493 db_session: AsyncSession,
494 auth_headers: StrDict,
495 ) -> None:
496 # Verify that a label created via POST is immediately visible via GET
497 repo = await _db_repo(db_session)
498 await db_session.commit()
499
500 await client.post(
501 f"/api/repos/{repo.repo_id}/labels",
502 json={"name": "freshly-created", "color": "#aabbcc"},
503 headers=auth_headers,
504 )
505
506 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
507 assert resp.status_code == 200
508 names = [i["name"] for i in resp.json()["items"]]
509 assert "freshly-created" in names
510
511 @pytest.mark.anyio
512 async def test_update_then_list_reflects_change(
513 self,
514 client: AsyncClient,
515 db_session: AsyncSession,
516 auth_headers: StrDict,
517 ) -> None:
518 repo = await _db_repo(db_session)
519 label = await _db_label(db_session, repo.repo_id, name="original")
520 await db_session.commit()
521
522 await client.patch(
523 f"/api/repos/{repo.repo_id}/labels/{label.id}",
524 json={"name": "renamed"},
525 headers=auth_headers,
526 )
527
528 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
529 names = [i["name"] for i in resp.json()["items"]]
530 assert "renamed" in names
531 assert "original" not in names
532
533 @pytest.mark.anyio
534 async def test_delete_then_list_removes_label(
535 self,
536 client: AsyncClient,
537 db_session: AsyncSession,
538 auth_headers: StrDict,
539 ) -> None:
540 repo = await _db_repo(db_session)
541 label = await _db_label(db_session, repo.repo_id, name="gone")
542 await db_session.commit()
543
544 await client.delete(
545 f"/api/repos/{repo.repo_id}/labels/{label.id}",
546 headers=auth_headers,
547 )
548
549 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
550 names = [i["name"] for i in resp.json()["items"]]
551 assert "gone" not in names
552
553 @pytest.mark.anyio
554 async def test_assign_issue_labels_via_issues_route(
555 self,
556 client: AsyncClient,
557 db_session: AsyncSession,
558 auth_headers: StrDict,
559 ) -> None:
560 # Note: POST /repos/{id}/issues/{n}/labels is handled by issues.py (registered
561 # alphabetically before labels.py). That route takes {"labels": [name, ...]} — free-form
562 # string labels stored on the issue's JSON field.
563 repo = await _db_repo(db_session)
564 issue = await _db_issue(db_session, repo.repo_id, number=1)
565 await db_session.commit()
566
567 resp = await client.post(
568 f"/api/repos/{repo.repo_id}/issues/{issue.number}/labels",
569 json={"labels": ["bug", "needs-review"]},
570 headers=auth_headers,
571 )
572 assert resp.status_code == 200
573 data = resp.json()
574 assert "bug" in data.get("labels", [])
575
576 @pytest.mark.anyio
577 async def test_assign_labels_to_proposal(
578 self,
579 client: AsyncClient,
580 db_session: AsyncSession,
581 auth_headers: StrDict,
582 ) -> None:
583 repo = await _db_repo(db_session)
584 label = await _db_label(db_session, repo.repo_id, name="proposal-label")
585 proposal = await _db_proposal(db_session, repo.repo_id, proposal_number=1)
586 await db_session.commit()
587
588 resp = await client.post(
589 f"/api/repos/{repo.repo_id}/proposals/{proposal.proposal_id}/labels",
590 json={"label_ids": [label.id]},
591 headers=auth_headers,
592 )
593 assert resp.status_code == 200
594 assigned = resp.json()
595 assert len(assigned) == 1
596
597 @pytest.mark.anyio
598 async def test_assign_labels_to_proposal_not_found(
599 self,
600 client: AsyncClient,
601 db_session: AsyncSession,
602 auth_headers: StrDict,
603 ) -> None:
604 repo = await _db_repo(db_session)
605 label = await _db_label(db_session, repo.repo_id, name="proposal-label")
606 await db_session.commit()
607
608 resp = await client.post(
609 f"/api/repos/{repo.repo_id}/proposals/nonexistent-proposal/labels",
610 json={"label_ids": [label.id]},
611 headers=auth_headers,
612 )
613 assert resp.status_code == 404
614
615 @pytest.mark.anyio
616 async def test_remove_label_from_proposal(
617 self,
618 client: AsyncClient,
619 db_session: AsyncSession,
620 auth_headers: StrDict,
621 ) -> None:
622 repo = await _db_repo(db_session)
623 label = await _db_label(db_session, repo.repo_id, name="proposal-removable")
624 proposal = await _db_proposal(db_session, repo.repo_id, proposal_number=1)
625 prl = MusehubProposalLabel(proposal_id=proposal.proposal_id, label_id=label.id)
626 db_session.add(prl)
627 await db_session.commit()
628
629 resp = await client.delete(
630 f"/api/repos/{repo.repo_id}/proposals/{proposal.proposal_id}/labels/{label.id}",
631 headers=auth_headers,
632 )
633 assert resp.status_code == 204
634
635 @pytest.mark.anyio
636 async def test_create_label_returns_label_fields(
637 self,
638 client: AsyncClient,
639 db_session: AsyncSession,
640 auth_headers: StrDict,
641 ) -> None:
642 repo = await _db_repo(db_session)
643 await db_session.commit()
644
645 resp = await client.post(
646 f"/api/repos/{repo.repo_id}/labels",
647 json={"name": "full-check", "color": "#123456", "description": "desc"},
648 headers=auth_headers,
649 )
650 assert resp.status_code == 201
651 data = resp.json()
652 assert data["name"] == "full-check"
653 assert data["color"] == "#123456"
654 assert data["description"] == "desc"
655 assert data["repo_id"] == repo.repo_id
656
657
658 # ═══════════════════════════════════════════════════════════════════════════════
659 # Layer 4 — Stress
660 # ═══════════════════════════════════════════════════════════════════════════════
661
662
663 class TestStressLabels:
664 @pytest.mark.anyio
665 async def test_bulk_create_labels(
666 self,
667 client: AsyncClient,
668 db_session: AsyncSession,
669 auth_headers: StrDict,
670 ) -> None:
671 repo = await _db_repo(db_session)
672 await db_session.commit()
673
674 n = 20
675 for i in range(n):
676 resp = await client.post(
677 f"/api/repos/{repo.repo_id}/labels",
678 json={"name": f"label-{i}", "color": "#aabbcc"},
679 headers=auth_headers,
680 )
681 assert resp.status_code == 201
682
683 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
684 assert resp.json()["total"] == n
685
686 @pytest.mark.anyio
687 async def test_sequential_label_creates(
688 self,
689 client: AsyncClient,
690 db_session: AsyncSession,
691 auth_headers: StrDict,
692 ) -> None:
693 # Sequential creates validate that the endpoint handles repeated calls correctly.
694 repo = await _db_repo(db_session)
695 await db_session.commit()
696
697 for i in range(10):
698 resp = await client.post(
699 f"/api/repos/{repo.repo_id}/labels",
700 json={"name": f"sequential-{i}", "color": "#aabbcc"},
701 headers=auth_headers,
702 )
703 assert resp.status_code == 201
704
705 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
706 assert resp.json()["total"] == 10
707
708 @pytest.mark.anyio
709 async def test_assign_many_labels_to_proposal(
710 self,
711 client: AsyncClient,
712 db_session: AsyncSession,
713 auth_headers: StrDict,
714 ) -> None:
715 repo = await _db_repo(db_session)
716 labels = [await _db_label(db_session, repo.repo_id, name=f"lbl-{i}") for i in range(5)]
717 proposal = await _db_proposal(db_session, repo.repo_id, proposal_number=1)
718 await db_session.commit()
719
720 resp = await client.post(
721 f"/api/repos/{repo.repo_id}/proposals/{proposal.proposal_id}/labels",
722 json={"label_ids": [l.id for l in labels]},
723 headers=auth_headers,
724 )
725 assert resp.status_code == 200
726 assert len(resp.json()) == 5
727
728
729 # ═══════════════════════════════════════════════════════════════════════════════
730 # Layer 5 — Data Integrity
731 # ═══════════════════════════════════════════════════════════════════════════════
732
733
734 class TestDataIntegrityLabels:
735 @pytest.mark.anyio
736 async def test_label_unique_constraint_db(self, db_session: AsyncSession) -> None:
737 from sqlalchemy.exc import IntegrityError
738
739 repo = await _db_repo(db_session)
740 await _db_label(db_session, repo.repo_id, name="unique-check")
741 await db_session.flush()
742
743 dupe = MusehubLabel(
744 id=_uid(),
745 repo_id=repo.repo_id,
746 name="unique-check",
747 color="#ffffff",
748 )
749 db_session.add(dupe)
750 with pytest.raises(IntegrityError):
751 await db_session.flush()
752
753 @pytest.mark.anyio
754 async def test_issue_label_composite_pk(self, db_session: AsyncSession) -> None:
755 from sqlalchemy import insert
756 from sqlalchemy.exc import IntegrityError
757
758 repo = await _db_repo(db_session)
759 label = await _db_label(db_session, repo.repo_id)
760 issue = await _db_issue(db_session, repo.repo_id)
761 await db_session.commit()
762
763 il1 = MusehubIssueLabel(issue_id=issue.issue_id, label_id=label.id)
764 db_session.add(il1)
765 await db_session.flush()
766
767 # Use a raw INSERT to bypass SQLAlchemy's identity map (which already
768 # holds il1 under this PK) and hit the DB-level PK constraint directly.
769 with pytest.raises(IntegrityError):
770 await db_session.execute(
771 insert(MusehubIssueLabel).values(
772 issue_id=issue.issue_id, label_id=label.id
773 )
774 )
775 await db_session.flush()
776
777 @pytest.mark.anyio
778 async def test_delete_label_removes_issue_associations(
779 self,
780 client: AsyncClient,
781 db_session: AsyncSession,
782 auth_headers: StrDict,
783 ) -> None:
784 repo = await _db_repo(db_session)
785 label = await _db_label(db_session, repo.repo_id, name="to-remove")
786 issue = await _db_issue(db_session, repo.repo_id, number=1)
787
788 # assign via DB (labels.py issue-label route is shadowed by issues.py)
789 il = MusehubIssueLabel(issue_id=issue.issue_id, label_id=label.id)
790 db_session.add(il)
791 await db_session.commit()
792
793 # delete label
794 resp = await client.delete(
795 f"/api/repos/{repo.repo_id}/labels/{label.id}",
796 headers=auth_headers,
797 )
798 assert resp.status_code == 204
799
800 # verify issue_label row gone
801
802 result = await db_session.execute(
803 text("SELECT COUNT(*) FROM musehub_issue_labels WHERE label_id = :lid"),
804 {"lid": label.id},
805 )
806 assert result.scalar_one() == 0
807
808 @pytest.mark.anyio
809 async def test_remove_label_from_issue_idempotent(
810 self,
811 client: AsyncClient,
812 db_session: AsyncSession,
813 auth_headers: StrDict,
814 ) -> None:
815 repo = await _db_repo(db_session)
816 label = await _db_label(db_session, repo.repo_id, name="idem-remove")
817 proposal = await _db_proposal(db_session, repo.repo_id, proposal_number=1)
818 await db_session.commit()
819
820 # First delete (not assigned) should still return 204 (idempotent)
821 resp = await client.delete(
822 f"/api/repos/{repo.repo_id}/proposals/{proposal.proposal_id}/labels/{label.id}",
823 headers=auth_headers,
824 )
825 assert resp.status_code == 204
826
827 @pytest.mark.anyio
828 async def test_label_color_stored_correctly(self, db_session: AsyncSession) -> None:
829 repo = await _db_repo(db_session)
830 label = await _db_label(db_session, repo.repo_id, name="colorcheck", color="#112233")
831 await db_session.commit()
832
833
834 result = await db_session.execute(
835 text("SELECT color FROM musehub_labels WHERE id = :lid"), {"lid": label.id}
836 )
837 assert result.scalar_one() == "#112233"
838
839
840 # ═══════════════════════════════════════════════════════════════════════════════
841 # Layer 6 — Security
842 # ═══════════════════════════════════════════════════════════════════════════════
843
844
845 class TestSecurityLabels:
846 @pytest.mark.anyio
847 async def test_create_label_requires_auth(
848 self, client: AsyncClient, db_session: AsyncSession
849 ) -> None:
850 repo = await _db_repo(db_session)
851 await db_session.commit()
852
853 resp = await client.post(
854 f"/api/repos/{repo.repo_id}/labels",
855 json={"name": "unauth", "color": "#aaaaaa"},
856 )
857 assert resp.status_code == 401
858
859 @pytest.mark.anyio
860 async def test_update_label_requires_auth(
861 self, client: AsyncClient, db_session: AsyncSession
862 ) -> None:
863 repo = await _db_repo(db_session)
864 label = await _db_label(db_session, repo.repo_id, name="secure")
865 await db_session.commit()
866
867 resp = await client.patch(
868 f"/api/repos/{repo.repo_id}/labels/{label.id}",
869 json={"name": "hacked"},
870 )
871 assert resp.status_code == 401
872
873 @pytest.mark.anyio
874 async def test_delete_label_requires_auth(
875 self, client: AsyncClient, db_session: AsyncSession
876 ) -> None:
877 repo = await _db_repo(db_session)
878 label = await _db_label(db_session, repo.repo_id, name="secure")
879 await db_session.commit()
880
881 resp = await client.delete(
882 f"/api/repos/{repo.repo_id}/labels/{label.id}"
883 )
884 assert resp.status_code == 401
885
886 @pytest.mark.anyio
887 async def test_assign_labels_to_proposal_requires_auth(
888 self, client: AsyncClient, db_session: AsyncSession
889 ) -> None:
890 repo = await _db_repo(db_session)
891 label = await _db_label(db_session, repo.repo_id, name="bug")
892 proposal = await _db_proposal(db_session, repo.repo_id, proposal_number=1)
893 await db_session.commit()
894
895 resp = await client.post(
896 f"/api/repos/{repo.repo_id}/proposals/{proposal.proposal_id}/labels",
897 json={"label_ids": [label.id]},
898 )
899 assert resp.status_code == 401
900
901 @pytest.mark.anyio
902 async def test_list_labels_public_no_auth(
903 self, client: AsyncClient, db_session: AsyncSession
904 ) -> None:
905 repo = await _db_repo(db_session, visibility="public")
906 await _db_label(db_session, repo.repo_id, name="open")
907 await db_session.commit()
908
909
910 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
911 assert resp.status_code == 200
912
913 @pytest.mark.anyio
914 async def test_create_label_wrong_repo_404(
915 self,
916 client: AsyncClient,
917 db_session: AsyncSession,
918 auth_headers: StrDict,
919 ) -> None:
920 resp = await client.post(
921 "/api/repos/nonexistent-repo/labels",
922 json={"name": "bug", "color": "#ff0000"},
923 headers=auth_headers,
924 )
925 assert resp.status_code == 404
926
927 @pytest.mark.anyio
928 async def test_assign_label_from_different_repo_to_proposal_404(
929 self,
930 client: AsyncClient,
931 db_session: AsyncSession,
932 auth_headers: StrDict,
933 ) -> None:
934 """Label belonging to repo2 cannot be assigned on repo1's proposal."""
935 repo1 = await _db_repo(db_session)
936 repo2 = await _db_repo(db_session)
937 label_in_repo2 = await _db_label(db_session, repo2.repo_id, name="foreign-label")
938 proposal = await _db_proposal(db_session, repo1.repo_id, proposal_number=1)
939 await db_session.commit()
940
941 resp = await client.post(
942 f"/api/repos/{repo1.repo_id}/proposals/{proposal.proposal_id}/labels",
943 json={"label_ids": [label_in_repo2.id]},
944 headers=auth_headers,
945 )
946 # _get_label_or_404 checks (repo_id, label_id) pair — should 404
947 assert resp.status_code == 404
948
949 @pytest.mark.anyio
950 async def test_remove_label_from_proposal_requires_auth(
951 self, client: AsyncClient, db_session: AsyncSession
952 ) -> None:
953 repo = await _db_repo(db_session)
954 label = await _db_label(db_session, repo.repo_id, name="secure-proposal")
955 proposal = await _db_proposal(db_session, repo.repo_id, proposal_number=1)
956 await db_session.commit()
957
958 resp = await client.delete(
959 f"/api/repos/{repo.repo_id}/proposals/{proposal.proposal_id}/labels/{label.id}"
960 )
961 assert resp.status_code == 401
962
963
964 # ═══════════════════════════════════════════════════════════════════════════════
965 # Layer 7 — Performance
966 # ═══════════════════════════════════════════════════════════════════════════════
967
968
969 class TestPerformanceLabels:
970 @pytest.mark.anyio
971 async def test_list_labels_latency(
972 self, client: AsyncClient, db_session: AsyncSession
973 ) -> None:
974 repo = await _db_repo(db_session)
975 for i in range(15):
976 await _db_label(db_session, repo.repo_id, name=f"perf-{i}")
977 await db_session.commit()
978
979
980 start = time.perf_counter()
981 resp = await client.get(f"/api/repos/{repo.repo_id}/labels")
982 elapsed = time.perf_counter() - start
983
984 assert resp.status_code == 200
985 assert elapsed < 0.5
986
987 @pytest.mark.anyio
988 async def test_create_label_latency(
989 self,
990 client: AsyncClient,
991 db_session: AsyncSession,
992 auth_headers: StrDict,
993 ) -> None:
994 repo = await _db_repo(db_session)
995 await db_session.commit()
996
997 start = time.perf_counter()
998 resp = await client.post(
999 f"/api/repos/{repo.repo_id}/labels",
1000 json={"name": "perf-label", "color": "#112233"},
1001 headers=auth_headers,
1002 )
1003 elapsed = time.perf_counter() - start
1004
1005 assert resp.status_code == 201
1006 assert elapsed < 0.5
1007
1008 @pytest.mark.anyio
1009 async def test_delete_label_latency(
1010 self,
1011 client: AsyncClient,
1012 db_session: AsyncSession,
1013 auth_headers: StrDict,
1014 ) -> None:
1015 repo = await _db_repo(db_session)
1016 label = await _db_label(db_session, repo.repo_id, name="perf-delete")
1017 await db_session.commit()
1018
1019 start = time.perf_counter()
1020 resp = await client.delete(
1021 f"/api/repos/{repo.repo_id}/labels/{label.id}",
1022 headers=auth_headers,
1023 )
1024 elapsed = time.perf_counter() - start
1025
1026 assert resp.status_code == 204
1027 assert elapsed < 0.5
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago