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
|
### 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
|
- **`drone rm` — provider-agnostic safe delete** — a contained recursive delete
|
||||||
that lets agents clean up scratch dirs without tripping the `rm -rf` block.
|
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
|
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