"""TDD — promote all commit_meta fields to first-class columns. Each WireCommit field gets its own column. commit_meta is dropped. Tests are written RED-first against the current schema and pass once migration 0020 and the wire service are updated. """ from __future__ import annotations import pytest from datetime import datetime, timezone from sqlalchemy.ext.asyncio import AsyncSession import sqlalchemy as sa from muse.core.types import blob_id, long_id, now_utc_iso from musehub.db import musehub_models as db from musehub.types.json_types import JSONObject, JSONValue def _utc() -> datetime: return datetime.now(tz=timezone.utc) async def _make_repo(session: AsyncSession, slug: str) -> str: from musehub.core.genesis import compute_identity_id, compute_repo_id owner = "gabriel" owner_user_id = compute_identity_id(owner.encode()) created_at = _utc() repo_id = compute_repo_id(owner_user_id, slug, "code", created_at.isoformat()) session.add(db.MusehubRepo( repo_id=repo_id, name=slug, owner=owner, slug=slug, visibility="public", owner_user_id=owner_user_id, description="", tags=[], created_at=created_at, )) await session.commit() return repo_id # --------------------------------------------------------------------------- # ORM column existence # --------------------------------------------------------------------------- class TestMusehubCommitColumns: """MusehubCommit must declare each promoted column.""" def test_has_signature_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "signature" in cols def test_has_signer_public_key_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "signer_public_key" in cols def test_has_signer_key_id_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "signer_key_id" in cols def test_has_sem_ver_bump_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "sem_ver_bump" in cols def test_has_breaking_changes_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "breaking_changes" in cols def test_has_toolchain_id_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "toolchain_id" in cols def test_has_prompt_hash_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "prompt_hash" in cols def test_has_reviewed_by_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "reviewed_by" in cols def test_has_test_runs_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "test_runs" in cols def test_has_no_commit_meta_column(self) -> None: cols = {c.key for c in db.MusehubCommit.__table__.columns} assert "commit_meta" not in cols, "commit_meta must be dropped" # --------------------------------------------------------------------------- # Wire push populates columns # --------------------------------------------------------------------------- def _make_wire_commit(snapshot_id: str, **overrides: JSONValue) -> JSONObject: import time cid = blob_id(f"meta-promo-{time.time()}".encode()) base = { "commit_id": cid, "parent_ids": [], "parent_commit_id": None, "parent2_commit_id": None, "snapshot_id": snapshot_id, "branch": "task/my-feature", "message": "feat: test commit", "author": "gabriel", "committed_at": now_utc_iso(), "signature": "", "signer_public_key": "", "signer_key_id": "", "agent_id": "claude-code", "model_id": "claude-sonnet-4-6", "toolchain_id": "cursor-agent-v2", "prompt_hash": long_id("a" * 64), "sem_ver_bump": "minor", "breaking_changes": ["removed foo()"], "reviewed_by": ["gabriel"], "test_runs": 3, "metadata": {}, "structured_delta": None, "format_version": 7, } base.update(overrides) return base def _make_snapshot(snapshot_id: str) -> JSONObject: return {"snapshot_id": snapshot_id, "manifest": {}, "committed_at": now_utc_iso()} def _stub_r2(monkeypatch: pytest.MonkeyPatch) -> None: from unittest.mock import AsyncMock store: dict[str, bytes] = {} async def _exists(oid: str) -> bool: return oid in store async def _put(oid: str, data: bytes, **kwargs: JSONValue) -> str: store[oid] = data return f"https://r2.fake/{oid}" async def _get(oid: str) -> bytes | None: return store.get(oid) backend = AsyncMock() backend.exists = _exists backend.put = _put backend.get = _get monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) def _stub_verify(monkeypatch: pytest.MonkeyPatch) -> None: """Bypass Ed25519 crypto gate so tests can use dummy key/sig values.""" monkeypatch.setattr("musehub.services.musehub_wire.verify_commit_ed25519", lambda *_: True) async def _push_commit(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, commit: JSONObject, snap: JSONObject, repo_id: str) -> None: import msgpack from musehub.services.musehub_wire import wire_push_stream from musehub.models.wire import SFRAME_HEADER, SFRAME_COMMIT_PACK, SFRAME_END from muse.core.mpack import MuseWireFrameWriter _fw = MuseWireFrameWriter() def _wrap(ft: str, payload: JSONObject) -> bytes: body = msgpack.packb(payload, use_bin_type=True) return _fw.wrap(frame_type=ft, payload=body) header_frame = _wrap(SFRAME_HEADER, { "t": SFRAME_HEADER, "branch": "dev", "force": False, "have": [], "n_objects": 0, "n_commits": 1, }) commit_frame = _wrap(SFRAME_COMMIT_PACK, {"t": SFRAME_COMMIT_PACK, "commits": [commit], "snapshots": [snap]}) end_frame = _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": 0, "n_commits": 1}) async def body(): yield header_frame + commit_frame + end_frame frames = [] async for chunk in wire_push_stream(db_session, repo_id, body(), "gabriel"): frames.append(chunk) @pytest.mark.asyncio async def test_push_stores_signature(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: """wire_push_stream writes signature to its own column.""" repo_id = await _make_repo(db_session, "meta-promo-sig") snap_id = blob_id(b"snap-sig") commit = _make_wire_commit(snap_id, signature="ed25519:MYSIG", signer_public_key="ed25519:PUBKEY") _stub_r2(monkeypatch) _stub_verify(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.signature == "ed25519:MYSIG" @pytest.mark.asyncio async def test_push_stores_signer_public_key(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-spk") snap_id = blob_id(b"snap-spk") commit = _make_wire_commit(snap_id, signature="ed25519:SIG", signer_public_key="ed25519:PUBKEY") _stub_r2(monkeypatch) _stub_verify(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.signer_public_key == "ed25519:PUBKEY" @pytest.mark.asyncio async def test_push_stores_signer_key_id(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-skid") snap_id = blob_id(b"snap-skid") commit = _make_wire_commit(snap_id, signer_key_id="ed25519:KEYID") _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.signer_key_id == "ed25519:KEYID" @pytest.mark.asyncio async def test_push_stores_sem_ver_bump(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-semver") snap_id = blob_id(b"snap-semver") commit = _make_wire_commit(snap_id, sem_ver_bump="major") _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.sem_ver_bump == "major" @pytest.mark.asyncio async def test_push_stores_breaking_changes(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-breaking") snap_id = blob_id(b"snap-breaking") commit = _make_wire_commit(snap_id, breaking_changes=["removed bar()", "renamed baz()"]) _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.breaking_changes == ["removed bar()", "renamed baz()"] @pytest.mark.asyncio async def test_push_stores_toolchain_id(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-toolchain") snap_id = blob_id(b"snap-toolchain") commit = _make_wire_commit(snap_id, toolchain_id="cursor-agent-v2") _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.toolchain_id == "cursor-agent-v2" @pytest.mark.asyncio async def test_push_stores_prompt_hash(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-prompt") snap_id = blob_id(b"snap-prompt") ph = long_id("b" * 64) commit = _make_wire_commit(snap_id, prompt_hash=ph) _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.prompt_hash == ph @pytest.mark.asyncio async def test_push_stores_reviewed_by(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-reviewed") snap_id = blob_id(b"snap-reviewed") commit = _make_wire_commit(snap_id, reviewed_by=["gabriel", "alice"]) _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.reviewed_by == ["gabriel", "alice"] @pytest.mark.asyncio async def test_push_stores_test_runs(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: repo_id = await _make_repo(db_session, "meta-promo-testruns") snap_id = blob_id(b"snap-testruns") commit = _make_wire_commit(snap_id, test_runs=7) _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert row.test_runs == 7 @pytest.mark.asyncio async def test_push_does_not_store_commit_meta(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> None: """After promotion, commit_meta column must not exist on pushed rows.""" repo_id = await _make_repo(db_session, "meta-promo-no-meta") snap_id = blob_id(b"snap-no-meta") commit = _make_wire_commit(snap_id) _stub_r2(monkeypatch) await _push_commit(db_session, monkeypatch, commit, _make_snapshot(snap_id), repo_id) row = (await db_session.execute( sa.select(db.MusehubCommit).where(db.MusehubCommit.commit_id == commit["commit_id"]) )).scalar_one() assert not hasattr(row, "commit_meta"), "commit_meta must be gone" # --------------------------------------------------------------------------- # Route query — promoted columns appear in symbol timeline context # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_symbol_timeline_query_includes_signature( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch ) -> None: """ui_symbols route selects signature from MusehubCommit — no AttributeError on row.signature.""" from musehub.db import musehub_models as db_models from musehub.core.genesis import compute_identity_id, compute_repo_id from sqlalchemy import select # Build a minimal repo + commit + history entry so the route has data to query owner = "gabriel" slug = "meta-promo-route-test" owner_user_id = compute_identity_id(owner.encode()) created_at = _utc() repo_id = compute_repo_id(owner_user_id, slug, "code", created_at.isoformat()) db_session.add(db_models.MusehubRepo( repo_id=repo_id, name=slug, owner=owner, slug=slug, visibility="public", owner_user_id=owner_user_id, description="", tags=[], created_at=created_at, )) await db_session.commit() commit_id = blob_id(b"route-test-commit") committed_at = _utc() db_session.add(db_models.MusehubCommit( commit_id=commit_id, repo_id=repo_id, branch="dev", parent_ids=[], message="feat: route test", author="gabriel", timestamp=committed_at, agent_id="claude-code", model_id="claude-sonnet-4-6", commit_branch="task/route-test", signature="ed25519:ROUTESIG", )) db_session.add(db_models.MusehubSymbolHistoryEntry( repo_id=repo_id, address="src/route.py::fn", commit_id=commit_id, committed_at=committed_at, author="gabriel", op="add", content_id=blob_id(b"body-route"), )) await db_session.commit() # Execute the same SELECT the route uses — must include signature without AttributeError result = await db_session.execute( select( db_models.MusehubCommit.commit_id, db_models.MusehubCommit.message, db_models.MusehubCommit.author, db_models.MusehubCommit.branch, db_models.MusehubCommit.commit_branch, db_models.MusehubCommit.agent_id, db_models.MusehubCommit.model_id, db_models.MusehubCommit.signature, ).where( db_models.MusehubCommit.repo_id == repo_id, db_models.MusehubCommit.commit_id == commit_id, ) ) row = result.one() assert row.signature == "ed25519:ROUTESIG" assert row.agent_id == "claude-code" assert row.commit_branch == "task/route-test" # --------------------------------------------------------------------------- # _to_wire_commit reads from first-class columns, not commit_meta # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_to_wire_commit_reads_first_class_columns( db_session: AsyncSession, ) -> None: """_to_wire_commit must read provenance from ORM columns, not commit_meta.""" from musehub.services.musehub_wire import _to_wire_commit repo_id = await _make_repo(db_session, "wire-commit-columns-001") commit_id = blob_id(b"wire-commit-cols-test") row = db.MusehubCommit( commit_id=commit_id, repo_id=repo_id, branch="dev", parent_ids=[blob_id(b"parent-commit")], message="feat: first-class columns", author="gabriel", timestamp=_utc(), agent_id="claude-code", model_id="claude-sonnet-4-6", toolchain_id="muse-1.0", commit_branch="task/my-feature", signature="ed25519:SIG123", signer_public_key="ed25519:PUB456", signer_key_id="key-id-789", sem_ver_bump="minor", breaking_changes=[], reviewed_by=[], test_runs=3, prompt_hash=blob_id(b"test-prompt"), ) db_session.add(row) await db_session.commit() await db_session.refresh(row) wc = _to_wire_commit(row) assert wc.commit_id == commit_id assert wc.agent_id == "claude-code" assert wc.model_id == "claude-sonnet-4-6" assert wc.toolchain_id == "muse-1.0" assert wc.signature == "ed25519:SIG123" assert wc.signer_public_key == "ed25519:PUB456" assert wc.signer_key_id == "key-id-789" assert wc.sem_ver_bump == "minor" assert wc.test_runs == 3 assert wc.prompt_hash == blob_id(b"test-prompt") assert wc.parent_commit_id == blob_id(b"parent-commit")