ops-mason/tests/test_plan_survey.py

"""plan-survey reads metadata only, never prints a token, and fails loudly."""
import datetime as dt
import importlib.util
import json
import sys
from pathlib import Path

import pytest

spec = importlib.util.spec_from_file_location("plan_survey", Path(__file__).parents[1] / "scripts/plan-survey.py")
survey = importlib.util.module_from_spec(spec)
sys.modules["plan_survey"] = survey
spec.loader.exec_module(survey)

TOKEN = "hvs.SYNTHETICgrantTOKENvalue0123456789"
NOW = dt.datetime(2026, 10, 5, 12, 0, tzinfo=dt.timezone.utc)
ARGS = ["--kv-path", "platform/workloads/inbound-library/imap",
        "--policy", "workload-kv-read-inbound-library-imap",
        "--approle", "inbound-library", "--consumer", "inbound-library", "--keyword", "imap"]

CATALOG = """
version: 1
entries:
  - id: inbound-library-imap
    title: inbound-library IMAP mailbox
    need_keywords: [inbound-library, imap]
    owner_repo: ops-mason
    path_template: platform/workloads/inbound-library/imap
    status: draft
  - id: npm-publish-lane
    title: something else
    need_keywords: [npm]
    owner_repo: railiance-platform
    status: active
"""

POLICY_GRANTING = '''path "platform/data/workloads/inbound-library/imap" {
  capabilities = ["read"]
}
path "platform/metadata/workloads/inbound-library/imap" {
  capabilities = ["read"]
}
'''


def lookup(policies=("default", "ops-mason-survey")):
    return {"data": {"id": TOKEN, "accessor": "acc-SECRET", "display_name": "token-claude-mason-wp-0008",
                     "policies": list(policies), "ttl": 2000}}


class FakeBao:
    """Records every argv; answers from a table keyed by positional words."""

    def __init__(self, table=None, denied=(), missing=()):
        self.calls = []
        self.table = {
            ("token", "lookup"): lookup(),
            ("secrets", "list"): {"platform/": {"type": "kv", "options": {"version": "2"}, "description": "workloads"},
                                  "sys/": {"type": "system"}},
            ("auth", "list"): {"approle/": {"type": "approle"}, "kubernetes/": {"type": "kubernetes"},
                               "token/": {"type": "token"}},
            ("policy", "list"): ["default", "root", "ops-mason-survey", "workload-kv-read-inbound-library-imap",
                                 "workload-kv-read-other"],
            ("list", "auth/approle/role"): ["inbound-library", "rein-openweights"],
            ("list", "auth/kubernetes/role"): ["external-secrets-activity-core"],
            ("read", "auth/approle/role/inbound-library"): {"data": {
                "token_policies": ["workload-kv-read-inbound-library-imap"], "token_ttl": 900}},
            ("policy", "read", "workload-kv-read-inbound-library-imap"): POLICY_GRANTING,
            ("kv", "metadata", "get", "platform/workloads/inbound-library/imap"): {"data": {
                "current_version": 1, "created_time": "2026-10-01T00:00:00Z",
                "custom_metadata": {"description": "IMAP login for inbound-library"}}},
            ("kv", "list", "platform/workloads/inbound-library"): ["imap"],
        }
        self.table.update(table or {})
        self.denied, self.missing = set(denied), set(missing)

    def __call__(self, argv, env):
        assert argv[0] == "bao"
        assert env["BAO_TOKEN"] == TOKEN
        self.calls.append(argv)
        key = tuple(a for a in argv[1:] if not a.startswith("-"))
        if key in self.denied:
            return 2, "", "Error: permission denied"
        if key in self.missing or key not in self.table:
            return 2, "", "No value found"
        value = self.table[key]
        return 0, value if isinstance(value, str) else json.dumps(value), ""


@pytest.fixture
def catalog(tmp_path):
    p = tmp_path / "catalog.yaml"
    p.write_text(CATALOG)
    return p


def run(fake, catalog, capsys, args=ARGS):
    rc = survey.main([*args, "--catalog", str(catalog)], runner=fake,
                     environ={"BAO_TOKEN": TOKEN, "BAO_ADDR": "http://127.0.0.1:18200"}, now=NOW)
    return rc, capsys.readouterr().out


def assert_metadata_only(calls):
    for argv in calls:
        pos = [a for a in argv[1:] if not a.startswith("-")]
        assert not (pos[:1] == ["kv"] and "get" in pos and "metadata" not in pos), argv
        assert pos[:2] != ["kv", "get"], argv
        assert not any("/data/" in a or a.startswith("data/") for a in pos), argv
        assert not any(seg in ("role-id", "secret-id") for a in pos for seg in a.split("/")), argv


def test_existing_lane_found_and_only_metadata_read(catalog, capsys):
    fake = FakeBao()
    rc, out = run(fake, catalog, capsys)
    assert rc == 0
    assert_metadata_only(fake.calls)
    assert "## 2. Existing-structure survey" in out
    assert "**Draft answer: yes.**" in out
    assert "token-claude-mason-wp-0008" in out and "BAO_ADDR=http://127.0.0.1:18200" in out
    assert "2026-10-05T12:00:00Z" in out
    assert "(5 policies, 1 relevant)" in out
    assert "inbound-library-imap  owner=ops-mason" in out and "npm-publish-lane" not in out
    assert "rein-openweights" not in out
    assert TOKEN not in out and "acc-SECRET" not in out


def test_no_existing_lane(catalog, capsys, tmp_path):
    empty = tmp_path / "empty.yaml"
    empty.write_text("version: 1\nentries: []\n")
    fake = FakeBao(table={("policy", "list"): ["default"], ("list", "auth/approle/role"): ["other"]},
                   missing={("kv", "metadata", "get", "platform/workloads/inbound-library/imap"),
                            ("kv", "list", "platform/workloads/inbound-library")})
    rc, out = run(fake, empty, capsys)
    assert rc == 0
    assert "**Draft answer: no.**" in out
    assert "no metadata (path unused)" in out
    assert_metadata_only(fake.calls)


def test_policy_without_bound_role_is_partial(catalog, capsys):
    fake = FakeBao(table={("list", "auth/approle/role"): ["other"]})
    rc, out = run(fake, catalog, capsys)
    assert "**Draft answer: partial.**" in out
    assert "no surveyed role binds it" in out


def test_missing_capability_degrades_and_is_reported(catalog, capsys):
    fake = FakeBao(denied={("list", "auth/approle/role"), ("list", "auth/kubernetes/role"),
                           ("policy", "read", "workload-kv-read-inbound-library-imap")})
    rc, out = run(fake, catalog, capsys)
    assert rc == 0
    assert "not permitted by ops-mason-survey" in out
    assert "incomplete:" in out


def test_no_grant_fails_loudly(monkeypatch, tmp_path):
    monkeypatch.setattr(survey.os.path, "expanduser", lambda p: str(tmp_path / "absent"))
    with pytest.raises(survey.SurveyError, match="no ops-mason-survey grant"):
        survey.main(ARGS, runner=FakeBao(), environ={})


@pytest.mark.parametrize("policies", [("root",), ("default", "ops-mason-survey", "platform-admin"), ("default",)])
def test_non_grant_token_is_refused(catalog, policies):
    fake = FakeBao(table={("token", "lookup"): lookup(policies)})
    with pytest.raises(survey.SurveyError, match="not an ops-mason-survey grant") as exc:
        survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN})
    assert TOKEN not in str(exc.value)
    assert len(fake.calls) == 1


def test_expired_grant_fails_loudly(catalog):
    fake = FakeBao(denied={("token", "lookup")})
    with pytest.raises(survey.SurveyError, match="lookup failed"):
        survey.main([*ARGS, "--catalog", str(catalog)], runner=fake, environ={"BAO_TOKEN": TOKEN})


def test_script_exits_nonzero_without_grant(tmp_path):
    import os
    import subprocess
    env = {k: v for k, v in os.environ.items() if k not in ("BAO_TOKEN", "VAULT_TOKEN")}
    env["HOME"] = str(tmp_path)
    p = subprocess.run([sys.executable, str(Path(__file__).parents[1] / "scripts/plan-survey.py"), *ARGS],
                       capture_output=True, text=True, env=env)
    assert p.returncode == 2
    assert "ERROR: no ops-mason-survey grant" in p.stderr


@pytest.mark.parametrize("args", [
    ["kv", "get", "platform/workloads/x"],
    ["kv", "get", "-field=password", "platform/workloads/x"],
    ["read", "platform/data/workloads/x"],
    ["list", "platform/data/workloads"],
    ["kv", "metadata", "get", "platform/data/x"],
    ["read", "auth/approle/role/x/role-id"],
    ["write", "auth/approle/role/x/secret-id"],
    ["read", "sys/raw/x"],
    ["policy", "write", "x", "-"],
    ["token", "create"],
])
def test_guard_refuses_value_reads_and_writes(args):
    assert not survey.allowed_bao_call(args)
    bao = survey.Bao(FakeBao(), {"BAO_TOKEN": TOKEN})
    with pytest.raises(survey.SurveyError, match="refused"):
        bao.call(*args)


def test_kv_path_must_not_be_an_api_data_path(catalog):
    with pytest.raises(survey.SurveyError, match="logical KV path"):
        survey.main(["--kv-path", "platform/data/workloads/x", "--policy", "p", "--consumer", "c"],
                    runner=FakeBao(), environ={"BAO_TOKEN": TOKEN})


def test_policy_glob_matching():
    assert survey.path_matches("platform/data/workloads/*", "platform/data/workloads/a/b")
    assert survey.path_matches("platform/data/+/a/b", "platform/data/workloads/a/b")
    assert not survey.path_matches("platform/data/workloads/a", "platform/data/workloads/a/b")