secrets-engine/tests/test_lifecycle.py

import copy

import pytest

from secrets_engine.catalog import get_entry, validate_entry
from secrets_engine.config import repo_root
from secrets_engine.errors import PolicyGuardError
from secrets_engine.lifecycle import (
    apply_lifecycle_plan,
    build_lifecycle_plan,
    build_native_deactivation_plan,
    require_destroy_confirmation,
)
from tests.test_catalog import VALID


class RecordingLifecycleClient:
    def __init__(self):
        self.calls = []

    def delete_approle(self, role_name):
        self.calls.append(("delete-approle", role_name))

    def delete_policy(self, policy_name):
        self.calls.append(("delete-policy", policy_name))

    def kv_delete_metadata(self, mount, path):
        self.calls.append(("delete-kv-metadata", f"{mount}/{path}"))


def _existing_kv_entry(*, auth_management="engine"):
    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"}
            ],
        }
    )
    if auth_management == "existing":
        data["delivery_auth"] = {
            "method": "approle",
            "management": "existing",
            "role_name": "external-example-role",
            "policy_name": "external-example-policy",
        }
    return validate_entry(data)


def test_kv_revoke_plan_deactivates_native_auth_and_preserves_custody():
    entry = _existing_kv_entry()
    plan = build_native_deactivation_plan(entry)

    assert [(a.kind, a.target, a.mutation) for a in plan.actions] == [
        ("delete-approle", entry.role_name, True),
        ("delete-policy", entry.policy_name, True),
        ("preserve-kv-custody", "platform/workloads/example/runtime", False),
        ("preserve-workload-delivery", entry.id, False),
    ]

    client = RecordingLifecycleClient()
    result = apply_lifecycle_plan(client, plan)
    assert client.calls == [
        ("delete-approle", entry.role_name),
        ("delete-policy", entry.policy_name),
    ]
    assert "platform/workloads/example/runtime" in result.preserved


def test_revoke_plan_never_mutates_externally_managed_delivery_auth():
    entry = _existing_kv_entry(auth_management="existing")
    plan = build_native_deactivation_plan(entry)
    assert not any(action.mutation for action in plan.actions)

    client = RecordingLifecycleClient()
    result = apply_lifecycle_plan(client, plan)
    assert client.calls == []
    assert entry.role_name in result.preserved
    assert entry.policy_name in result.preserved


def test_suspend_removes_only_approle_and_preserves_policy_and_kv():
    entry = _existing_kv_entry()
    plan = build_lifecycle_plan(entry, "suspend")
    client = RecordingLifecycleClient()

    result = apply_lifecycle_plan(client, plan)

    assert client.calls == [("delete-approle", entry.role_name)]
    assert entry.policy_name in result.preserved
    assert f"{entry.mount}/{entry.path}" in result.preserved


def test_destroy_plan_is_explicit_and_deletes_auth_before_kv_metadata():
    entry = _existing_kv_entry()
    plan = build_lifecycle_plan(entry, "destroy")
    client = RecordingLifecycleClient()

    result = apply_lifecycle_plan(client, plan)

    assert plan.operation == "destroy"
    assert client.calls == [
        ("delete-approle", entry.role_name),
        ("delete-policy", entry.policy_name),
        ("delete-kv-metadata", f"{entry.mount}/{entry.path}"),
    ]
    assert list(result.applied) == [
        entry.role_name,
        entry.policy_name,
        f"{entry.mount}/{entry.path}",
    ]


def test_destroy_requires_exact_confirmation_and_kv_lane():
    entry = _existing_kv_entry()
    with pytest.raises(PolicyGuardError, match="exact catalog id"):
        require_destroy_confirmation(entry, "wrong-lane")
    require_destroy_confirmation(entry, entry.id)

    auth_entry = get_entry(repo_root() / "catalog", "warden-sign")
    with pytest.raises(PolicyGuardError, match="no KV custody"):
        build_lifecycle_plan(auth_entry, "destroy")


def test_auth_capability_revoke_deletes_only_role_and_policy():
    entry = get_entry(repo_root() / "catalog", "warden-sign")
    plan = build_native_deactivation_plan(entry)
    client = RecordingLifecycleClient()

    apply_lifecycle_plan(client, plan)

    assert client.calls == [
        ("delete-approle", entry.role_name),
        ("delete-policy", entry.policy_name),
    ]


def test_rendered_and_applied_mutation_targets_are_identical():
    entry = _existing_kv_entry()
    plan = build_native_deactivation_plan(entry)
    rendered = plan.render()
    expected_targets = [action.target for action in plan.actions if action.mutation]
    assert all(target in rendered for target in expected_targets)

    client = RecordingLifecycleClient()
    result = apply_lifecycle_plan(client, plan)
    assert list(result.applied) == expected_targets