tegwick
7d54d049c1
The warden admin wrapper refuses any output; the first founder run was refused (nothing written, all entries still drafts). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 327203@bnt-lap001 Assistant-Session: 55028ae3-4914-4bd1-8f5c-815f3527aef0
51 lines
2.2 KiB
Python
51 lines
2.2 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
_spec = importlib.util.spec_from_file_location("custody_describe", Path(__file__).parents[1] / "scripts" / "custody-describe.py")
|
|
cd = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(cd)
|
|
|
|
FULL = dict(description="d", owner="o", used_by="u", rotation="r", on_loss="l")
|
|
|
|
|
|
def test_apply_patches_only_accepted_complete_entries_and_only_metadata(tmp_path, capsys):
|
|
doc = {
|
|
"platform/workloads/a/b": {"review": "accepted", **FULL},
|
|
"platform/workloads/c/d": {"review": "draft", **FULL},
|
|
"operators/e": {"review": "accepted", **dict(FULL, used_by="")},
|
|
}
|
|
with patch.object(cd.subprocess, "run", return_value=MagicMock(returncode=0)) as run:
|
|
assert cd.apply(doc, receipt=str(tmp_path / 'r.json')) == 0
|
|
assert capsys.readouterr() == ('', '')
|
|
import json, os, stat
|
|
r = tmp_path / 'r.json'
|
|
assert json.load(open(r)) == {'platform/workloads/a/b': 'patched', 'platform/workloads/c/d': 'skipped', 'operators/e': 'skipped'}
|
|
assert stat.S_IMODE(os.stat(r).st_mode) == 0o600
|
|
assert run.call_count == 1
|
|
args = run.call_args.args[0]
|
|
assert args[:5] == ["bao", "kv", "metadata", "patch", "-mount=platform"]
|
|
assert args[-1] == "workloads/a/b"
|
|
assert not any(a in ("get", "put", "patch-data") or "/data/" in a for a in args)
|
|
assert run.call_args.kwargs["stdout"] is cd.subprocess.DEVNULL and run.call_args.kwargs["stderr"] is cd.subprocess.DEVNULL
|
|
|
|
|
|
def test_apply_reports_failure_without_output(tmp_path, capsys):
|
|
doc = {"platform/workloads/a/b": {"review": "accepted", **FULL}}
|
|
with patch.object(cd.subprocess, "run", return_value=MagicMock(returncode=2)):
|
|
assert cd.apply(doc, receipt=str(tmp_path / 'r.json')) == 1
|
|
assert capsys.readouterr() == ('', '')
|
|
|
|
|
|
def test_check_refuses_drafts_missing_and_overlong_fields():
|
|
doc = {
|
|
"p/ok": {"review": "accepted", **FULL},
|
|
"p/draft": {"review": "draft", **FULL},
|
|
"p/long": {"review": "accepted", **dict(FULL, on_loss="x" * 600)},
|
|
}
|
|
assert cd.check(doc) == 1
|
|
assert cd.check({"p/ok": doc["p/ok"]}) == 0
|
|
|
|
|
|
def test_clip_respects_openbao_value_limit():
|
|
assert len(cd._clip("y" * 2000)) == cd.MAX_VALUE
|