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