gabriel / musehub public
test_mcp_mist_tools.py python
1,096 lines 42.7 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """Section 16 — MCP Mist Tools: 7-layer test suite.
2
3 Covers the mist MCP executors:
4 Write tools (write_tools/mists.py):
5 execute_create_mist, execute_update_mist, execute_fork_mist, execute_delete_mist
6 Read tools (services/musehub_mcp_executor.py):
7 execute_read_mist, execute_list_mists, execute_read_mist_embed
8 Resource handlers (mcp/resources.py):
9 _read_mist, _read_owner_mists (via musehub://mists/... URIs)
10
11 Seven layers:
12
13 Layer 1 Unit:
14 - muse_mist_* tool names appear in correct catalogue sets
15 - _mist_data serialises MistResponse to correct dict keys
16 - execute_create_mist: empty actor → forbidden
17 - execute_create_mist: empty filename → missing_args
18 - execute_create_mist: empty content → missing_args
19 - execute_update_mist: empty actor → forbidden
20 - execute_fork_mist: empty actor → forbidden
21 - execute_delete_mist: empty actor → forbidden
22
23 Layer 2 Integration:
24 - execute_create_mist: happy path returns mist_id, artifact_type, content
25 - execute_create_mist: duplicate content → already_exists
26 - execute_update_mist: title change persisted
27 - execute_update_mist: non-owner → not_found
28 - execute_update_mist: visibility change to secret
29 - execute_delete_mist: happy path returns deleted=True
30 - execute_delete_mist: non-owner → not_found
31 - execute_delete_mist: unknown mist_id → not_found
32 - execute_fork_mist: happy path returns new mist_id and fork_parent_id
33 - execute_fork_mist: unknown source → not_found
34 - execute_read_mist: public mist readable by anon
35 - execute_read_mist: secret mist readable by owner
36 - execute_read_mist: secret mist blocked for non-owner
37 - execute_read_mist: unknown → not_found
38 - execute_list_mists: explore mode returns public mists
39 - execute_list_mists: owner mode returns owner's mists
40 - execute_list_mists: secret excluded for anon, included for owner
41 - execute_read_mist_embed: returns iframe, javascript, badge strings
42 - execute_read_mist_embed: secret mist → forbidden
43 - execute_read_mist_embed: unknown mist → not_found
44
45 Layer 3 E2E (HTTP tools/call):
46 - Anonymous muse_mist_create → 401
47 - Anonymous muse_mist_update → 401
48 - Anonymous muse_mist_fork → 401
49 - Anonymous muse_mist_delete → 401
50 - Authenticated muse_mist_create → isError=False, mist_id present
51 - Authenticated muse_mist_list (read tool) → isError=False, mists list
52
53 Layer 4 Stress:
54 - 10 sequential creates under 1000 ms
55
56 Layer 5 Data Integrity:
57 - Created mist retrievable via execute_read_mist
58 - Created mist appears in execute_list_mists(owner=...)
59 - Updated title persisted after execute_update_mist
60 - Deleted mist not found via execute_read_mist
61 - Fork parent_id correct + source fork_count incremented
62
63 Layer 6 Security:
64 - muse_mist_create/update/fork/delete in MUSEHUB_WRITE_TOOL_NAMES
65 - muse_mist_read/list/embed in read set (not in MUSEHUB_WRITE_TOOL_NAMES)
66 - Secret mist inaccessible via read executor to non-owner
67 - Secret mist excluded from explore listing
68 - Content returned as-is (no XSS transformation)
69
70 Layer 7 Performance:
71 - 10 sequential creates under 1000 ms
72 """
73 from __future__ import annotations
74
75 import json
76 import secrets
77 import time
78 from datetime import datetime, timezone, timedelta
79
80 import pytest
81 import pytest_asyncio
82 from httpx import AsyncClient, ASGITransport
83 from sqlalchemy.ext.asyncio import AsyncSession
84
85 from musehub.core.genesis import compute_identity_id, compute_repo_id
86 from musehub.db import musehub_models as db
87 from musehub.main import app
88 from musehub.mcp.tools.musehub import MUSEHUB_WRITE_TOOL_NAMES, MUSEHUB_TOOL_NAMES
89 from musehub.types.json_types import JSONObject, StrDict
90 from musehub.mcp.write_tools.mists import (
91 _mist_data,
92 execute_create_mist,
93 execute_delete_mist,
94 execute_fork_mist,
95 execute_update_mist,
96 )
97 from musehub.services.musehub_mcp_executor import (
98 execute_list_mists,
99 execute_list_mist_forks,
100 execute_read_mist,
101 execute_read_mist_embed,
102 execute_read_mist_raw,
103 )
104 from musehub.mcp.resources import read_resource
105
106
107 # ── Fixtures ──────────────────────────────────────────────────────────────────
108
109
110 @pytest.fixture
111 def anyio_backend() -> str:
112 return "asyncio"
113
114
115 @pytest_asyncio.fixture
116 async def http_client(db_session: AsyncSession) -> AsyncClient:
117 async with AsyncClient(
118 transport=ASGITransport(app=app),
119 base_url="http://localhost",
120 ) as c:
121 yield c
122
123
124 # ── Helpers ───────────────────────────────────────────────────────────────────
125
126 _OWNER = "alice"
127 _PY_CONTENT = "def validate(x: str) -> bool:\n return bool(x)\n"
128 _PY_FILENAME = "validate.py"
129
130
131 def _uid() -> str:
132 return secrets.token_hex(16)
133
134
135 def _unique_content() -> str:
136 """Return content unique enough that its mist_id won't collide."""
137 return f"{_PY_CONTENT}# salt={secrets.token_hex(16)}"
138
139
140 def _tools_call(name: str, arguments: JSONObject) -> JSONObject:
141 return {
142 "jsonrpc": "2.0",
143 "id": 1,
144 "method": "tools/call",
145 "params": {"name": name, "arguments": arguments},
146 }
147
148
149 def _unwrap_tool_text(text: str) -> str:
150 text = text.strip()
151 if text.startswith("<musehub_tool_result>"):
152 text = text[len("<musehub_tool_result>"):].strip()
153 if text.endswith("</musehub_tool_result>"):
154 text = text[: -len("</musehub_tool_result>")].strip()
155 return text
156
157
158 async def _create(
159 content: str | None = None,
160 filename: str = _PY_FILENAME,
161 visibility: str = "public",
162 actor: str = _OWNER,
163 title: str = "",
164 ) -> "MusehubToolResult": # type: ignore[name-defined]
165 return await execute_create_mist(
166 filename=filename,
167 content=content or _unique_content(),
168 actor=actor,
169 title=title,
170 visibility=visibility,
171 )
172
173
174 # ── Layer 1 — Unit ────────────────────────────────────────────────────────────
175
176
177 class TestUnitToolCatalogue:
178 def test_mist_write_tools_in_write_set(self) -> None:
179 expected = {"muse_mist_create", "muse_mist_update", "muse_mist_fork", "muse_mist_delete"}
180 missing = expected - MUSEHUB_WRITE_TOOL_NAMES
181 assert not missing, f"Missing from write set: {missing}"
182
183 def test_mist_read_tools_NOT_in_write_set(self) -> None:
184 read_tools = {"muse_mist_read", "muse_mist_list", "muse_mist_embed"}
185 in_write = read_tools & MUSEHUB_WRITE_TOOL_NAMES
186 assert not in_write, f"Read tools incorrectly in write set: {in_write}"
187
188 def test_all_mist_tools_in_tool_names(self) -> None:
189 expected = {
190 "muse_mist_create", "muse_mist_update", "muse_mist_fork",
191 "muse_mist_delete", "muse_mist_read", "muse_mist_list", "muse_mist_embed",
192 }
193 missing = expected - MUSEHUB_TOOL_NAMES
194 assert not missing, f"Missing from MUSEHUB_TOOL_NAMES: {missing}"
195
196
197 class TestUnitMistDataHelper:
198 async def test_mist_data_keys(self, db_session: AsyncSession) -> None:
199 result = await _create()
200 assert result.ok is True
201 data = result.data
202 for key in ("mist_id", "owner", "artifact_type", "language", "filename",
203 "content", "size_bytes", "version", "visibility", "tags",
204 "symbol_anchors", "created_at", "updated_at"):
205 assert key in data, f"Missing key: {key}"
206
207
208 class TestUnitInputValidation:
209 async def test_create_empty_actor_returns_forbidden(self) -> None:
210 result = await execute_create_mist(filename="f.py", content="x", actor="")
211 assert result.ok is False
212 assert result.error_code == "forbidden"
213
214 async def test_create_empty_filename_returns_missing_args(self) -> None:
215 result = await execute_create_mist(filename="", content="x", actor=_OWNER)
216 assert result.ok is False
217 assert result.error_code == "missing_args"
218
219 async def test_create_empty_content_returns_missing_args(self) -> None:
220 result = await execute_create_mist(filename="f.py", content="", actor=_OWNER)
221 assert result.ok is False
222 assert result.error_code == "missing_args"
223
224 async def test_update_empty_actor_returns_forbidden(self) -> None:
225 result = await execute_update_mist(mist_id="aB3xKq9dPwNm", actor="")
226 assert result.ok is False
227 assert result.error_code == "forbidden"
228
229 async def test_fork_empty_actor_returns_forbidden(self) -> None:
230 result = await execute_fork_mist(mist_id="aB3xKq9dPwNm", actor="")
231 assert result.ok is False
232 assert result.error_code == "forbidden"
233
234 async def test_delete_empty_actor_returns_forbidden(self) -> None:
235 result = await execute_delete_mist(mist_id="aB3xKq9dPwNm", actor="")
236 assert result.ok is False
237 assert result.error_code == "forbidden"
238
239
240 # ── Layer 2 — Integration ─────────────────────────────────────────────────────
241
242
243 class TestIntegrationCreate:
244 async def test_create_happy_path(self, db_session: AsyncSession) -> None:
245 result = await _create()
246 assert result.ok is True
247 data = result.data
248 assert len(data["mist_id"]) == 12
249 assert data["artifact_type"] == "code"
250 assert data["language"] == "python"
251 assert data["owner"] == _OWNER
252 assert data["visibility"] == "public"
253 assert data["version"] == 1
254
255 async def test_create_duplicate_content_returns_already_exists(
256 self, db_session: AsyncSession
257 ) -> None:
258 content = _unique_content()
259 r1 = await execute_create_mist(filename=_PY_FILENAME, content=content, actor=_OWNER)
260 assert r1.ok is True
261 r2 = await execute_create_mist(filename=_PY_FILENAME, content=content, actor=_OWNER)
262 assert r2.ok is False
263 assert r2.error_code == "already_exists"
264
265 async def test_create_with_title_and_tags(self, db_session: AsyncSession) -> None:
266 result = await execute_create_mist(
267 filename=_PY_FILENAME,
268 content=_unique_content(),
269 actor=_OWNER,
270 title="My helper",
271 tags=["utils", "security"],
272 )
273 assert result.ok is True
274 assert result.data["title"] == "My helper"
275 assert result.data["tags"] == ["utils", "security"]
276
277 async def test_create_secret_mist(self, db_session: AsyncSession) -> None:
278 result = await execute_create_mist(
279 filename=_PY_FILENAME,
280 content=_unique_content(),
281 actor=_OWNER,
282 visibility="secret",
283 )
284 assert result.ok is True
285 assert result.data["visibility"] == "secret"
286
287
288 class TestIntegrationUpdate:
289 async def test_update_title(self, db_session: AsyncSession) -> None:
290 created = await _create()
291 mid = created.data["mist_id"]
292 result = await execute_update_mist(mist_id=mid, actor=_OWNER, title="New title")
293 assert result.ok is True
294 assert result.data["title"] == "New title"
295
296 async def test_update_visibility_to_secret(self, db_session: AsyncSession) -> None:
297 created = await _create()
298 mid = created.data["mist_id"]
299 result = await execute_update_mist(mist_id=mid, actor=_OWNER, visibility="secret")
300 assert result.ok is True
301 assert result.data["visibility"] == "secret"
302
303 async def test_update_content_increments_version(self, db_session: AsyncSession) -> None:
304 created = await _create()
305 mid = created.data["mist_id"]
306 result = await execute_update_mist(
307 mist_id=mid, actor=_OWNER, content="# new content\n"
308 )
309 assert result.ok is True
310 assert result.data["version"] == 2
311
312 async def test_update_non_owner_returns_not_found(self, db_session: AsyncSession) -> None:
313 created = await _create()
314 mid = created.data["mist_id"]
315 result = await execute_update_mist(mist_id=mid, actor="bob", title="Stolen")
316 assert result.ok is False
317 assert result.error_code == "not_found"
318
319 async def test_update_unknown_mist_returns_not_found(self, db_session: AsyncSession) -> None:
320 result = await execute_update_mist(mist_id="unknown12345", actor=_OWNER, title="X")
321 assert result.ok is False
322 assert result.error_code == "not_found"
323
324
325 class TestIntegrationDelete:
326 async def test_delete_happy_path(self, db_session: AsyncSession) -> None:
327 created = await _create()
328 mid = created.data["mist_id"]
329 result = await execute_delete_mist(mist_id=mid, actor=_OWNER)
330 assert result.ok is True
331 assert result.data["deleted"] is True
332 assert result.data["mist_id"] == mid
333
334 async def test_delete_non_owner_returns_not_found(self, db_session: AsyncSession) -> None:
335 created = await _create()
336 mid = created.data["mist_id"]
337 result = await execute_delete_mist(mist_id=mid, actor="bob")
338 assert result.ok is False
339 assert result.error_code == "not_found"
340
341 async def test_delete_unknown_returns_not_found(self, db_session: AsyncSession) -> None:
342 result = await execute_delete_mist(mist_id="unknown12345", actor=_OWNER)
343 assert result.ok is False
344 assert result.error_code == "not_found"
345
346
347 class TestIntegrationFork:
348 async def test_fork_happy_path(self, db_session: AsyncSession) -> None:
349 source = await _create(actor=_OWNER)
350 mid = source.data["mist_id"]
351 result = await execute_fork_mist(mist_id=mid, actor="bob")
352 assert result.ok is True
353 assert result.data["fork_parent_id"] == mid
354 assert result.data["owner"] == "bob"
355 assert result.data["mist_id"] != mid
356
357 async def test_fork_unknown_returns_not_found(self, db_session: AsyncSession) -> None:
358 result = await execute_fork_mist(mist_id="unknown12345", actor="bob")
359 assert result.ok is False
360 assert result.error_code == "not_found"
361
362
363 class TestIntegrationReadMist:
364 async def test_read_public_mist_anon(self, db_session: AsyncSession) -> None:
365 created = await _create()
366 mid = created.data["mist_id"]
367 result = await execute_read_mist(mid, actor="")
368 assert result.ok is True
369 assert result.data["mist_id"] == mid
370 assert "content" in result.data
371
372 async def test_read_secret_mist_as_owner(self, db_session: AsyncSession) -> None:
373 created = await _create(visibility="secret")
374 mid = created.data["mist_id"]
375 result = await execute_read_mist(mid, actor=_OWNER)
376 assert result.ok is True
377
378 async def test_read_secret_mist_as_non_owner_returns_forbidden(
379 self, db_session: AsyncSession
380 ) -> None:
381 created = await _create(visibility="secret")
382 mid = created.data["mist_id"]
383 result = await execute_read_mist(mid, actor="bob")
384 assert result.ok is False
385 assert result.error_code == "forbidden"
386
387 async def test_read_unknown_returns_not_found(self, db_session: AsyncSession) -> None:
388 result = await execute_read_mist("unknown12345")
389 assert result.ok is False
390 assert result.error_code == "not_found"
391
392
393 class TestIntegrationListMists:
394 async def test_explore_returns_public(self, db_session: AsyncSession) -> None:
395 created = await _create(actor=_OWNER)
396 mid = created.data["mist_id"]
397 result = await execute_list_mists(owner=None)
398 assert result.ok is True
399 ids = {m["mist_id"] for m in result.data["mists"]}
400 assert mid in ids
401
402 async def test_explore_excludes_secret(self, db_session: AsyncSession) -> None:
403 created = await _create(visibility="secret")
404 mid = created.data["mist_id"]
405 result = await execute_list_mists(owner=None)
406 assert result.ok is True
407 ids = {m["mist_id"] for m in result.data["mists"]}
408 assert mid not in ids
409
410 async def test_owner_mode_includes_public(self, db_session: AsyncSession) -> None:
411 created = await _create(actor=_OWNER)
412 mid = created.data["mist_id"]
413 result = await execute_list_mists(owner=_OWNER)
414 assert result.ok is True
415 ids = {m["mist_id"] for m in result.data["mists"]}
416 assert mid in ids
417
418 async def test_owner_mode_excludes_secret_for_anon(self, db_session: AsyncSession) -> None:
419 created = await _create(visibility="secret")
420 mid = created.data["mist_id"]
421 result = await execute_list_mists(owner=_OWNER, include_secret=True, actor="bob")
422 assert result.ok is True
423 ids = {m["mist_id"] for m in result.data["mists"]}
424 assert mid not in ids
425
426 async def test_owner_mode_includes_secret_for_owner(self, db_session: AsyncSession) -> None:
427 created = await _create(visibility="secret", actor=_OWNER)
428 mid = created.data["mist_id"]
429 result = await execute_list_mists(owner=_OWNER, include_secret=True, actor=_OWNER)
430 assert result.ok is True
431 ids = {m["mist_id"] for m in result.data["mists"]}
432 assert mid in ids
433
434
435 class TestIntegrationEmbed:
436 async def test_embed_public_mist(self, db_session: AsyncSession) -> None:
437 created = await _create()
438 mid = created.data["mist_id"]
439 result = await execute_read_mist_embed(mid, owner=_OWNER)
440 assert result.ok is True
441 data = result.data
442 assert "iframe" in data
443 assert "javascript" in data
444 assert "badge" in data
445 assert mid in data["iframe"]
446
447 async def test_embed_secret_mist_returns_forbidden(self, db_session: AsyncSession) -> None:
448 created = await _create(visibility="secret")
449 mid = created.data["mist_id"]
450 result = await execute_read_mist_embed(mid, owner=_OWNER)
451 assert result.ok is False
452 assert result.error_code == "forbidden"
453
454 async def test_embed_unknown_mist_returns_not_found(self, db_session: AsyncSession) -> None:
455 result = await execute_read_mist_embed("unknown12345", owner="nobody")
456 assert result.ok is False
457 assert result.error_code == "not_found"
458
459
460 class TestIntegrationResource:
461 async def test_read_resource_single_mist(self, db_session: AsyncSession) -> None:
462 created = await _create()
463 mid = created.data["mist_id"]
464 data = await read_resource(f"musehub://mists/{_OWNER}/{mid}")
465 assert "error" not in data
466 assert data["mist_id"] == mid
467 assert "content" in data
468
469 async def test_read_resource_owner_mists(self, db_session: AsyncSession) -> None:
470 created = await _create(actor=_OWNER)
471 mid = created.data["mist_id"]
472 data = await read_resource(f"musehub://mists/{_OWNER}")
473 assert "error" not in data
474 ids = {m["mist_id"] for m in data["mists"]}
475 assert mid in ids
476
477 async def test_read_resource_unknown_mist(self, db_session: AsyncSession) -> None:
478 data = await read_resource("musehub://mists/nobody/unknown12345")
479 assert "error" in data
480
481 async def test_read_resource_secret_mist_blocked_for_anon(
482 self, db_session: AsyncSession
483 ) -> None:
484 created = await _create(visibility="secret")
485 mid = created.data["mist_id"]
486 data = await read_resource(f"musehub://mists/{_OWNER}/{mid}", user_id=None)
487 assert "error" in data
488
489
490 # ── Layer 3 — End-to-End ──────────────────────────────────────────────────────
491
492
493 class TestE2EAuthGate:
494 """Write tool calls without auth must return 401."""
495
496 async def test_create_mist_no_auth(self, http_client: AsyncClient) -> None:
497 resp = await http_client.post(
498 "/mcp",
499 json=_tools_call("muse_mist_create", {"filename": "f.py", "content": "x"}),
500 headers={"Content-Type": "application/json"},
501 )
502 assert resp.status_code == 401
503
504 async def test_update_mist_no_auth(self, http_client: AsyncClient) -> None:
505 resp = await http_client.post(
506 "/mcp",
507 json=_tools_call("muse_mist_update", {"mist_id": "aB3xKq9dPwNm"}),
508 headers={"Content-Type": "application/json"},
509 )
510 assert resp.status_code == 401
511
512 async def test_fork_mist_no_auth(self, http_client: AsyncClient) -> None:
513 resp = await http_client.post(
514 "/mcp",
515 json=_tools_call("muse_mist_fork", {"mist_id": "aB3xKq9dPwNm"}),
516 headers={"Content-Type": "application/json"},
517 )
518 assert resp.status_code == 401
519
520 async def test_delete_mist_no_auth(self, http_client: AsyncClient) -> None:
521 resp = await http_client.post(
522 "/mcp",
523 json=_tools_call("muse_mist_delete", {"mist_id": "aB3xKq9dPwNm"}),
524 headers={"Content-Type": "application/json"},
525 )
526 assert resp.status_code == 401
527
528 async def test_create_mist_with_auth(
529 self, http_client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict
530 ) -> None:
531 content = _unique_content()
532 resp = await http_client.post(
533 "/mcp",
534 json=_tools_call("muse_mist_create", {
535 "filename": _PY_FILENAME,
536 "content": content,
537 "title": "E2E mist",
538 }),
539 headers=auth_headers,
540 )
541 assert resp.status_code == 200
542 result = resp.json()["result"]
543 assert result["isError"] is False
544 payload = json.loads(_unwrap_tool_text(result["content"][0]["text"]))
545 assert "mist_id" in payload
546 assert payload["title"] == "E2E mist"
547
548 async def test_list_mists_read_tool_no_auth(self, http_client: AsyncClient) -> None:
549 """muse_mist_list is a read tool — accessible without auth."""
550 resp = await http_client.post(
551 "/mcp",
552 json=_tools_call("muse_mist_list", {}),
553 headers={"Content-Type": "application/json"},
554 )
555 # Read tools don't require auth at the HTTP layer
556 assert resp.status_code in (200, 401)
557 if resp.status_code == 200:
558 result = resp.json()["result"]
559 assert result["isError"] is False
560
561
562 # ── Layer 4 — Stress ──────────────────────────────────────────────────────────
563
564
565 class TestStressMistTools:
566 async def test_10_sequential_creates(self, db_session: AsyncSession) -> None:
567 start = time.monotonic()
568 ids: list[str] = []
569 for _ in range(10):
570 result = await execute_create_mist(
571 filename=_PY_FILENAME,
572 content=_unique_content(),
573 actor=_OWNER,
574 )
575 assert result.ok is True
576 ids.append(result.data["mist_id"])
577 elapsed = time.monotonic() - start
578 assert elapsed < 1.0, f"10 creates took {elapsed:.2f}s (> 1s)"
579 assert len(set(ids)) == 10, "All mist IDs must be unique"
580
581
582 # ── Layer 5 — Data Integrity ──────────────────────────────────────────────────
583
584
585 class TestDataIntegrity:
586 async def test_created_mist_retrievable(self, db_session: AsyncSession) -> None:
587 created = await _create(title="Persistent")
588 mid = created.data["mist_id"]
589 read = await execute_read_mist(mid, actor=_OWNER)
590 assert read.ok is True
591 assert read.data["mist_id"] == mid
592 assert read.data["title"] == "Persistent"
593
594 async def test_created_mist_in_owner_list(self, db_session: AsyncSession) -> None:
595 created = await _create(actor=_OWNER)
596 mid = created.data["mist_id"]
597 result = await execute_list_mists(owner=_OWNER)
598 assert result.ok is True
599 ids = {m["mist_id"] for m in result.data["mists"]}
600 assert mid in ids
601
602 async def test_update_title_persisted(self, db_session: AsyncSession) -> None:
603 created = await _create()
604 mid = created.data["mist_id"]
605 await execute_update_mist(mist_id=mid, actor=_OWNER, title="Persisted title")
606 read = await execute_read_mist(mid)
607 assert read.ok is True
608 assert read.data["title"] == "Persisted title"
609
610 async def test_deleted_mist_not_found(self, db_session: AsyncSession) -> None:
611 created = await _create()
612 mid = created.data["mist_id"]
613 del_result = await execute_delete_mist(mist_id=mid, actor=_OWNER)
614 assert del_result.ok is True
615 read = await execute_read_mist(mid)
616 assert read.ok is False
617 assert read.error_code == "not_found"
618
619 async def test_fork_parent_id_and_source_fork_count(self, db_session: AsyncSession) -> None:
620 source = await _create(actor=_OWNER)
621 mid = source.data["mist_id"]
622 fork = await execute_fork_mist(mist_id=mid, actor="bob")
623 assert fork.ok is True
624 assert fork.data["fork_parent_id"] == mid
625 # Source fork_count incremented — verify via read
626 read = await execute_read_mist(mid, actor=_OWNER)
627 assert read.ok is True
628 assert read.data["fork_count"] >= 1
629
630 async def test_view_count_increments_on_read(self, db_session: AsyncSession) -> None:
631 created = await _create()
632 mid = created.data["mist_id"]
633 r1 = await execute_read_mist(mid)
634 r2 = await execute_read_mist(mid)
635 assert r2.data["view_count"] > r1.data["view_count"]
636
637 async def test_embed_count_increments_on_embed(self, db_session: AsyncSession) -> None:
638 created = await _create()
639 mid = created.data["mist_id"]
640 r1 = await execute_read_mist(mid)
641 await execute_read_mist_embed(mid, owner=_OWNER)
642 r2 = await execute_read_mist(mid)
643 assert r2.data["embed_count"] > r1.data["embed_count"]
644
645
646 # ── Layer 6 — Security ────────────────────────────────────────────────────────
647
648
649 class TestSecurity:
650 def test_write_tools_in_auth_gate_set(self) -> None:
651 write_tools = {"muse_mist_create", "muse_mist_update", "muse_mist_fork", "muse_mist_delete"}
652 missing = write_tools - MUSEHUB_WRITE_TOOL_NAMES
653 assert not missing, f"Write tools missing from auth gate: {missing}"
654
655 def test_read_tools_not_in_write_set(self) -> None:
656 read_tools = {"muse_mist_read", "muse_mist_list", "muse_mist_embed"}
657 in_write = read_tools & MUSEHUB_WRITE_TOOL_NAMES
658 assert not in_write, f"Read tools in write auth gate (shouldn't be): {in_write}"
659
660 async def test_secret_mist_not_in_explore(self, db_session: AsyncSession) -> None:
661 created = await _create(visibility="secret")
662 mid = created.data["mist_id"]
663 result = await execute_list_mists(owner=None, actor="")
664 ids = {m["mist_id"] for m in result.data["mists"]}
665 assert mid not in ids, "Secret mist must not appear in explore feed"
666
667 async def test_secret_mist_blocked_for_non_owner_read(self, db_session: AsyncSession) -> None:
668 created = await _create(visibility="secret")
669 mid = created.data["mist_id"]
670 result = await execute_read_mist(mid, actor="bob")
671 assert result.ok is False
672 assert result.error_code == "forbidden"
673
674 async def test_content_returned_verbatim_no_xss_transform(
675 self, db_session: AsyncSession
676 ) -> None:
677 """Content is returned verbatim — XSS prevention is a renderer concern."""
678 xss_payload = '<script>alert("xss")</script>'
679 created = await execute_create_mist(
680 filename="test.html",
681 content=xss_payload,
682 actor=_OWNER,
683 )
684 assert created.ok is True
685 mid = created.data["mist_id"]
686 read = await execute_read_mist(mid, actor=_OWNER)
687 assert read.ok is True
688 assert read.data["content"] == xss_payload
689
690 async def test_agent_id_stored_verbatim(self, db_session: AsyncSession) -> None:
691 """agent_id is stored as opaque string — no injection risk in storage."""
692 agent = "agentception-worker-42; DROP TABLE mists;--"
693 created = await execute_create_mist(
694 filename=_PY_FILENAME,
695 content=_unique_content(),
696 actor=_OWNER,
697 agent_id=agent,
698 )
699 assert created.ok is True
700 assert created.data["agent_id"] == agent
701
702
703 # ── Layer 7 — Performance ─────────────────────────────────────────────────────
704
705
706 class TestPerformance:
707 async def test_10_creates_under_500ms(self, db_session: AsyncSession) -> None:
708 start = time.monotonic()
709 for _ in range(10):
710 result = await execute_create_mist(
711 filename=_PY_FILENAME,
712 content=_unique_content(),
713 actor=_OWNER,
714 )
715 assert result.ok is True
716 elapsed = time.monotonic() - start
717 assert elapsed < 0.5, f"10 creates took {elapsed:.2f}s (> 500ms)"
718
719 async def test_list_100_mists_under_200ms(self, db_session: AsyncSession) -> None: # noqa: E501
720 from muse.plugins.mist.plugin import compute_mist_id
721
722 base_time = datetime.now(tz=timezone.utc)
723 unique_type = f"perf_{secrets.token_hex(4)}"
724 owner_id = compute_identity_id(_OWNER.encode())
725 for i in range(20):
726 content = f"perf_{i}_{secrets.token_hex(16)}"
727 mid = compute_mist_id(content.encode())
728 from musehub.db.musehub_models import MusehubMist
729 from musehub.db import musehub_models as dbm
730 slug = secrets.token_hex(6)
731 created_at = base_time
732 repo_id = compute_repo_id(owner_id, slug, "code", created_at.isoformat())
733 repo = dbm.MusehubRepo(
734 repo_id=repo_id,
735 name=slug,
736 owner=_OWNER,
737 slug=slug,
738 visibility="public",
739 owner_user_id=owner_id,
740 created_at=created_at,
741 updated_at=created_at,
742 )
743 db_session.add(repo)
744 await db_session.flush()
745 await db_session.refresh(repo)
746 row = MusehubMist(
747 mist_id=mid,
748 repo_id=str(repo.repo_id),
749 owner=_OWNER,
750 filename="p.py",
751 content=content,
752 artifact_type=unique_type,
753 language="python",
754 visibility="public",
755 tags=[],
756 symbol_anchors=[],
757 created_at=base_time + timedelta(seconds=i),
758 updated_at=base_time + timedelta(seconds=i),
759 )
760 db_session.add(row)
761 await db_session.commit()
762
763 start = time.monotonic()
764 result = await execute_list_mists(
765 artifact_type=unique_type,
766 limit=20,
767 )
768 elapsed = time.monotonic() - start
769 assert result.ok is True
770 assert elapsed < 0.2, f"list 20 mists took {elapsed:.2f}s (> 200ms)"
771
772
773 # ── execute_list_mist_forks tests ─────────────────────────────────────────────
774
775
776 @pytest.mark.anyio
777 class TestListMistForks:
778 """Tests for execute_list_mist_forks — all 8 tiers.
779
780 Covers: empty mist_id guard, not_found, forbidden (secret parent,
781 non-owner actor), happy path with zero forks, happy path with forks,
782 limit clamping, and performance (<200ms for 10 forks).
783 """
784
785 async def test_empty_mist_id_returns_missing_args(
786 self, db_session: AsyncSession
787 ) -> None:
788 """Empty mist_id returns missing_args immediately without a DB hit."""
789 result = await execute_list_mist_forks("")
790 assert result.ok is False
791 assert result.error_code == "missing_args"
792
793 async def test_unknown_mist_returns_not_found(
794 self, db_session: AsyncSession
795 ) -> None:
796 """Non-existent parent mist returns not_found."""
797 result = await execute_list_mist_forks("NoSuchMistXX")
798 assert result.ok is False
799 assert result.error_code == "not_found"
800
801 async def test_secret_parent_anon_returns_forbidden(
802 self, db_session: AsyncSession
803 ) -> None:
804 """Secret parent mist with anonymous actor returns forbidden."""
805 created = await execute_create_mist(
806 filename=_PY_FILENAME,
807 content=_unique_content(),
808 actor=_OWNER,
809 visibility="secret",
810 )
811 assert created.ok is True
812 mist_id = created.data["mist_id"]
813
814 result = await execute_list_mist_forks(mist_id, actor="")
815 assert result.ok is False
816 assert result.error_code == "forbidden"
817
818 async def test_public_parent_no_forks_returns_empty_list(
819 self, db_session: AsyncSession
820 ) -> None:
821 """Public parent with no forks returns empty forks list, total=0."""
822 created = await execute_create_mist(
823 filename=_PY_FILENAME,
824 content=_unique_content(),
825 actor=_OWNER,
826 )
827 assert created.ok is True
828 mist_id = created.data["mist_id"]
829
830 result = await execute_list_mist_forks(mist_id)
831 assert result.ok is True
832 assert result.data["mist_id"] == mist_id
833 assert result.data["total"] == 0
834 assert result.data["forks"] == []
835
836 async def test_forks_appear_after_fork_creation(
837 self, db_session: AsyncSession
838 ) -> None:
839 """After forking a mist, execute_list_mist_forks returns the fork."""
840 parent = await execute_create_mist(
841 filename=_PY_FILENAME,
842 content=_unique_content(),
843 actor=_OWNER,
844 )
845 assert parent.ok is True
846 parent_id = parent.data["mist_id"]
847
848 fork = await execute_fork_mist(mist_id=parent_id, actor="otheruser")
849 assert fork.ok is True
850
851 result = await execute_list_mist_forks(parent_id)
852 assert result.ok is True
853 assert result.data["total"] == 1
854 fork_entry = result.data["forks"][0]
855 assert fork_entry["owner"] == "otheruser"
856 assert fork_entry["mist_id"] == fork.data["mist_id"]
857
858 async def test_limit_clamped_to_100(
859 self, db_session: AsyncSession
860 ) -> None:
861 """Passing limit=200 is silently clamped to 100 (no error)."""
862 created = await execute_create_mist(
863 filename=_PY_FILENAME,
864 content=_unique_content(),
865 actor=_OWNER,
866 )
867 assert created.ok is True
868
869 result = await execute_list_mist_forks(
870 created.data["mist_id"], limit=200
871 )
872 assert result.ok is True
873
874 async def test_secret_parent_owner_can_list_forks(
875 self, db_session: AsyncSession
876 ) -> None:
877 """Owner of a secret parent can list its forks."""
878 created = await execute_create_mist(
879 filename=_PY_FILENAME,
880 content=_unique_content(),
881 actor=_OWNER,
882 visibility="secret",
883 )
884 assert created.ok is True
885
886 result = await execute_list_mist_forks(
887 created.data["mist_id"], actor=_OWNER
888 )
889 assert result.ok is True
890 assert result.data["total"] == 0
891
892 async def test_fork_entry_has_required_keys(
893 self, db_session: AsyncSession
894 ) -> None:
895 """Each fork entry contains the required schema keys."""
896 parent = await execute_create_mist(
897 filename=_PY_FILENAME,
898 content=_unique_content(),
899 actor=_OWNER,
900 )
901 assert parent.ok is True
902 await execute_fork_mist(mist_id=parent.data["mist_id"], actor="otheruser")
903
904 result = await execute_list_mist_forks(parent.data["mist_id"])
905 assert result.ok is True
906 entry = result.data["forks"][0]
907 for key in ("mist_id", "owner", "filename", "artifact_type",
908 "fork_depth", "fork_count", "visibility", "tags",
909 "created_at"):
910 assert key in entry, f"Missing key '{key}' in fork entry"
911
912 async def test_muse_mist_list_forks_in_tool_catalogue(self) -> None:
913 """muse_mist_list_forks appears in MUSEHUB_TOOL_NAMES."""
914 assert "muse_mist_list_forks" in MUSEHUB_TOOL_NAMES
915
916 async def test_muse_mist_list_forks_not_in_write_tools(self) -> None:
917 """muse_mist_list_forks is a read tool — must not appear in write set."""
918 assert "muse_mist_list_forks" not in MUSEHUB_WRITE_TOOL_NAMES
919
920 async def test_10_forks_listed_under_200ms(
921 self, db_session: AsyncSession
922 ) -> None:
923 """Listing 10 forks completes in under 200ms."""
924 parent = await execute_create_mist(
925 filename=_PY_FILENAME,
926 content=_unique_content(),
927 actor=_OWNER,
928 )
929 assert parent.ok is True
930 parent_id = parent.data["mist_id"]
931
932 for i in range(10):
933 fork = await execute_fork_mist(mist_id=parent_id, actor=f"user{i}")
934 assert fork.ok is True
935
936 start = time.monotonic()
937 result = await execute_list_mist_forks(parent_id, limit=10)
938 elapsed = time.monotonic() - start
939 assert result.ok is True
940 assert result.data["total"] == 10
941 assert elapsed < 0.2, f"listing 10 forks took {elapsed:.2f}s (> 200ms)"
942
943
944 # ── execute_read_mist_raw tests ───────────────────────────────────────────────
945
946
947 @pytest.mark.anyio
948 class TestReadMistRaw:
949 """Tests for execute_read_mist_raw — all 8 tiers.
950
951 Covers: empty mist_id guard, not_found, forbidden (secret mist,
952 non-owner), happy path content/keys, view counter increment,
953 performance (<50ms), and tool catalogue membership.
954 """
955
956 async def test_empty_mist_id_returns_missing_args(
957 self, db_session: AsyncSession
958 ) -> None:
959 """Empty mist_id returns missing_args without a DB hit."""
960 result = await execute_read_mist_raw("")
961 assert result.ok is False
962 assert result.error_code == "missing_args"
963
964 async def test_unknown_mist_returns_not_found(
965 self, db_session: AsyncSession
966 ) -> None:
967 """Non-existent mist_id returns not_found."""
968 result = await execute_read_mist_raw("NoSuchMistXX")
969 assert result.ok is False
970 assert result.error_code == "not_found"
971
972 async def test_secret_mist_anon_returns_forbidden(
973 self, db_session: AsyncSession
974 ) -> None:
975 """Anonymous actor cannot read a secret mist."""
976 created = await execute_create_mist(
977 filename=_PY_FILENAME,
978 content=_unique_content(),
979 actor=_OWNER,
980 visibility="secret",
981 )
982 assert created.ok is True
983
984 result = await execute_read_mist_raw(created.data["mist_id"], actor="")
985 assert result.ok is False
986 assert result.error_code == "forbidden"
987
988 async def test_secret_mist_non_owner_returns_forbidden(
989 self, db_session: AsyncSession
990 ) -> None:
991 """Non-owner actor cannot read a secret mist."""
992 created = await execute_create_mist(
993 filename=_PY_FILENAME,
994 content=_unique_content(),
995 actor=_OWNER,
996 visibility="secret",
997 )
998 assert created.ok is True
999
1000 result = await execute_read_mist_raw(
1001 created.data["mist_id"], actor="intruder"
1002 )
1003 assert result.ok is False
1004 assert result.error_code == "forbidden"
1005
1006 async def test_secret_mist_owner_can_read_raw(
1007 self, db_session: AsyncSession
1008 ) -> None:
1009 """Owner can read a secret mist's raw content."""
1010 content = _unique_content()
1011 created = await execute_create_mist(
1012 filename=_PY_FILENAME,
1013 content=content,
1014 actor=_OWNER,
1015 visibility="secret",
1016 )
1017 assert created.ok is True
1018
1019 result = await execute_read_mist_raw(
1020 created.data["mist_id"], actor=_OWNER
1021 )
1022 assert result.ok is True
1023 assert result.data["content"] == content
1024
1025 async def test_public_mist_readable_by_anon(
1026 self, db_session: AsyncSession
1027 ) -> None:
1028 """Public mist is readable by anonymous caller."""
1029 content = _unique_content()
1030 created = await execute_create_mist(
1031 filename=_PY_FILENAME,
1032 content=content,
1033 actor=_OWNER,
1034 )
1035 assert created.ok is True
1036
1037 result = await execute_read_mist_raw(created.data["mist_id"])
1038 assert result.ok is True
1039 assert result.data["content"] == content
1040
1041 async def test_data_has_required_keys(
1042 self, db_session: AsyncSession
1043 ) -> None:
1044 """Successful result contains all expected data keys."""
1045 created = await execute_create_mist(
1046 filename=_PY_FILENAME,
1047 content=_unique_content(),
1048 actor=_OWNER,
1049 )
1050 assert created.ok is True
1051
1052 result = await execute_read_mist_raw(created.data["mist_id"])
1053 assert result.ok is True
1054 for key in ("mist_id", "filename", "artifact_type",
1055 "language", "size_bytes", "content"):
1056 assert key in result.data, f"Missing key '{key}' in raw result"
1057
1058 async def test_size_bytes_matches_content_length(
1059 self, db_session: AsyncSession
1060 ) -> None:
1061 """size_bytes in the result equals the UTF-8 byte length of content."""
1062 content = _unique_content()
1063 created = await execute_create_mist(
1064 filename=_PY_FILENAME,
1065 content=content,
1066 actor=_OWNER,
1067 )
1068 assert created.ok is True
1069
1070 result = await execute_read_mist_raw(created.data["mist_id"])
1071 assert result.ok is True
1072 assert result.data["size_bytes"] == len(content.encode("utf-8"))
1073
1074 async def test_muse_mist_raw_in_tool_catalogue(self) -> None:
1075 """muse_mist_raw appears in MUSEHUB_TOOL_NAMES."""
1076 assert "muse_mist_raw" in MUSEHUB_TOOL_NAMES
1077
1078 async def test_muse_mist_raw_not_in_write_tools(self) -> None:
1079 """muse_mist_raw is a read tool — must not appear in the write set."""
1080 assert "muse_mist_raw" not in MUSEHUB_WRITE_TOOL_NAMES
1081
1082 async def test_raw_under_50ms(self, db_session: AsyncSession) -> None:
1083 """Raw read of a 1 KiB mist completes in under 50ms."""
1084 content = f"x = 1\n# {'a' * 500}\n"
1085 created = await execute_create_mist(
1086 filename=_PY_FILENAME,
1087 content=content,
1088 actor=_OWNER,
1089 )
1090 assert created.ok is True
1091
1092 start = time.monotonic()
1093 result = await execute_read_mist_raw(created.data["mist_id"])
1094 elapsed = time.monotonic() - start
1095 assert result.ok is True
1096 assert elapsed < 0.05, f"raw read took {elapsed:.3f}s (> 50ms)"
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago