gabriel / musehub public
test_mpack_index_job_phase4.py python
324 lines 11.8 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 49 days ago
1 """TDD — Phase 4: observability and return-dict invariants.
2
3 process_mpack_index_job must return a result dict that a monitoring dashboard
4 or alerting rule can consume without re-parsing log lines.
5
6 Phase 4 invariants:
7 1. Return dict contains the canonical five keys (counts + timing).
8 2. All count values are non-negative integers.
9 3. mpack_size_bytes matches the byte length of what was stored.
10 4. elapsed_ms is a positive float.
11 5. The structured summary log line is emitted with index_rows, graph_rows,
12 and elapsed.
13
14 Canonical return dict keys from process_mpack_index_job:
15 mpack_index_written — rows upserted into MusehubMPackIndex (int)
16 byte_ranges_computed — number of byte-range entries computed (int)
17 commit_graph_written — rows upserted into MusehubCommitGraph (int)
18 mpack_size_bytes — raw wire size of the stored mpack (int)
19 elapsed_ms — wall-clock duration of the job (float ms)
20 """
21 from __future__ import annotations
22
23 import datetime
24 import logging
25 import pathlib
26
27 import msgpack
28 import pytest
29 import pytest_asyncio
30
31 from httpx import AsyncClient, ASGITransport
32 from sqlalchemy import select
33 from sqlalchemy.ext.asyncio import AsyncSession
34
35 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
36 from musehub.db.database import get_db
37 from musehub.db.musehub_repo_models import MusehubRepo
38 from musehub.main import app
39
40 from muse.core.object_store import write_object
41 from muse.core.mpack import build_mpack
42 from muse.core.paths import muse_dir
43 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
44 from muse.core.commits import CommitRecord, write_commit
45 from muse.core.refs import write_branch_ref
46 from muse.core.snapshots import SnapshotRecord, write_snapshot
47 from muse.core.types import blob_id
48
49
50 _AUTH_CTX = MSignContext(
51 handle="gabriel",
52 identity_id="sha256:" + "0" * 64,
53 is_agent=False,
54 is_admin=True,
55 )
56
57 _N_FILES = 8
58 _N_COMMITS = 4
59 _FILES_CHANGED = 2
60 _BLOB_SIZE = 128
61
62 # Canonical return dict keys for process_mpack_index_job.
63 _REQUIRED_KEYS: frozenset[str] = frozenset({
64 "mpack_index_written",
65 "byte_ranges_computed",
66 "commit_graph_written",
67 "mpack_size_bytes",
68 "elapsed_ms",
69 })
70
71
72 # ── fixtures ────────────────────────────────────────────────────────────────
73
74 @pytest_asyncio.fixture()
75 async def client(db_session: AsyncSession) -> None:
76 async def _override_get_db() -> None:
77 yield db_session
78
79 app.dependency_overrides[get_db] = _override_get_db
80 app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
81 app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX
82
83 async with AsyncClient(
84 transport=ASGITransport(app=app),
85 base_url="https://localhost:1337",
86 ) as c:
87 yield c
88
89 app.dependency_overrides.clear()
90
91
92 @pytest_asyncio.fixture()
93 async def repo(client: AsyncClient) -> None:
94 resp = await client.post(
95 "/api/repos",
96 json={"owner": "gabriel", "name": "phase4-obs-test", "visibility": "public", "initialize": False},
97 )
98 assert resp.status_code in (200, 201), resp.text
99 data = resp.json()
100 yield data["slug"]
101 await client.delete(f"/api/repos/{data['repoId']}")
102
103
104 def _make_repo(tmp: pathlib.Path) -> tuple[pathlib.Path, str, dict]:
105 tmp.mkdir(parents=True, exist_ok=True)
106 dot = muse_dir(tmp)
107 dot.mkdir()
108 (dot / "repo.json").write_text('{"repo_id":"phase4-test","owner":"gabriel"}')
109 for d in ("commits", "snapshots", "objects"):
110 (dot / d).mkdir()
111 (dot / "refs" / "heads").mkdir(parents=True)
112 (dot / "HEAD").write_text("ref: refs/heads/main\n")
113 (dot / "config.toml").write_text("")
114
115 blob_ids: list[str] = []
116 for i in range(_N_FILES):
117 data = f"base-{i:04d}".encode() + b"x" * _BLOB_SIZE
118 oid = blob_id(data)
119 write_object(tmp, oid, data)
120 blob_ids.append(oid)
121
122 base_manifest = {f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)}
123 parent = None
124 tip = ""
125 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
126
127 for i in range(_N_COMMITS):
128 manifest = dict(base_manifest)
129 for j in range(_FILES_CHANGED):
130 idx = (i * _FILES_CHANGED + j) % _N_FILES
131 raw = f"c{i:04d}-f{j}".encode() + b"y" * _BLOB_SIZE
132 oid = blob_id(raw)
133 write_object(tmp, oid, raw)
134 manifest[f"src/file_{idx:04d}.py"] = oid
135
136 sid = compute_snapshot_id(manifest)
137 write_snapshot(tmp, SnapshotRecord(snapshot_id=sid, manifest=manifest))
138 msg = f"commit-{i:05d}"
139 cid = compute_commit_id(
140 parent_ids=[parent] if parent else [],
141 snapshot_id=sid,
142 message=msg,
143 committed_at_iso=ts.isoformat(),
144 author="gabriel",
145 )
146 write_commit(tmp, CommitRecord(
147 commit_id=cid, branch="main",
148 snapshot_id=sid, message=msg, committed_at=ts,
149 parent_commit_id=parent, parent2_commit_id=None,
150 author="gabriel", metadata={}, structured_delta=None,
151 sem_ver_bump="none", breaking_changes=[],
152 agent_id="", model_id="", toolchain_id="",
153 prompt_hash="", signature="", signer_key_id="",
154 ))
155 parent = cid
156 tip = cid
157 ts += datetime.timedelta(seconds=60)
158
159 write_branch_ref(tmp, "main", tip)
160 mpack = build_mpack(tmp, [tip], have=[])
161 return tmp, tip, mpack
162
163
164 async def _push_mpack(
165 repo_slug: str,
166 mpack: dict,
167 head: str,
168 db_session: AsyncSession,
169 ) -> tuple[str, int]:
170 """Store mpack in MemoryBackend and create a mpack.index job row.
171
172 Returns (job_id, stored_byte_length) so tests can verify mpack_size_bytes.
173 """
174 import musehub.storage.backends as _backends_mod
175 from datetime import datetime, timezone
176 from musehub.core.genesis import compute_job_id as _compute_job_id
177 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
178
179 repo_row = (await db_session.execute(
180 select(MusehubRepo).where(MusehubRepo.slug == repo_slug)
181 )).scalar_one()
182 repo_id = repo_row.repo_id
183
184 wire_bytes = msgpack.packb(mpack, use_bin_type=True)
185 mpack_key = "sha256:" + __import__("hashlib").sha256(wire_bytes).hexdigest()
186 n_objects = len(mpack.get("blobs") or [])
187
188 backend = _backends_mod.get_backend()
189 await backend.put_mpack(mpack_key, wire_bytes)
190
191 now = datetime.now(tz=timezone.utc)
192 job_id = _compute_job_id(repo_id, "mpack.index", now.isoformat())
193 db_session.add(MusehubBackgroundJob(
194 job_id=job_id,
195 repo_id=repo_id,
196 job_type="mpack.index",
197 payload={
198 "mpack_key": mpack_key,
199 "pusher_id": "sha256:" + "0" * 64,
200 "branch": "main",
201 "head": head,
202 "force": False,
203 "declared_objects_count": n_objects,
204 },
205 status="pending",
206 created_at=now,
207 attempt=0,
208 ))
209 await db_session.flush()
210 return job_id, len(wire_bytes)
211
212
213 # ── Phase 4 tests ───────────────────────────────────────────────────────────
214
215 @pytest.mark.asyncio
216 async def test_return_dict_has_required_keys(
217 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
218 ) -> None:
219 """Return dict contains all five canonical keys.
220
221 These keys let monitoring dashboards query job results without parsing
222 log lines. Any missing key is a regression in the observability contract.
223 """
224 _, head, mpack = _make_repo(tmp_path / "repo")
225 job_id, _ = await _push_mpack(repo, mpack, head, db_session)
226
227 from musehub.services.musehub_wire import process_mpack_index_job
228 result = await process_mpack_index_job(db_session, job_id)
229 await db_session.commit()
230
231 missing = _REQUIRED_KEYS - set(result.keys())
232 assert not missing, (
233 f"process_mpack_index_job return dict is missing keys: {missing}\n"
234 f"Got keys: {sorted(result.keys())}"
235 )
236
237
238 @pytest.mark.asyncio
239 async def test_all_counts_are_non_negative(
240 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
241 ) -> None:
242 """Every count value in the return dict is a non-negative integer."""
243 _, head, mpack = _make_repo(tmp_path / "repo")
244 job_id, _ = await _push_mpack(repo, mpack, head, db_session)
245
246 from musehub.services.musehub_wire import process_mpack_index_job
247 result = await process_mpack_index_job(db_session, job_id)
248 await db_session.commit()
249
250 count_keys = ("mpack_index_written", "byte_ranges_computed", "commit_graph_written", "mpack_size_bytes")
251 bad = {k: result[k] for k in count_keys if result[k] < 0}
252 assert not bad, f"Negative count values in result: {bad}"
253
254
255 @pytest.mark.asyncio
256 async def test_mpack_size_bytes_matches_stored_bytes(
257 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
258 ) -> None:
259 """mpack_size_bytes matches the byte length of what was stored in the backend."""
260 _, head, mpack = _make_repo(tmp_path / "repo")
261 job_id, stored_byte_length = await _push_mpack(repo, mpack, head, db_session)
262
263 from musehub.services.musehub_wire import process_mpack_index_job
264 result = await process_mpack_index_job(db_session, job_id)
265 await db_session.commit()
266
267 assert "mpack_size_bytes" in result, f"mpack_size_bytes missing. Got: {sorted(result.keys())}"
268 assert result["mpack_size_bytes"] == stored_byte_length, (
269 f"mpack_size_bytes {result['mpack_size_bytes']} != stored {stored_byte_length}"
270 )
271
272
273 @pytest.mark.asyncio
274 async def test_elapsed_ms_is_positive(
275 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
276 ) -> None:
277 """elapsed_ms is a positive float — zero would indicate a timer bug."""
278 _, head, mpack = _make_repo(tmp_path / "repo")
279 job_id, _ = await _push_mpack(repo, mpack, head, db_session)
280
281 from musehub.services.musehub_wire import process_mpack_index_job
282 result = await process_mpack_index_job(db_session, job_id)
283 await db_session.commit()
284
285 assert result["elapsed_ms"] > 0, (
286 f"elapsed_ms={result['elapsed_ms']} — must be positive (timer checkpoint bug?)"
287 )
288
289
290 @pytest.mark.asyncio
291 async def test_summary_log_contains_index_and_graph_counts(
292 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
293 caplog: pytest.LogCaptureFixture,
294 ) -> None:
295 """The ✅ done summary log line contains index_rows, graph_rows, and elapsed.
296
297 Ops teams grep a single log line per job for the full result —
298 not seven separate lines. The summary must contain these three fields
299 so log-based alerts stay in sync with the return dict.
300 """
301 _, head, mpack = _make_repo(tmp_path / "repo")
302 job_id, _ = await _push_mpack(repo, mpack, head, db_session)
303
304 from musehub.services.musehub_wire import process_mpack_index_job
305 with caplog.at_level(logging.INFO, logger="musehub.services.musehub_wire_shared"):
306 result = await process_mpack_index_job(db_session, job_id)
307 await db_session.commit()
308
309 summary_lines = [
310 r.message for r in caplog.records
311 if "mpack.index" in r.message and "done" in r.message
312 ]
313 assert summary_lines, (
314 "No '✅ [mpack.index] done' summary log line found.\n"
315 f"All mpack.index log lines: {[r.message for r in caplog.records if 'mpack.index' in r.message]}"
316 )
317
318 summary = summary_lines[-1]
319 _REQUIRED_LOG_FIELDS = ("index_rows", "graph_rows", "elapsed")
320 missing_fields = [f for f in _REQUIRED_LOG_FIELDS if f not in summary]
321 assert not missing_fields, (
322 f"Summary log line missing fields: {missing_fields}\n"
323 f"Summary: {summary!r}"
324 )
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 49 days ago