Seed tenant-engine sender token into its dedicated KV path (MASON-WP-0007 row 7)
Attended, silent, check-and-set. Evidence 2026-10-02: created, version 1, no values. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: sonnet Assistant-Process: 292700@bnt-lap001 Assistant-Session: 42de6b79-9019-4cfd-b49a-6db3bbfe9341
This commit is contained in:
parent
f0dff91eb5
commit
e7ee6d0694
4 changed files with 308 additions and 0 deletions
103
tests/test_seed_tenant_engine_audit_core_sender.py
Normal file
103
tests/test_seed_tenant_engine_audit_core_sender.py
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
"""Seed script for the tenant-engine sender token (MASON-WP-0007 row 7).
|
||||
|
||||
Uses a fake OpenBao and obviously fake token strings. No real credential.
|
||||
"""
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
SCRIPT = Path(__file__).parents[1] / "scripts" / "seed-tenant-engine-audit-core-sender.py"
|
||||
SPEC = importlib.util.spec_from_file_location("seed_tenant_engine_sender", SCRIPT)
|
||||
SEED = importlib.util.module_from_spec(SPEC)
|
||||
sys.modules[SPEC.name] = SEED
|
||||
SPEC.loader.exec_module(SEED)
|
||||
|
||||
FAKE_OLD = "fake-token-old-not-a-secret"
|
||||
FAKE_NEW = "fake-token-new-not-a-secret"
|
||||
|
||||
|
||||
class FakeBao:
|
||||
"""Minimal KV v2 behaviour: cas enforcement, versions, NotFound."""
|
||||
|
||||
def __init__(self, existing=None):
|
||||
self.version = 0
|
||||
self.value = None
|
||||
self.writes = []
|
||||
if existing is not None:
|
||||
self.version, self.value = 1, existing
|
||||
|
||||
def __call__(self, args, *, payload=None):
|
||||
path = args[-2] if args[0] == "write" else args[-1]
|
||||
assert path == SEED.DEST_DATA
|
||||
if args[0] == "read":
|
||||
if self.version == 0:
|
||||
raise SEED.NotFound()
|
||||
return {"data": {"metadata": {"version": self.version}, "data": {"token": self.value}}}
|
||||
assert args[0] == "write"
|
||||
assert payload["options"]["cas"] == self.version, "cas must match the version read"
|
||||
self.writes.append(payload)
|
||||
self.version += 1
|
||||
self.value = payload["data"]["token"]
|
||||
return {}
|
||||
|
||||
|
||||
def test_pick_token_takes_newest_and_rejects_missing():
|
||||
rows = [
|
||||
{"name": "other", "tokens": ["x"]},
|
||||
{"name": "tenant-engine", "tokens": [FAKE_OLD, FAKE_NEW]},
|
||||
]
|
||||
assert SEED.pick_token(rows) == FAKE_NEW
|
||||
with pytest.raises(ValueError, match="sender_not_in_registry"):
|
||||
SEED.pick_token([{"name": "other", "tokens": ["x"]}])
|
||||
with pytest.raises(ValueError, match="sender_has_no_tokens"):
|
||||
SEED.pick_token([{"name": "tenant-engine", "tokens": []}])
|
||||
|
||||
|
||||
def test_creates_with_cas_zero():
|
||||
fake = FakeBao()
|
||||
result = SEED.seed(FAKE_NEW, call=fake)
|
||||
assert result == {"action": "created", "version": 1}
|
||||
assert fake.writes[0]["options"]["cas"] == 0
|
||||
assert fake.writes[0]["data"] == {"token": FAKE_NEW}
|
||||
|
||||
|
||||
def test_rerun_with_same_value_does_not_write():
|
||||
fake = FakeBao(existing=FAKE_NEW)
|
||||
assert SEED.seed(FAKE_NEW, call=fake) == {"action": "already_seeded", "version": 1}
|
||||
assert fake.writes == []
|
||||
|
||||
|
||||
def test_rotation_reseeds_against_current_version():
|
||||
fake = FakeBao(existing=FAKE_OLD)
|
||||
assert SEED.seed(FAKE_NEW, call=fake) == {"action": "reseeded", "version": 2}
|
||||
assert fake.value == FAKE_NEW
|
||||
|
||||
|
||||
def test_readback_mismatch_fails():
|
||||
class Lossy(FakeBao):
|
||||
def __call__(self, args, *, payload=None):
|
||||
out = super().__call__(args, payload=payload)
|
||||
if args[0] == "read" and self.writes:
|
||||
out["data"]["data"]["token"] = "something-else"
|
||||
return out
|
||||
|
||||
with pytest.raises(RuntimeError, match="readback_mismatch"):
|
||||
SEED.seed(FAKE_NEW, call=Lossy())
|
||||
|
||||
|
||||
def test_evidence_never_contains_a_token(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(SEED, "EVIDENCE_DIR", tmp_path)
|
||||
SEED._evidence("done", {"destination_version": 1, "action": "created", "field": "token"})
|
||||
SEED._fail("seed_failed", {"error_class": "RuntimeError"})
|
||||
text = "".join(p.read_text() for p in tmp_path.iterdir())
|
||||
assert FAKE_NEW not in text and FAKE_OLD not in text
|
||||
assert json.loads(next(tmp_path.iterdir()).read_text())["credential_values_emitted"] is False
|
||||
|
||||
|
||||
def test_source_never_prints_or_passes_the_token_on_a_command_line():
|
||||
src = SCRIPT.read_text()
|
||||
assert "print(" not in src
|
||||
assert "sys.stdout" not in src and "sys.stderr" not in src
|
||||
Loading…
Add table
Add a link
Reference in a new issue