test_knowtation_attestation.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Tests for muse.plugins.knowtation.attestation — Phase 7.2. |
| 2 | |
| 3 | Rule #0 tier coverage: |
| 4 | |
| 5 | * unit — id derivation, inbox detection, request-body builder, |
| 6 | URL validator, config loader. |
| 7 | * integration — register provider, compute → anchor → verify with a |
| 8 | mocked endpoint. |
| 9 | * security — air.required hard rejection, inbox bypass, malformed |
| 10 | URL rejection, placeholder-sig is provably-bad, |
| 11 | missing HMAC key is hard error not silent True. |
| 12 | * data-integrity — cross-language parity: 200 fixtures produce identical |
| 13 | UTF-8 request bodies between Python and air.mjs (Node |
| 14 | subprocess; xfail when Node is unavailable). |
| 15 | |
| 16 | Stress / performance / e2e tiers omitted — this layer is a thin port over |
| 17 | ``urllib`` with no async dispatch; the hot-path stress test belongs to |
| 18 | Phase 7.4 (the ICP push hook) where 100 concurrent calls actually exercise |
| 19 | something interesting. |
| 20 | """ |
| 21 | |
| 22 | from __future__ import annotations |
| 23 | |
| 24 | import http.server |
| 25 | import json |
| 26 | import os |
| 27 | import shutil |
| 28 | import subprocess |
| 29 | import threading |
| 30 | from typing import Any, cast |
| 31 | |
| 32 | import pytest |
| 33 | |
| 34 | from muse.core.attestation import ( |
| 35 | AnchorReceipt, |
| 36 | AttestationRequiredError, |
| 37 | AttestationVerifyError, |
| 38 | ImmutableReceiptError, |
| 39 | get_attestation_provider, |
| 40 | unregister_attestation_provider, |
| 41 | ) |
| 42 | from muse.plugins.knowtation.attestation import ( |
| 43 | ALGORITHM, |
| 44 | PLACEHOLDER_SIG, |
| 45 | PROVIDER_DOMAIN, |
| 46 | AirConfig, |
| 47 | KnowtationHmacAttestationProvider, |
| 48 | build_air_request_body, |
| 49 | canonical_record_bytes, |
| 50 | compute_attestation_id, |
| 51 | is_inbox_path, |
| 52 | load_air_config, |
| 53 | register_knowtation_attestation_provider, |
| 54 | ) |
| 55 | |
| 56 | |
| 57 | # --------------------------------------------------------------------------- |
| 58 | # Cleanup fixture — keep registry pristine. |
| 59 | # --------------------------------------------------------------------------- |
| 60 | |
| 61 | |
| 62 | @pytest.fixture(autouse=True) |
| 63 | def _clean_registry() -> None: |
| 64 | """Unregister the knowtation provider after each test.""" |
| 65 | yield |
| 66 | unregister_attestation_provider(PROVIDER_DOMAIN) |
| 67 | |
| 68 | |
| 69 | @pytest.fixture(autouse=True) |
| 70 | def _clean_env(monkeypatch: pytest.MonkeyPatch) -> None: |
| 71 | """Strip AIR-related env vars so tests are deterministic.""" |
| 72 | monkeypatch.delenv("KNOWTATION_AIR_ENDPOINT", raising=False) |
| 73 | monkeypatch.delenv("MUSE_AIR_HMAC_KEY", raising=False) |
| 74 | monkeypatch.delenv("MUSE_AIR_CONFIG", raising=False) |
| 75 | |
| 76 | |
| 77 | # =========================================================================== |
| 78 | # UNIT TIER |
| 79 | # =========================================================================== |
| 80 | |
| 81 | |
| 82 | class TestInboxDetection: |
| 83 | """Exact parity with isInboxPath() in air.mjs.""" |
| 84 | |
| 85 | @pytest.mark.parametrize( |
| 86 | "path,expected", |
| 87 | [ |
| 88 | ("inbox", True), |
| 89 | ("inbox/foo.md", True), |
| 90 | ("inbox/sub/bar.md", True), |
| 91 | ("projects/alpha/inbox", True), |
| 92 | ("projects/alpha/inbox/note.md", True), |
| 93 | ("projects/alpha-beta/inbox/x", True), |
| 94 | ("notes/inbox.md", False), |
| 95 | ("projects/inbox/note.md", False), |
| 96 | ("projects/alpha/sub/inbox/note.md", False), |
| 97 | ("inbox-archive/x", False), |
| 98 | ("", False), |
| 99 | ], |
| 100 | ) |
| 101 | def test_classification(self, path: str, expected: bool) -> None: |
| 102 | """Direct verbatim parity with the JS regex.""" |
| 103 | assert is_inbox_path(path) is expected |
| 104 | |
| 105 | def test_backslash_normalisation(self) -> None: |
| 106 | """Windows paths are normalised before matching.""" |
| 107 | assert is_inbox_path("inbox\\foo.md") is True |
| 108 | assert is_inbox_path("projects\\alpha\\inbox\\x") is True |
| 109 | |
| 110 | |
| 111 | class TestIdDerivation: |
| 112 | """compute_attestation_id is deterministic and unique.""" |
| 113 | |
| 114 | def test_deterministic(self) -> None: |
| 115 | """Same inputs → same id, every call.""" |
| 116 | a = compute_attestation_id("snap-1", "notes/x.md") |
| 117 | b = compute_attestation_id("snap-1", "notes/x.md") |
| 118 | assert a == b |
| 119 | assert len(a) == 64 |
| 120 | assert a.islower() |
| 121 | |
| 122 | def test_distinct_for_different_inputs(self) -> None: |
| 123 | """Distinct (snapshot,path) pairs collide with negligible probability.""" |
| 124 | ids = { |
| 125 | compute_attestation_id("snap-1", "notes/x.md"), |
| 126 | compute_attestation_id("snap-2", "notes/x.md"), |
| 127 | compute_attestation_id("snap-1", "notes/y.md"), |
| 128 | } |
| 129 | assert len(ids) == 3 |
| 130 | |
| 131 | def test_separator_injection_resistance(self) -> None: |
| 132 | """Null-byte separator means concatenation aliases cannot collide.""" |
| 133 | with pytest.raises(ValueError): |
| 134 | compute_attestation_id("snap\x00inj", "notes/x.md") |
| 135 | |
| 136 | def test_rejects_non_strings(self) -> None: |
| 137 | """Non-string args are rejected at the boundary.""" |
| 138 | with pytest.raises(ValueError): |
| 139 | compute_attestation_id(cast(str, 1), "x") # type: ignore[arg-type] |
| 140 | |
| 141 | |
| 142 | class TestRequestBodyBuilder: |
| 143 | """build_air_request_body produces canonical JSON.""" |
| 144 | |
| 145 | def test_keys_in_alphabetical_order(self) -> None: |
| 146 | """JSON keys appear in canonical alphabetical order.""" |
| 147 | body = build_air_request_body("write", "notes/x.md", "abc") |
| 148 | # Decoded text order is what air.mjs (and any strict JSON consumer) |
| 149 | # observes; assert the *string* form keeps the canonical order. |
| 150 | text = body.decode("utf-8") |
| 151 | assert text == '{"action":"write","content_hash":"abc","path":"notes/x.md"}' |
| 152 | |
| 153 | def test_unicode_path_passthrough(self) -> None: |
| 154 | """Unicode paths survive round-trip without escaping.""" |
| 155 | body = build_air_request_body("write", "notes/résumé.md", "abc") |
| 156 | decoded = json.loads(body) |
| 157 | assert decoded == {"action": "write", "content_hash": "abc", "path": "notes/résumé.md"} |
| 158 | |
| 159 | def test_rejects_null_byte(self) -> None: |
| 160 | """Null-byte values would break canonical encoding — rejected.""" |
| 161 | with pytest.raises(ValueError, match="null"): |
| 162 | build_air_request_body("write", "x\x00y", "abc") |
| 163 | |
| 164 | def test_rejects_non_string(self) -> None: |
| 165 | """Non-string args are rejected at the boundary.""" |
| 166 | with pytest.raises(ValueError): |
| 167 | build_air_request_body(cast(str, 1), "x", "abc") # type: ignore[arg-type] |
| 168 | |
| 169 | |
| 170 | class TestCanonicalRecordBytes: |
| 171 | """canonical_record_bytes is the input to HMAC sign/verify.""" |
| 172 | |
| 173 | def test_includes_required_fields(self) -> None: |
| 174 | """All five canonical fields appear, null-byte separated, correct order.""" |
| 175 | record = { |
| 176 | "id": "abc" * 21 + "x", |
| 177 | "action": "write", |
| 178 | "path": "notes/x.md", |
| 179 | "content_hash": "deadbeef", |
| 180 | "timestamp": "2026-01-01T00:00:00Z", |
| 181 | "sig": "should-not-appear", |
| 182 | "algorithm": ALGORITHM, |
| 183 | "provider_id": PROVIDER_DOMAIN, |
| 184 | } |
| 185 | out = canonical_record_bytes(cast(Any, record)) |
| 186 | decoded = out.decode("utf-8") |
| 187 | parts = decoded.split("\x00") |
| 188 | assert parts == ["write", "notes/x.md", "deadbeef", "2026-01-01T00:00:00Z", record["id"]] |
| 189 | assert "should-not-appear" not in decoded |
| 190 | |
| 191 | |
| 192 | # =========================================================================== |
| 193 | # INTEGRATION TIER |
| 194 | # =========================================================================== |
| 195 | |
| 196 | |
| 197 | class _LocalAir: |
| 198 | """Tiny in-process HTTP server used by integration tests.""" |
| 199 | |
| 200 | def __init__(self, response: dict[str, Any] | None = None, status: int = 200) -> None: |
| 201 | self.response = response or {"id": "ignored", "air_sig": "deadbeef-sig"} |
| 202 | self.status = status |
| 203 | self.requests: list[bytes] = [] |
| 204 | outer = self |
| 205 | |
| 206 | class _Handler(http.server.BaseHTTPRequestHandler): |
| 207 | def do_POST(self) -> None: # noqa: N802 — stdlib name |
| 208 | length = int(self.headers.get("Content-Length", "0")) |
| 209 | outer.requests.append(self.rfile.read(length)) |
| 210 | self.send_response(outer.status) |
| 211 | self.send_header("Content-Type", "application/json") |
| 212 | body = json.dumps(outer.response).encode("utf-8") |
| 213 | self.send_header("Content-Length", str(len(body))) |
| 214 | self.end_headers() |
| 215 | self.wfile.write(body) |
| 216 | |
| 217 | def log_message(self, fmt: str, *args: Any) -> None: # noqa: ARG002 |
| 218 | return None |
| 219 | |
| 220 | self._server = http.server.HTTPServer(("127.0.0.1", 0), _Handler) |
| 221 | self.endpoint = f"http://127.0.0.1:{self._server.server_address[1]}" |
| 222 | self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) |
| 223 | self._thread.start() |
| 224 | |
| 225 | def stop(self) -> None: |
| 226 | self._server.shutdown() |
| 227 | self._server.server_close() |
| 228 | self._thread.join(timeout=2.0) |
| 229 | |
| 230 | |
| 231 | @pytest.fixture |
| 232 | def local_air() -> Any: |
| 233 | """Local AIR endpoint that returns a real signature.""" |
| 234 | server = _LocalAir() |
| 235 | yield server |
| 236 | server.stop() |
| 237 | |
| 238 | |
| 239 | class TestRoundTrip: |
| 240 | """register → compute → anchor → verify with a real local endpoint.""" |
| 241 | |
| 242 | def test_compute_anchor_returns_valid_record(self, local_air: _LocalAir) -> None: |
| 243 | """Happy path: configured endpoint yields a non-placeholder record.""" |
| 244 | cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} |
| 245 | provider = register_knowtation_attestation_provider(cfg) |
| 246 | |
| 247 | record = provider.compute_attestation( |
| 248 | "snap-abc", {"path": "notes/x.md", "content_hash": "deadbeef"} |
| 249 | ) |
| 250 | assert record["id"] == compute_attestation_id("snap-abc", "notes/x.md") |
| 251 | assert record["sig"] == "deadbeef-sig" |
| 252 | assert record["algorithm"] == ALGORITHM |
| 253 | assert record["provider_id"] == PROVIDER_DOMAIN |
| 254 | |
| 255 | receipt = provider.anchor(record) |
| 256 | assert receipt["tx_id"] == f"air-local:{record['id']}" |
| 257 | assert receipt["provider_id"] == PROVIDER_DOMAIN |
| 258 | |
| 259 | def test_request_body_is_canonical(self, local_air: _LocalAir) -> None: |
| 260 | """Endpoint receives exactly the canonical body shape.""" |
| 261 | cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} |
| 262 | provider = KnowtationHmacAttestationProvider(cfg) |
| 263 | provider.compute_attestation( |
| 264 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 265 | ) |
| 266 | assert local_air.requests |
| 267 | body = local_air.requests[0].decode("utf-8") |
| 268 | assert body == '{"action":"write","content_hash":"abc","path":"notes/x.md"}' |
| 269 | |
| 270 | def test_get_via_registry(self) -> None: |
| 271 | """register_knowtation_attestation_provider plumbs into the registry.""" |
| 272 | provider = register_knowtation_attestation_provider({"enabled": False}) |
| 273 | assert get_attestation_provider(PROVIDER_DOMAIN) is provider |
| 274 | |
| 275 | |
| 276 | class TestVerifyWithKey: |
| 277 | """Round-trip verify when both sides know the HMAC key.""" |
| 278 | |
| 279 | def test_local_hmac_verify_round_trip(self, monkeypatch: pytest.MonkeyPatch) -> None: |
| 280 | """Sign with key → verify with same key → valid=True.""" |
| 281 | # Use a known key, sign locally, then verify. |
| 282 | key_hex = "00" * 32 |
| 283 | provider = KnowtationHmacAttestationProvider({"hmac_key": key_hex}) |
| 284 | record = provider.compute_attestation( |
| 285 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 286 | ) |
| 287 | # Build a synthetic record with a real HMAC sig so verify succeeds. |
| 288 | from muse.plugins.knowtation.attestation import ( |
| 289 | _hmac_sha256_hex, |
| 290 | canonical_record_bytes, |
| 291 | ) |
| 292 | record["timestamp"] = "2026-01-01T00:00:00Z" |
| 293 | sig = _hmac_sha256_hex(key_hex, canonical_record_bytes(record)) |
| 294 | record["sig"] = sig |
| 295 | record["algorithm"] = ALGORITHM |
| 296 | receipt = provider.anchor(record) |
| 297 | result = provider.verify(record, receipt) |
| 298 | assert result["valid"] is True |
| 299 | assert result["depth"] == 0 |
| 300 | |
| 301 | def test_tampered_record_returns_valid_false(self, monkeypatch: pytest.MonkeyPatch) -> None: |
| 302 | """Mutating any HMAC-bound field after signing flips valid to False.""" |
| 303 | key_hex = "00" * 32 |
| 304 | provider = KnowtationHmacAttestationProvider({"hmac_key": key_hex}) |
| 305 | record = provider.compute_attestation( |
| 306 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 307 | ) |
| 308 | from muse.plugins.knowtation.attestation import ( |
| 309 | _hmac_sha256_hex, |
| 310 | canonical_record_bytes, |
| 311 | ) |
| 312 | record["timestamp"] = "2026-01-01T00:00:00Z" |
| 313 | sig = _hmac_sha256_hex(key_hex, canonical_record_bytes(record)) |
| 314 | record["sig"] = sig |
| 315 | receipt = provider.anchor(record) |
| 316 | record["content_hash"] = "tampered-hash" |
| 317 | result = provider.verify(record, receipt) |
| 318 | assert result["valid"] is False |
| 319 | assert "HMAC" in result["reason"] |
| 320 | |
| 321 | |
| 322 | # =========================================================================== |
| 323 | # SECURITY TIER (mandatory) |
| 324 | # =========================================================================== |
| 325 | |
| 326 | |
| 327 | class TestSecurity: |
| 328 | """Hardening contracts — Rule #0 mandates this tier for Phase 7.""" |
| 329 | |
| 330 | def test_inbox_bypass_skips_endpoint(self, local_air: _LocalAir) -> None: |
| 331 | """Inbox paths never call the AIR endpoint.""" |
| 332 | cfg: AirConfig = {"enabled": True, "required": True, "endpoint": local_air.endpoint} |
| 333 | provider = KnowtationHmacAttestationProvider(cfg) |
| 334 | record = provider.compute_attestation( |
| 335 | "snap", {"path": "inbox/note.md", "content_hash": "x"} |
| 336 | ) |
| 337 | assert record["algorithm"] == "bypass-inbox" |
| 338 | assert record["sig"] == "" |
| 339 | assert local_air.requests == [] |
| 340 | |
| 341 | def test_required_true_endpoint_500_raises(self) -> None: |
| 342 | """500 from endpoint with required=True → AttestationRequiredError.""" |
| 343 | server = _LocalAir(status=500) |
| 344 | try: |
| 345 | cfg: AirConfig = {"enabled": True, "required": True, "endpoint": server.endpoint} |
| 346 | provider = KnowtationHmacAttestationProvider(cfg) |
| 347 | with pytest.raises(AttestationRequiredError): |
| 348 | provider.compute_attestation( |
| 349 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 350 | ) |
| 351 | finally: |
| 352 | server.stop() |
| 353 | |
| 354 | def test_required_false_endpoint_500_returns_placeholder(self) -> None: |
| 355 | """500 from endpoint with required=False → placeholder, no raise.""" |
| 356 | server = _LocalAir(status=500) |
| 357 | try: |
| 358 | cfg: AirConfig = {"enabled": True, "required": False, "endpoint": server.endpoint} |
| 359 | provider = KnowtationHmacAttestationProvider(cfg) |
| 360 | record = provider.compute_attestation( |
| 361 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 362 | ) |
| 363 | assert record["sig"] == PLACEHOLDER_SIG |
| 364 | assert record["algorithm"].startswith("placeholder-") |
| 365 | finally: |
| 366 | server.stop() |
| 367 | |
| 368 | def test_required_true_no_endpoint_raises(self) -> None: |
| 369 | """No endpoint configured + required=True → AttestationRequiredError.""" |
| 370 | cfg: AirConfig = {"enabled": True, "required": True} |
| 371 | provider = KnowtationHmacAttestationProvider(cfg) |
| 372 | with pytest.raises(AttestationRequiredError): |
| 373 | provider.compute_attestation( |
| 374 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 375 | ) |
| 376 | |
| 377 | def test_malformed_endpoint_url_raises(self) -> None: |
| 378 | """file:// and other non-http(s) schemes are rejected — no SSRF.""" |
| 379 | cfg: AirConfig = {"enabled": True, "required": True, "endpoint": "file:///etc/passwd"} |
| 380 | provider = KnowtationHmacAttestationProvider(cfg) |
| 381 | with pytest.raises(AttestationRequiredError): |
| 382 | provider.compute_attestation( |
| 383 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 384 | ) |
| 385 | |
| 386 | def test_anchor_is_write_once(self, local_air: _LocalAir) -> None: |
| 387 | """Re-anchoring the same record raises ImmutableReceiptError.""" |
| 388 | cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} |
| 389 | provider = KnowtationHmacAttestationProvider(cfg) |
| 390 | record = provider.compute_attestation( |
| 391 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 392 | ) |
| 393 | provider.anchor(record) |
| 394 | with pytest.raises(ImmutableReceiptError): |
| 395 | provider.anchor(record) |
| 396 | |
| 397 | def test_verify_placeholder_returns_valid_false_not_true(self) -> None: |
| 398 | """Placeholder sig is provably-bad: valid=False, never silent True.""" |
| 399 | provider = KnowtationHmacAttestationProvider({}) |
| 400 | record = provider.compute_attestation( |
| 401 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 402 | ) |
| 403 | receipt = provider.anchor(record) |
| 404 | result = provider.verify(record, receipt) |
| 405 | assert result["valid"] is False |
| 406 | assert "placeholder" in result["reason"].lower() |
| 407 | |
| 408 | def test_verify_missing_hmac_key_raises_not_returns( |
| 409 | self, monkeypatch: pytest.MonkeyPatch, local_air: _LocalAir |
| 410 | ) -> None: |
| 411 | """Missing HMAC key → AttestationVerifyError, never silent True.""" |
| 412 | cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} |
| 413 | provider = KnowtationHmacAttestationProvider(cfg) |
| 414 | record = provider.compute_attestation( |
| 415 | "snap", {"path": "notes/x.md", "content_hash": "abc"} |
| 416 | ) |
| 417 | receipt = provider.anchor(record) |
| 418 | with pytest.raises(AttestationVerifyError) as exc: |
| 419 | provider.verify(record, receipt) |
| 420 | assert exc.value.code == "KEY_MISSING" |
| 421 | |
| 422 | def test_verify_inbox_record_returns_valid_false(self) -> None: |
| 423 | """Inbox-bypass records cannot be verified — valid=False, never True.""" |
| 424 | provider = KnowtationHmacAttestationProvider({"enabled": True}) |
| 425 | record = provider.compute_attestation( |
| 426 | "snap", {"path": "inbox/x.md", "content_hash": "abc"} |
| 427 | ) |
| 428 | receipt = provider.anchor(record) |
| 429 | result = provider.verify(record, receipt) |
| 430 | assert result["valid"] is False |
| 431 | |
| 432 | def test_verify_provider_id_mismatch_raises(self) -> None: |
| 433 | """Records from a different provider raise, never return valid=True.""" |
| 434 | provider = KnowtationHmacAttestationProvider({"hmac_key": "00" * 32}) |
| 435 | evil_record: Any = { |
| 436 | "id": "a" * 64, |
| 437 | "action": "write", |
| 438 | "path": "notes/x.md", |
| 439 | "content_hash": "x", |
| 440 | "timestamp": "2026-01-01T00:00:00Z", |
| 441 | "sig": "any", |
| 442 | "algorithm": ALGORITHM, |
| 443 | "provider_id": "evil-domain", |
| 444 | } |
| 445 | evil_receipt: Any = { |
| 446 | "tx_id": "x", "anchor_url": "https://evil.example/x", |
| 447 | "anchored_at": "2026-01-01T00:00:00Z", "provider_id": "evil-domain", |
| 448 | } |
| 449 | with pytest.raises(AttestationVerifyError) as exc: |
| 450 | provider.verify(evil_record, evil_receipt) |
| 451 | assert exc.value.code == "PROVIDER_MISMATCH" |
| 452 | |
| 453 | |
| 454 | # =========================================================================== |
| 455 | # DATA-INTEGRITY TIER — cross-language parity with air.mjs |
| 456 | # =========================================================================== |
| 457 | |
| 458 | |
| 459 | _NODE = shutil.which("node") |
| 460 | _AIR_MJS = "/Users/aaronrenecarvajal/knowtation/lib/air.mjs" |
| 461 | |
| 462 | |
| 463 | def _node_build_body(action: str, path: str, content_hash: str) -> bytes: |
| 464 | """Invoke node to call buildAirRequestBody — returns the same bytes Python should.""" |
| 465 | script = ( |
| 466 | f"import('{_AIR_MJS}').then(m => " |
| 467 | "{ " |
| 468 | f"const out = m.buildAirRequestBody({json.dumps(action)}, " |
| 469 | f"{json.dumps(path)}, {json.dumps(content_hash)}); " |
| 470 | "process.stdout.write(out); " |
| 471 | "}).catch(e => { console.error(e); process.exit(2); });" |
| 472 | ) |
| 473 | proc = subprocess.run( |
| 474 | [_NODE, "--input-type=module", "-e", script], |
| 475 | capture_output=True, |
| 476 | timeout=10, |
| 477 | check=False, |
| 478 | ) |
| 479 | if proc.returncode != 0: |
| 480 | raise RuntimeError(f"node failed: {proc.stderr.decode('utf-8', errors='replace')}") |
| 481 | return proc.stdout |
| 482 | |
| 483 | |
| 484 | @pytest.mark.skipif( |
| 485 | _NODE is None or not os.path.exists(_AIR_MJS), |
| 486 | reason="Node or air.mjs unavailable — cross-language parity test gated", |
| 487 | ) |
| 488 | class TestCrossLanguageParity: |
| 489 | """200 fixture inputs produce identical Python and JS request bodies.""" |
| 490 | |
| 491 | def _fixtures(self) -> list[tuple[str, str, str]]: |
| 492 | """Generate 200 deterministic (action, path, content_hash) fixtures.""" |
| 493 | actions = ["write", "export", "import", "consolidation"] |
| 494 | paths = [ |
| 495 | "notes/a.md", |
| 496 | "projects/alpha/note.md", |
| 497 | "deep/nested/path/with-dashes_underscores.md", |
| 498 | "unicode/résumé.md", |
| 499 | "spaces in name.md", |
| 500 | "", |
| 501 | "single", |
| 502 | ] |
| 503 | out: list[tuple[str, str, str]] = [] |
| 504 | i = 0 |
| 505 | while len(out) < 200: |
| 506 | action = actions[i % len(actions)] |
| 507 | path = paths[(i // len(actions)) % len(paths)] |
| 508 | ch = f"{i:064x}" |
| 509 | out.append((action, path, ch)) |
| 510 | i += 1 |
| 511 | return out |
| 512 | |
| 513 | def test_200_fixtures_byte_identical(self) -> None: |
| 514 | """For every fixture, Python and JS produce the same UTF-8 bytes.""" |
| 515 | mismatches: list[tuple[tuple[str, str, str], bytes, bytes]] = [] |
| 516 | for fix in self._fixtures(): |
| 517 | py = build_air_request_body(*fix) |
| 518 | js = _node_build_body(*fix) |
| 519 | if py != js: |
| 520 | mismatches.append((fix, py, js)) |
| 521 | assert not mismatches, ( |
| 522 | f"{len(mismatches)} fixtures produced different bytes; " |
| 523 | f"first: {mismatches[0]!r}" |
| 524 | ) |
| 525 | |
| 526 | |
| 527 | # =========================================================================== |
| 528 | # CONFIG LOADER |
| 529 | # =========================================================================== |
| 530 | |
| 531 | |
| 532 | class TestConfigLoader: |
| 533 | """load_air_config reads the YAML and respects env-var fallbacks.""" |
| 534 | |
| 535 | def test_missing_file_returns_empty(self, tmp_path: Any) -> None: |
| 536 | """Missing file → empty AirConfig, no exception.""" |
| 537 | cfg = load_air_config(tmp_path / "nope.yaml") |
| 538 | assert cfg == {} |
| 539 | |
| 540 | def test_yaml_parsed(self, tmp_path: Any) -> None: |
| 541 | """Valid YAML is parsed into the AirConfig fields.""" |
| 542 | path = tmp_path / "air.yaml" |
| 543 | path.write_text( |
| 544 | "air:\n enabled: true\n required: false\n endpoint: https://x.example/air\n", |
| 545 | encoding="utf-8", |
| 546 | ) |
| 547 | cfg = load_air_config(path) |
| 548 | assert cfg["enabled"] is True |
| 549 | assert cfg["required"] is False |
| 550 | assert cfg["endpoint"] == "https://x.example/air" |
| 551 | |
| 552 | def test_env_var_path_used(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None: |
| 553 | """MUSE_AIR_CONFIG env var overrides the default path.""" |
| 554 | path = tmp_path / "custom.yaml" |
| 555 | path.write_text("air:\n enabled: true\n", encoding="utf-8") |
| 556 | monkeypatch.setenv("MUSE_AIR_CONFIG", str(path)) |
| 557 | cfg = load_air_config() |
| 558 | assert cfg["enabled"] is True |
| 559 | |
| 560 | def test_malformed_root_raises(self, tmp_path: Any) -> None: |
| 561 | """Top-level YAML not a mapping → AttestationError.""" |
| 562 | from muse.core.attestation import AttestationError |
| 563 | path = tmp_path / "bad.yaml" |
| 564 | path.write_text("- a\n- b\n", encoding="utf-8") |
| 565 | with pytest.raises(AttestationError): |
| 566 | load_air_config(path) |
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