secrets-engine/tests/test_openrouter_binding.py

"""A named native lane must not hide a changed custody or delivery target."""
import copy
from pathlib import Path

import pytest

from secrets_engine.authorization import build_action_request, approval_binding_digest, validate_decision_envelope
from secrets_engine.catalog import load_entry, validate_entry
from secrets_engine.errors import DecisionError
from secrets_engine.plan import build_plan
from tests.test_action_authorization import _envelope

CATALOG = Path(__file__).resolve().parents[1] / "catalog/openrouter-llm-connect.yaml"


def request(entry):
    return build_action_request(entry, "apply", subject_id="secrets-engine",
        subject_type="service", request_id="check:openrouter-apply", purpose="IR-WP-0004 native OpenRouter access",
        policy_targets=[entry.policy_name], auth_targets=[entry.role_name])


@pytest.mark.parametrize("change", ["path", "mount", "repo", "consumer", "ttl", "uses", "field", "delivery"])
def test_old_approval_cannot_follow_changed_native_target(change):
    entry = load_entry(CATALOG)
    original = request(entry)
    data = copy.deepcopy(entry.raw)
    if change in {"path", "mount", "repo"}: data[change] += "-other"
    elif change == "consumer": data["consumers"][1]["name"] = "unrelated-recipient"
    elif change == "ttl": data["delivery_auth"]["token_max_ttl"] = "24h"
    elif change == "uses": data["delivery_auth"]["secret_id_num_uses"] = 0
    elif change == "field": data["fields"].append("OTHER_KEY")
    elif change == "delivery": data["delivery_modes"].append("exec-file")
    changed = request(validate_entry(data))
    assert approval_binding_digest(original) != approval_binding_digest(changed)
    with pytest.raises(DecisionError, match="submitted request digest"):
        validate_decision_envelope(_envelope(original), changed,
            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
            accepted_policy_versions={"v2"})


def test_issuing_approval_does_not_change_its_own_target():
    entry = load_entry(CATALOG)
    original = request(entry)
    data = copy.deepcopy(entry.raw)
    data["approval"]["authorization_id"] = "new-approval-object"
    assert approval_binding_digest(original) == approval_binding_digest(request(validate_entry(data)))
    entry.delivery_auth["token_max_ttl"] = "24h"
    assert original["context"]["catalog_target"]["delivery_auth"]["token_max_ttl"] == "30m"


def test_openrouter_plan_exposes_all_limits_and_preserves_existing_custody():
    entry = load_entry(CATALOG)
    plan = build_plan(entry, "prod")
    assert [a.kind for a in plan.actions] == ["kv-mount-check", "policy", "approle"]
    assert plan.policy_name == plan.role_name == "se-prod-openrouter-llm-connect"
    assert entry.kv_data_path in plan.policy_hcl
    assert plan.actions[-1].detail == {
        "token_policies": plan.policy_name, "auth": "approle",
        "token_ttl": "15m", "token_max_ttl": "30m", "token_num_uses": 8,
        "secret_id_ttl": "15m", "secret_id_num_uses": 1,
    }


@pytest.mark.parametrize("action", ["apply", "verify", "exec"])
def test_cli_submits_the_exact_reviewed_openrouter_targets(action, tmp_path, monkeypatch):
    import json
    from secrets_engine import cli
    from secrets_engine.authorization import digest_material
    from secrets_engine.config import Config
    root = Path(__file__).resolve().parents[1]
    entry = load_entry(root / "docs/proposals/openrouter-key-check.yaml")
    expected = json.loads((root / f"docs/evidence/2026-09-14-openrouter-final-{action}-request.json").read_text())
    class Observed(Exception):
        pass
    def observe(cfg, actual_entry, actual_action, decision, **kwargs):
        actual = build_action_request(actual_entry, actual_action,
            subject_id="secrets-engine", subject_type="service",
            purpose=actual_entry.approval["purpose"], **kwargs)
        assert digest_material(actual) == digest_material(expected)
        raise Observed
    monkeypatch.setattr(cli, "authorize_action", observe)
    with pytest.raises(Observed):
        cli._require_lane_approval(Config.load(), entry, action,
            fields=() if action == "apply" else tuple(entry.fields))