gabriel / musehub public
test_mist_models_service.py python
1,248 lines 43.0 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Section 15 — Mists: 8-layer test suite.
2
3 Tests span the ORM model (MusehubMist), the seven Pydantic wire models in
4 musehub.models.mists, and the nine service functions in
5 musehub.services.musehub_mists.
6
7 Layer 1 Unit
8 - TestUnitPydanticModels: MistResponse, MistListEntry, MistListResponse,
9 MistCreateRequest, MistUpdateRequest, MistForkResponse, MistEmbedResponse
10 construction and field defaults.
11 - TestUnitValidators: MistCreateRequest.validate_visibility,
12 validate_tags (count, length, null-byte, HTML-special),
13 validate_filename (delegates to _validate_mist_filename).
14
15 Layer 2 Integration
16 - TestIntegrationCreate: create_mist persists all fields, returns MistResponse.
17 - TestIntegrationGet: get_mist returns populated response or None.
18 - TestIntegrationList: list_mists — owner filter, secret exclusion, artifact_type
19 filter, total counter, next_cursor pagination.
20 - TestIntegrationFork: fork_mist copies content, sets fork_parent_id,
21 increments source fork_count atomically.
22 - TestIntegrationUpdate: update_mist patches fields, owner guard.
23 - TestIntegrationDelete: delete_mist hard-delete, owner guard.
24 - TestIntegrationCounters: increment_mist_view / increment_mist_embed atomic.
25 - TestIntegrationForkList: get_mist_forks returns direct children.
26
27 Layer 3 Edge Cases
28 - TestEdgeCases: create duplicate mist_id → IntegrityError; fork with None
29 parent → None; fork depth limit; update content increments version;
30 list with bad cursor string is ignored.
31
32 Layer 4 Stress
33 - TestStress: 50 mists created, list returns expected totals; 5-level fork chain.
34
35 Layer 5 Data Integrity
36 - TestDataIntegrity: list total stays consistent after delete; counters are
37 independent per mist; fork inherits tags and symbol_anchors; fork keeps
38 parent visibility.
39
40 Layer 6 Performance
41 - TestPerformance: list 50 mists <500ms; count 100 forks <500ms.
42
43 Layer 7 Security
44 - TestSecurity: update/delete rejected for non-owner; secret mists hidden from
45 public list; fork depth limit blocks chain attacks.
46
47 Layer 8 Docstrings / API
48 - TestDocstrings: every public service function has a docstring; every Pydantic
49 model has a class docstring.
50 """
51
52 from __future__ import annotations
53
54 import time
55 import uuid
56
57 import pytest
58 from sqlalchemy.exc import IntegrityError
59 from sqlalchemy.ext.asyncio import AsyncSession
60
61 from musehub.db.musehub_models import MusehubMist, MusehubRepo
62 from musehub.models.mists import (
63 MistCreateRequest,
64 MistEmbedResponse,
65 MistForkResponse,
66 MistListEntry,
67 MistListResponse,
68 MistResponse,
69 MistUpdateRequest,
70 )
71 from musehub.services.musehub_mists import (
72 create_mist,
73 delete_mist,
74 fork_mist,
75 get_mist,
76 get_mist_forks,
77 increment_mist_embed,
78 increment_mist_view,
79 list_mists,
80 update_mist,
81 )
82
83 # ===========================================================================
84 # Helpers
85 # ===========================================================================
86
87 _OWNER = "gabriel"
88 _OTHER = "alice"
89
90
91 def _uid() -> str:
92 return str(uuid.uuid4())
93
94
95 def _mist_id() -> str:
96 """Return a unique 12-character fake mist_id."""
97 return uuid.uuid4().hex[:12]
98
99
100 async def _repo(
101 session: AsyncSession,
102 slug: str | None = None,
103 owner: str = _OWNER,
104 visibility: str = "public",
105 ) -> MusehubRepo:
106 slug = slug or _uid()
107 repo = MusehubRepo(
108 name=slug,
109 owner=owner,
110 slug=slug,
111 visibility=visibility,
112 owner_user_id=f"uid-{owner}",
113 )
114 session.add(repo)
115 await session.flush()
116 await session.refresh(repo)
117 return repo
118
119
120 async def _mist(
121 session: AsyncSession,
122 repo_id: str,
123 *,
124 mist_id: str | None = None,
125 owner: str = _OWNER,
126 filename: str = "hello.py",
127 content: str = "print('hello')",
128 artifact_type: str = "code",
129 language: str = "python",
130 visibility: str = "public",
131 tags: list[str] | None = None,
132 symbol_anchors: list[str] | None = None,
133 title: str = "",
134 description: str = "",
135 ) -> MistResponse:
136 return await create_mist(
137 session,
138 mist_id=mist_id or _mist_id(),
139 filename=filename,
140 content=content,
141 owner=owner,
142 repo_id=repo_id,
143 artifact_type=artifact_type,
144 language=language,
145 size_bytes=len(content.encode()),
146 visibility=visibility,
147 tags=tags or [],
148 symbol_anchors=symbol_anchors or [],
149 title=title,
150 description=description,
151 )
152
153
154 # ===========================================================================
155 # Layer 1 — Unit: Pydantic models
156 # ===========================================================================
157
158
159 class TestUnitPydanticModels:
160 """MistResponse and friends construct with expected defaults."""
161
162 def test_mist_response_required_fields(self) -> None:
163 from datetime import datetime, timezone
164
165 now = datetime.now(tz=timezone.utc)
166 resp = MistResponse(
167 mist_id="abc123def456",
168 owner="gabriel",
169 artifact_type="code",
170 filename="script.py",
171 created_at=now,
172 updated_at=now,
173 )
174 assert resp.mist_id == "abc123def456"
175 assert resp.owner == "gabriel"
176 assert resp.content == ""
177 assert resp.signed is False
178 assert resp.fork_parent_id is None
179 assert resp.fork_depth == 0
180 assert resp.visibility == "public"
181 assert resp.tags == []
182 assert resp.symbol_anchors == []
183
184 def test_mist_list_entry_primary_symbol_none(self) -> None:
185 from datetime import datetime, timezone
186
187 now = datetime.now(tz=timezone.utc)
188 entry = MistListEntry(
189 mist_id="abc123def456",
190 owner="gabriel",
191 artifact_type="prose",
192 filename="essay.md",
193 created_at=now,
194 updated_at=now,
195 )
196 assert entry.primary_symbol is None
197 assert entry.language == ""
198 assert entry.title == ""
199
200 def test_mist_list_response_defaults(self) -> None:
201 resp = MistListResponse()
202 assert resp.total == 0
203 assert resp.next_cursor is None
204 assert resp.mists == []
205
206 def test_mist_create_request_minimal(self) -> None:
207 req = MistCreateRequest(filename="score.mid", content="binary")
208 assert req.visibility == "public"
209 assert req.tags == []
210 assert req.agent_id == ""
211 assert req.gpg_signature is None
212
213 def test_mist_update_request_all_none(self) -> None:
214 req = MistUpdateRequest()
215 assert req.title is None
216 assert req.content is None
217 assert req.visibility is None
218
219 def test_mist_fork_response_fields(self) -> None:
220 from datetime import datetime, timezone
221
222 now = datetime.now(tz=timezone.utc)
223 resp = MistForkResponse(
224 mist_id="fork11111111",
225 owner="alice",
226 fork_parent_id="orig11111111",
227 artifact_type="code",
228 filename="main.py",
229 created_at=now,
230 )
231 assert resp.url == ""
232 assert resp.language == ""
233
234 def test_mist_embed_response_fields(self) -> None:
235 resp = MistEmbedResponse(
236 mist_id="abc123def456",
237 owner="gabriel",
238 iframe="<iframe/>",
239 js="<script/>",
240 badge="[![Mist](url)](link)",
241 )
242 assert resp.mist_id == "abc123def456"
243
244 def test_mist_response_camel_alias(self) -> None:
245 from datetime import datetime, timezone
246
247 now = datetime.now(tz=timezone.utc)
248 resp = MistResponse(
249 mist_id="abc123def456",
250 owner="gabriel",
251 artifact_type="code",
252 filename="main.py",
253 created_at=now,
254 updated_at=now,
255 fork_parent_id="parent11111",
256 symbol_anchors=["main.py::foo"],
257 )
258 d = resp.model_dump(by_alias=True)
259 assert "mistId" in d
260 assert "forkParentId" in d
261 assert "symbolAnchors" in d
262
263
264 class TestUnitValidators:
265 """MistCreateRequest validators reject bad input."""
266
267 def test_visibility_invalid(self) -> None:
268 with pytest.raises(Exception):
269 MistCreateRequest(filename="f.py", content="x", visibility="private")
270
271 def test_visibility_valid_values(self) -> None:
272 MistCreateRequest(filename="f.py", content="x", visibility="public")
273 MistCreateRequest(filename="f.py", content="x", visibility="secret")
274
275 def test_too_many_tags(self) -> None:
276 with pytest.raises(Exception):
277 MistCreateRequest(
278 filename="f.py",
279 content="x",
280 tags=["t"] * 11,
281 )
282
283 def test_tag_too_long(self) -> None:
284 with pytest.raises(Exception):
285 MistCreateRequest(
286 filename="f.py",
287 content="x",
288 tags=["a" * 65],
289 )
290
291 def test_tag_null_byte(self) -> None:
292 with pytest.raises(Exception):
293 MistCreateRequest(
294 filename="f.py",
295 content="x",
296 tags=["bad\x00tag"],
297 )
298
299 def test_tag_html_special(self) -> None:
300 for ch in ("<", ">", '"', "'", "&"):
301 with pytest.raises(Exception):
302 MistCreateRequest(
303 filename="f.py",
304 content="x",
305 tags=[f"tag{ch}val"],
306 )
307
308 def test_tag_valid(self) -> None:
309 req = MistCreateRequest(
310 filename="f.py",
311 content="x",
312 tags=["python", "audio", "ai"],
313 )
314 assert len(req.tags) == 3
315
316 def test_filename_traversal_rejected(self) -> None:
317 with pytest.raises(Exception):
318 MistCreateRequest(filename="../etc/passwd", content="x")
319
320 def test_filename_path_separator_rejected(self) -> None:
321 with pytest.raises(Exception):
322 MistCreateRequest(filename="a/b.py", content="x")
323
324 def test_filename_null_byte_rejected(self) -> None:
325 with pytest.raises(Exception):
326 MistCreateRequest(filename="f\x00ile.py", content="x")
327
328 def test_update_visibility_invalid(self) -> None:
329 with pytest.raises(Exception):
330 MistUpdateRequest(visibility="admin")
331
332 def test_update_visibility_none_allowed(self) -> None:
333 req = MistUpdateRequest(visibility=None)
334 assert req.visibility is None
335
336
337 # ===========================================================================
338 # Layer 2 — Integration
339 # ===========================================================================
340
341
342 class TestIntegrationCreate:
343 """create_mist persists a row and returns MistResponse."""
344
345 @pytest.mark.asyncio
346 async def test_create_stores_all_fields(self, db_session: AsyncSession) -> None:
347 repo = await _repo(db_session)
348 mid = _mist_id()
349 resp = await create_mist(
350 db_session,
351 mist_id=mid,
352 filename="score.mid",
353 content="MThd...",
354 owner=_OWNER,
355 repo_id=str(repo.repo_id),
356 artifact_type="midi",
357 language="",
358 size_bytes=7,
359 title="My Score",
360 description="A midi score",
361 visibility="public",
362 tags=["music", "midi"],
363 symbol_anchors=[],
364 agent_id="agent-1",
365 model_id="claude-sonnet-4-6",
366 gpg_signature="SIG",
367 )
368 assert resp.mist_id == mid
369 assert resp.filename == "score.mid"
370 assert resp.artifact_type == "midi"
371 assert resp.title == "My Score"
372 assert resp.description == "A midi score"
373 assert resp.tags == ["music", "midi"]
374 assert resp.signed is True
375 assert resp.agent_id == "agent-1"
376 assert resp.model_id == "claude-sonnet-4-6"
377 assert resp.fork_depth == 0
378 assert resp.fork_count == 0
379 assert resp.view_count == 0
380 assert resp.version == 1
381
382 @pytest.mark.asyncio
383 async def test_create_defaults(self, db_session: AsyncSession) -> None:
384 repo = await _repo(db_session)
385 resp = await create_mist(
386 db_session,
387 mist_id=_mist_id(),
388 filename="minimal.txt",
389 content="hello",
390 owner=_OWNER,
391 repo_id=str(repo.repo_id),
392 )
393 assert resp.artifact_type == "unknown"
394 assert resp.language == ""
395 assert resp.title == ""
396 assert resp.description == ""
397 assert resp.visibility == "public"
398 assert resp.tags == []
399 assert resp.signed is False
400
401 @pytest.mark.asyncio
402 async def test_create_returns_mist_response(self, db_session: AsyncSession) -> None:
403 repo = await _repo(db_session)
404 resp = await _mist(db_session, str(repo.repo_id))
405 assert isinstance(resp, MistResponse)
406 assert resp.created_at is not None
407 assert resp.updated_at is not None
408
409 @pytest.mark.asyncio
410 async def test_create_with_base_url(self, db_session: AsyncSession) -> None:
411 repo = await _repo(db_session)
412 mid = _mist_id()
413 resp = await create_mist(
414 db_session,
415 mist_id=mid,
416 filename="f.py",
417 content="x",
418 owner=_OWNER,
419 repo_id=str(repo.repo_id),
420 base_url="https://musehub.ai",
421 )
422 assert resp.url == f"https://musehub.ai/{_OWNER}/mists/{mid}"
423
424 @pytest.mark.asyncio
425 async def test_create_symbol_anchors_stored(self, db_session: AsyncSession) -> None:
426 repo = await _repo(db_session)
427 anchors = ["main.py::foo", "main.py::bar"]
428 resp = await _mist(
429 db_session,
430 str(repo.repo_id),
431 symbol_anchors=anchors,
432 )
433 assert resp.symbol_anchors == anchors
434
435
436 class TestIntegrationGet:
437 """get_mist returns MistResponse or None."""
438
439 @pytest.mark.asyncio
440 async def test_get_existing(self, db_session: AsyncSession) -> None:
441 repo = await _repo(db_session)
442 created = await _mist(db_session, str(repo.repo_id))
443 fetched = await get_mist(db_session, created.mist_id)
444 assert fetched is not None
445 assert fetched.mist_id == created.mist_id
446 assert fetched.content == created.content
447
448 @pytest.mark.asyncio
449 async def test_get_not_found(self, db_session: AsyncSession) -> None:
450 result = await get_mist(db_session, "notexist000")
451 assert result is None
452
453 @pytest.mark.asyncio
454 async def test_get_with_base_url(self, db_session: AsyncSession) -> None:
455 repo = await _repo(db_session)
456 created = await _mist(db_session, str(repo.repo_id))
457 fetched = await get_mist(db_session, created.mist_id, base_url="https://musehub.ai")
458 assert fetched is not None
459 assert fetched.url.startswith("https://musehub.ai/")
460
461
462 class TestIntegrationList:
463 """list_mists — filters, totals, pagination."""
464
465 @pytest.mark.asyncio
466 async def test_list_by_owner(self, db_session: AsyncSession) -> None:
467 r1 = await _repo(db_session, owner=_OWNER)
468 r2 = await _repo(db_session, owner=_OTHER)
469 for _ in range(3):
470 await _mist(db_session, str(r1.repo_id), owner=_OWNER)
471 for _ in range(2):
472 await _mist(db_session, str(r2.repo_id), owner=_OTHER)
473
474 result = await list_mists(db_session, _OWNER)
475 assert result.total == 3
476 assert all(e.owner == _OWNER for e in result.mists)
477
478 @pytest.mark.asyncio
479 async def test_list_secret_hidden_by_default(self, db_session: AsyncSession) -> None:
480 repo = await _repo(db_session)
481 await _mist(db_session, str(repo.repo_id), visibility="public")
482 await _mist(db_session, str(repo.repo_id), visibility="secret")
483
484 result = await list_mists(db_session, _OWNER, include_secret=False)
485 assert result.total == 1
486 assert all(e.visibility == "public" for e in result.mists)
487
488 @pytest.mark.asyncio
489 async def test_list_secret_visible_when_owner(self, db_session: AsyncSession) -> None:
490 repo = await _repo(db_session)
491 await _mist(db_session, str(repo.repo_id), visibility="public")
492 await _mist(db_session, str(repo.repo_id), visibility="secret")
493
494 result = await list_mists(db_session, _OWNER, include_secret=True)
495 assert result.total == 2
496
497 @pytest.mark.asyncio
498 async def test_list_artifact_type_filter(self, db_session: AsyncSession) -> None:
499 repo = await _repo(db_session)
500 await _mist(db_session, str(repo.repo_id), artifact_type="code")
501 await _mist(db_session, str(repo.repo_id), artifact_type="midi")
502 await _mist(db_session, str(repo.repo_id), artifact_type="midi")
503
504 result = await list_mists(db_session, _OWNER, artifact_type="midi")
505 assert result.total == 2
506 assert all(e.artifact_type == "midi" for e in result.mists)
507
508 @pytest.mark.asyncio
509 async def test_list_pagination_next_cursor(self, db_session: AsyncSession) -> None:
510 repo = await _repo(db_session)
511 for _ in range(5):
512 await _mist(db_session, str(repo.repo_id))
513
514 page1 = await list_mists(db_session, _OWNER, limit=3)
515 assert len(page1.mists) == 3
516 assert page1.next_cursor is not None
517
518 page2 = await list_mists(db_session, _OWNER, limit=3, cursor=page1.next_cursor)
519 assert len(page2.mists) == 2
520 assert page2.next_cursor is None
521
522 @pytest.mark.asyncio
523 async def test_list_global_explore(self, db_session: AsyncSession) -> None:
524 r1 = await _repo(db_session, owner=_OWNER)
525 r2 = await _repo(db_session, owner=_OTHER)
526 await _mist(db_session, str(r1.repo_id), owner=_OWNER)
527 await _mist(db_session, str(r2.repo_id), owner=_OTHER, visibility="secret")
528
529 # explore — owner=None, include_secret=False → only 1 public
530 result = await list_mists(db_session, owner=None, include_secret=False)
531 assert result.total == 1
532
533 @pytest.mark.asyncio
534 async def test_list_newest_first(self, db_session: AsyncSession) -> None:
535 import asyncio
536
537 repo = await _repo(db_session)
538 for i in range(3):
539 await _mist(db_session, str(repo.repo_id), content=f"v{i}")
540 await asyncio.sleep(0.01)
541
542 result = await list_mists(db_session, _OWNER, limit=10)
543 dates = [e.created_at for e in result.mists]
544 assert dates == sorted(dates, reverse=True)
545
546
547 class TestIntegrationFork:
548 """fork_mist creates a copy with correct linkage."""
549
550 @pytest.mark.asyncio
551 async def test_fork_creates_linked_copy(self, db_session: AsyncSession) -> None:
552 repo = await _repo(db_session)
553 original = await _mist(db_session, str(repo.repo_id), tags=["a", "b"])
554 fork_repo = await _repo(db_session, owner=_OTHER)
555 fork_id = _mist_id()
556
557 resp = await fork_mist(
558 db_session,
559 original.mist_id,
560 new_mist_id=fork_id,
561 new_owner=_OTHER,
562 new_repo_id=str(fork_repo.repo_id),
563 )
564 assert resp is not None
565 assert resp.mist_id == fork_id
566 assert resp.fork_parent_id == original.mist_id
567 assert resp.owner == _OTHER
568
569 @pytest.mark.asyncio
570 async def test_fork_increments_source_fork_count(self, db_session: AsyncSession) -> None:
571 repo = await _repo(db_session)
572 original = await _mist(db_session, str(repo.repo_id))
573 fork_repo = await _repo(db_session, owner=_OTHER)
574
575 await fork_mist(
576 db_session,
577 original.mist_id,
578 new_mist_id=_mist_id(),
579 new_owner=_OTHER,
580 new_repo_id=str(fork_repo.repo_id),
581 )
582 await db_session.commit()
583
584 source = await get_mist(db_session, original.mist_id)
585 assert source is not None
586 assert source.fork_count == 1
587
588 @pytest.mark.asyncio
589 async def test_fork_inherits_content(self, db_session: AsyncSession) -> None:
590 repo = await _repo(db_session)
591 original = await _mist(
592 db_session,
593 str(repo.repo_id),
594 content="def foo(): pass",
595 symbol_anchors=["main.py::foo"],
596 tags=["ai"],
597 )
598 fork_repo = await _repo(db_session, owner=_OTHER)
599 fork_id = _mist_id()
600 await fork_mist(
601 db_session,
602 original.mist_id,
603 new_mist_id=fork_id,
604 new_owner=_OTHER,
605 new_repo_id=str(fork_repo.repo_id),
606 )
607 fork = await get_mist(db_session, fork_id)
608 assert fork is not None
609 assert fork.content == "def foo(): pass"
610 assert fork.symbol_anchors == ["main.py::foo"]
611 assert fork.tags == ["ai"]
612
613 @pytest.mark.asyncio
614 async def test_fork_increments_depth(self, db_session: AsyncSession) -> None:
615 repo = await _repo(db_session)
616 original = await _mist(db_session, str(repo.repo_id))
617 fork_repo = await _repo(db_session, owner=_OTHER)
618 fork_id = _mist_id()
619
620 resp = await fork_mist(
621 db_session,
622 original.mist_id,
623 new_mist_id=fork_id,
624 new_owner=_OTHER,
625 new_repo_id=str(fork_repo.repo_id),
626 )
627 assert resp is not None
628 fork = await get_mist(db_session, fork_id)
629 assert fork is not None
630 assert fork.fork_depth == 1
631
632 @pytest.mark.asyncio
633 async def test_fork_nonexistent_returns_none(self, db_session: AsyncSession) -> None:
634 repo = await _repo(db_session)
635 result = await fork_mist(
636 db_session,
637 "doesnotexist",
638 new_mist_id=_mist_id(),
639 new_owner=_OTHER,
640 new_repo_id=str(repo.repo_id),
641 )
642 assert result is None
643
644
645 class TestIntegrationUpdate:
646 """update_mist patches fields; owner guard blocks others."""
647
648 @pytest.mark.asyncio
649 async def test_update_title(self, db_session: AsyncSession) -> None:
650 repo = await _repo(db_session)
651 original = await _mist(db_session, str(repo.repo_id), title="old")
652
653 updated = await update_mist(
654 db_session, original.mist_id, _OWNER, title="new title"
655 )
656 assert updated is not None
657 assert updated.title == "new title"
658
659 @pytest.mark.asyncio
660 async def test_update_visibility(self, db_session: AsyncSession) -> None:
661 repo = await _repo(db_session)
662 original = await _mist(db_session, str(repo.repo_id), visibility="public")
663
664 updated = await update_mist(
665 db_session, original.mist_id, _OWNER, visibility="secret"
666 )
667 assert updated is not None
668 assert updated.visibility == "secret"
669
670 @pytest.mark.asyncio
671 async def test_update_content_increments_version(self, db_session: AsyncSession) -> None:
672 repo = await _repo(db_session)
673 original = await _mist(db_session, str(repo.repo_id), content="v1")
674
675 updated = await update_mist(
676 db_session, original.mist_id, _OWNER, content="v2"
677 )
678 assert updated is not None
679 assert updated.content == "v2"
680 assert updated.version == 2
681
682 @pytest.mark.asyncio
683 async def test_update_non_owner_returns_none(self, db_session: AsyncSession) -> None:
684 repo = await _repo(db_session)
685 original = await _mist(db_session, str(repo.repo_id))
686
687 result = await update_mist(
688 db_session, original.mist_id, _OTHER, title="hacked"
689 )
690 assert result is None
691
692 @pytest.mark.asyncio
693 async def test_update_not_found_returns_none(self, db_session: AsyncSession) -> None:
694 result = await update_mist(
695 db_session, "notexist000", _OWNER, title="x"
696 )
697 assert result is None
698
699 @pytest.mark.asyncio
700 async def test_update_none_fields_unchanged(self, db_session: AsyncSession) -> None:
701 repo = await _repo(db_session)
702 original = await _mist(
703 db_session, str(repo.repo_id), title="keep", tags=["x"]
704 )
705
706 updated = await update_mist(
707 db_session, original.mist_id, _OWNER, description="new desc"
708 )
709 assert updated is not None
710 assert updated.title == "keep"
711 assert updated.tags == ["x"]
712 assert updated.description == "new desc"
713
714
715 class TestIntegrationDelete:
716 """delete_mist hard-deletes; owner guard blocks others."""
717
718 @pytest.mark.asyncio
719 async def test_delete_own_mist(self, db_session: AsyncSession) -> None:
720 repo = await _repo(db_session)
721 m = await _mist(db_session, str(repo.repo_id))
722
723 ok = await delete_mist(db_session, m.mist_id, _OWNER)
724 assert ok is True
725
726 gone = await get_mist(db_session, m.mist_id)
727 assert gone is None
728
729 @pytest.mark.asyncio
730 async def test_delete_non_owner_fails(self, db_session: AsyncSession) -> None:
731 repo = await _repo(db_session)
732 m = await _mist(db_session, str(repo.repo_id))
733
734 ok = await delete_mist(db_session, m.mist_id, _OTHER)
735 assert ok is False
736 still_there = await get_mist(db_session, m.mist_id)
737 assert still_there is not None
738
739 @pytest.mark.asyncio
740 async def test_delete_not_found_returns_false(self, db_session: AsyncSession) -> None:
741 ok = await delete_mist(db_session, "notexist000", _OWNER)
742 assert ok is False
743
744
745 class TestIntegrationCounters:
746 """Atomic view/embed counter increments."""
747
748 @pytest.mark.asyncio
749 async def test_increment_view_count(self, db_session: AsyncSession) -> None:
750 repo = await _repo(db_session)
751 m = await _mist(db_session, str(repo.repo_id))
752
753 await increment_mist_view(db_session, m.mist_id)
754 await increment_mist_view(db_session, m.mist_id)
755 await db_session.commit()
756
757 fetched = await get_mist(db_session, m.mist_id)
758 assert fetched is not None
759 assert fetched.view_count == 2
760
761 @pytest.mark.asyncio
762 async def test_increment_embed_count(self, db_session: AsyncSession) -> None:
763 repo = await _repo(db_session)
764 m = await _mist(db_session, str(repo.repo_id))
765
766 await increment_mist_embed(db_session, m.mist_id)
767 await db_session.commit()
768
769 fetched = await get_mist(db_session, m.mist_id)
770 assert fetched is not None
771 assert fetched.embed_count == 1
772
773 @pytest.mark.asyncio
774 async def test_increment_nonexistent_noop(self, db_session: AsyncSession) -> None:
775 """Incrementing a missing mist_id is a silent no-op."""
776 await increment_mist_view(db_session, "notexist000")
777 await increment_mist_embed(db_session, "notexist000")
778
779
780 class TestIntegrationForkList:
781 """get_mist_forks returns direct children only."""
782
783 @pytest.mark.asyncio
784 async def test_forks_of_original(self, db_session: AsyncSession) -> None:
785 repo = await _repo(db_session)
786 original = await _mist(db_session, str(repo.repo_id))
787
788 for _ in range(3):
789 fork_repo = await _repo(db_session, owner=_OTHER)
790 await fork_mist(
791 db_session,
792 original.mist_id,
793 new_mist_id=_mist_id(),
794 new_owner=_OTHER,
795 new_repo_id=str(fork_repo.repo_id),
796 )
797
798 forks = await get_mist_forks(db_session, original.mist_id)
799 assert len(forks) == 3
800 assert all(isinstance(f, MistListEntry) for f in forks)
801 assert all(f.fork_parent_id == original.mist_id for f in forks)
802
803 @pytest.mark.asyncio
804 async def test_no_forks_returns_empty(self, db_session: AsyncSession) -> None:
805 repo = await _repo(db_session)
806 m = await _mist(db_session, str(repo.repo_id))
807 forks = await get_mist_forks(db_session, m.mist_id)
808 assert forks == []
809
810
811 # ===========================================================================
812 # Layer 3 — Edge Cases
813 # ===========================================================================
814
815
816 class TestEdgeCases:
817 """Boundary and error conditions."""
818
819 @pytest.mark.asyncio
820 async def test_duplicate_mist_id_raises_integrity_error(
821 self, db_session: AsyncSession
822 ) -> None:
823 repo = await _repo(db_session)
824 mid = _mist_id()
825 await create_mist(
826 db_session,
827 mist_id=mid,
828 filename="a.py",
829 content="x",
830 owner=_OWNER,
831 repo_id=str(repo.repo_id),
832 )
833 with pytest.raises(IntegrityError):
834 await create_mist(
835 db_session,
836 mist_id=mid,
837 filename="b.py",
838 content="y",
839 owner=_OWNER,
840 repo_id=str(repo.repo_id),
841 )
842
843 @pytest.mark.asyncio
844 async def test_fork_depth_limit_enforced(self, db_session: AsyncSession) -> None:
845 """Fork chain at depth 5 cannot be forked further."""
846 from musehub.services.musehub_mists import _FORK_DEPTH_LIMIT
847
848 repo = await _repo(db_session)
849 current_id = (await _mist(db_session, str(repo.repo_id))).mist_id
850
851 for depth in range(_FORK_DEPTH_LIMIT):
852 fork_repo = await _repo(db_session, owner=_OTHER)
853 fork_id = _mist_id()
854 resp = await fork_mist(
855 db_session,
856 current_id,
857 new_mist_id=fork_id,
858 new_owner=_OTHER,
859 new_repo_id=str(fork_repo.repo_id),
860 )
861 assert resp is not None, f"Expected fork at depth {depth + 1} to succeed"
862 current_id = fork_id
863
864 # Now at depth 5 — next fork must be rejected
865 final_repo = await _repo(db_session, owner=_OTHER)
866 result = await fork_mist(
867 db_session,
868 current_id,
869 new_mist_id=_mist_id(),
870 new_owner=_OTHER,
871 new_repo_id=str(final_repo.repo_id),
872 )
873 assert result is None
874
875 @pytest.mark.asyncio
876 async def test_update_content_twice_increments_version_twice(
877 self, db_session: AsyncSession
878 ) -> None:
879 repo = await _repo(db_session)
880 m = await _mist(db_session, str(repo.repo_id))
881
882 await update_mist(db_session, m.mist_id, _OWNER, content="v2")
883 await update_mist(db_session, m.mist_id, _OWNER, content="v3")
884
885 final = await get_mist(db_session, m.mist_id)
886 assert final is not None
887 assert final.version == 3
888 assert final.content == "v3"
889
890 @pytest.mark.asyncio
891 async def test_list_bad_cursor_ignored(self, db_session: AsyncSession) -> None:
892 """A non-ISO-8601 cursor string is silently ignored (returns full list)."""
893 repo = await _repo(db_session)
894 await _mist(db_session, str(repo.repo_id))
895 await _mist(db_session, str(repo.repo_id))
896
897 result = await list_mists(db_session, _OWNER, cursor="not-a-date")
898 assert result.total == 2
899
900 @pytest.mark.asyncio
901 async def test_create_empty_tags_stored_as_list(self, db_session: AsyncSession) -> None:
902 repo = await _repo(db_session)
903 m = await _mist(db_session, str(repo.repo_id))
904 assert m.tags == []
905
906 @pytest.mark.asyncio
907 async def test_create_with_secret_visibility(self, db_session: AsyncSession) -> None:
908 repo = await _repo(db_session)
909 m = await _mist(db_session, str(repo.repo_id), visibility="secret")
910 assert m.visibility == "secret"
911 fetched = await get_mist(db_session, m.mist_id)
912 assert fetched is not None
913 assert fetched.visibility == "secret"
914
915 @pytest.mark.asyncio
916 async def test_mist_list_entry_primary_symbol_from_anchors(
917 self, db_session: AsyncSession
918 ) -> None:
919 repo = await _repo(db_session)
920 anchors = ["utils.py::compute", "utils.py::clean"]
921 await _mist(db_session, str(repo.repo_id), symbol_anchors=anchors)
922
923 result = await list_mists(db_session, _OWNER)
924 assert result.total == 1
925 assert result.mists[0].primary_symbol == "utils.py::compute"
926
927
928 # ===========================================================================
929 # Layer 4 — Stress
930 # ===========================================================================
931
932
933 class TestStress:
934 """Bulk create and list under load."""
935
936 @pytest.mark.asyncio
937 async def test_create_50_mists(self, db_session: AsyncSession) -> None:
938 repo = await _repo(db_session)
939 for i in range(50):
940 await _mist(db_session, str(repo.repo_id), content=f"print({i})")
941
942 result = await list_mists(db_session, _OWNER, limit=50)
943 assert result.total == 50
944 assert len(result.mists) == 50
945
946 @pytest.mark.asyncio
947 async def test_five_level_fork_chain(self, db_session: AsyncSession) -> None:
948 """Create a 5-level deep fork chain and verify depths."""
949 repo = await _repo(db_session)
950 root_id = (await _mist(db_session, str(repo.repo_id))).mist_id
951 current_id = root_id
952
953 for expected_depth in range(1, 6):
954 fork_repo = await _repo(db_session, owner=_OTHER)
955 fork_id = _mist_id()
956 resp = await fork_mist(
957 db_session,
958 current_id,
959 new_mist_id=fork_id,
960 new_owner=_OTHER,
961 new_repo_id=str(fork_repo.repo_id),
962 )
963 assert resp is not None
964 fork_row = await get_mist(db_session, fork_id)
965 assert fork_row is not None
966 assert fork_row.fork_depth == expected_depth
967 current_id = fork_id
968
969 @pytest.mark.asyncio
970 async def test_list_pagination_covers_all_pages(self, db_session: AsyncSession) -> None:
971 """Paginate through 25 mists with page size 10."""
972 repo = await _repo(db_session)
973 for i in range(25):
974 await _mist(db_session, str(repo.repo_id), content=f"item {i}")
975
976 collected: list = []
977 cursor = None
978 while True:
979 page = await list_mists(db_session, _OWNER, limit=10, cursor=cursor)
980 collected.extend(page.mists)
981 if page.next_cursor is None:
982 break
983 cursor = page.next_cursor
984
985 assert len(collected) == 25
986
987
988 # ===========================================================================
989 # Layer 5 — Data Integrity
990 # ===========================================================================
991
992
993 class TestDataIntegrity:
994 """Counters are independent; list total is consistent after delete."""
995
996 @pytest.mark.asyncio
997 async def test_total_decrements_after_delete(self, db_session: AsyncSession) -> None:
998 repo = await _repo(db_session)
999 m1 = await _mist(db_session, str(repo.repo_id))
1000 await _mist(db_session, str(repo.repo_id))
1001
1002 await delete_mist(db_session, m1.mist_id, _OWNER)
1003
1004 result = await list_mists(db_session, _OWNER)
1005 assert result.total == 1
1006
1007 @pytest.mark.asyncio
1008 async def test_counters_independent_per_mist(self, db_session: AsyncSession) -> None:
1009 repo = await _repo(db_session)
1010 m1 = await _mist(db_session, str(repo.repo_id))
1011 m2 = await _mist(db_session, str(repo.repo_id))
1012
1013 await increment_mist_view(db_session, m1.mist_id)
1014 await increment_mist_embed(db_session, m2.mist_id)
1015 await db_session.commit()
1016
1017 r1 = await get_mist(db_session, m1.mist_id)
1018 r2 = await get_mist(db_session, m2.mist_id)
1019 assert r1 is not None and r2 is not None
1020 assert r1.view_count == 1 and r1.embed_count == 0
1021 assert r2.view_count == 0 and r2.embed_count == 1
1022
1023 @pytest.mark.asyncio
1024 async def test_fork_inherits_parent_visibility(self, db_session: AsyncSession) -> None:
1025 repo = await _repo(db_session)
1026 original = await _mist(db_session, str(repo.repo_id), visibility="secret")
1027 fork_repo = await _repo(db_session, owner=_OTHER)
1028 fork_id = _mist_id()
1029
1030 await fork_mist(
1031 db_session,
1032 original.mist_id,
1033 new_mist_id=fork_id,
1034 new_owner=_OTHER,
1035 new_repo_id=str(fork_repo.repo_id),
1036 )
1037
1038 fork = await get_mist(db_session, fork_id)
1039 assert fork is not None
1040 assert fork.visibility == "secret"
1041
1042 @pytest.mark.asyncio
1043 async def test_fork_count_multi(self, db_session: AsyncSession) -> None:
1044 """fork_count on original equals number of forks created."""
1045 repo = await _repo(db_session)
1046 original = await _mist(db_session, str(repo.repo_id))
1047
1048 for _ in range(4):
1049 fr = await _repo(db_session, owner=_OTHER)
1050 await fork_mist(
1051 db_session,
1052 original.mist_id,
1053 new_mist_id=_mist_id(),
1054 new_owner=_OTHER,
1055 new_repo_id=str(fr.repo_id),
1056 )
1057 await db_session.commit()
1058
1059 source = await get_mist(db_session, original.mist_id)
1060 assert source is not None
1061 assert source.fork_count == 4
1062
1063 @pytest.mark.asyncio
1064 async def test_update_tags_replaces_list(self, db_session: AsyncSession) -> None:
1065 repo = await _repo(db_session)
1066 m = await _mist(db_session, str(repo.repo_id), tags=["a", "b"])
1067
1068 updated = await update_mist(db_session, m.mist_id, _OWNER, tags=["c"])
1069 assert updated is not None
1070 assert updated.tags == ["c"]
1071
1072
1073 # ===========================================================================
1074 # Layer 6 — Performance
1075 # ===========================================================================
1076
1077
1078 class TestPerformance:
1079 """Bulk operations must complete within generous thresholds."""
1080
1081 @pytest.mark.asyncio
1082 async def test_list_50_mists_under_500ms(self, db_session: AsyncSession) -> None:
1083 repo = await _repo(db_session)
1084 for i in range(50):
1085 await _mist(db_session, str(repo.repo_id), content=f"c{i}")
1086
1087 t0 = time.perf_counter()
1088 result = await list_mists(db_session, _OWNER, limit=50)
1089 elapsed = time.perf_counter() - t0
1090
1091 assert result.total == 50
1092 assert elapsed < 0.5, f"list_mists took {elapsed:.3f}s — too slow"
1093
1094 @pytest.mark.asyncio
1095 async def test_get_100_times_under_500ms(self, db_session: AsyncSession) -> None:
1096 repo = await _repo(db_session)
1097 m = await _mist(db_session, str(repo.repo_id))
1098
1099 t0 = time.perf_counter()
1100 for _ in range(100):
1101 await get_mist(db_session, m.mist_id)
1102 elapsed = time.perf_counter() - t0
1103
1104 assert elapsed < 0.5, f"100x get_mist took {elapsed:.3f}s"
1105
1106
1107 # ===========================================================================
1108 # Layer 7 — Security
1109 # ===========================================================================
1110
1111
1112 class TestSecurity:
1113 """Owner guard and visibility rules are enforced."""
1114
1115 @pytest.mark.asyncio
1116 async def test_update_by_non_owner_rejected(self, db_session: AsyncSession) -> None:
1117 repo = await _repo(db_session)
1118 m = await _mist(db_session, str(repo.repo_id))
1119
1120 result = await update_mist(db_session, m.mist_id, "attacker", content="evil")
1121 assert result is None
1122
1123 # Content must be unchanged
1124 original = await get_mist(db_session, m.mist_id)
1125 assert original is not None
1126 assert original.content == m.content
1127
1128 @pytest.mark.asyncio
1129 async def test_delete_by_non_owner_rejected(self, db_session: AsyncSession) -> None:
1130 repo = await _repo(db_session)
1131 m = await _mist(db_session, str(repo.repo_id))
1132
1133 ok = await delete_mist(db_session, m.mist_id, "attacker")
1134 assert ok is False
1135 assert await get_mist(db_session, m.mist_id) is not None
1136
1137 @pytest.mark.asyncio
1138 async def test_secret_mist_hidden_from_explore(self, db_session: AsyncSession) -> None:
1139 repo = await _repo(db_session)
1140 await _mist(db_session, str(repo.repo_id), visibility="secret")
1141
1142 result = await list_mists(db_session, owner=None, include_secret=False)
1143 assert result.total == 0
1144
1145 @pytest.mark.asyncio
1146 async def test_fork_depth_limit_prevents_deep_chain(
1147 self, db_session: AsyncSession
1148 ) -> None:
1149 from musehub.services.musehub_mists import _FORK_DEPTH_LIMIT
1150
1151 repo = await _repo(db_session)
1152 current_id = (await _mist(db_session, str(repo.repo_id))).mist_id
1153
1154 for _ in range(_FORK_DEPTH_LIMIT):
1155 fr = await _repo(db_session, owner=_OTHER)
1156 resp = await fork_mist(
1157 db_session,
1158 current_id,
1159 new_mist_id=_mist_id(),
1160 new_owner=_OTHER,
1161 new_repo_id=str(fr.repo_id),
1162 )
1163 assert resp is not None
1164 current_id = resp.mist_id
1165
1166 # Attempt to exceed the limit
1167 final_repo = await _repo(db_session, owner=_OTHER)
1168 over_limit = await fork_mist(
1169 db_session,
1170 current_id,
1171 new_mist_id=_mist_id(),
1172 new_owner=_OTHER,
1173 new_repo_id=str(final_repo.repo_id),
1174 )
1175 assert over_limit is None
1176
1177 @pytest.mark.asyncio
1178 async def test_secret_visible_only_with_include_secret(
1179 self, db_session: AsyncSession
1180 ) -> None:
1181 repo = await _repo(db_session)
1182 await _mist(db_session, str(repo.repo_id), visibility="secret")
1183
1184 hidden = await list_mists(db_session, _OWNER, include_secret=False)
1185 visible = await list_mists(db_session, _OWNER, include_secret=True)
1186
1187 assert hidden.total == 0
1188 assert visible.total == 1
1189
1190
1191 # ===========================================================================
1192 # Layer 8 — Docstrings / API surface
1193 # ===========================================================================
1194
1195
1196 class TestDocstrings:
1197 """Public API surface has docstrings."""
1198
1199 def test_service_functions_have_docstrings(self) -> None:
1200 import musehub.services.musehub_mists as svc
1201
1202 fns = [
1203 svc.create_mist,
1204 svc.get_mist,
1205 svc.list_mists,
1206 svc.fork_mist,
1207 svc.update_mist,
1208 svc.delete_mist,
1209 svc.increment_mist_view,
1210 svc.increment_mist_embed,
1211 svc.get_mist_forks,
1212 ]
1213 missing = [f.__name__ for f in fns if not (f.__doc__ or "").strip()]
1214 assert missing == [], f"Service functions missing docstrings: {missing}"
1215
1216 def test_pydantic_models_have_docstrings(self) -> None:
1217 import musehub.models.mists as m
1218
1219 models = [
1220 m.MistResponse,
1221 m.MistListEntry,
1222 m.MistListResponse,
1223 m.MistCreateRequest,
1224 m.MistUpdateRequest,
1225 m.MistForkResponse,
1226 m.MistEmbedResponse,
1227 ]
1228 missing = [cls.__name__ for cls in models if not (cls.__doc__ or "").strip()]
1229 assert missing == [], f"Pydantic models missing docstrings: {missing}"
1230
1231 def test_orm_model_has_tablename(self) -> None:
1232 assert MusehubMist.__tablename__ == "musehub_mists"
1233
1234 def test_orm_model_has_expected_columns(self) -> None:
1235 expected = {
1236 "mist_id", "repo_id", "owner", "artifact_type", "language",
1237 "filename", "title", "description", "content", "size_bytes",
1238 "commit_id", "snapshot_id", "version", "agent_id", "model_id",
1239 "gpg_signature", "fork_parent_id", "fork_depth", "fork_count",
1240 "view_count", "embed_count", "visibility", "tags",
1241 "symbol_anchors", "created_at", "updated_at",
1242 }
1243 actual = {c.key for c in MusehubMist.__table__.columns}
1244 assert expected == actual
1245
1246 def test_fork_depth_limit_constant(self) -> None:
1247 from musehub.services.musehub_mists import _FORK_DEPTH_LIMIT
1248 assert _FORK_DEPTH_LIMIT == 5
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago