tegwick
84beefc406
Founder run 4 merged 27 accepted descriptions. Two failed with HTTP 400: used_by was 514 bytes because clipping counted characters and appended a 3-byte ellipsis. Clip and check now count bytes; manifest re-clipped. 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
81 lines
3.8 KiB
Python
81 lines
3.8 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
import pwd
|
|
import stat
|
|
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 _r(rc=0, out="", err=""):
|
|
return MagicMock(returncode=rc, stdout=out, stderr=err)
|
|
|
|
|
|
def test_apply_patches_only_accepted_complete_entries_silently(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=_r()) as run:
|
|
assert cd.apply(doc, receipt=str(tmp_path / "r.json")) == 0
|
|
assert capsys.readouterr() == ("", "")
|
|
assert run.call_count == 1
|
|
args = run.call_args.args[0]
|
|
assert args[:5] == ["bao", "kv", "metadata", "patch", "-mount=platform"] and args[-1] == "workloads/a/b"
|
|
assert not any(a in ("get", "put") or "/data/" in a for a in args)
|
|
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
|
|
|
|
|
|
def test_patch_refused_falls_back_to_merge_keeping_existing_metadata(tmp_path):
|
|
existing = json.dumps({"data": {"custom_metadata": {"ccr": "CCR-2026-0033", "owner": "old"}}})
|
|
calls = [_r(2, err="Error ... Code: 403. Errors: permission denied"), _r(0, out=existing), _r(0)]
|
|
doc = {"platform/workloads/a/b": {"review": "accepted", **FULL}}
|
|
with patch.object(cd.subprocess, "run", side_effect=calls) as run:
|
|
assert cd.apply(doc, receipt=str(tmp_path / "r.json")) == 0
|
|
get_args, put_args = run.call_args_list[1].args[0], run.call_args_list[2].args[0]
|
|
assert get_args[1:4] == ["kv", "metadata", "get"]
|
|
assert put_args[1:4] == ["kv", "metadata", "put"]
|
|
assert "-custom-metadata=ccr=CCR-2026-0033" in put_args and "-custom-metadata=owner=o" in put_args
|
|
assert json.load(open(tmp_path / "r.json")) == {"platform/workloads/a/b": "merged"}
|
|
|
|
|
|
def test_failure_records_only_http_code(tmp_path, capsys):
|
|
doc = {"platform/workloads/a/b": {"review": "accepted", **FULL}}
|
|
with patch.object(cd.subprocess, "run", return_value=_r(2, err="Code: 500. secret-ish detail")):
|
|
assert cd.apply(doc, receipt=str(tmp_path / "r.json")) == 1
|
|
assert capsys.readouterr() == ("", "")
|
|
assert json.load(open(tmp_path / "r.json")) == {"platform/workloads/a/b": "failed patch http=500"}
|
|
|
|
|
|
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_byte_limit():
|
|
assert len(cd._clip("y" * 2000).encode()) <= cd.MAX_VALUE
|
|
assert len(cd._clip("→" * 400).encode()) <= cd.MAX_VALUE
|
|
assert cd.check({"p/x": {"review": "accepted", **dict(FULL, used_by="y" * 510 + "…")}}) == 1
|
|
|
|
|
|
def test_receipt_ignores_temporary_home_and_bao_addr_defaults(tmp_path, monkeypatch):
|
|
assert cd.RECEIPT.startswith(pwd.getpwuid(os.getuid()).pw_dir)
|
|
monkeypatch.delenv("BAO_ADDR", raising=False)
|
|
doc = {"platform/workloads/a/b": {"review": "accepted", **FULL}}
|
|
with patch.object(cd.subprocess, "run", return_value=_r()) as run:
|
|
cd.apply(doc, receipt=str(tmp_path / "r.json"))
|
|
assert run.call_args.kwargs["env"]["BAO_ADDR"] == "http://127.0.0.1:18200"
|