ops-mason/tests/test_executor.py

import json
from unittest.mock import MagicMock, patch

import pytest

from ops_mason.executor import (
    AppRoleKVSpec,
    BuildError,
    BuildRefused,
    _policy_hcl,
    build_approle_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_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)