gabriel / muse public
test_commit_attest_flag.py python
266 lines 10.1 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Tests for ``muse commit --attest`` — Phase 7.3.
2
3 Rule #0 tier coverage:
4
5 * unit — flag parsing, skip-if-no-provider, metadata storage shape.
6 * integration — end-to-end: commit --attest with a registered stub provider
7 produces a record under commit.metadata['attestation'].
8 * security — provider raises AttestationRequiredError → commit aborted
9 with ExitCode.ATTESTATION_REQUIRED (not silent skip);
10 required=False non-required errors → commit succeeds without
11 attestation; sensitive metadata never leaks into stderr.
12 """
13
14 from __future__ import annotations
15
16 import argparse
17 import json
18 import os
19 import pathlib
20 import subprocess
21 from typing import Any
22
23 import pytest
24
25 from muse.cli.commands import commit as commit_cmd
26 from muse.core.attestation import (
27 AnchorReceipt,
28 AttestationRecord,
29 AttestationRequiredError,
30 MuseAttestationProvider,
31 VerifyResult,
32 register_attestation_provider,
33 unregister_attestation_provider,
34 )
35 from muse.core.errors import ExitCode
36
37
38 # ---------------------------------------------------------------------------
39 # Stub providers for fault injection
40 # ---------------------------------------------------------------------------
41
42
43 class _OkStub:
44 """Always-succeeds provider — verifies happy-path behaviour."""
45
46 domain = "knowtation"
47
48 def __init__(self) -> None:
49 self.calls: list[tuple[str, dict[str, object]]] = []
50
51 def compute_attestation(
52 self, snapshot_id: str, commit_meta: dict[str, object]
53 ) -> AttestationRecord:
54 self.calls.append((snapshot_id, dict(commit_meta)))
55 return {
56 "id": "a" * 64,
57 "action": "write",
58 "path": str(commit_meta.get("path", "")),
59 "timestamp": str(commit_meta.get("timestamp", "")),
60 "content_hash": snapshot_id,
61 "sig": "stub-sig",
62 "algorithm": "stub",
63 "provider_id": "knowtation",
64 }
65
66 def anchor(self, record: AttestationRecord) -> AnchorReceipt:
67 return {
68 "tx_id": "stub", "anchor_url": "https://stub.example/x",
69 "anchored_at": "", "provider_id": "knowtation",
70 }
71
72 def verify(self, record: AttestationRecord, receipt: AnchorReceipt) -> VerifyResult:
73 return {"valid": True, "reason": "", "depth": 0, "provider_id": "knowtation"}
74
75
76 class _RequiredStub(_OkStub):
77 """Provider that always raises AttestationRequiredError."""
78
79 def compute_attestation(
80 self, snapshot_id: str, commit_meta: dict[str, object]
81 ) -> AttestationRecord:
82 raise AttestationRequiredError("simulated air.required=true rejection")
83
84
85 class _SoftFailStub(_OkStub):
86 """Provider that raises a non-required AttestationError — should be soft-skipped."""
87
88 def compute_attestation(
89 self, snapshot_id: str, commit_meta: dict[str, object]
90 ) -> AttestationRecord:
91 from muse.core.attestation import AttestationError
92 raise AttestationError("transient endpoint failure")
93
94
95 @pytest.fixture(autouse=True)
96 def _registry_cleanup() -> None:
97 """Ensure each test starts and ends with no provider registered."""
98 unregister_attestation_provider("knowtation")
99 yield
100 unregister_attestation_provider("knowtation")
101
102
103 # ---------------------------------------------------------------------------
104 # Helpers
105 # ---------------------------------------------------------------------------
106
107
108 def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path:
109 """Initialise a knowtation-domain Muse repo at *tmp_path* and chdir into it."""
110 os.chdir(tmp_path)
111 from muse.cli.commands.init import run as init_run
112 args = argparse.Namespace(
113 bare=False, template=None, default_branch="main",
114 force=False, domain="knowtation", as_json=False,
115 )
116 init_run(args)
117 (tmp_path / "notes").mkdir()
118 (tmp_path / "notes" / "seed.md").write_text("# seed\n", encoding="utf-8")
119 return tmp_path
120
121
122 def _commit(
123 tmp_path: pathlib.Path,
124 *,
125 message: str = "first",
126 attest: bool = False,
127 sign: bool = False,
128 agent_id: str = "",
129 ) -> dict[str, Any]:
130 """Run muse commit and return the parsed JSON result."""
131 args = argparse.Namespace(
132 message=message, allow_empty=False, dry_run=False,
133 section=None, track=None, emotion=None, author=None,
134 agent_id=agent_id or None, model_id=None, toolchain_id=None,
135 event_type=None, meta=None,
136 sign=sign, attest=attest, attest_config=None,
137 fmt="json",
138 )
139 captured: list[str] = []
140 import builtins
141 real_print = builtins.print
142 try:
143 builtins.print = lambda *a, **kw: captured.append(" ".join(str(x) for x in a)) # type: ignore[assignment]
144 commit_cmd.run(args)
145 finally:
146 builtins.print = real_print
147 payload_lines = [c for c in captured if c.strip().startswith("{")]
148 assert payload_lines, f"commit produced no JSON output; got: {captured!r}"
149 return json.loads(payload_lines[-1])
150
151
152 # ===========================================================================
153 # UNIT TIER
154 # ===========================================================================
155
156
157 class TestFlagParsing:
158 """argparse should accept the new --attest and --attest-config flags."""
159
160 def test_attest_flag_in_parser(self) -> None:
161 """Help text mentions --attest."""
162 parser = argparse.ArgumentParser()
163 sp = parser.add_subparsers(dest="cmd")
164 commit_cmd.register(sp)
165 help_text = parser.format_help()
166 # The subparser help text is populated lazily; ensure the option
167 # is at least present in the parser's actions.
168 commit_parser = sp.choices["commit"]
169 opts = {a.option_strings[0] for a in commit_parser._actions if a.option_strings}
170 assert "--attest" in opts
171 assert "--attest-config" in opts
172
173
174 # ===========================================================================
175 # INTEGRATION TIER
176 # ===========================================================================
177
178
179 class TestEndToEnd:
180 """Real init + commit flow with a registered provider."""
181
182 def test_attest_skipped_when_no_provider(self, tmp_path: pathlib.Path) -> None:
183 """No provider → commit succeeds, no attestation key in metadata."""
184 from muse.core.store import read_commit
185 _init_repo(tmp_path)
186 result = _commit(tmp_path, attest=True)
187 assert result["dry_run"] is False
188 commit_record = read_commit(tmp_path, result["commit_id"])
189 assert commit_record is not None
190 assert "attestation" not in (commit_record.metadata or {})
191
192 def test_attest_attaches_record_when_provider_present(
193 self, tmp_path: pathlib.Path
194 ) -> None:
195 """Registered provider → metadata['attestation'] is canonical JSON."""
196 from muse.core.store import read_commit
197 provider = _OkStub()
198 register_attestation_provider("knowtation", provider)
199 _init_repo(tmp_path)
200 result = _commit(tmp_path, attest=True)
201 commit_record = read_commit(tmp_path, result["commit_id"])
202 assert commit_record is not None
203 attestation_blob = commit_record.metadata["attestation"]
204 parsed = json.loads(attestation_blob)
205 assert parsed["provider_id"] == "knowtation"
206 assert parsed["sig"] == "stub-sig"
207 assert provider.calls
208 snap_id, meta = provider.calls[0]
209 assert meta["action"] == "write"
210 # Picks the first changed path in canonical sort order — covers both
211 # `.museattributes` (created by init) and `notes/seed.md`.
212 assert meta["path"] in (".museattributes", "notes/seed.md")
213
214 def test_no_attest_flag_no_provider_call(self, tmp_path: pathlib.Path) -> None:
215 """Without --attest the provider is not called even when registered."""
216 provider = _OkStub()
217 register_attestation_provider("knowtation", provider)
218 _init_repo(tmp_path)
219 _commit(tmp_path, attest=False)
220 assert provider.calls == []
221
222
223 # ===========================================================================
224 # SECURITY TIER (mandatory)
225 # ===========================================================================
226
227
228 class TestSecurity:
229 """Hardening contracts."""
230
231 def test_required_error_aborts_commit(self, tmp_path: pathlib.Path) -> None:
232 """AttestationRequiredError → exit ATTESTATION_REQUIRED, no commit written."""
233 from muse.core.store import get_head_commit_id, read_current_branch
234 register_attestation_provider("knowtation", _RequiredStub())
235 _init_repo(tmp_path)
236 # Snapshot HEAD before — must not advance.
237 branch = read_current_branch(tmp_path)
238 head_before = get_head_commit_id(tmp_path, branch)
239 with pytest.raises(SystemExit) as exc:
240 _commit(tmp_path, attest=True)
241 assert exc.value.code == ExitCode.ATTESTATION_REQUIRED
242 head_after = get_head_commit_id(tmp_path, branch)
243 assert head_after == head_before # HEAD unchanged: commit aborted.
244
245 def test_soft_failure_does_not_abort(self, tmp_path: pathlib.Path) -> None:
246 """Non-required AttestationError → commit succeeds, no record attached."""
247 from muse.core.store import read_commit
248 register_attestation_provider("knowtation", _SoftFailStub())
249 _init_repo(tmp_path)
250 result = _commit(tmp_path, attest=True)
251 commit_record = read_commit(tmp_path, result["commit_id"])
252 assert commit_record is not None
253 assert "attestation" not in (commit_record.metadata or {})
254
255 def test_attestation_record_canonical_json(self, tmp_path: pathlib.Path) -> None:
256 """The stored attestation blob is sort_keys=True compact JSON — no NaN."""
257 from muse.core.store import read_commit
258 register_attestation_provider("knowtation", _OkStub())
259 _init_repo(tmp_path)
260 result = _commit(tmp_path, attest=True)
261 commit_record = read_commit(tmp_path, result["commit_id"])
262 assert commit_record is not None
263 blob = commit_record.metadata["attestation"]
264 # Re-encoding sort_keys must equal the original — round-trip stable.
265 roundtrip = json.dumps(json.loads(blob), sort_keys=True, separators=(",", ":"))
266 assert blob == roundtrip
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago