generated from coulomb/repo-seed
tegwick
be3b4a2a86
resolve_fetch_command used shlex.split, which treated `|` as a literal argument — breaking reuse-surface-hub-write-token (kubectl | base64 -d). Piped commands now run via shell=True with inherited stdio for --fetch.
115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
"""Tests for the dev-tier contract-double fixture library (WP-0015 T4)."""
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from warden.doubles import (
|
|
SYNTHETIC_PREFIX,
|
|
available_doubles,
|
|
doubles_path_prepended,
|
|
materialize_doubles,
|
|
)
|
|
|
|
|
|
def test_available_doubles_includes_routed_subsystems():
|
|
names = available_doubles()
|
|
assert "bao" in names
|
|
assert "key-cape" in names
|
|
|
|
|
|
def test_materialize_writes_executables(tmp_path):
|
|
paths = materialize_doubles(tmp_path)
|
|
assert set(paths) == set(available_doubles())
|
|
for p in paths.values():
|
|
assert p.exists()
|
|
import os
|
|
|
|
assert os.access(p, os.X_OK)
|
|
|
|
|
|
def test_bao_kv_get_emits_synthetic_value(tmp_path):
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "bao"), "kv", "get", "-field=NPM_AUTH_TOKEN", "platform/x/y"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
value = out.stdout.strip()
|
|
assert value.startswith(SYNTHETIC_PREFIX)
|
|
assert "NPM_AUTH_TOKEN" in value
|
|
|
|
|
|
def test_bao_login_emits_synthetic_token(tmp_path):
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "bao"), "login", "-method=oidc"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
assert out.stdout.strip().startswith(SYNTHETIC_PREFIX)
|
|
|
|
|
|
def test_keycape_login_emits_synthetic_session(tmp_path):
|
|
materialize_doubles(tmp_path, ["key-cape"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "key-cape"), "login"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
assert out.stdout.strip().startswith(SYNTHETIC_PREFIX)
|
|
|
|
|
|
def test_double_rejects_unknown_contract(tmp_path):
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
out = subprocess.run(
|
|
[str(tmp_path / "bao"), "write", "secret/x"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert out.returncode == 2
|
|
|
|
|
|
def test_unknown_double_raises(tmp_path):
|
|
with pytest.raises(KeyError):
|
|
materialize_doubles(tmp_path, ["nonesuch"])
|
|
|
|
|
|
def test_path_prepended_puts_doubles_first(tmp_path):
|
|
path = doubles_path_prepended(tmp_path, base_path="/usr/bin")
|
|
assert path.split(":")[0] == str(tmp_path)
|
|
|
|
|
|
def test_proxy_fetch_runs_fully_offline_against_double(tmp_path):
|
|
"""End-to-end: the proxy fetch lane resolves `bao` from the doubles dir."""
|
|
import os
|
|
|
|
materialize_doubles(tmp_path, ["bao"])
|
|
from warden.proxy import resolve_fetch_command
|
|
from warden.routing.models import RouteEntry
|
|
|
|
entry = RouteEntry(
|
|
id="openbao-api-key",
|
|
title="API key",
|
|
need_keywords=["npm"],
|
|
owner_repo="railiance-platform",
|
|
subsystem="OpenBao",
|
|
warden_executes=False,
|
|
wiki_ref="w",
|
|
canon_ref="c",
|
|
reviewed="2026-06-27",
|
|
status="active",
|
|
path_template="platform/x/y/z",
|
|
fetch_command="bao kv get -field=<FIELD> <path_template>",
|
|
exec_capable=True,
|
|
)
|
|
resolved = resolve_fetch_command(entry, field="API_KEY", path="platform/x/y/z")
|
|
env = dict(os.environ, PATH=doubles_path_prepended(tmp_path))
|
|
# proxy_fetch inherits stdout; run it in a child so we can capture the stream.
|
|
result = subprocess.run(resolved.argv, capture_output=True, text=True, env=env, check=True)
|
|
assert result.stdout.strip().startswith(SYNTHETIC_PREFIX)
|