import json from unittest.mock import MagicMock, patch import pytest from ops_mason.executor import ( AppRoleKVSpec, BuildError, BuildRefused, KubernetesKVSpec, _policy_hcl, build_approle_kv_lane, build_kubernetes_kv_lane, ) def test_policy_hcl_uses_kv_v2_data_and_metadata_paths() -> None: hcl = _policy_hcl("reins/rein-openweights/openrouter", ("read",)) assert 'path "reins/data/rein-openweights/openrouter" {' in hcl assert 'path "reins/metadata/rein-openweights/openrouter" {' in hcl assert 'path "reins/rein-openweights/openrouter" {' not in hcl assert 'capabilities = ["read"]' in hcl from ops_mason.plan import ConstructionPlan def _plan(tmp_path, *, status="approved", approved_by="bernd", approved_at="2026-07-27"): text = ( "---\n" "id: test-lane\n" f"status: {status}\n" + (f'approved_by: "{approved_by}"\n' if approved_by else "approved_by: null\n") + (f'approved_at: "{approved_at}"\n' if approved_at else "approved_at: null\n") + "---\n# Plan\n" ) p = tmp_path / "plan.md" p.write_text(text) return ConstructionPlan.load(str(p)) def _spec(tmp_path) -> AppRoleKVSpec: return AppRoleKVSpec( policy_name="workload-kv-read-test-lane", kv_path="reins/test/openrouter", approle_name="test-lane", token_num_uses=8, delivery_dir=tmp_path / "delivery", audit_log_path=tmp_path / "audit.jsonl", ) def test_token_num_uses_has_no_silently_unbounded_default() -> None: import inspect sig = inspect.signature(AppRoleKVSpec) assert sig.parameters["token_num_uses"].default is inspect.Parameter.empty def test_refuses_when_not_approved(tmp_path) -> None: plan = _plan(tmp_path, status="draft", approved_by=None, approved_at=None) with pytest.raises(BuildRefused, match="not approved"): build_approle_kv_lane(plan, _spec(tmp_path)) def test_refuses_when_approved_status_but_missing_approver(tmp_path) -> None: plan = _plan(tmp_path, status="approved", approved_by=None, approved_at="2026-07-27") with pytest.raises(BuildRefused, match="not approved"): build_approle_kv_lane(plan, _spec(tmp_path)) def test_refusal_never_calls_bao(tmp_path) -> None: plan = _plan(tmp_path, status="reviewed", approved_by=None, approved_at=None) with patch("ops_mason.executor.subprocess.run") as run: with pytest.raises(BuildRefused): build_approle_kv_lane(plan, _spec(tmp_path)) run.assert_not_called() def test_build_writes_policy_approle_and_delivers_credentials(tmp_path) -> None: plan = _plan(tmp_path) spec = _spec(tmp_path) def fake_run(cmd, input=None, capture_output=True, text=True, timeout=30): result = MagicMock(returncode=0, stderr="") if cmd[1:3] == ["read", "-field=role_id"]: result.stdout = "role-id-value\n" elif "-field=secret_id" in cmd: result.stdout = "secret-id-value\n" else: result.stdout = "" return result with patch("ops_mason.executor.subprocess.run", side_effect=fake_run) as run: objects = build_approle_kv_lane(plan, spec) assert objects["policy_name"] == "workload-kv-read-test-lane" assert objects["approle_name"] == "test-lane" assert objects["kv_path"] == "reins/test/openrouter" role_id_file = spec.delivery_dir / "role_id" secret_id_file = spec.delivery_dir / "secret_id" assert role_id_file.read_text() == "role-id-value\n" assert secret_id_file.read_text() == "secret-id-value\n" assert oct(role_id_file.stat().st_mode)[-3:] == "600" assert oct(secret_id_file.stat().st_mode)[-3:] == "600" # policy write call carried the HCL on stdin, not argv -- never in a log line # -- KV v2 shape: data/ and metadata/ sub-paths, not the bare path policy_call = next(c for c in run.call_args_list if c.args[0][:2] == ["bao", "policy"]) assert 'path "reins/data/test/openrouter"' in policy_call.kwargs["input"] assert 'path "reins/metadata/test/openrouter"' in policy_call.kwargs["input"] assert 'path "reins/test/openrouter"' not in policy_call.kwargs["input"] # audit record landed at the explicit path, metadata only, no role_id/secret_id values audit_text = spec.audit_log_path.read_text() assert "policy_name" in audit_text assert "role-id-value" not in audit_text assert "secret-id-value" not in audit_text def test_reuse_policy_does_not_rewrite_existing_policy(tmp_path) -> None: plan = _plan(tmp_path) spec = _spec(tmp_path) spec.reuse_policy = True def fake_run(cmd, input=None, capture_output=True, text=True, timeout=30): result = MagicMock(returncode=0, stderr="") if cmd[1:3] == ["read", "-field=role_id"]: result.stdout = "role-id-value\n" elif "-field=secret_id" in cmd: result.stdout = "secret-id-value\n" else: result.stdout = "" return result with patch("ops_mason.executor.subprocess.run", side_effect=fake_run) as run: build_approle_kv_lane(plan, spec) bao_cmds = [c.args[0] for c in run.call_args_list] assert not any(cmd[:2] == ["bao", "policy"] for cmd in bao_cmds) assert any(cmd[1:3] == ["write", "auth/approle/role/test-lane"] for cmd in bao_cmds) def test_build_appends_audit_record(tmp_path) -> None: plan = _plan(tmp_path) spec = _spec(tmp_path) def fake_run(cmd, input=None, capture_output=True, text=True, timeout=30): result = MagicMock(returncode=0, stderr="") result.stdout = "value\n" return result with ( patch("ops_mason.executor.subprocess.run", side_effect=fake_run), patch("ops_mason.executor.record_build") as record_build_mock, ): build_approle_kv_lane(plan, spec) record_build_mock.assert_called_once() kwargs = record_build_mock.call_args.kwargs assert kwargs["plan_id"] == "test-lane" assert kwargs["approved_by"] == "bernd" assert "policy_name" in kwargs["objects"] assert kwargs["log_path"] == spec.audit_log_path def test_bao_failure_raises_build_error(tmp_path) -> None: plan = _plan(tmp_path) spec = _spec(tmp_path) fake_result = MagicMock(returncode=1, stderr="permission denied", stdout="") with patch("ops_mason.executor.subprocess.run", return_value=fake_result): with pytest.raises(BuildError, match="permission denied"): build_approle_kv_lane(plan, spec) def _kubernetes_spec(tmp_path) -> KubernetesKVSpec: return KubernetesKVSpec( policy_name="workload-kv-read-binky-qonto-api", role_name="external-secrets-rapp-qonto", service_account_names=("external-secrets",), service_account_namespaces=("external-secrets",), audit_log_path=tmp_path / "audit.jsonl", ) def test_kubernetes_lane_refusal_never_calls_bao(tmp_path) -> None: plan = _plan(tmp_path, status="reviewed", approved_by=None, approved_at=None) with patch("ops_mason.executor.subprocess.run") as run: with pytest.raises(BuildRefused): build_kubernetes_kv_lane(plan, _kubernetes_spec(tmp_path)) run.assert_not_called() def test_kubernetes_lane_builds_exact_service_account_binding(tmp_path) -> None: plan = _plan(tmp_path) spec = _kubernetes_spec(tmp_path) result = MagicMock(returncode=0, stderr="", stdout="") with ( patch("ops_mason.executor.subprocess.run", return_value=result) as run, patch("ops_mason.executor.record_build") as audit, ): objects = build_kubernetes_kv_lane(plan, spec) command = run.call_args.args[0] assert command == [ "bao", "write", "auth/kubernetes/role/external-secrets-rapp-qonto", "bound_service_account_names=external-secrets", "bound_service_account_namespaces=external-secrets", "policies=workload-kv-read-binky-qonto-api", "ttl=15m", ] assert objects["kubernetes_role_name"] == "external-secrets-rapp-qonto" audit.assert_called_once() def test_kubernetes_lane_requires_nonempty_bindings(tmp_path) -> None: plan = _plan(tmp_path) spec = KubernetesKVSpec( policy_name="policy", role_name="role", service_account_names=(), service_account_namespaces=("external-secrets",), ) with patch("ops_mason.executor.subprocess.run") as run: with pytest.raises(BuildRefused, match="explicit service account"): build_kubernetes_kv_lane(plan, spec) run.assert_not_called()