activity-core/tests/test_custodian_operations.py

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()