test(e2e): cross-OS wiring harness + 3-OS CI, red-first (FPLAN-0239)
Build-wheel -> install-clean -> assert 4-tier wiring ladder (install/init/ hooks-fire/drone-route). Validated green on Linux; Windows red-first by design (symlink/venv-path/tmp gaps = DPLAN-0194 P3 fix-list). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
3ad0580070
commit
cd1af34be8
@@ -0,0 +1,50 @@
|
||||
name: e2e-wheel
|
||||
|
||||
# Cross-OS end-to-end WIRING test (FPLAN-0239, P1 of DPLAN-0194).
|
||||
# Builds the wheel, installs it into a clean venv (handled by the pytest
|
||||
# fixtures in tests/e2e/conftest.py), and runs the 4-tier wiring ladder.
|
||||
#
|
||||
# RED-FIRST: Windows is EXPECTED to fail in known places (symlink init,
|
||||
# bin-vs-Scripts, /tmp). Do not "fix" Windows here — the red is the deliverable.
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main, dev]
|
||||
paths:
|
||||
- "tests/e2e/**"
|
||||
- ".github/workflows/e2e-wheel.yml"
|
||||
- "pyproject.toml"
|
||||
- "src/**"
|
||||
pull_request:
|
||||
paths:
|
||||
- "tests/e2e/**"
|
||||
- ".github/workflows/e2e-wheel.yml"
|
||||
- "pyproject.toml"
|
||||
- "src/**"
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
e2e-wheel:
|
||||
name: e2e-wheel (${{ matrix.os }})
|
||||
runs-on: ${{ matrix.os }}
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
os: [ubuntu-latest, windows-latest, macos-latest]
|
||||
python-version: ["3.12"]
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
|
||||
- name: Set up Python ${{ matrix.python-version }}
|
||||
uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
|
||||
- name: Install build tooling
|
||||
run: python -m pip install --upgrade pip build pytest
|
||||
|
||||
- name: Run cross-OS e2e wiring harness
|
||||
# conftest.py builds the wheel + clean venv internally; the outer env
|
||||
# only needs build + pytest.
|
||||
run: python -m pytest tests/e2e -v
|
||||
@@ -12,6 +12,16 @@ and this project uses [Calendar Versioning](https://calver.org/) in the format
|
||||
|
||||
### Added
|
||||
|
||||
- **Cross-OS end-to-end WIRING test (`tests/e2e/`, `e2e-wheel.yml`)** — the first
|
||||
CI gate that proves real AIPass *wiring* (not units-with-mocks) by building the
|
||||
wheel, installing it into a clean venv, and asserting a 4-tier ladder: package
|
||||
install + console scripts (T0), `aipass init` scaffolding (T1), a hook actually
|
||||
firing via the bridge with an observable `engine.jsonl` record (T2a), and
|
||||
`drone` resolving + subprocess-executing a real branch (T3). Runs on a 3-OS
|
||||
matrix (ubuntu/windows/macos, `fail-fast: false`). Validated green on Linux;
|
||||
Windows is **red-first by design** — the expected failures (unguarded `.venv`
|
||||
symlink, `.venv/bin` vs `Scripts`, hardcoded `/tmp`) are the portability
|
||||
fix-list, not regressions. (DPLAN-0194 / FPLAN-0239)
|
||||
- **`drone rm` — provider-agnostic safe delete** — a contained recursive delete
|
||||
that lets agents clean up scratch dirs without tripping the `rm -rf` block.
|
||||
Deletes are confined to the project root and the system temp dirs (`/tmp` and
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# DPLAN-0194 — P1 cross-OS e2e wiring harness PROTOTYPE (Linux/Docker dev loop).
|
||||
# Runs INSIDE a clean container with the repo bind-mounted read-only at /repo.
|
||||
# Builds a wheel, installs into a CLEAN venv, then asserts the 4 tiers:
|
||||
# T0 install+binaries T1 aipass init scaffold T2a synthetic hook fire T3 drone routing
|
||||
# Tolerant: never exits on first failure — runs every assertion so we see the full red/green ladder.
|
||||
#
|
||||
set -uo pipefail
|
||||
|
||||
P=0; F=0
|
||||
ok(){ echo " ok $1"; P=$((P+1)); }
|
||||
no(){ echo " XX $1"; F=$((F+1)); }
|
||||
chk(){ if eval "$2" >/dev/null 2>&1; then ok "$1"; else no "$1"; fi; }
|
||||
hdr(){ echo; echo "=== $1 ==="; }
|
||||
|
||||
SRC=~/src
|
||||
BUILDENV=/tmp/buildenv
|
||||
CLEANENV=/tmp/cleanenv
|
||||
DIST=/tmp/dist
|
||||
PY=$CLEANENV/bin/python
|
||||
AIPASS=$CLEANENV/bin/aipass
|
||||
DRONE=$CLEANENV/bin/drone
|
||||
|
||||
hdr "SETUP — copy repo (writable), build wheel"
|
||||
rm -rf "$SRC" "$DIST" "$BUILDENV" "$CLEANENV"
|
||||
cp -r /repo "$SRC" 2>/dev/null || true # .trinity memory files are perm-restricted; harmless, code copies fine
|
||||
cd "$SRC"
|
||||
# A real fresh-clone runs setup.sh to GENERATE the registry (the host's AIPASS_REGISTRY.json
|
||||
# is mode 0600 and won't copy across uids anyway). Synthesize a minimal one pointing at the
|
||||
# copied branches — faithful to what setup.sh produces, lets Tier 3 prove routing plumbing.
|
||||
cat > "$SRC/AIPASS_REGISTRY.json" <<JSON
|
||||
{ "metadata": { "name": "AIPASS", "version": "1.0.0", "total_branches": 2 },
|
||||
"branches": [
|
||||
{ "name": "drone", "path": "$SRC/src/aipass/drone" },
|
||||
{ "name": "seedgo", "path": "$SRC/src/aipass/seedgo" }
|
||||
] }
|
||||
JSON
|
||||
python3 -m venv "$BUILDENV"
|
||||
"$BUILDENV/bin/pip" -q install --upgrade pip build 2>&1 | tail -2
|
||||
echo " building wheel..."
|
||||
"$BUILDENV/bin/python" -m build --wheel --outdir "$DIST" . 2>&1 | tail -4
|
||||
WHEEL=$(ls "$DIST"/*.whl 2>/dev/null | head -1)
|
||||
echo " wheel: ${WHEEL:-<NONE>}"
|
||||
|
||||
hdr "TIER 0 — clean-venv wheel install + binaries"
|
||||
python3 -m venv "$CLEANENV"
|
||||
if [ -n "${WHEEL:-}" ]; then
|
||||
"$CLEANENV/bin/pip" -q install "$WHEEL" 2>&1 | tail -3
|
||||
fi
|
||||
chk "wheel built" "[ -n '${WHEEL:-}' ]"
|
||||
chk "clean venv has pip (not silent-broken venv, #495)" "[ -x '$CLEANENV/bin/pip' ]"
|
||||
chk "aipass console_script installed" "[ -x '$AIPASS' ]"
|
||||
chk "drone console_script installed" "[ -x '$DRONE' ]"
|
||||
chk "drone --version runs" "'$DRONE' --version"
|
||||
chk "aipass entrypoint imports (aipass init --help)" "'$AIPASS' init --help"
|
||||
|
||||
hdr "TIER 1 — aipass init scaffolds correctly"
|
||||
PROJ=/tmp/proj; rm -rf "$PROJ"
|
||||
# AIPASS_HOME left UNSET on purpose: tests core scaffold independent of venv/templates,
|
||||
# and sidesteps the .venv symlink (the symlink bug is a Windows-only failure — N/A on Linux).
|
||||
"$AIPASS" init "$PROJ" demo > /tmp/init.out 2>&1
|
||||
echo " init exit=$? (see /tmp/init.out)"; tail -3 /tmp/init.out | sed 's/^/ | /'
|
||||
chk "DEMO_REGISTRY.json exists" "[ -f '$PROJ/DEMO_REGISTRY.json' ]"
|
||||
chk "DEMO_REGISTRY.json is valid JSON" "jq -e . '$PROJ/DEMO_REGISTRY.json'"
|
||||
chk "registry metadata.name == DEMO" "[ \"\$(jq -r .metadata.name '$PROJ/DEMO_REGISTRY.json')\" = DEMO ]"
|
||||
chk ".claude/settings.json exists" "[ -f '$PROJ/.claude/settings.json' ]"
|
||||
chk "settings deny has EnterPlanMode" "jq -e '.permissions.deny|index(\"EnterPlanMode\")' '$PROJ/.claude/settings.json'"
|
||||
chk "src/demo/__init__.py exists" "[ -f '$PROJ/src/demo/__init__.py' ]"
|
||||
chk ".gitignore mentions .venv" "grep -q '.venv' '$PROJ/.gitignore'"
|
||||
chk ".trinity/ NOT created (projects!=citizens)" "[ ! -e '$PROJ/.trinity' ]"
|
||||
chk "no passport.json created" "[ ! -e '$PROJ/.trinity/passport.json' ]"
|
||||
|
||||
hdr "TIER 2a — synthetic hook fire (module form, sentinel UUID, engine.jsonl)"
|
||||
HOOKP=/tmp/hookproj; rm -rf "$HOOKP"; mkdir -p "$HOOKP/.aipass"
|
||||
# minimal isolated config: ONLY rm_gate enabled -> no git_gate/sound noise
|
||||
cat > "$HOOKP/.aipass/hooks.json" <<'JSON'
|
||||
{ "hooks_enabled": true,
|
||||
"PreToolUse": {
|
||||
"rm_gate": { "enabled": true, "handler": "aipass.hooks.apps.handlers.security.rm_gate.handle", "matcher": "Bash" }
|
||||
} }
|
||||
JSON
|
||||
UUID="PROTOUUID12345"
|
||||
LOG=$(find "$CLEANENV" -path '*/aipass/hooks/logs/engine.jsonl' 2>/dev/null | head -1)
|
||||
LOGDIR=$(dirname "$(find "$CLEANENV" -path '*/aipass/hooks' -type d 2>/dev/null | head -1)")
|
||||
[ -n "$LOG" ] && : > "$LOG" # truncate if present
|
||||
# fire: rm -rf -> expect block (exit 2)
|
||||
OUT=$(cd "$HOOKP" && echo "{\"tool_name\":\"Bash\",\"tool_input\":{\"command\":\"rm -rf /tmp/x\"},\"agent_id\":\"$UUID\"}" \
|
||||
| AIPASS_HOME="$HOOKP" "$PY" -m aipass.hooks.apps.handlers.bridges.claude "PreToolUse:rm_gate" 2>/tmp/hook.err)
|
||||
HX=$?
|
||||
# relocate LOG now if it didn't exist before
|
||||
[ -z "$LOG" ] && LOG=$(find "$CLEANENV" -path '*/aipass/hooks/logs/engine.jsonl' 2>/dev/null | head -1)
|
||||
printf '%s' "$OUT" > /tmp/hook.out # write to file: never eval-interpolate captured JSON
|
||||
echo " hook exit=$HX stdout=${OUT:0:80}"
|
||||
[ -s /tmp/hook.err ] && echo " stderr: $(head -1 /tmp/hook.err)"
|
||||
# REAL contract (discovered by prototype): rm_gate blocks via {"decision":"block"} on STDOUT, exit 0 — NOT exit 2.
|
||||
chk "rm_gate decision==block (stdout JSON)" "jq -e '.decision==\"block\"' /tmp/hook.out"
|
||||
chk "bridge exit 0 (block via JSON not code)" "[ '$HX' = 0 ]"
|
||||
chk "engine.jsonl exists" "[ -n '$LOG' ] && [ -f '$LOG' ]"
|
||||
chk "engine.jsonl logged sentinel UUID" "[ -n '$LOG' ] && grep -q '$UUID' '$LOG'"
|
||||
chk "logged record hook==rm_gate" "[ -n '$LOG' ] && grep '$UUID' '$LOG' | grep -q rm_gate"
|
||||
# negative: harmless echo -> allow (exit 0)
|
||||
OUT2=$(cd "$HOOKP" && echo "{\"tool_name\":\"Bash\",\"tool_input\":{\"command\":\"echo hi\"},\"agent_id\":\"$UUID-neg\"}" \
|
||||
| AIPASS_HOME="$HOOKP" "$PY" -m aipass.hooks.apps.handlers.bridges.claude "PreToolUse:rm_gate" 2>/dev/null)
|
||||
HX2=$?
|
||||
chk "rm_gate allows echo (exit 0)" "[ '$HX2' = 0 ]"
|
||||
|
||||
hdr "TIER 3 — drone routing (against real repo registry)"
|
||||
chk "drone systems runs (reads registry)" "cd '$SRC' && '$DRONE' systems"
|
||||
chk "drone systems lists a known branch" "cd '$SRC' && '$DRONE' systems 2>/dev/null | grep -qi seedgo"
|
||||
chk "drone @drone --help routes" "cd '$SRC' && '$DRONE' @drone --help"
|
||||
|
||||
hdr "RESULT"
|
||||
echo " PASS=$P FAIL=$F"
|
||||
[ "$F" -eq 0 ] && echo " ALL GREEN" || echo " $F red — that's the truth we wanted"
|
||||
exit 0
|
||||
@@ -0,0 +1,139 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: conftest.py
|
||||
# Description: Session-scoped fixtures for the cross-OS e2e wiring harness
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-06-03
|
||||
# =============================================
|
||||
|
||||
"""Session-scoped pytest fixtures for the cross-OS end-to-end WIRING harness.
|
||||
|
||||
These fixtures build the aipass wheel and install it into a FRESH, clean venv
|
||||
(never the repo .venv, never ``pip install -e``) so the tests in this package
|
||||
exercise the real installed package the way a contributor on any OS would.
|
||||
|
||||
CRITICAL cross-OS-harness rule: this harness must itself run on Windows. The
|
||||
venv binary directory is ``Scripts`` on Windows and ``bin`` on POSIX, and the
|
||||
script extension is ``.exe`` on Windows. We resolve those from ``os.name`` /
|
||||
``sys.executable`` and never hardcode — the harness must NOT contain the very
|
||||
bugs it tests for.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# Repo root = three levels up from this file: <repo>/tests/e2e/conftest.py
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def _venv_bin_dir(venv_root: Path) -> Path:
|
||||
"""Return the venv directory that holds executables for THIS platform.
|
||||
|
||||
Windows venvs put scripts in ``Scripts``; POSIX venvs use ``bin``. This is
|
||||
exactly the bin-vs-Scripts split the harness exists to expose, so the
|
||||
harness must resolve it correctly itself.
|
||||
"""
|
||||
return venv_root / ("Scripts" if os.name == "nt" else "bin")
|
||||
|
||||
|
||||
def _exe(name: str) -> str:
|
||||
"""Append the Windows executable suffix to a console-script name."""
|
||||
return f"{name}.exe" if os.name == "nt" else name
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CleanVenv:
|
||||
"""Paths into a clean venv with the aipass wheel installed."""
|
||||
|
||||
root: Path
|
||||
python: Path
|
||||
pip: Path
|
||||
aipass: Path
|
||||
drone: Path
|
||||
site_packages: Path
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def wheel(tmp_path_factory: pytest.TempPathFactory) -> Path:
|
||||
"""Build the aipass wheel from the repo root and return its path.
|
||||
|
||||
Uses ``python -m build --wheel`` (build backend = hatchling, package =
|
||||
aipass 2.5.0). The outer environment running pytest only needs ``build``
|
||||
and ``pytest`` installed — this fixture produces the wheel that the
|
||||
clean-venv fixture then installs.
|
||||
"""
|
||||
dist_dir = tmp_path_factory.mktemp("wheel_dist")
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "build", "--wheel", "--outdir", str(dist_dir), str(REPO_ROOT)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(
|
||||
"wheel build failed (exit "
|
||||
f"{result.returncode}).\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
|
||||
)
|
||||
|
||||
wheels = sorted(dist_dir.glob("*.whl"))
|
||||
if not wheels:
|
||||
raise RuntimeError(f"no wheel produced in {dist_dir}. build output:\n{result.stdout}")
|
||||
return wheels[0]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def clean_venv(wheel: Path, tmp_path_factory: pytest.TempPathFactory) -> CleanVenv:
|
||||
"""Create a fresh venv and install the wheel into it.
|
||||
|
||||
NOT the repo .venv, NOT an editable install — a brand-new venv with the
|
||||
built wheel installed, so we test the wiring of the real package.
|
||||
"""
|
||||
venv_root = tmp_path_factory.mktemp("clean_venv")
|
||||
|
||||
# Build the venv with the current interpreter — no hardcoded "python".
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "venv", str(venv_root)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"venv creation failed:\n{result.stdout}\n{result.stderr}")
|
||||
|
||||
bin_dir = _venv_bin_dir(venv_root)
|
||||
py = bin_dir / _exe("python")
|
||||
pip = bin_dir / _exe("pip")
|
||||
|
||||
install = subprocess.run(
|
||||
[str(py), "-m", "pip", "install", str(wheel)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
if install.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"wheel install into clean venv failed:\nSTDOUT:\n{install.stdout}\nSTDERR:\n{install.stderr}"
|
||||
)
|
||||
|
||||
# Resolve site-packages from the venv's own interpreter — portable across
|
||||
# OSes and python minor versions instead of guessing lib/pythonX.Y.
|
||||
sp = subprocess.run(
|
||||
[str(py), "-c", "import sysconfig; print(sysconfig.get_path('purelib'))"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
if sp.returncode != 0:
|
||||
raise RuntimeError(f"could not resolve site-packages:\n{sp.stdout}\n{sp.stderr}")
|
||||
site_packages = Path(sp.stdout.strip())
|
||||
|
||||
return CleanVenv(
|
||||
root=venv_root,
|
||||
python=py,
|
||||
pip=pip,
|
||||
aipass=bin_dir / _exe("aipass"),
|
||||
drone=bin_dir / _exe("drone"),
|
||||
site_packages=site_packages,
|
||||
)
|
||||
@@ -0,0 +1,378 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: test_wiring.py
|
||||
# Description: Cross-OS end-to-end WIRING tests against the installed wheel
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-06-03
|
||||
# =============================================
|
||||
|
||||
"""Cross-OS end-to-end WIRING tests (FPLAN-0239, P1 of DPLAN-0194).
|
||||
|
||||
This proves AIPass *wiring* — not units-with-mocks — by building the wheel,
|
||||
installing it into a clean venv (see ``conftest.py``), and asserting a 4-tier
|
||||
ladder against the real installed package:
|
||||
|
||||
T0 install + console scripts exist and run
|
||||
T1 ``aipass init`` scaffolds a project correctly
|
||||
T2a a hook actually fires, blocks, and leaves a sentinel record
|
||||
T3 ``drone`` resolves a real branch and executes it via subprocess
|
||||
|
||||
It is RED-FIRST: it is expected to fail on Windows in known places (symlink
|
||||
init, bin-vs-Scripts, /tmp). The harness itself is written to be cross-OS so
|
||||
those reds reflect AIPass bugs, not harness bugs.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from conftest import REPO_ROOT, CleanVenv
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _run(cmd: list[str], **kwargs) -> subprocess.CompletedProcess:
|
||||
"""Run a subprocess capturing text output with a bounded timeout."""
|
||||
kwargs.setdefault("capture_output", True)
|
||||
kwargs.setdefault("text", True)
|
||||
kwargs.setdefault("timeout", 60)
|
||||
return subprocess.run(cmd, **kwargs)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# TIER 0 — clean-venv wheel install + binaries
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
def test_t0_wheel_built(wheel: Path) -> None:
|
||||
"""The wheel built and is named for aipass."""
|
||||
assert wheel.exists()
|
||||
assert wheel.suffix == ".whl"
|
||||
assert wheel.name.startswith("aipass-")
|
||||
|
||||
|
||||
def test_t0_venv_has_pip(clean_venv: CleanVenv) -> None:
|
||||
"""Clean venv has a working pip — not a silently-broken venv (ref #495)."""
|
||||
result = _run([str(clean_venv.python), "-m", "pip", "--version"])
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_t0_console_scripts_exist(clean_venv: CleanVenv) -> None:
|
||||
"""Both ``aipass`` and ``drone`` console scripts are installed."""
|
||||
assert clean_venv.aipass.exists(), f"missing aipass at {clean_venv.aipass}"
|
||||
assert clean_venv.drone.exists(), f"missing drone at {clean_venv.drone}"
|
||||
|
||||
|
||||
def test_t0_drone_version_runs(clean_venv: CleanVenv) -> None:
|
||||
"""``drone --version`` runs (entry point imports cleanly)."""
|
||||
result = _run([str(clean_venv.drone), "--version"])
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_t0_aipass_init_help_runs(clean_venv: CleanVenv) -> None:
|
||||
"""``aipass init --help`` runs (the init entry point imports cleanly)."""
|
||||
result = _run([str(clean_venv.aipass), "init", "--help"])
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# TIER 1 — aipass init scaffolds correctly
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def init_project(clean_venv: CleanVenv, tmp_path_factory: pytest.TempPathFactory) -> Path:
|
||||
"""Run ``aipass init <proj> demo`` in a clean neutral dir, return the project.
|
||||
|
||||
``aipass init`` REFUSES when a ``*_REGISTRY.json`` sits in or above CWD
|
||||
(``_guard_init``), so we run it from a pristine tmp dir whose parents have
|
||||
no registry. ``AIPASS_HOME`` is left UNSET so the .venv-symlink step (a
|
||||
Windows-only failure tracked separately) is skipped — this test must pass
|
||||
on Linux.
|
||||
"""
|
||||
neutral = tmp_path_factory.mktemp("neutral_init_cwd")
|
||||
proj = neutral / "proj"
|
||||
|
||||
env = {
|
||||
"PATH": _path_env(),
|
||||
"HOME": str(neutral),
|
||||
}
|
||||
if sys.platform == "win32":
|
||||
# Windows needs a few base vars for subprocess/venv tooling to work.
|
||||
import os as _os
|
||||
|
||||
for key in ("SYSTEMROOT", "TEMP", "TMP", "USERPROFILE", "PATHEXT", "COMSPEC"):
|
||||
if key in _os.environ:
|
||||
env[key] = _os.environ[key]
|
||||
|
||||
result = subprocess.run(
|
||||
[str(clean_venv.aipass), "init", str(proj), "demo"],
|
||||
cwd=str(neutral),
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120,
|
||||
)
|
||||
assert result.returncode == 0, (
|
||||
f"aipass init failed (exit {result.returncode}).\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
|
||||
)
|
||||
return proj
|
||||
|
||||
|
||||
def _path_env() -> str:
|
||||
"""Minimal PATH for the init subprocess (portable across OSes)."""
|
||||
import os
|
||||
|
||||
return os.environ.get("PATH", "")
|
||||
|
||||
|
||||
def test_t1_registry_created_and_valid(init_project: Path) -> None:
|
||||
"""DEMO_REGISTRY.json exists, is valid JSON, and names DEMO."""
|
||||
registry = init_project / "DEMO_REGISTRY.json"
|
||||
assert registry.is_file(), f"missing {registry}"
|
||||
data = json.loads(registry.read_text(encoding="utf-8"))
|
||||
assert data["metadata"]["name"] == "DEMO"
|
||||
|
||||
|
||||
def test_t1_claude_settings_deny_enterplanmode(init_project: Path) -> None:
|
||||
""".claude/settings.json exists and its permissions.deny blocks EnterPlanMode."""
|
||||
settings = init_project / ".claude" / "settings.json"
|
||||
assert settings.is_file(), f"missing {settings}"
|
||||
data = json.loads(settings.read_text(encoding="utf-8"))
|
||||
assert "EnterPlanMode" in data["permissions"]["deny"]
|
||||
|
||||
|
||||
def test_t1_src_package_scaffolded(init_project: Path) -> None:
|
||||
"""src/demo/__init__.py is created."""
|
||||
assert (init_project / "src" / "demo" / "__init__.py").is_file()
|
||||
|
||||
|
||||
def test_t1_gitignore_mentions_venv(init_project: Path) -> None:
|
||||
""".gitignore mentions .venv."""
|
||||
gitignore = init_project / ".gitignore"
|
||||
assert gitignore.is_file()
|
||||
assert ".venv" in gitignore.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_t1_no_trinity_no_passport(init_project: Path) -> None:
|
||||
"""Projects are NOT citizens: no .trinity/ dir and no passport.json."""
|
||||
assert not (init_project / ".trinity").exists()
|
||||
assert not (init_project / ".trinity" / "passport.json").exists()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# TIER 2a — synthetic hook fire (module form, sentinel UUID, engine.jsonl)
|
||||
# ===========================================================================
|
||||
|
||||
_RM_GATE_CONFIG = {
|
||||
"hooks_enabled": True,
|
||||
"PreToolUse": {
|
||||
"rm_gate": {
|
||||
"enabled": True,
|
||||
"handler": "aipass.hooks.apps.handlers.security.rm_gate.handle",
|
||||
"matcher": "Bash",
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _fire_hook(clean_venv: CleanVenv, work: Path, command: str, agent_id: str) -> subprocess.CompletedProcess:
|
||||
"""Invoke the Claude bridge in MODULE form for a single PreToolUse event.
|
||||
|
||||
Module form (``python -m aipass.hooks...``) is venv-portable and avoids the
|
||||
bin/Scripts split — the very bug this suite tests for.
|
||||
"""
|
||||
payload = json.dumps(
|
||||
{
|
||||
"tool_name": "Bash",
|
||||
"tool_input": {"command": command},
|
||||
"agent_id": agent_id,
|
||||
}
|
||||
)
|
||||
env = _hook_env(work)
|
||||
return subprocess.run(
|
||||
[
|
||||
str(clean_venv.python),
|
||||
"-m",
|
||||
"aipass.hooks.apps.handlers.bridges.claude",
|
||||
"PreToolUse:rm_gate",
|
||||
],
|
||||
input=payload,
|
||||
cwd=str(work),
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
def _hook_env(work: Path) -> dict:
|
||||
"""Build a minimal env for the hook subprocess, with AIPASS_HOME isolated."""
|
||||
import os
|
||||
|
||||
env = dict(os.environ)
|
||||
env["AIPASS_HOME"] = str(work)
|
||||
return env
|
||||
|
||||
|
||||
def _engine_log(clean_venv: CleanVenv) -> Path | None:
|
||||
"""Glob the installed package for aipass/hooks/logs/engine.jsonl."""
|
||||
hits = sorted(clean_venv.site_packages.glob("aipass/hooks/logs/engine.jsonl"))
|
||||
return hits[0] if hits else None
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def hook_workspace(tmp_path_factory: pytest.TempPathFactory) -> Path:
|
||||
"""A tmp workspace holding an isolated .aipass/hooks.json with only rm_gate."""
|
||||
work = tmp_path_factory.mktemp("hook_ws")
|
||||
aipass_dir = work / ".aipass"
|
||||
aipass_dir.mkdir(parents=True, exist_ok=True)
|
||||
(aipass_dir / "hooks.json").write_text(json.dumps(_RM_GATE_CONFIG), encoding="utf-8")
|
||||
return work
|
||||
|
||||
|
||||
def test_t2a_rm_gate_blocks(clean_venv: CleanVenv, hook_workspace: Path) -> None:
|
||||
"""rm -rf is blocked via {"decision":"block"} on STDOUT with exit code 0.
|
||||
|
||||
Corrected contract (NOT exit 2): the bridge writes the engine's block JSON
|
||||
to stdout and exits 0.
|
||||
"""
|
||||
sentinel = f"e2e-block-{uuid.uuid4()}"
|
||||
proc = _fire_hook(clean_venv, hook_workspace, "rm -rf /tmp/x", sentinel)
|
||||
|
||||
assert proc.returncode == 0, f"expected exit 0, got {proc.returncode}. stderr:\n{proc.stderr}"
|
||||
decision = json.loads(proc.stdout)
|
||||
assert decision.get("decision") == "block", f"stdout was: {proc.stdout!r}"
|
||||
|
||||
# Oracle: the engine log must hold a record with OUR sentinel AND hook==rm_gate.
|
||||
log = _engine_log(clean_venv)
|
||||
assert log is not None and log.is_file(), "engine.jsonl not found in installed package"
|
||||
assert _log_has_sentinel_for_hook(log, sentinel, "rm_gate"), (
|
||||
f"no engine.jsonl record found for sentinel {sentinel} + hook rm_gate"
|
||||
)
|
||||
|
||||
|
||||
def _log_has_sentinel_for_hook(log: Path, sentinel: str, hook: str) -> bool:
|
||||
"""Return True if any JSONL record has this agent_id AND this hook name.
|
||||
|
||||
Asserts on the sentinel, never line counts — the log is shared with live
|
||||
sessions.
|
||||
"""
|
||||
for line in log.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
rec = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if rec.get("agent_id") == sentinel and rec.get("hook") == hook:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def test_t2a_rm_gate_allows_echo(clean_venv: CleanVenv, hook_workspace: Path) -> None:
|
||||
"""A harmless command is allowed: exit 0 and decision is not "block"."""
|
||||
sentinel = f"e2e-allow-{uuid.uuid4()}"
|
||||
proc = _fire_hook(clean_venv, hook_workspace, "echo hi", sentinel)
|
||||
|
||||
assert proc.returncode == 0, proc.stderr
|
||||
if proc.stdout.strip():
|
||||
try:
|
||||
decision = json.loads(proc.stdout)
|
||||
assert decision.get("decision") != "block", f"unexpected block: {proc.stdout!r}"
|
||||
except json.JSONDecodeError:
|
||||
# Non-JSON / empty output is also a valid "allow" signal.
|
||||
pass
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# TIER 3 — drone routing (real resolve -> subprocess -> execute)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def routing_root(clean_venv: CleanVenv) -> Iterator[Path]:
|
||||
"""Generate a minimal registry at the repo root pointing at real branches.
|
||||
|
||||
The repo's own AIPASS_REGISTRY.json is mode 0600 / host-absolute and won't
|
||||
relocate (its paths are the developer's home, absent in CI), so we GENERATE
|
||||
a minimal registry — faithful to what setup.sh produces — pointing at REAL
|
||||
branch dirs under this checkout.
|
||||
|
||||
drone validates path containment against the registry's PARENT dir, so the
|
||||
registry must sit at the repo root for the ``src/aipass/*`` branch paths to
|
||||
validate. We back up any existing registry and restore it on teardown so a
|
||||
local run never mutates the working tree permanently.
|
||||
|
||||
We target ``ai_mail`` — a real BRANCH (not an in-process drone module) — so
|
||||
``drone @ai_mail --help`` exercises the full resolve -> subprocess ->
|
||||
execute path, the proof we never previously got green.
|
||||
"""
|
||||
src = REPO_ROOT / "src" / "aipass"
|
||||
registry = {
|
||||
"metadata": {"name": "AIPASS", "version": "1.0.0", "total_branches": 3},
|
||||
"branches": [
|
||||
{"name": "ai_mail", "path": str(src / "ai_mail"), "status": "active"},
|
||||
{"name": "seedgo", "path": str(src / "seedgo"), "status": "active"},
|
||||
{"name": "drone", "path": str(src / "drone"), "status": "active"},
|
||||
],
|
||||
}
|
||||
|
||||
registry_path = REPO_ROOT / "AIPASS_REGISTRY.json"
|
||||
backup = registry_path.read_bytes() if registry_path.exists() else None
|
||||
try:
|
||||
registry_path.write_text(json.dumps(registry, indent=2), encoding="utf-8")
|
||||
yield REPO_ROOT
|
||||
finally:
|
||||
if backup is not None:
|
||||
registry_path.write_bytes(backup)
|
||||
elif registry_path.exists():
|
||||
registry_path.unlink()
|
||||
|
||||
|
||||
def _drone_env() -> dict:
|
||||
"""Env for drone subprocesses (inherits PATH etc.)."""
|
||||
import os
|
||||
|
||||
return dict(os.environ)
|
||||
|
||||
|
||||
def test_t3_drone_systems_lists_branch(clean_venv: CleanVenv, routing_root: Path) -> None:
|
||||
"""``drone systems`` reads the registry and lists a known branch."""
|
||||
proc = _run(
|
||||
[str(clean_venv.drone), "systems"],
|
||||
cwd=str(routing_root),
|
||||
env=_drone_env(),
|
||||
)
|
||||
assert proc.returncode == 0, f"drone systems failed:\n{proc.stdout}\n{proc.stderr}"
|
||||
out = proc.stdout.lower()
|
||||
assert "seedgo" in out or "ai_mail" in out, f"no known branch listed:\n{proc.stdout}"
|
||||
|
||||
|
||||
def test_t3_drone_routes_to_real_branch(clean_venv: CleanVenv, routing_root: Path) -> None:
|
||||
"""``drone @ai_mail --help`` resolves -> subprocesses -> returns help text.
|
||||
|
||||
This is the real integration proof: ai_mail is a registry branch (not an
|
||||
in-process drone module), so a non-empty help body proves drone resolved
|
||||
the @name, found apps/ai_mail.py, ran it in a subprocess, and captured its
|
||||
output.
|
||||
"""
|
||||
proc = _run(
|
||||
[str(clean_venv.drone), "@ai_mail", "--help"],
|
||||
cwd=str(routing_root),
|
||||
env=_drone_env(),
|
||||
)
|
||||
combined = (proc.stdout + proc.stderr).lower()
|
||||
assert proc.returncode == 0, f"drone @ai_mail --help failed:\n{proc.stdout}\n{proc.stderr}"
|
||||
assert proc.stdout.strip(), f"no help text returned:\nSTDOUT:\n{proc.stdout}\nSTDERR:\n{proc.stderr}"
|
||||
assert "ai_mail" in combined or "mail" in combined, f"help text unexpected:\n{proc.stdout}"
|
||||
Reference in New Issue
Block a user