import json import hashlib from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch import pytest from activity_core import custodian_operations as runner from activity_core.bounded_operations import normalize_operation_result, validate_bounded_operations from activity_core.definition_parser import parse_file @pytest.fixture def runtime(tmp_path, monkeypatch): root, fleet, state = (tmp_path / x for x in ("owner", "fleet", "state")) for path in (root / "scripts", fleet, state, state / "security-flow"): path.mkdir(parents=True, exist_ok=True) state.chmod(0o700) (state / "security-flow").chmod(0o700) for key, value in {"ROOT": root, "FLEET_ROOT": fleet, "STATE_ROOT": state}.items(): monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_" + key, str(value)) monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_HUB", "http://127.0.0.1:8000") monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_REPOS", "the-custodian") (fleet / "the-custodian/workplans").mkdir(parents=True) (fleet / "the-custodian/workplans/task.md").write_text("task") manifest = {"schema": "repo-manager.fleet-input.v1", "repositories": [{ "slug": "the-custodian", "commit": "a" * 40, "files": [{"path": "workplans/task.md", "bytes": 4, "sha256": hashlib.sha256(b"task").hexdigest()}]}]} raw = json.dumps(manifest).encode() (fleet / "manifest.json").write_bytes(raw) (fleet / "COMPLETE").write_text(hashlib.sha256(raw).hexdigest()) policy = state / "policy.json" policy.write_text(json.dumps({"schema": "activity-core.custodian-input-policy.v1", "repositories": ["the-custodian"], "max_observation_age_seconds": 7200, "max_future_skew_seconds": 300, "max_snapshot_bytes": 1024, "expected_hub_instance": "railiance01"})) receipts = state / "receipts.json" receipts.write_text(json.dumps([{"schema": "repo-manager.freshness-receipt.v1", "repo_slug": "the-custodian", "outcome": "applied", "finished_at": datetime.now(timezone.utc).isoformat(), "hub": {"verified_primary": True, "instance_role": "primary", "instance_label": "railiance01"}, **{key: "a" * 40 for key in ("forge_commit", "expected_commit", "derived_commit", "projected_commit")}}])) monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY", str(policy)) monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS", str(receipts)) for name in ("security_flow_daily.py", "weekly_allocation_report.py"): (root / "scripts" / name).write_text("# owner fixture\n") return root, fleet, state def request(runtime, monkeypatch): root, fleet, state = runtime inputs = {} for key in ("cluster", "allocation", "telemetry", "inventory", "provenance"): path = fleet / (key + ".json") path.write_text("{}") inputs[key] = str(path) data = {"schema": "custodian.weekly-allocation-request.v1", "as_of": "2026-10-19T00:00:00Z", **inputs, "output": str(state / "week.json")} path = state / "request.json" path.write_text(json.dumps(data)) monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST", str(path)) return path, data def test_fixed_daily_argv_and_missing_binding_before_execution(runtime, monkeypatch): root, fleet, state = runtime argv, cwd, output, request_data = runner.build_command("custodian_security_flow", {"apply": True}) assert argv[1:] == [str(root / "scripts/security_flow_daily.py"), "--root", str(fleet), "--repos", "the-custodian", "--hub", "http://127.0.0.1:8000", "--state-dir", str(state / "security-flow"), "--apply"] monkeypatch.delenv("ACTIVITY_CORE_CUSTODIAN_HUB") with patch.object(runner, "_execute") as execute, pytest.raises(ValueError): runner.run_custodian_operation("custodian_security_flow", {"apply": True}) execute.assert_not_called() @pytest.mark.parametrize("params", [{"apply": "true"}, {"apply": True, "command": "echo x"}, {"apply": True, "timeout_seconds": True}, {"apply": True, "timeout_seconds": 241}, {"apply": True, "root": "/tmp"}]) def test_invalid_definition_parameters_refused(params): with pytest.raises(ValueError): runner.validate_params("custodian_security_flow", params) def test_definition_no_profile_no_tasks_and_evidence_required(): root = Path(__file__).resolve().parents[1] for name in ("daily-security-flow", "weekly-railiance-allocation-review"): definition = parse_file(root / "activity-definitions" / f"{name}.md") assert definition.enabled is False assert definition.rules == definition.instructions == [] sources = definition.context_sources validate_bounded_operations(sources, []) sources[0]["params"].pop("evidence_sinks") with pytest.raises(ValueError): validate_bounded_operations(sources, []) def test_symlink_state_and_output_refused(runtime, monkeypatch, tmp_path): root, fleet, state = runtime path, data = request(runtime, monkeypatch) external = tmp_path / "outside" external.mkdir() (state / "redirect").symlink_to(external, target_is_directory=True) data["output"] = str(state / "redirect/week.json") path.write_text(json.dumps(data)) with pytest.raises(ValueError, match="symlink"): runner.build_command("custodian_allocation_review", {}) data["output"] = str(external / "week.json") path.write_text(json.dumps(data)) with pytest.raises(ValueError, match="outside"): runner.build_command("custodian_allocation_review", {}) def test_unbound_request_and_alias_input_refused(runtime, monkeypatch): path, data = request(runtime, monkeypatch) data["cluster"] = None path.write_text(json.dumps(data)) with pytest.raises(ValueError, match="unbound"): runner.build_command("custodian_allocation_review", {}) path, data = request(runtime, monkeypatch) data["output"] = str(path) path.write_text(json.dumps(data)) with pytest.raises(ValueError, match="request"): runner.build_command("custodian_allocation_review", {}) def test_fixed_fixture_execution_retains_only_compact_receipt(runtime): root, fleet, state = runtime (root / "scripts/security_flow_daily.py").write_text(''' import argparse,json,pathlib p=argparse.ArgumentParser() p.add_argument('--root'); p.add_argument('--repos'); p.add_argument('--hub'); p.add_argument('--state-dir'); p.add_argument('--apply',action='store_true') a=p.parse_args() pathlib.Path(a.state_dir,'report.json').write_text(json.dumps({k:[] for k in ['READY','UNBLOCKED','ATTENDED','STALLED','DEPENDENT']})) print('fixture-output-must-not-enter-receipt') ''') receipt = runner.run_custodian_operation("custodian_security_flow", {"apply": False}) assert receipt["status"] == "completed" assert receipt["counts"]["ready"] == 0 assert "fixture-output" not in json.dumps(receipt) assert len(receipt["report_sha256"]) == 64 def test_nonzero_exit_is_not_success_and_error_is_sanitized(runtime): with patch.object(runner, "_execute", return_value=7), pytest.raises(RuntimeError, match="exit 7"): runner.run_custodian_operation("custodian_security_flow", {"apply": False}) def test_stale_report_is_not_command_success(runtime): _, _, state = runtime output = state / "security-flow/report.json" output.write_text('{}') with patch.object(runner, "_execute", return_value=0), pytest.raises(RuntimeError, match="not refreshed"): runner.run_custodian_operation("custodian_security_flow", {"apply": False}) def test_weekly_fixture_execution_and_receipt_projection(runtime, monkeypatch): root, _, state = runtime request(runtime, monkeypatch) (root / "scripts/weekly_allocation_report.py").write_text(''' import argparse,json,pathlib p=argparse.ArgumentParser();p.add_argument('--request');a=p.parse_args() r=json.loads(pathlib.Path(a.request).read_text()) pathlib.Path(r['output']).write_text(json.dumps({'schema':'custodian.weekly-allocation.v1','recommendations':[]})) ''') result = runner.run_custodian_operation("custodian_allocation_review", {}) assert result["counts"] == {"recommendations": 0} assert "secret" not in normalize_operation_result("custodian_allocation_review", result | {"secret": "discard"}) @pytest.mark.asyncio async def test_temporal_dispatch_uses_existing_stage(monkeypatch): from activity_core.activities import execute_bounded_operation monkeypatch.setattr(runner, "run_custodian_operation", lambda operation, params: { "kind": operation, "status": "completed", "counts": {}, "unknown": "discard"}) result = await execute_bounded_operation({"source": {"type": "shell", "query": "custodian_allocation_review", "operation": "custodian_allocation_review", "params": {}, "bind_to": "context.custodian_result"}}) assert result["custodian_result"]["status"] == "completed" assert "unknown" not in result["custodian_result"] def test_subprocess_timeout_kills_group(): import subprocess process = __import__('unittest.mock', fromlist=['MagicMock']).MagicMock() process.__enter__.return_value = process process.pid = 12345 process.wait.side_effect = [subprocess.TimeoutExpired('fixed-owner-command', 1), 0] with patch.object(runner.subprocess, 'Popen', return_value=process), patch.object(runner.os, 'killpg') as kill: with pytest.raises(RuntimeError, match='timed out'): runner._execute(['fixed-owner-command'], Path('/tmp'), 1) kill.assert_called_once_with(12345, runner.signal.SIGKILL) def test_configured_freshness_admission_refuses_before_subprocess(runtime, monkeypatch): root, fleet, state = runtime policy = state / "policy.json" receipts = state / "receipts.json" policy.write_text("{}") receipts.write_text("[]") monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY", str(policy)) monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS", str(receipts)) with patch.object(runner, "_execute") as execute, pytest.raises(ValueError): runner.run_custodian_operation("custodian_security_flow", {"apply": True}) execute.assert_not_called() def test_missing_policy_refuses_before_subprocess(runtime, monkeypatch): monkeypatch.delenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY") with patch.object(runner, "_execute") as execute, pytest.raises(ValueError): runner.run_custodian_operation("custodian_security_flow", {"apply": True}) execute.assert_not_called()