gabriel / musehub public
test_mist_phase1_intel_pipeline.py python
377 lines 13.5 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago
1 """Phase 1 TDD: Mist domain intel pipeline — job dispatch + MistProvider.
2
3 Tests are written RED first. Run them before touching musehub_intel_providers.py
4 to confirm they fail for the right reason, then implement to make them green.
5
6 Coverage:
7 1. job_types_for_push("mist") includes "intel.mist"
8 2. MistProvider.compute extracts anchors and persists intel results
9 3. MistProvider.compute handles binary / anchor-free artifacts gracefully
10 4. Regression: code and midi dispatch are unaffected by the mist branch
11 5. MistProvider is registered in _PROVIDER_REGISTRY under "intel.mist"
12 """
13 from __future__ import annotations
14
15 import secrets
16 from datetime import datetime, timezone
17
18 import pytest
19 from sqlalchemy import select
20 from sqlalchemy.ext.asyncio import AsyncSession
21
22 from musehub.db import musehub_models as db
23 from musehub.core.genesis import compute_identity_id, compute_repo_id
24
25
26 # ---------------------------------------------------------------------------
27 # Helpers
28 # ---------------------------------------------------------------------------
29
30 def _now() -> datetime:
31 return datetime.now(tz=timezone.utc)
32
33
34 def _uid() -> str:
35 return secrets.token_hex(16)
36
37
38 def _repo_id(owner: str, slug: str) -> str:
39 return compute_repo_id(
40 compute_identity_id(owner.encode()),
41 slug,
42 "mist",
43 _now().isoformat(),
44 )
45
46
47 async def _seed_mist(
48 session: AsyncSession,
49 *,
50 owner: str = "testuser",
51 filename: str = "snippet.py",
52 content: str = "def hello():\n return 'world'\n",
53 artifact_type: str = "code",
54 symbol_anchors: list[str] | None = None,
55 mist_id: str | None = None,
56 ) -> tuple[db.MusehubRepo, db.MusehubMist]:
57 """Create a MusehubRepo (domain_id='mist') and a linked MusehubMist row."""
58 slug = mist_id or f"mist-{secrets.token_hex(4)}"
59 owner_id = compute_identity_id(owner.encode())
60 created_at = _now()
61 repo_id = compute_repo_id(owner_id, slug, "mist", created_at.isoformat())
62
63 repo = db.MusehubRepo(
64 repo_id=repo_id,
65 name=slug,
66 owner=owner,
67 slug=slug,
68 visibility="public",
69 owner_user_id=owner_id,
70 domain_id="mist",
71 description="test mist repo",
72 tags=[],
73 created_at=created_at,
74 )
75 session.add(repo)
76 await session.flush()
77
78 actual_mist_id = mist_id or f"Abc{secrets.token_hex(5)[:9]}"
79 mist = db.MusehubMist(
80 mist_id=actual_mist_id,
81 repo_id=repo_id,
82 owner=owner,
83 filename=filename,
84 content=content,
85 artifact_type=artifact_type,
86 language="python" if filename.endswith(".py") else "",
87 size_bytes=len(content.encode()),
88 symbol_anchors=symbol_anchors or [],
89 )
90 session.add(mist)
91 await session.commit()
92 await session.refresh(repo)
93 await session.refresh(mist)
94 return repo, mist
95
96
97 # ---------------------------------------------------------------------------
98 # 1. job_types_for_push dispatch
99 # ---------------------------------------------------------------------------
100
101 class TestJobTypesForPush:
102 def test_mist_domain_dispatches_intel_mist(self) -> None:
103 from musehub.services.musehub_intel_providers import job_types_for_push
104
105 types = job_types_for_push("mist")
106 assert "intel.mist" in types, (
107 f"job_types_for_push('mist') must include 'intel.mist'; got {types}"
108 )
109
110 def test_mist_domain_always_includes_structural(self) -> None:
111 from musehub.services.musehub_intel_providers import job_types_for_push
112
113 types = job_types_for_push("mist")
114 assert "intel.structural" in types
115
116 def test_mist_domain_always_includes_gc(self) -> None:
117 from musehub.services.musehub_intel_providers import job_types_for_push
118
119 types = job_types_for_push("mist")
120 assert "gc" in types
121
122 def test_mist_domain_does_not_include_intel_code(self) -> None:
123 from musehub.services.musehub_intel_providers import job_types_for_push
124
125 types = job_types_for_push("mist")
126 assert "intel.code" not in types, (
127 "mist domain must not trigger code intel job"
128 )
129
130 def test_code_domain_unaffected(self) -> None:
131 from musehub.services.musehub_intel_providers import job_types_for_push
132
133 types = job_types_for_push("code")
134 assert "intel.code" in types
135 assert "intel.mist" not in types
136
137 def test_midi_domain_unaffected(self) -> None:
138 from musehub.services.musehub_intel_providers import job_types_for_push
139
140 types = job_types_for_push("midi")
141 assert "intel.midi" in types
142 assert "intel.mist" not in types
143
144 def test_none_domain_defaults_to_code(self) -> None:
145 from musehub.services.musehub_intel_providers import job_types_for_push
146
147 types = job_types_for_push(None)
148 assert "intel.code" in types
149 assert "intel.mist" not in types
150
151
152 # ---------------------------------------------------------------------------
153 # 2. _PROVIDER_REGISTRY contains "intel.mist"
154 # ---------------------------------------------------------------------------
155
156 class TestProviderRegistry:
157 def test_intel_mist_is_registered(self) -> None:
158 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
159
160 assert "intel.mist" in _PROVIDER_REGISTRY, (
161 "'intel.mist' must be in _PROVIDER_REGISTRY"
162 )
163
164 def test_intel_mist_satisfies_protocol(self) -> None:
165 from musehub.services.musehub_intel_providers import (
166 _PROVIDER_REGISTRY,
167 IntelProvider,
168 )
169
170 provider = _PROVIDER_REGISTRY["intel.mist"]
171 assert isinstance(provider, IntelProvider), (
172 "MistProvider must satisfy the IntelProvider protocol"
173 )
174
175
176 # ---------------------------------------------------------------------------
177 # 3. MistProvider.compute — anchor extraction
178 # ---------------------------------------------------------------------------
179
180 class TestMistProviderCompute:
181 @pytest.mark.asyncio
182 async def test_extracts_anchors_for_python_artifact(
183 self, db_session: AsyncSession, test_user: db.MusehubIdentity
184 ) -> None:
185 """Provider returns mist.anchors result with correct symbol addresses."""
186 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
187
188 provider = _PROVIDER_REGISTRY["intel.mist"]
189 repo, mist = await _seed_mist(
190 db_session,
191 owner=test_user.handle,
192 filename="utils.py",
193 content="def add(a, b):\n return a + b\n\ndef subtract(a, b):\n return a - b\n",
194 artifact_type="code",
195 )
196
197 results = await provider.compute(db_session, repo.repo_id, "HEAD", {})
198
199 assert len(results) == 1
200 intel_type, data = results[0]
201 assert intel_type == "mist.anchors"
202 assert data["mist_id"] == mist.mist_id
203 assert data["filename"] == "utils.py"
204 assert data["artifact_type"] == "code"
205 anchors: list[str] = data["symbol_anchors"]
206 assert any("add" in a for a in anchors), f"Expected 'add' anchor; got {anchors}"
207 assert any("subtract" in a for a in anchors), f"Expected 'subtract' anchor; got {anchors}"
208 assert data["anchor_count"] == len(anchors)
209
210 @pytest.mark.asyncio
211 async def test_anchor_count_matches_symbol_anchors_length(
212 self, db_session: AsyncSession, test_user: db.MusehubIdentity
213 ) -> None:
214 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
215
216 provider = _PROVIDER_REGISTRY["intel.mist"]
217 repo, _ = await _seed_mist(
218 db_session,
219 owner=test_user.handle,
220 filename="calc.py",
221 content=(
222 "class Calc:\n"
223 " def mul(self, a, b): return a * b\n"
224 " def div(self, a, b): return a / b\n"
225 ),
226 )
227
228 results = await provider.compute(db_session, repo.repo_id, "HEAD", {})
229 _, data = results[0]
230 assert data["anchor_count"] == len(data["symbol_anchors"])
231
232 @pytest.mark.asyncio
233 async def test_binary_artifact_produces_zero_anchors(
234 self, db_session: AsyncSession, test_user: db.MusehubIdentity
235 ) -> None:
236 """Binary content (e.g. base64) with no parsable symbols → zero anchors, no crash."""
237 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
238
239 provider = _PROVIDER_REGISTRY["intel.mist"]
240 # Base64-encoded PNG header — not parseable as Python/JS/TS
241 binary_content = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk"
242 repo, _ = await _seed_mist(
243 db_session,
244 owner=test_user.handle,
245 filename="image.png",
246 content=binary_content,
247 artifact_type="image",
248 )
249
250 results = await provider.compute(db_session, repo.repo_id, "HEAD", {})
251
252 assert len(results) == 1
253 _, data = results[0]
254 assert data["symbol_anchors"] == []
255 assert data["anchor_count"] == 0
256
257 @pytest.mark.asyncio
258 async def test_no_mist_for_repo_returns_empty(
259 self, db_session: AsyncSession, test_user: db.MusehubIdentity
260 ) -> None:
261 """A repo with no mist row (edge case) → empty results, no crash."""
262 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
263
264 provider = _PROVIDER_REGISTRY["intel.mist"]
265 owner_id = compute_identity_id(test_user.handle.encode())
266 created_at = _now()
267 repo_id = compute_repo_id(owner_id, "orphan-repo", "mist", created_at.isoformat())
268 repo = db.MusehubRepo(
269 repo_id=repo_id,
270 name="orphan-repo",
271 owner=test_user.handle,
272 slug="orphan-repo",
273 visibility="public",
274 owner_user_id=owner_id,
275 domain_id="mist",
276 description="",
277 tags=[],
278 created_at=created_at,
279 )
280 db_session.add(repo)
281 await db_session.commit()
282
283 results = await provider.compute(db_session, repo_id, "HEAD", {})
284 assert results == []
285
286 @pytest.mark.asyncio
287 async def test_updates_symbol_anchors_on_mist_row(
288 self, db_session: AsyncSession, test_user: db.MusehubIdentity
289 ) -> None:
290 """Provider refreshes mist.symbol_anchors in the DB if they were stale."""
291 from musehub.services.musehub_intel_providers import _PROVIDER_REGISTRY
292
293 provider = _PROVIDER_REGISTRY["intel.mist"]
294 # Seed with deliberately empty symbol_anchors
295 repo, mist = await _seed_mist(
296 db_session,
297 owner=test_user.handle,
298 filename="module.py",
299 content="def process(data):\n return data\n",
300 symbol_anchors=[], # stale — will be refreshed by provider
301 )
302
303 await provider.compute(db_session, repo.repo_id, "HEAD", {})
304 await db_session.commit()
305
306 await db_session.refresh(mist)
307 assert any("process" in a for a in mist.symbol_anchors), (
308 f"mist.symbol_anchors should be refreshed; got {mist.symbol_anchors}"
309 )
310
311 @pytest.mark.asyncio
312 async def test_results_persisted_via_persist_intel_results(
313 self, db_session: AsyncSession, test_user: db.MusehubIdentity
314 ) -> None:
315 """Full pipeline: compute → persist_intel_results → row in musehub_intel_results."""
316 from musehub.services.musehub_intel_providers import (
317 _PROVIDER_REGISTRY,
318 persist_intel_results,
319 )
320
321 provider = _PROVIDER_REGISTRY["intel.mist"]
322 repo, _ = await _seed_mist(
323 db_session,
324 owner=test_user.handle,
325 filename="api.py",
326 content="async def handle(request):\n pass\n",
327 )
328
329 results = await provider.compute(db_session, repo.repo_id, "HEAD", {})
330 await persist_intel_results(db_session, repo.repo_id, "HEAD", results)
331 await db_session.commit()
332
333 row = (await db_session.execute(
334 select(db.MusehubIntelResult).where(
335 db.MusehubIntelResult.repo_id == repo.repo_id,
336 db.MusehubIntelResult.intel_type == "mist.anchors",
337 )
338 )).scalar_one_or_none()
339
340 assert row is not None, "intel result row must exist after persist_intel_results"
341 import json
342 data = json.loads(row.data_json)
343 assert data["mist_id"] is not None
344 assert "symbol_anchors" in data
345
346 @pytest.mark.asyncio
347 async def test_persist_is_idempotent(
348 self, db_session: AsyncSession, test_user: db.MusehubIdentity
349 ) -> None:
350 """Running compute + persist twice for the same repo produces exactly one row."""
351 from musehub.services.musehub_intel_providers import (
352 _PROVIDER_REGISTRY,
353 persist_intel_results,
354 )
355 from sqlalchemy import func
356
357 provider = _PROVIDER_REGISTRY["intel.mist"]
358 repo, _ = await _seed_mist(
359 db_session,
360 owner=test_user.handle,
361 filename="idempotent.py",
362 content="def noop(): pass\n",
363 )
364
365 for _ in range(2):
366 results = await provider.compute(db_session, repo.repo_id, "HEAD", {})
367 await persist_intel_results(db_session, repo.repo_id, "HEAD", results)
368 await db_session.commit()
369
370 count = (await db_session.execute(
371 select(func.count()).select_from(db.MusehubIntelResult).where(
372 db.MusehubIntelResult.repo_id == repo.repo_id,
373 db.MusehubIntelResult.intel_type == "mist.anchors",
374 )
375 )).scalar_one()
376
377 assert count == 1, f"Idempotent upsert must produce exactly 1 row; got {count}"
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago