from __future__ import annotations import argparse import importlib.util import os import shutil import sys import tempfile import unittest from pathlib import Path from unittest import mock REPO_DIR = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_DIR / "scripts")) SPEC = importlib.util.spec_from_file_location( "credential_change", REPO_DIR / "scripts/credential-change.py" ) credential_change = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None sys.modules[SPEC.name] = credential_change SPEC.loader.exec_module(credential_change) import ccr_intake # noqa: E402 cc = credential_change STANDARD = "CCR-2026-0051-mas-tem-relay-smtp.yaml" POLICY = "openbao/policies/workload-kv-read-mas-tem-relay.hcl" class CcrIntakeTests(unittest.TestCase): """Run against a scratch repo root so nothing is written into the real tree.""" def setUp(self) -> None: self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, True) self.ccrs = self.root / "credential-change-requests" self.ccrs.mkdir() (self.root / "openbao/policies").mkdir(parents=True) shutil.copy2(REPO_DIR / "credential-change-requests" / STANDARD, self.ccrs / STANDARD) shutil.copy2(REPO_DIR / POLICY, self.root / POLICY) for target in (cc, ccr_intake): patcher = mock.patch.object(target, "REPO_DIR", self.root) patcher.start() self.addCleanup(patcher.stop) env = mock.patch.dict(os.environ, {"CCR_DIR": str(self.ccrs)}) env.start() self.addCleanup(env.stop) self.profile, _digest = ccr_intake.load_profile() self.path = self.ccrs / STANDARD def edit(self, change) -> None: data = cc.load_yaml(self.path) change(data) cc.dump_yaml(self.path, data) def violations(self) -> list[str]: return ccr_intake.standard_violations(cc.load_yaml(self.path), self.profile, self.ccrs, cc) def assert_violation(self, change, fragment: str) -> None: self.edit(change) found = self.violations() self.assertTrue(any(fragment in item for item in found), found) def test_standard_lane_passes_and_records_review_without_approval(self) -> None: self.assertEqual(ccr_intake.review_standard(self.path, cc), []) data = cc.load_yaml(self.path) self.assertEqual(data["status"], "proposed") comment = data["review"]["comments"][-1] self.assertEqual(comment["decision"], "approved") self.assertIn("ccr-standard-lane v1 sha256:", comment["comment"]) self.assertTrue(comment["reviewer"].startswith("railiance-platform-owner")) self.assertIn("apply requires status approved, got proposed", cc.apply_blockers(data)) def test_check_mode_does_not_write(self) -> None: before = self.path.read_text() ccr_intake.review_standard(self.path, cc, write=False) self.assertEqual(self.path.read_text(), before) def test_out_of_profile_records_needs_platform_review(self) -> None: self.edit(lambda d: d["risk"].update(classification="high")) self.assertTrue(ccr_intake.review_standard(self.path, cc)) comment = cc.load_yaml(self.path)["review"]["comments"][-1] self.assertEqual(comment["decision"], "needs_platform_review") self.assertEqual(cc.load_yaml(self.path)["status"], "proposed") def test_requires_proposed_status(self) -> None: self.edit(lambda d: d.update(status="approved")) with self.assertRaises(SystemExit): ccr_intake.review_standard(self.path, cc) def test_metadata_read_rejected(self) -> None: self.assert_violation(lambda d: d["openbao"].update(metadata_read=True), "metadata_read") def test_kv_path_outside_prefixes_rejected(self) -> None: self.assert_violation( lambda d: d["openbao"].update(kv_path="platform/operators/founder/x"), "outside the standard" ) def test_supplier_path_needs_exactly_two_segments(self) -> None: self.assert_violation( lambda d: d["openbao"].update(kv_path="platform/suppliers/scaleway/a/b"), "outside the standard" ) def test_policy_file_drift_rejected(self) -> None: (self.root / POLICY).write_text('path "platform/data/suppliers/*" {\n capabilities = ["read"]\n}\n') self.assertTrue(any("differs" in v for v in self.violations())) def test_long_ttl_rejected(self) -> None: self.assert_violation(lambda d: d["openbao"]["auth"].update(token_max_ttl="1h"), "token_max_ttl") def test_default_policy_rejected(self) -> None: self.assert_violation( lambda d: d["openbao"]["auth"].update(token_no_default_policy=False), "token_no_default_policy" ) def test_role_prefix_and_audience_enforced(self) -> None: self.assert_violation(lambda d: d["openbao"]["auth"].update(role="mas-eso"), "role must start") self.assert_violation(lambda d: d["openbao"]["auth"].update(audience="vault"), "audience") def test_two_service_accounts_rejected(self) -> None: self.assert_violation( lambda d: d["openbao"]["auth"]["bound_claims"]["service_account_names"].append("other"), "service account", ) def test_reserved_namespace_rejected(self) -> None: self.assert_violation( lambda d: d["openbao"]["auth"]["bound_claims"].update(service_account_namespaces=["external-secrets"]), "reserved", ) def test_second_reader_of_path_rejected(self) -> None: other = cc.load_yaml(self.path) other.update(id="CCR-2026-0052", status="active") cc.dump_yaml(self.ccrs / "CCR-2026-0052-other.yaml", other) self.assertTrue(any("already has a reader: CCR-2026-0052" in v for v in self.violations())) def test_cancelled_reader_is_not_a_conflict(self) -> None: other = cc.load_yaml(self.path) other.update(id="CCR-2026-0052", status="cancelled") cc.dump_yaml(self.ccrs / "CCR-2026-0052-other.yaml", other) self.assertEqual(self.violations(), []) def test_next_id_and_duplicate_detection(self) -> None: self.assertEqual(ccr_intake.next_ccr_id(self.ccrs, 2026), "CCR-2026-0052") self.assertEqual(ccr_intake.next_ccr_id(self.ccrs, 2027), "CCR-2027-0001") self.assertEqual(ccr_intake.duplicate_ids(self.ccrs, cc), {}) shutil.copy2(self.path, self.ccrs / "CCR-2026-0051-race.yaml") self.assertIn("CCR-2026-0051", ccr_intake.duplicate_ids(self.ccrs, cc)) def submit_args(self, **overrides) -> argparse.Namespace: draft_dir = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, draft_dir, True) draft = cc.load_yaml(self.path) draft.update(id="CCR-DRAFT-MAS", status="draft") draft["openbao"].update( kv_path="platform/suppliers/scaleway/other-relay", policy_name="workload-kv-read-other-relay", policy_file="openbao/policies/workload-kv-read-other-relay.hcl", ) draft["openbao"]["auth"]["policies"] = ["workload-kv-read-other-relay"] draft["review"]["comments"] = [{"reviewer": "someone", "decision": "approved", "comment": "x"}] cc.dump_yaml(draft_dir / "draft.yaml", draft) (draft_dir / "policy.hcl").write_text( (self.root / POLICY).read_text().replace("mas-tem-relay", "other-relay") ) values = dict(ccr=str(draft_dir / "draft.yaml"), policy=str(draft_dir / "policy.hcl"), slug="other-relay", requester="ops-mason", profile=str(ccr_intake.DEFAULT_PROFILE)) values.update(overrides) return argparse.Namespace(**values) def test_submit_allocates_id_copies_policy_and_reviews(self) -> None: destination, violations = ccr_intake.submit(self.submit_args(), cc) self.assertEqual(violations, []) self.assertEqual(destination.name, "CCR-2026-0052-other-relay.yaml") data = cc.load_yaml(destination) self.assertEqual((data["id"], data["status"]), ("CCR-2026-0052", "proposed")) self.assertEqual(len(data["review"]["comments"]), 1) self.assertTrue(data["review"]["comments"][0]["reviewer"].startswith("railiance-platform-owner")) self.assertTrue((self.root / "openbao/policies/workload-kv-read-other-relay.hcl").exists()) def test_submit_refuses_unlisted_requester(self) -> None: with self.assertRaises(SystemExit): ccr_intake.submit(self.submit_args(requester="random-agent"), cc) def test_submit_refuses_policy_overwrite(self) -> None: (self.root / "openbao/policies/workload-kv-read-other-relay.hcl").write_text("different\n") with self.assertRaises(SystemExit): ccr_intake.submit(self.submit_args(), cc) def test_submit_refuses_secret_markers(self) -> None: args = self.submit_args() Path(args.ccr).write_text(Path(args.ccr).read_text() + "# hvs.notarealtoken\n") with self.assertRaises(SystemExit): ccr_intake.submit(args, cc) def submitted_branch(self) -> tuple[Path, str]: main_root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, main_root, True) shutil.copytree(self.root, main_root, dirs_exist_ok=True) destination, _violations = ccr_intake.submit(self.submit_args(), cc) return main_root, destination.stem def test_branch_gate_accepts_clean_submission(self) -> None: main_root, stem = self.submitted_branch() self.assertEqual(ccr_intake.branch_gate(main_root, stem, cc), []) def test_branch_gate_refuses_extra_changes(self) -> None: main_root, stem = self.submitted_branch() (self.root / "registry").mkdir() (self.root / "registry/ccr-standard-profile.yaml").write_text("submitters: [anyone]\n") errors = ccr_intake.branch_gate(main_root, stem, cc) self.assertIn("ccr branch may not change registry/ccr-standard-profile.yaml", errors) def test_branch_gate_refuses_modified_existing_policy(self) -> None: main_root, stem = self.submitted_branch() path = self.ccrs / f"{stem}.yaml" data = cc.load_yaml(path) data["openbao"]["policy_file"] = POLICY cc.dump_yaml(path, data) (self.root / POLICY).write_text((self.root / POLICY).read_text() + "# widened\n") errors = ccr_intake.branch_gate(main_root, stem, cc) self.assertTrue(any("may not modify" in e for e in errors), errors) def test_branch_gate_refuses_hand_edited_review(self) -> None: main_root, stem = self.submitted_branch() path = self.ccrs / f"{stem}.yaml" data = cc.load_yaml(path) data["review"]["comments"].append({"reviewer": "founder", "decision": "approved", "comment": "ok"}) cc.dump_yaml(path, data) errors = ccr_intake.branch_gate(main_root, stem, cc) self.assertTrue(any("policy-as-code approval" in e for e in errors), errors) def test_branch_gate_refuses_unlisted_requester(self) -> None: main_root, stem = self.submitted_branch() path = self.ccrs / f"{stem}.yaml" data = cc.load_yaml(path) data["requester"]["agent"] = "someone-else" cc.dump_yaml(path, data) errors = ccr_intake.branch_gate(main_root, stem, cc) self.assertTrue(any("listed submitter" in e for e in errors), errors) def test_branch_gate_refuses_existing_ccr(self) -> None: main_root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, main_root, True) shutil.copytree(self.root, main_root, dirs_exist_ok=True) errors = ccr_intake.branch_gate(main_root, Path(STANDARD).stem, cc) self.assertTrue(any("already exists on main" in e for e in errors), errors) def test_request_decision_records_id_after_platform_review(self) -> None: ccr_intake.review_standard(self.path, cc) with mock.patch.object(cc, "state_hub_post_json", return_value={"id": "dec-1", "status": "open"}) as post: ccr_intake.request_decision(self.path, cc, "http://hub") payload = post.call_args.args[2] self.assertIn("APPROVE: CCR-2026-0051", payload["description"]) self.assertNotIn("status", payload) state_hub = cc.load_yaml(self.path)["state_hub"] self.assertEqual(state_hub["decision_id"], "dec-1") self.assertEqual(cc.load_yaml(self.path)["status"], "proposed") with self.assertRaises(SystemExit): ccr_intake.request_decision(self.path, cc, "http://hub") def test_request_decision_requires_platform_approval(self) -> None: self.edit(lambda d: d["risk"].update(classification="high")) ccr_intake.review_standard(self.path, cc) with mock.patch.object(cc, "state_hub_post_json") as post, self.assertRaises(SystemExit): ccr_intake.request_decision(self.path, cc, "http://hub") post.assert_not_called() def test_request_decision_refuses_pre_resolved_decision(self) -> None: ccr_intake.review_standard(self.path, cc) resolved = {"id": "dec-2", "status": "resolved"} with mock.patch.object(cc, "state_hub_post_json", return_value=resolved), self.assertRaises(SystemExit): ccr_intake.request_decision(self.path, cc, "http://hub") if __name__ == "__main__": unittest.main()