activity-core/tests/test_bounded_operations.py

from __future__ import annotations

import pytest

from activity_core.bounded_operations import (
    BOUNDED_OPERATION_REGISTRY,
    normalize_operation_result,
    operation_sources,
    validate_bounded_operations,
)


def _evidence_sinks(event_type: str) -> list[dict[str, str]]:
    return [{"type": "state-hub-progress", "event_type": event_type}]


def test_registry_names_only_the_three_accepted_operations() -> None:
    assert set(BOUNDED_OPERATION_REGISTRY) == {
        "sbom_nexus_ingest",
        "forgejo_package_prune",
        "cnpg_option_a_backup",
    }


def test_operation_result_projection_drops_subprocess_output() -> None:
    projected = normalize_operation_result(
        "cnpg_option_a_backup",
        {
            "overall": "pass",
            "dumped": 3,
            "uploaded": 3,
            "failed": 0,
            "script_exit_code": 0,
            "log_tail": "must not persist",
            "tool_output": "must not persist",
            "credential": "must not persist",
        },
    )

    assert projected == {
        "kind": "cnpg_option_a_backup",
        "overall": "pass",
        "dry_run": False,
        "dumped": 3,
        "uploaded": 3,
        "failed": 0,
        "script_exit_code": 0,
    }
    assert "must not persist" not in str(projected)


def test_valid_prune_operation_is_admitted() -> None:
    sources = [
        {
            "type": "shell",
            "query": "forgejo_package_prune",
            "operation": "forgejo_package_prune",
            "params": {
                "prune_script": "/opt/railiance-platform/tools/cmd/forgejo-package-prune",
                "live_images_file": "/opt/railiance-platform/live-images.txt",
                "apply": True,
                "max_versions": 3,
                "evidence_sinks": _evidence_sinks("forgejo_package_prune"),
            },
        }
    ]

    validate_bounded_operations(sources, [])
    assert operation_sources(sources)[0][1].temporal_max_attempts == 1


@pytest.mark.parametrize(
    ("source", "error"),
    [
        (
            {"type": "shell", "query": "run_whatever", "params": {}},
            "not registered",
        ),
        (
            {
                "type": "shell",
                "query": "forgejo_package_prune",
                "params": {},
            },
            "must declare operation",
        ),
        (
            {
                "type": "shell",
                "query": "forgejo_package_prune",
                "operation": "forgejo_package_prune",
                "params": {
                    "prune_script": "/tmp/arbitrary-command",
                    "apply": False,
                    "max_versions": 3,
                    "evidence_sinks": _evidence_sinks("forgejo_package_prune"),
                },
            },
            "canonical path",
        ),
    ],
)
def test_shell_admission_fails_closed(source: dict, error: str) -> None:
    with pytest.raises(ValueError, match=error):
        validate_bounded_operations([source], [])


def test_sbom_requires_bounded_limit_and_report_evidence() -> None:
    source = {
        "type": "sbom-nexus",
        "query": "catch_up",
        "operation": "sbom_nexus_ingest",
        "params": {"apply": True, "limit": 4},
    }
    instruction = {
        "report_sinks": _evidence_sinks("sbom_catchup"),
    }

    with pytest.raises(ValueError, match="limit must be in 1..3"):
        validate_bounded_operations([source], [instruction])

    source["params"]["limit"] = 3
    with pytest.raises(ValueError, match="requires an instruction report"):
        validate_bounded_operations([source], [])

    validate_bounded_operations([source], [instruction])


def test_backup_requires_explicit_safe_targets_and_evidence() -> None:
    source = {
        "type": "shell",
        "query": "cnpg_option_a_backup",
        "operation": "cnpg_option_a_backup",
        "params": {
            "backup_script": "/opt/railiance-platform/tools/cmd/cnpg-option-a-backup",
            "dry_run": False,
            "timeout_seconds": 7200,
            "targets": "r01-forgejo-db,r01-state-hub-db",
            "evidence_sinks": _evidence_sinks("cnpg_option_a_backup"),
        },
    }

    validate_bounded_operations([source], [])
    source["params"]["targets"] = "ok,$(unsafe)"
    with pytest.raises(ValueError, match="safe names"):
        validate_bounded_operations([source], [])

    source["params"]["targets"] = "r01-forgejo-db"
    source["params"]["evidence_sinks"] = [{"type": "state-hub-progress"}]
    with pytest.raises(ValueError, match="event_type: cnpg_option_a_backup"):
        validate_bounded_operations([source], [])


def test_definition_cannot_combine_two_bounded_operations() -> None:
    prune = {
        "type": "shell",
        "query": "forgejo_package_prune",
        "operation": "forgejo_package_prune",
    }
    backup = {
        "type": "shell",
        "query": "cnpg_option_a_backup",
        "operation": "cnpg_option_a_backup",
    }

    with pytest.raises(ValueError, match="at most one bounded operation"):
        validate_bounded_operations([prune, backup], [])