secrets-engine/tests/test_existing_lane_admission.py

import copy
import os

from secrets_engine.apply import apply_plan
from secrets_engine.catalog import load_catalog, validate_entry
from secrets_engine.config import repo_root
from secrets_engine.plan import build_plan
from secrets_engine.provision import provision_from_file
from secrets_engine.routing import route_lane
from tests.test_catalog import VALID


class RecordingApplyClient:
    def __init__(self):
        self.policy_writes = []
        self.approle_writes = []

    def kv_mount_exists(self, _mount):
        raise AssertionError("existing mount must not be inspected for mutation")

    def ensure_kv_mount(self, _mount):
        raise AssertionError("existing mount must not be created")

    def read_policy(self, _name):
        return None

    def write_policy(self, name, hcl):
        self.policy_writes.append((name, hcl))

    def ensure_approle_enabled(self):
        pass

    def write_approle(self, role_name, policies, ttl="30m", **_kwargs):
        self.approle_writes.append((role_name, policies, ttl))


class RecordingProvisionClient:
    def __init__(self):
        self.patches = []

    def ensure_kv_mount(self, _mount):
        raise AssertionError("existing mount must not be created during provision")

    def kv_patch_fields(self, mount, path, values):
        self.patches.append((mount, path, values))


def _existing_mount_entry():
    data = copy.deepcopy(VALID)
    data.update(
        {
            "stage": "prod",
            "mount": "platform",
            "path": "workloads/example/runtime",
            "mount_management": "existing",
            "workload_delivery": [
                {"mode": "external-secrets", "owner": "rapp-example"}
            ],
        }
    )
    return validate_entry(data)


def test_apply_existing_mount_only_adds_approved_delivery_auth():
    entry = _existing_mount_entry()
    plan = build_plan(entry, "prod", decision_id="approved")
    client = RecordingApplyClient()
    result = apply_plan(client, entry, plan)
    assert any("externally managed; no mutation" in item for item in result.skipped)
    assert len(client.policy_writes) == 1
    assert len(client.approle_writes) == 1


def test_apply_existing_delivery_auth_is_fully_non_mutating():
    data = copy.deepcopy(VALID)
    data["mount_management"] = "existing"
    data["delivery_auth"] = {
        "method": "approle",
        "management": "existing",
        "role_name": "existing-exact-role",
        "policy_name": "existing-exact-policy",
    }
    entry = validate_entry(data)
    plan = build_plan(entry, "test", decision_id="approved")
    client = RecordingApplyClient()
    result = apply_plan(client, entry, plan)
    assert client.policy_writes == []
    assert client.approle_writes == []
    assert len(result.skipped) == 3
    assert all("no mutation" in item for item in result.skipped[:2])


def test_provision_existing_mount_never_attempts_mount_creation(tmp_path):
    entry = _existing_mount_entry()
    value_file = tmp_path / "value"
    value_file.write_text("test-only-value", encoding="utf-8")
    os.chmod(value_file, 0o600)
    client = RecordingProvisionClient()
    provision_from_file(client, entry, "api_token", value_file)
    assert client.patches == [
        (
            "platform",
            "workloads/example/runtime",
            {"api_token": "test-only-value"},
        )
    ]


def test_provision_existing_multi_field_path_uses_merge_safe_backend(tmp_path):
    data = copy.deepcopy(VALID)
    data.update(
        {
            "stage": "prod",
            "mount": "platform",
            "path": "workloads/example/runtime",
            "mount_management": "existing",
            "fields": ["api_token", "webhook_secret"],
            "workload_delivery": [
                {"mode": "external-secrets", "owner": "rapp-example"}
            ],
        }
    )
    entry = validate_entry(data)
    value_file = tmp_path / "value"
    value_file.write_text("test-only-value", encoding="utf-8")
    os.chmod(value_file, 0o600)
    client = RecordingProvisionClient()

    provision_from_file(client, entry, "api_token", value_file)

    assert client.patches == [
        (
            "platform",
            "workloads/example/runtime",
            {"api_token": "test-only-value"},
        )
    ]


def test_admitted_lanes_fail_closed_without_resolved_ccr(tmp_path):
    entries = load_catalog(repo_root() / "catalog")
    for lane_id in (
        "issue-core-ingestion-api-key",
        "reuse-surface-hub-write-token",
        "openrouter-llm-connect",
        "forgejo-admin-api-token",
        "email-connect-transactional",
    ):
        result = route_lane(entries[lane_id], hub_url="", repo_root=tmp_path, client=None)
        assert result.ready is False
        assert result.missing.startswith("approved decision for 'CCR-2026-")
        assert result.next_command.startswith("secrets-engine decision inspect CCR-2026-")