secrets-engine/tests/test_wp0016_manifestations.py

"""SECRETS-WP-0016: plan stays off the store client; apply still calls it.

Later cases cover one-stage process binding, the in-process service, and the
refusal of a local allow flag. Fixture credentials stay out of captured output.
"""
from __future__ import annotations

import copy
import io
import json
import sys
from pathlib import Path

import pytest

from secrets_engine import cli, lifecycle
from secrets_engine.apply import apply_plan
from secrets_engine.catalog import validate_entry
from secrets_engine.config import Config, repo_root
from secrets_engine.openbao import OpenBaoClient
from secrets_engine.plan import build_plan
from secrets_engine.service import run_service
from secrets_engine.stage_binding import STAGE_TOKEN_FILES, StartupError
from tests.test_catalog import VALID

SENTINEL = "fixture-stage-credential-wp0016"
STAGE_ENV = tuple(STAGE_TOKEN_FILES.values())


class RecordingApplyClient:
    def __init__(self):
        self.calls: list[str] = []

    def kv_mount_exists(self, mount):
        self.calls.append("kv_mount_exists")
        assert mount
        return False

    def ensure_kv_mount(self, mount):
        self.calls.append("ensure_kv_mount")
        assert mount

    def read_policy(self, name):
        self.calls.append("read_policy")
        assert name
        return None

    def write_policy(self, name, hcl):
        self.calls.append("write_policy")
        assert name and hcl

    def ensure_approle_enabled(self):
        self.calls.append("ensure_approle_enabled")

    def write_approle(self, role_name, policies, ttl="30m", **kwargs):
        self.calls.append("write_approle")
        assert role_name and policies and ttl
        del kwargs


def _clear_stage_env(monkeypatch):
    for key in STAGE_ENV:
        monkeypatch.delenv(key, raising=False)
    monkeypatch.delenv("SECRETS_ENGINE_LOCAL_ALLOW", raising=False)


def _quiet_runtime(monkeypatch, tmp_path: Path):
    """Keep the test off the hub, the PDP, and the repo evidence directory."""
    monkeypatch.setenv("SECRETS_ENGINE_HUB_URL", "")
    monkeypatch.setenv("SECRETS_ENGINE_EVIDENCE", str(tmp_path / "evidence"))
    for key in (
        "SECRETS_ENGINE_APPROVAL_URL",
        "SECRETS_ENGINE_APPROVAL_TOKEN_FILE",
        "SECRETS_ENGINE_PDP_URL",
        "SECRETS_ENGINE_PDP_TOKEN_FILE",
        "SECRETS_ENGINE_PDP_TRUST_FILE",
        "SECRETS_ENGINE_PDP_VERIFIER",
        "SECRETS_ENGINE_OPENBAO_JWT_LOGIN",
        "BAO_TOKEN",
        "VAULT_TOKEN",
    ):
        monkeypatch.delenv(key, raising=False)


def _isolate_authority(monkeypatch, tmp_path: Path):
    """No serving PDP, no published stance file, no stage credential."""
    _clear_stage_env(monkeypatch)
    _quiet_runtime(monkeypatch, tmp_path)
    monkeypatch.setenv("SECRETS_ENGINE_PEP_STANCE", str(tmp_path / "missing-stance.yaml"))


class _Auth:
    auth = "env"
    bootstrap_token_file = None


class _LifecycleClient:
    def __init__(self):
        self.calls: list[str] = []

    def delete_approle(self, name):
        self.calls.append(f"delete_approle:{name}")

    def delete_policy(self, name):
        raise AssertionError(f"unexpected policy delete {name}")

    def kv_delete_metadata(self, mount, path):
        raise AssertionError(f"unexpected kv delete {mount}/{path}")


def test_plan_path_constructs_no_openbao_client(monkeypatch):
    constructed: list[str] = []

    def _refuse_init(self, *args, **kwargs):
        constructed.append("init")
        raise AssertionError("plan path constructed an OpenBao client")

    def _refuse_resolve(cls, *args, **kwargs):
        constructed.append("resolve")
        raise AssertionError("plan path resolved an OpenBao client")

    monkeypatch.setattr(OpenBaoClient, "__init__", _refuse_init)
    monkeypatch.setattr(OpenBaoClient, "resolve", _refuse_resolve)
    entry = validate_entry(copy.deepcopy(VALID))
    plan = build_plan(entry, "test", decision_id="d1")
    assert plan.catalog_id == entry.id
    assert plan.actions
    assert constructed == []


def test_apply_path_issues_openbao_calls():
    entry = validate_entry(copy.deepcopy(VALID))
    plan = build_plan(entry, "test", decision_id="d1")
    client = RecordingApplyClient()
    result = apply_plan(client, entry, plan)
    for name in ("ensure_kv_mount", "write_policy", "write_approle"):
        assert name in client.calls
    assert any(name.startswith("kv-mount") for name in result.applied)
    assert any(name.startswith("policy") for name in result.applied)
    assert any(name.startswith("approle") for name in result.applied)


def _write_stage_file(tmp_path: Path, name: str) -> Path:
    path = tmp_path / name
    path.write_text(SENTINEL + "\n", encoding="utf-8")
    return path


def _assert_sentinel_absent(captured):
    assert SENTINEL not in captured.out
    assert SENTINEL not in captured.err


@pytest.mark.parametrize("stage", ("build", "test", "production"))
def test_cli_binds_one_stage_and_hides_the_credential(stage, tmp_path, monkeypatch, capsys):
    _clear_stage_env(monkeypatch)
    path = _write_stage_file(tmp_path, stage)
    monkeypatch.setenv(STAGE_TOKEN_FILES[stage], str(path))
    for _ in range(2):
        capsys.readouterr()
        code = cli.main(["instance"])
        captured = capsys.readouterr()
        assert code == 0
        assert captured.out.strip() == f"stage={stage}"
        _assert_sentinel_absent(captured)


def test_cli_refuses_two_stage_credentials(tmp_path, monkeypatch, capsys):
    _clear_stage_env(monkeypatch)
    monkeypatch.setenv(STAGE_TOKEN_FILES["build"], str(_write_stage_file(tmp_path, "build")))
    monkeypatch.setenv(STAGE_TOKEN_FILES["test"], str(_write_stage_file(tmp_path, "test")))
    for _ in range(2):
        capsys.readouterr()
        code = cli.main(["instance"])
        captured = capsys.readouterr()
        assert code != 0
        assert "more than one stage credential is present" in captured.err
        _assert_sentinel_absent(captured)


def _bind_stage(monkeypatch, tmp_path: Path, stage: str) -> None:
    _clear_stage_env(monkeypatch)
    monkeypatch.setenv(STAGE_TOKEN_FILES[stage], str(_write_stage_file(tmp_path, stage)))


def test_service_lifecycle_uses_the_cli_library_and_waits(tmp_path, monkeypatch):
    _quiet_runtime(monkeypatch, tmp_path)
    _bind_stage(monkeypatch, tmp_path, "build")
    client = _LifecycleClient()
    monkeypatch.setattr(OpenBaoClient, "resolve", lambda *_a, **_k: client)
    order: list[str] = []
    real_build = lifecycle.build_lifecycle_plan
    real_apply = lifecycle.apply_lifecycle_plan

    def spy_build(entry, operation):
        order.append(f"build:{operation}")
        return real_build(entry, operation)

    def spy_apply(bound_client, plan):
        order.append(f"apply:{plan.operation}")
        return real_apply(bound_client, plan)

    monkeypatch.setattr(lifecycle, "build_lifecycle_plan", spy_build)
    monkeypatch.setattr(lifecycle, "apply_lifecycle_plan", spy_apply)

    def jobs():
        order.append("yield-1")
        yield {
            "catalog_id": "example-build-test-token",
            "operation": "suspend",
            "confirm_destroy": "",
        }
        order.append("yield-2")
        yield {
            "catalog_id": "example-build-test-token",
            "operation": "suspend",
            "confirm_destroy": "",
        }

    outcomes = run_service(
        Config.load(), jobs(), allow=True, auth_args=_Auth()
    )
    assert order == [
        "yield-1",
        "build:suspend",
        "apply:suspend",
        "yield-2",
        "build:suspend",
        "apply:suspend",
    ]
    assert len(client.calls) == 2
    assert all(call.startswith("delete_approle:") for call in client.calls)
    rendered = "\n".join(plan.render() + result.render() for plan, result in outcomes)
    assert SENTINEL not in rendered
    assert "applied:" in rendered


def test_serve_reads_a_second_job_from_stdin(tmp_path, monkeypatch, capsys):
    _quiet_runtime(monkeypatch, tmp_path)
    _bind_stage(monkeypatch, tmp_path, "build")
    client = _LifecycleClient()
    monkeypatch.setattr(OpenBaoClient, "resolve", lambda *_a, **_k: client)
    extra = json.dumps(
        {"catalog_id": "example-build-test-token", "operation": "suspend"}
    )
    monkeypatch.setattr(sys, "stdin", io.StringIO(extra + "\n"))
    code = cli.main(
        [
            "serve",
            "example-build-test-token",
            "--operation",
            "suspend",
            "--auth",
            "env",
            "--allow",
        ]
    )
    captured = capsys.readouterr()
    assert code == 0
    assert len(client.calls) == 2
    assert captured.out.count("applied:") == 2
    _assert_sentinel_absent(captured)


def test_service_refuses_a_second_stage_credential_file(tmp_path, monkeypatch, capsys):
    _clear_stage_env(monkeypatch)
    monkeypatch.setenv(STAGE_TOKEN_FILES["build"], str(_write_stage_file(tmp_path, "build")))
    monkeypatch.setenv(
        STAGE_TOKEN_FILES["production"],
        str(_write_stage_file(tmp_path, "production")),
    )

    pulled: list[str] = []

    def jobs():
        pulled.append("pulled")
        yield {
            "catalog_id": "example-build-test-token",
            "operation": "suspend",
            "confirm_destroy": "",
        }

    with pytest.raises(StartupError, match="more than one stage credential is present") as raised:
        run_service(Config.load(), jobs(), allow=True, auth_args=_Auth())
    assert pulled == []
    assert SENTINEL not in str(raised.value)

    capsys.readouterr()
    code = cli.main(
        ["serve", "example-build-test-token", "--operation", "suspend", "--auth", "env"]
    )
    captured = capsys.readouterr()
    assert code != 0
    assert "more than one stage credential is present" in captured.err
    _assert_sentinel_absent(captured)


def test_no_deployment_manifest_was_added():
    root = repo_root()
    for path in root.rglob("*"):
        if ".git" in path.parts or path.suffix not in {".yaml", ".yml"}:
            continue
        text = path.read_text(encoding="utf-8", errors="ignore")
        assert "kind: Deployment" not in text


def test_local_allow_does_not_authorize_protected_apply(tmp_path, monkeypatch, capsys):
    _isolate_authority(monkeypatch, tmp_path)
    monkeypatch.setenv("SECRETS_ENGINE_LOCAL_ALLOW", "1")
    monkeypatch.setattr(
        OpenBaoClient,
        "resolve",
        lambda *_a, **_k: pytest.fail("protected apply opened a backend"),
    )
    code = cli.main(
        [
            "apply",
            "whynot-design-npm-publish",
            "--stage",
            "prod",
            "--auth",
            "env",
            "--allow",
        ]
    )
    captured = capsys.readouterr()
    assert code != 0
    assert "applied:" not in captured.out
    assert "unable to load PEP stance" in captured.err or "DecisionError" in captured.err or "error:" in captured.err
    _assert_sentinel_absent(captured)


def test_serve_allow_flag_does_not_authorize(tmp_path, monkeypatch, capsys):
    _isolate_authority(monkeypatch, tmp_path)
    _bind_stage(monkeypatch, tmp_path, "production")
    monkeypatch.setenv("SECRETS_ENGINE_LOCAL_ALLOW", "1")
    monkeypatch.setattr(
        OpenBaoClient,
        "resolve",
        lambda *_a, **_k: pytest.fail("serve opened a backend"),
    )
    code = cli.main(
        [
            "serve",
            "whynot-design-npm-publish",
            "--operation",
            "suspend",
            "--auth",
            "env",
            "--allow",
        ]
    )
    captured = capsys.readouterr()
    assert code != 0
    assert "applied:" not in captured.out
    assert "unable to load PEP stance" in captured.err
    _assert_sentinel_absent(captured)


def test_use_claim_names_access_engine_and_stores_no_credential():
    path = repo_root() / "docs" / "engine-use-claim.md"
    text = path.read_text(encoding="utf-8")
    assert "access-engine" in text
    assert SENTINEL not in text
    lowered = text.lower()
    assert "token=" not in lowered
    assert "bao_token" not in lowered