diff --git a/.gitea/workflows/canaries.yml b/.gitea/workflows/canaries.yml new file mode 100644 index 0000000..0085b76 --- /dev/null +++ b/.gitea/workflows/canaries.yml @@ -0,0 +1,31 @@ +# Weekly canary suite. Catches upstream regressions (broken pipelock +# image packaging at the pinned digest, etc.) without coupling every +# dev push to upstream registry availability. +# +# Opt-in via CLAUDE_BOTTLE_RUN_CANARIES=1 so the same files can be run +# locally with the same gating. + +name: canaries + +on: + schedule: + # 12:00 UTC every Monday. + - cron: "0 12 * * 1" + workflow_dispatch: + +jobs: + canaries: + runs-on: ubuntu-latest + env: + CLAUDE_BOTTLE_RUN_CANARIES: "1" + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Run canaries + run: python3 -m unittest discover -t . -s tests/canaries -v diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index f08e038..ee2d590 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -1,10 +1,14 @@ -# Run the project's full test suite on every PR push and on push to main. +# Run the project's test suite on every PR push and on push to main. # -# The suite uses stdlib `unittest` (see tests/run_tests.py) — no external -# Python dependencies are required to execute it. Integration tests need a -# reachable Docker daemon; if Docker is unavailable on the runner those -# tests skip cleanly via tests/_docker.py:skip_unless_docker, so the job -# still passes (with skips visible in the run output). +# The suite uses stdlib `unittest` discovery — no external Python +# dependencies are required to execute it. Tests are split by directory: +# +# tests/unit/ — pure unit tests; always run +# tests/integration/ — need a reachable Docker daemon; skip cleanly +# (via tests/_docker.py:skip_unless_docker) when +# Docker isn't available on the runner +# tests/canaries/ — upstream regression canaries; run on a separate +# schedule (see canaries.yml), not here # # This workflow assumes the Gitea Actions runner exposes the host Docker # socket to the job container so `docker` commands inside the job can @@ -20,8 +24,21 @@ on: pull_request: jobs: - test: - name: run tests/run_tests.py + unit: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Run unit tests + run: python3 -m unittest discover -t . -s tests/unit -v + + integration: runs-on: ubuntu-latest steps: - name: Checkout @@ -41,5 +58,5 @@ jobs: echo "docker not on PATH — integration tests will skip" fi - - name: Run full test suite - run: python3 tests/run_tests.py + - name: Run integration tests + run: python3 -m unittest discover -t . -s tests/integration -v diff --git a/claude_bottle/backend/__init__.py b/claude_bottle/backend/__init__.py index cb0600a..d44633b 100644 --- a/claude_bottle/backend/__init__.py +++ b/claude_bottle/backend/__init__.py @@ -66,6 +66,13 @@ class BottlePlan(ABC): def print(self, *, remote_control: bool) -> None: """Render the y/N preflight summary to stderr.""" + @abstractmethod + def to_dict(self, *, remote_control: bool) -> dict[str, object]: + """Return the plan as a JSON-serializable dict for machine + consumption (used by `start --dry-run --format=json`). The key + set is part of the CLI's user-facing contract — adding fields + is fine, renaming or removing is a breaking change.""" + @dataclass(frozen=True) class BottleCleanupPlan(ABC): diff --git a/claude_bottle/backend/docker/backend.py b/claude_bottle/backend/docker/backend.py index 70a880a..3f97db3 100644 --- a/claude_bottle/backend/docker/backend.py +++ b/claude_bottle/backend/docker/backend.py @@ -104,7 +104,7 @@ class DockerBottleBackend(BottleBackend): prompt_file.write_text("") prompt_file.chmod(0o600) - proxy_plan = self.prepare_proxy(spec, stage_dir) + proxy_plan = self._proxy.prepare(bottle, slug, stage_dir) resolved = resolve_env(manifest, spec.agent_name) self._write_env_files(resolved, env_file, args_file) prompt_file.write_text(agent.prompt) @@ -151,16 +151,6 @@ class DockerBottleBackend(BottleBackend): args_lines = [f"-e\n{name}" for name in resolved.forwarded] args_file.write_text("\n".join(args_lines) + ("\n" if args_lines else "")) - def prepare_proxy(self, spec: BottleSpec, stage_dir: Path) -> pipelock.PipelockProxyPlan: - """Decide where the pipelock yaml lives in `stage_dir`, delegate - to PipelockProxy to write it, and return the resolved - PipelockProxyPlan for the launch step to consume. Stage-only: - no Docker resources created yet.""" - yaml_path = stage_dir / "pipelock.yaml" - bottle = spec.manifest.bottle_for(spec.agent_name) - slug = docker_mod.slugify(spec.agent_name) - return self._proxy.prepare(bottle, slug, yaml_path) - @contextmanager def launch(self, plan: BottlePlan) -> Iterator[DockerBottle]: """Build, launch, and provision a Docker bottle. Teardown on exit.""" diff --git a/claude_bottle/backend/docker/bottle_plan.py b/claude_bottle/backend/docker/bottle_plan.py index 63651e5..2e98f1e 100644 --- a/claude_bottle/backend/docker/bottle_plan.py +++ b/claude_bottle/backend/docker/bottle_plan.py @@ -12,7 +12,7 @@ from dataclasses import dataclass from pathlib import Path from ...log import info -from ...pipelock import PipelockProxyPlan +from ...pipelock import PipelockProxyPlan, pipelock_effective_allowlist from .. import BottlePlan @@ -75,3 +75,36 @@ class DockerBottlePlan(BottlePlan): ) info("remote-control : " + ("enabled" if remote_control else "disabled")) print(file=sys.stderr) + + def to_dict(self, *, remote_control: bool) -> dict[str, object]: + spec = self.spec + manifest = spec.manifest + agent = manifest.agents[spec.agent_name] + bottle = manifest.bottle_for(spec.agent_name) + + env_names = list(bottle.env.keys()) + if spec.forward_oauth_token: + env_names.append("CLAUDE_CODE_OAUTH_TOKEN") + + hosts = pipelock_effective_allowlist(bottle) + return { + "agent": spec.agent_name, + "bottle": agent.bottle, + "container_name": self.container_name, + "image": self.image, + "derived_image": self.derived_image, + "stage_dir": str(self.stage_dir), + "runtime": "runsc" if self.use_runsc else "runc", + "env_names": env_names, + "skills": list(agent.skills), + "ssh_hosts": [e.Host for e in bottle.ssh], + "egress": { + "host_count": len(hosts), + "hosts": hosts, + }, + "prompt": { + "length": len(agent.prompt), + "first_line": agent.prompt.splitlines()[0] if agent.prompt else "", + }, + "remote_control": remote_control, + } diff --git a/claude_bottle/cli/start.py b/claude_bottle/cli/start.py index 81b2c5e..585bd75 100644 --- a/claude_bottle/cli/start.py +++ b/claude_bottle/cli/start.py @@ -5,6 +5,7 @@ session ends.""" from __future__ import annotations import argparse +import json import os import shutil import sys @@ -12,7 +13,7 @@ import tempfile from pathlib import Path from ..backend import BottleSpec, get_bottle_backend -from ..log import info +from ..log import die, info from ..manifest import Manifest from ._common import PROG, USER_CWD, read_tty_line @@ -22,10 +23,18 @@ def cmd_start(argv: list[str]) -> int: parser.add_argument("--dry-run", action="store_true") parser.add_argument("--cwd", action="store_true", help="copy host cwd into a derived image") parser.add_argument("--remote-control", action="store_true") + parser.add_argument( + "--format", + choices=("text", "json"), + default="text", + help="preflight output format; --format=json requires --dry-run", + ) parser.add_argument("name", help="agent name defined in claude-bottle.json") args = parser.parse_args(argv) dry_run = args.dry_run or os.environ.get("CLAUDE_BOTTLE_DRY_RUN") == "1" + if args.format == "json" and not dry_run: + die("--format=json requires --dry-run") manifest = Manifest.resolve(USER_CWD) spec = BottleSpec( @@ -40,6 +49,12 @@ def cmd_start(argv: list[str]) -> int: try: backend = get_bottle_backend() plan = backend.prepare(spec, stage_dir=stage_dir) + + if args.format == "json": + json.dump(plan.to_dict(remote_control=args.remote_control), sys.stdout, indent=2) + sys.stdout.write("\n") + return 0 + plan.print(remote_control=args.remote_control) if dry_run: diff --git a/claude_bottle/pipelock.py b/claude_bottle/pipelock.py index 4d9967f..2a28041 100644 --- a/claude_bottle/pipelock.py +++ b/claude_bottle/pipelock.py @@ -15,6 +15,7 @@ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass from pathlib import Path +from typing import cast from .manifest import Bottle from .util import is_ipv4_literal @@ -85,6 +86,72 @@ def pipelock_allowlist_summary(bottle: Bottle) -> str: +# --- Config build + YAML render -------------------------------------------- + + +def pipelock_build_config(bottle: Bottle) -> dict[str, object]: + """Build the structured pipelock config dict the sidecar will load. + + Deliberately carries no env values, no secrets, no per-agent + customization beyond the resolved hostname list. The shape mirrors + the YAML pipelock expects on disk; `pipelock_render_yaml` serializes + it. Tests assert on this dict; production code renders it.""" + cfg: dict[str, object] = { + "version": 1, + "mode": "strict", + "enforce": True, + "api_allowlist": pipelock_effective_allowlist(bottle), + "forward_proxy": {"enabled": True}, + } + trusted = pipelock_bottle_ssh_trusted_domains(bottle) + if trusted: + cfg["trusted_domains"] = trusted + ip_cidrs = pipelock_bottle_ssh_ip_cidrs(bottle) + if ip_cidrs: + cfg["ssrf"] = {"ip_allowlist": ip_cidrs} + cfg["dlp"] = {"include_defaults": True, "scan_env": True} + return cfg + + +def pipelock_render_yaml(cfg: dict[str, object]) -> str: + """Render a pipelock config dict (as produced by + `pipelock_build_config`) as YAML. Hand-rolled so we don't take a + YAML-parser dependency for a fixed, narrow shape.""" + def _bool(b: object) -> str: + return "true" if b else "false" + + lines: list[str] = [] + lines.append(f"version: {cfg['version']}") + lines.append(f"mode: {cfg['mode']}") + lines.append(f"enforce: {_bool(cfg['enforce'])}") + lines.append("") + lines.append("api_allowlist:") + for h in cast(list[str], cfg["api_allowlist"]): + lines.append(f' - "{h}"') + lines.append("") + lines.append("forward_proxy:") + fp = cast(dict[str, object], cfg["forward_proxy"]) + lines.append(f" enabled: {_bool(fp['enabled'])}") + lines.append("") + if "trusted_domains" in cfg: + lines.append("trusted_domains:") + for td in cast(list[str], cfg["trusted_domains"]): + lines.append(f' - "{td}"') + lines.append("") + if "ssrf" in cfg: + lines.append("ssrf:") + ssrf = cast(dict[str, object], cfg["ssrf"]) + lines.append(" ip_allowlist:") + for cidr in cast(list[str], ssrf["ip_allowlist"]): + lines.append(f' - "{cidr}"') + lines.append("") + lines.append("dlp:") + dlp = cast(dict[str, object], cfg["dlp"]) + lines.append(f" include_defaults: {_bool(dlp['include_defaults'])}") + lines.append(f" scan_env: {_bool(dlp['scan_env'])}") + return "\n".join(lines) + "\n" + + # --- Proxy class ----------------------------------------------------------- @@ -110,9 +177,9 @@ class PipelockProxy(ABC): and lives on concrete subclasses.""" def prepare( - self, bottle: Bottle, slug: str, yaml_path: Path + self, bottle: Bottle, slug: str, stage_dir: Path ) -> PipelockProxyPlan: - """Write the pipelock yaml config (mode 600) to `yaml_path` + """Write the pipelock yaml config (mode 600) under `stage_dir` and return the plan for `.start`. `slug` is the agent-derived identifier (lowercased, @@ -121,51 +188,13 @@ class PipelockProxy(ABC): (`claude-bottle-pipelock-`), the internal/egress networks. It's stored on the returned plan so the backend's start step can derive the sidecar's container name.""" + yaml_path = stage_dir / "pipelock.yaml" self._build_pipelock_yaml(bottle, yaml_path) return PipelockProxyPlan(yaml_path=yaml_path, slug=slug) def _build_pipelock_yaml(self, bottle: Bottle, yaml_path: Path): - """Write the pipelock yaml config (mode 600) to `yaml_path` - for the sidecar to consume when it boots. Carries the - effective allowlist (bottle.egress.allowlist UNION - claude-bottle defaults UNION ssh hostnames), a fixed listen - port, strict mode + forward_proxy + DLP defaults + scan_env. - Deliberately contains no env values, no secrets, no per-agent - customization beyond the hostname list.""" - allowlist = pipelock_effective_allowlist(bottle) - trusted = pipelock_bottle_ssh_trusted_domains(bottle) - ip_cidrs = pipelock_bottle_ssh_ip_cidrs(bottle) - - lines: list[str] = [] - lines.append("version: 1") - lines.append("mode: strict") - lines.append("enforce: true") - lines.append("") - lines.append("# Hostnames the agent is allowed to reach. Effective list is") - lines.append("# claude-bottle defaults UNION bottle.egress.allowlist (sorted, deduped).") - lines.append("api_allowlist:") - for h in allowlist: - lines.append(f' - "{h}"') - lines.append("") - lines.append("forward_proxy:") - lines.append(" enabled: true") - lines.append("") - if trusted: - lines.append("trusted_domains:") - for td in trusted: - lines.append(f' - "{td}"') - lines.append("") - if ip_cidrs: - lines.append("ssrf:") - lines.append(" ip_allowlist:") - for cidr in ip_cidrs: - lines.append(f' - "{cidr}"') - lines.append("") - lines.append("dlp:") - lines.append(" include_defaults: true") - lines.append(" scan_env: true") - - yaml_path.write_text("\n".join(lines) + "\n") + """Write the pipelock yaml config (mode 600) to `yaml_path`.""" + yaml_path.write_text(pipelock_render_yaml(pipelock_build_config(bottle))) yaml_path.chmod(0o600) @abstractmethod diff --git a/tests/README.md b/tests/README.md index 150e669..9e8ef91 100644 --- a/tests/README.md +++ b/tests/README.md @@ -8,47 +8,58 @@ tests need Docker and skip cleanly otherwise. ``` tests/ - run_tests.py # entry point - fixtures.py # JSON manifest builders - _docker.py # docker-availability skip helper - test_pipelock_naming.py # unit - test_pipelock_classify.py # unit - test_pipelock_allowlist.py # unit - test_pipelock_yaml.py # unit - test_pipelock_image.py # integration - test_pipelock_sidecar_smoke.py # integration - test_dry_run_plan.py # integration - test_orphan_cleanup.py # integration + fixtures.py # JSON manifest builders (shared) + _docker.py # docker-availability skip helper (shared) + unit/ + test_pipelock_classify.py + test_pipelock_allowlist.py + test_pipelock_yaml.py + test_manifest_runtime.py + integration/ + test_pipelock_sidecar_smoke.py + test_dry_run_plan.py + test_orphan_cleanup.py + canaries/ + test_pipelock_image.py # opt-in; see below ``` +Classification falls out of the directory — no hand-maintained list to +keep in sync. + ## Running ```bash -tests/run_tests.py # everything -tests/run_tests.py unit # unit only -tests/run_tests.py integration # integration only -tests/run_tests.py tests/test_pipelock_yaml.py # one file +python -m unittest discover -t . -s tests/unit -v # unit only +python -m unittest discover -t . -s tests/integration -v # integration only +python -m unittest discover -t . -s tests -v # both (recursive) +python -m unittest tests.unit.test_pipelock_yaml # one file ``` -You can also run via `python -m unittest`: - -```bash -python -m unittest discover -s tests -python -m unittest tests.test_pipelock_yaml -``` +Discovery is invoked with `-t .` (top-level dir = repo root) so the +`claude_bottle` package on `sys.path` resolves correctly. ## What the integration tests cover -- `test_pipelock_image.py` — the pinned digest is reachable, ENTRYPOINT - is `/pipelock`, and `CMD` includes `run`. -- `test_pipelock_sidecar_smoke.py` — `docker create` + `docker cp` the - generated YAML to `/etc/pipelock.yaml` + `docker start`, then probe - `/health`. -- `test_dry_run_plan.py` — `cli.py start --dry-run` shows the resolved - egress allowlist and creates zero docker resources. -- `test_orphan_cleanup.py` — network_remove and pipelock_stop are - idempotent against missing resources, so the EXIT trap can call them - unconditionally. +- `test_pipelock_sidecar_smoke.py` — drives `DockerPipelockProxy.prepare` + + `.start` (the production code path) against a real Docker daemon and + probes the sidecar's `/health` from an in-network curl container. +- `test_dry_run_plan.py` — `cli.py start --dry-run --format=json` emits + a structured plan that contains the resolved egress allowlist and + the bottle's runtime, and creates zero Docker resources. +- `test_orphan_cleanup.py` — `network_remove` and `PipelockProxy.stop` + are idempotent against missing resources, so the EXIT trap can call + them unconditionally. + +## Canaries + +`tests/canaries/` holds upstream-regression checks (e.g. the pinned +pipelock digest's binary still runs). These are gated on +`CLAUDE_BOTTLE_RUN_CANARIES=1` and not part of the per-push suite. +They're invoked by the scheduled `canaries` workflow. + +```bash +CLAUDE_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v +``` ## What's NOT covered @@ -60,9 +71,10 @@ python -m unittest tests.test_pipelock_yaml ## Adding a test -1. Pick a filename: `test_.py`. Add it to `INTEGRATION_NAMES` - in `run_tests.py` if it needs Docker. -2. Boilerplate: +1. Pick the directory: `tests/unit/` for a pure unit test, + `tests/integration/` for one that needs Docker. +2. Filename: `test_.py`. +3. Boilerplate: ```python import unittest @@ -75,5 +87,5 @@ python -m unittest tests.test_pipelock_yaml if __name__ == "__main__": unittest.main() ``` -3. For Docker-dependent tests, decorate the class with +4. For Docker-dependent tests, decorate the class with `@skip_unless_docker()` from `tests._docker`. diff --git a/tests/canaries/__init__.py b/tests/canaries/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_pipelock_image.py b/tests/canaries/test_pipelock_image.py similarity index 62% rename from tests/test_pipelock_image.py rename to tests/canaries/test_pipelock_image.py index ffb23b1..72a54cf 100644 --- a/tests/test_pipelock_image.py +++ b/tests/canaries/test_pipelock_image.py @@ -1,7 +1,13 @@ -"""Integration: the pinned pipelock image's binary actually runs. -Catches a broken upstream packaging at the pinned digest. Requires -docker.""" +"""Canary: the pinned pipelock image's binary actually runs. +This test exists to catch a broken upstream packaging at the pinned +digest. It is NOT part of the per-push suite — that would couple every +dev push to upstream registry availability. Set +CLAUDE_BOTTLE_RUN_CANARIES=1 to opt in (a scheduled CI workflow does +this; humans can run it ad-hoc the same way). +""" + +import os import subprocess import unittest @@ -9,6 +15,10 @@ from claude_bottle.backend.docker.pipelock import PIPELOCK_IMAGE from tests._docker import skip_unless_docker +@unittest.skipUnless( + os.environ.get("CLAUDE_BOTTLE_RUN_CANARIES") == "1", + "canary suite is opt-in; set CLAUDE_BOTTLE_RUN_CANARIES=1 to run", +) @skip_unless_docker() class TestPipelockImage(unittest.TestCase): @classmethod diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_dry_run_plan.py b/tests/integration/test_dry_run_plan.py similarity index 54% rename from tests/test_dry_run_plan.py rename to tests/integration/test_dry_run_plan.py index 1cf00a2..1672faa 100644 --- a/tests/test_dry_run_plan.py +++ b/tests/integration/test_dry_run_plan.py @@ -1,10 +1,9 @@ -"""Integration: cli.py start --dry-run renders the planned shape and -does not create any docker resources. Confirms the preflight contract -from PRD 0001 (allowlist line in the plan, no docker side effects).""" +"""Integration: cli.py start --dry-run --format=json renders a stable +machine-readable plan and creates zero Docker resources. The shape of +the JSON document is part of the CLI's user-facing contract.""" import json import os -import re import subprocess import sys import tempfile @@ -13,12 +12,12 @@ from pathlib import Path from tests._docker import skip_unless_docker -REPO_ROOT = Path(__file__).resolve().parent.parent +REPO_ROOT = Path(__file__).resolve().parent.parent.parent @skip_unless_docker() class TestDryRunPlan(unittest.TestCase): - def test_dry_run(self): + def test_dry_run_emits_structured_plan(self): work_dir = Path(tempfile.mkdtemp()) try: manifest = work_dir / "claude-bottle.json" @@ -34,24 +33,43 @@ class TestDryRunPlan(unittest.TestCase): env = os.environ.copy() env["HOME"] = str(work_dir) - env["CLAUDE_BOTTLE_DRY_RUN"] = "1" + env.pop("CLAUDE_BOTTLE_DRY_RUN", None) result = subprocess.run( - [sys.executable, str(REPO_ROOT / "cli.py"), "start", "demo"], + [ + sys.executable, str(REPO_ROOT / "cli.py"), + "start", "--dry-run", "--format", "json", "demo", + ], cwd=work_dir, env=env, capture_output=True, text=True, ) - out = result.stdout + result.stderr + self.assertEqual( + 0, result.returncode, + f"start --dry-run failed: stderr={result.stderr}", + ) - self.assertIn("egress", out, "preflight: egress line present") - # 7 baked defaults + 1 bottle entry = 8. - self.assertRegex(out, r"8 hosts allowed", "preflight: bottle entry counted") - self.assertIn("api.anthropic.com", out, "preflight: baked default shown") - self.assertRegex(out, r"runtime\s*:\s*runc", "preflight: default runtime shown") - self.assertIn("dry-run requested", out, "dry-run banner present") - self.assertNotIn("/dev/tty", out, "dry-run exited before tty prompt") + plan = json.loads(result.stdout) + self.assertEqual("demo", plan["agent"]) + self.assertEqual("dev", plan["bottle"]) + self.assertEqual("runc", plan["runtime"], + "runsc isn't available on the CI runner") + self.assertEqual([], plan["skills"]) + self.assertEqual([], plan["ssh_hosts"]) + self.assertEqual(False, plan["remote_control"]) + self.assertEqual(0, plan["prompt"]["length"]) + + # User-declared host + a baked default both present; the union + # is sorted and deduplicated. + hosts = plan["egress"]["hosts"] + self.assertIn("example.org", hosts) + self.assertIn("api.anthropic.com", hosts) + self.assertEqual(plan["egress"]["host_count"], len(hosts)) + self.assertEqual(sorted(set(hosts)), hosts, + "hosts must be sorted and deduplicated") + + # No Docker side effects. self.assertEqual(nets_before, self._count_claude_bottle_networks(), "no networks created") self.assertEqual(ctrs_before, self._count_claude_bottle_containers(), diff --git a/tests/test_orphan_cleanup.py b/tests/integration/test_orphan_cleanup.py similarity index 100% rename from tests/test_orphan_cleanup.py rename to tests/integration/test_orphan_cleanup.py diff --git a/tests/integration/test_pipelock_sidecar_smoke.py b/tests/integration/test_pipelock_sidecar_smoke.py new file mode 100644 index 0000000..a501ed9 --- /dev/null +++ b/tests/integration/test_pipelock_sidecar_smoke.py @@ -0,0 +1,118 @@ +"""Integration: drive the production pipelock-sidecar bring-up +(`DockerPipelockProxy.prepare` → `.start`) and probe /health from a +sibling container on the same internal network. The point is that the +test exercises the production code path — if the docker create/cp/start +sequence in DockerPipelockProxy.start changes shape, this test should +notice. + +We don't probe /health from the host because the sidecar is created +attached to an `--internal` network with no published port (that's +the production topology). An in-network curl container reaches it the +same way the agent container would in production. +""" + +import dataclasses +import os +import shutil +import subprocess +import tempfile +import unittest +from pathlib import Path + +from claude_bottle.backend.docker.network import ( + network_create_egress, + network_create_internal, + network_remove, +) +from claude_bottle.backend.docker.pipelock import ( + PIPELOCK_PORT, + DockerPipelockProxy, + pipelock_container_name, +) +from tests._docker import skip_unless_docker +from tests.fixtures import fixture_minimal + +CURL_IMAGE = "curlimages/curl:latest" + + +@skip_unless_docker() +class TestPipelockSidecarSmoke(unittest.TestCase): + @classmethod + def setUpClass(cls): + # Pre-pull curlimages/curl so the per-test retry loop isn't + # racing the registry. Skip cleanly if the pull fails (the + # canary suite will surface a real registry outage separately). + result = subprocess.run( + ["docker", "pull", CURL_IMAGE], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if result.returncode != 0: + raise unittest.SkipTest(f"could not pull {CURL_IMAGE}") + + def setUp(self): + self.slug = f"cb-test-smoke-{os.getpid()}" + self.sidecar_name = "" + self.internal_net = "" + self.egress_net = "" + self.work_dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + if self.sidecar_name: + DockerPipelockProxy().stop(self.sidecar_name) + for n in (self.internal_net, self.egress_net): + if n: + network_remove(n) + shutil.rmtree(self.work_dir, ignore_errors=True) + + @unittest.skipIf( + os.environ.get("GITEA_ACTIONS") == "true", + "skipped under act_runner: docker socket mount topology breaks " + "in-process visibility of networks created on the host daemon", + ) + def test_prepare_and_start_yield_healthy_sidecar(self): + proxy = DockerPipelockProxy() + + prep = proxy.prepare(fixture_minimal().bottles["dev"], self.slug, self.work_dir) + + self.internal_net = network_create_internal(self.slug) + self.egress_net = network_create_egress(self.slug) + + plan = dataclasses.replace( + prep, + internal_network=self.internal_net, + egress_network=self.egress_net, + ) + + self.sidecar_name = proxy.start(plan) + self.assertEqual(pipelock_container_name(self.slug), self.sidecar_name) + + # Probe /health from a sibling container on the internal network — + # same access topology the agent container uses in production. + # curl retries on connection refused while pipelock is booting. + probe = subprocess.run( + [ + "docker", "run", "--rm", + "--network", self.internal_net, + CURL_IMAGE, + "-sf", "--max-time", "2", + "--retry", "15", + "--retry-delay", "1", + "--retry-connrefused", + f"http://{self.sidecar_name}:{PIPELOCK_PORT}/health", + ], + capture_output=True, + text=True, + timeout=60, + ) + self.assertEqual( + 0, probe.returncode, + f"health probe failed: stdout={probe.stdout!r} stderr={probe.stderr!r}", + ) + body = probe.stdout + self.assertIn('"status":"healthy"', body) + self.assertRegex(body, r'"version":"[0-9]+\.[0-9]+\.[0-9]+"') + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/run_tests.py b/tests/run_tests.py deleted file mode 100755 index f8edfab..0000000 --- a/tests/run_tests.py +++ /dev/null @@ -1,91 +0,0 @@ -#!/usr/bin/env python3 -"""Test runner. Wraps unittest's discovery so we can split unit / -integration the same way the bash runner did. - -Usage: - tests/run_tests.py # unit + integration - tests/run_tests.py unit # unit only (no docker) - tests/run_tests.py integration # integration only (need docker) - tests/run_tests.py tests/test_x.py # one specific file (or path) - -Tests are auto-classified as integration when their filename matches -one of INTEGRATION_NAMES below; everything else is a unit test. -""" - -from __future__ import annotations - -import sys -import unittest -from pathlib import Path - -REPO_ROOT = Path(__file__).resolve().parent.parent -TESTS_DIR = REPO_ROOT / "tests" - -INTEGRATION_NAMES = { - "test_dry_run_plan.py", - "test_orphan_cleanup.py", - "test_pipelock_image.py", - "test_pipelock_sidecar_smoke.py", -} - - -def _all_test_files() -> list[Path]: - return sorted(TESTS_DIR.glob("test_*.py")) - - -def _classify(path: Path) -> str: - return "integration" if path.name in INTEGRATION_NAMES else "unit" - - -def _modname(path: Path) -> str: - return f"tests.{path.stem}" - - -def _build_suite(files: list[Path]) -> unittest.TestSuite: - loader = unittest.TestLoader() - suite = unittest.TestSuite() - for f in files: - suite.addTests(loader.loadTestsFromName(_modname(f))) - return suite - - -def usage() -> None: - sys.stderr.write( - "usage: tests/run_tests.py [unit|integration|path/to/test.py]\n" - ) - - -def main(argv: list[str]) -> int: - sys.path.insert(0, str(REPO_ROOT)) - - if not argv: - files = _all_test_files() - else: - arg = argv[0] - if arg in ("-h", "--help"): - usage() - return 0 - if arg == "unit": - files = [f for f in _all_test_files() if _classify(f) == "unit"] - elif arg == "integration": - files = [f for f in _all_test_files() if _classify(f) == "integration"] - else: - p = Path(arg).resolve() - if not p.is_file(): - sys.stderr.write(f"no such file: {arg}\n") - usage() - return 2 - files = [p] - - if not files: - sys.stderr.write("no test files found\n") - return 2 - - suite = _build_suite(files) - runner = unittest.TextTestRunner(verbosity=2) - result = runner.run(suite) - return 0 if result.wasSuccessful() else 1 - - -if __name__ == "__main__": - sys.exit(main(sys.argv[1:])) diff --git a/tests/test_manifest_runtime.py b/tests/test_manifest_runtime.py deleted file mode 100644 index 963fabf..0000000 --- a/tests/test_manifest_runtime.py +++ /dev/null @@ -1,68 +0,0 @@ -"""Unit: bottle 'runtime' field is no longer supported (PRD 0003). - -gVisor is now auto-detected by the Docker factory. A manifest carrying -the legacy 'runtime' field must fail loudly with a message pointing the -user at the auto-detect behavior, rather than silently ignoring.""" - -import io -import sys -import unittest - -from claude_bottle.log import Die -from claude_bottle.manifest import Bottle, Manifest - - -_ABSENT = object() - - -def _manifest(runtime_value: object) -> dict: - """Build a minimal manifest JSON shape with one bottle whose runtime - field is set (or absent if `runtime_value is _ABSENT`).""" - bottle: dict = {} - if runtime_value is not _ABSENT: - bottle["runtime"] = runtime_value - return { - "bottles": {"dev": bottle}, - "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, - } - - -class TestManifestRuntimeRemoved(unittest.TestCase): - def test_loads_when_runtime_absent(self): - m = Manifest.from_json_obj(_manifest(_ABSENT)) - self.assertIn("dev", m.bottles) - - def test_bottle_dataclass_has_no_runtime_attribute(self): - """Structural check: the field has been removed from the dataclass.""" - b = Bottle() - self.assertFalse(hasattr(b, "runtime")) - - def test_rejects_runsc_value_with_helpful_message(self): - captured = io.StringIO() - old_stderr = sys.stderr - sys.stderr = captured - try: - with self.assertRaises(Die): - Manifest.from_json_obj(_manifest("runsc")) - finally: - sys.stderr = old_stderr - msg = captured.getvalue() - self.assertIn("'runtime'", msg, "error names the field") - self.assertIn("auto-detect", msg, "error points at the new behavior") - - def test_rejects_runc_value(self): - with self.assertRaises(Die): - Manifest.from_json_obj(_manifest("runc")) - - def test_rejects_unknown_value(self): - with self.assertRaises(Die): - Manifest.from_json_obj(_manifest("kata-runtime")) - - def test_rejects_non_string(self): - """Any presence of the field is an error; type is not consulted.""" - with self.assertRaises(Die): - Manifest.from_json_obj(_manifest(42)) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_pipelock_allowlist.py b/tests/test_pipelock_allowlist.py deleted file mode 100644 index 90847e3..0000000 --- a/tests/test_pipelock_allowlist.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Unit: allowlist resolution — pipelock_bottle_allowlist, -pipelock_bottle_ssh_hostnames, pipelock_bottle_ssh_ip_cidrs, -pipelock_bottle_ssh_trusted_domains, pipelock_effective_allowlist.""" - -import unittest - -from claude_bottle.log import Die -from claude_bottle.manifest import Manifest -from claude_bottle.pipelock import ( - pipelock_bottle_allowlist, - pipelock_bottle_ssh_hostnames, - pipelock_bottle_ssh_ip_cidrs, - pipelock_bottle_ssh_trusted_domains, - pipelock_effective_allowlist, -) -from tests.fixtures import fixture_minimal, fixture_with_egress, fixture_with_ssh - - -class TestBottleAllowlist(unittest.TestCase): - def test_egress_allowlist_present(self): - out = pipelock_bottle_allowlist(fixture_with_egress().bottles["dev"]) - self.assertIn("github.com", out) - self.assertIn("gitlab.com", out) - self.assertIn("registry.npmjs.org", out) - - def test_empty_when_no_egress_block(self): - out = pipelock_bottle_allowlist(fixture_minimal().bottles["dev"]) - self.assertEqual([], out) - - def test_rejects_non_string_entry(self): - bad = { - "bottles": {"dev": {"egress": {"allowlist": ["github.com", 42]}}}, - "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, - } - with self.assertRaises(Die): - Manifest.from_json_obj(bad) - - -class TestSSHHostnames(unittest.TestCase): - def test_hostnames_include_both(self): - hosts = pipelock_bottle_ssh_hostnames(fixture_with_ssh().bottles["dev"]) - self.assertIn("100.78.141.42", hosts) - self.assertIn("github.com", hosts) - - def test_ip_cidrs_only_ipv4(self): - cidrs = pipelock_bottle_ssh_ip_cidrs(fixture_with_ssh().bottles["dev"]) - self.assertIn("100.78.141.42/32", cidrs) - self.assertNotIn("github.com", cidrs) - - def test_trusted_domains_only_hostnames(self): - trusted = pipelock_bottle_ssh_trusted_domains(fixture_with_ssh().bottles["dev"]) - self.assertIn("github.com", trusted) - self.assertNotIn("100.78.141.42", trusted) - - -class TestEffectiveAllowlist(unittest.TestCase): - def test_union_and_dedup(self): - manifest = Manifest.from_json_obj({ - "bottles": { - "dev": { - "egress": {"allowlist": ["registry.npmjs.org"]}, - "ssh": [ - {"Host": "ts", "IdentityFile": "/dev/null", - "Hostname": "100.78.141.42", "User": "git", "Port": 30009}, - {"Host": "gh", "IdentityFile": "/dev/null", - "Hostname": "github.com", "User": "git", "Port": 22}, - ], - } - }, - "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, - }) - eff = pipelock_effective_allowlist(manifest.bottles["dev"]) - self.assertIn("api.anthropic.com", eff) - self.assertIn("registry.npmjs.org", eff) - self.assertIn("100.78.141.42", eff) - self.assertIn("github.com", eff) - self.assertEqual(len(eff), len(set(eff)), "deduplicated") - self.assertEqual(eff, sorted(eff), "sorted") - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_pipelock_sidecar_smoke.py b/tests/test_pipelock_sidecar_smoke.py deleted file mode 100644 index 06131a2..0000000 --- a/tests/test_pipelock_sidecar_smoke.py +++ /dev/null @@ -1,104 +0,0 @@ -"""Integration: full sidecar smoke test. Boots a pipelock container the -same way cli.py does (docker create + docker cp YAML + docker start), -then probes /health.""" - -import os -import re -import shutil -import subprocess -import tempfile -import time -import unittest -import urllib.request -from pathlib import Path - -from claude_bottle.backend.docker.pipelock import ( - PIPELOCK_IMAGE, - DockerPipelockProxy, -) -from tests._docker import skip_unless_docker -from tests.fixtures import fixture_minimal - - -@skip_unless_docker() -class TestPipelockSidecarSmoke(unittest.TestCase): - def setUp(self): - self.name = f"cb-test-pipelock-smoke-{os.getpid()}" - self.work_dir = Path(tempfile.mkdtemp()) - - def tearDown(self): - subprocess.run( - ["docker", "rm", "-f", self.name], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - shutil.rmtree(self.work_dir, ignore_errors=True) - - @unittest.skipIf( - os.environ.get("GITEA_ACTIONS") == "true", - "skipped under act_runner: published port is on the host's " - "loopback, not reachable from the job container's 127.0.0.1", - ) - def test_smoke(self): - yaml_path = self.work_dir / "pipelock.yaml" - DockerPipelockProxy().prepare(fixture_minimal().bottles["dev"], "demo", yaml_path) - - create = subprocess.run( - [ - "docker", "create", - "--name", self.name, - "-p", "0:8888", - PIPELOCK_IMAGE, - "run", "--config", "/etc/pipelock.yaml", - "--listen", "0.0.0.0:8888", - ], - stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, - text=True, - ) - self.assertEqual(0, create.returncode, f"docker create failed: {create.stderr}") - - # Guard against /etc/pipelock/ regressions: the path must be - # /etc/pipelock.yaml, since the image is distroless. - cp = subprocess.run( - ["docker", "cp", str(yaml_path), f"{self.name}:/etc/pipelock.yaml"], - stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, - text=True, - ) - self.assertEqual(0, cp.returncode, f"docker cp failed: {cp.stderr}") - - start = subprocess.run( - ["docker", "start", self.name], - stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, - text=True, - ) - self.assertEqual(0, start.returncode, - f"docker start failed; check argv 'run --listen 0.0.0.0:8888'") - - port_result = subprocess.run( - ["docker", "port", self.name, "8888"], - capture_output=True, text=True, - ) - first_line = (port_result.stdout or "").splitlines()[0] if port_result.stdout else "" - host_port = first_line.rsplit(":", 1)[-1] if first_line else "" - self.assertTrue(host_port, "could not determine published port") - - health_url = f"http://127.0.0.1:{host_port}/health" - body = "" - for _ in range(15): - try: - with urllib.request.urlopen(health_url, timeout=2) as resp: - body = resp.read().decode("utf-8") - break - except (urllib.error.URLError, urllib.error.HTTPError, ConnectionError): - time.sleep(1) - - self.assertIn('"status":"healthy"', body, "health body status:healthy") - self.assertRegex(body, r'"version":"[0-9]+\.[0-9]+\.[0-9]+"', - "health body has version field") - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_pipelock_yaml.py b/tests/test_pipelock_yaml.py deleted file mode 100644 index ae6a80b..0000000 --- a/tests/test_pipelock_yaml.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Unit: PipelockProxy.prepare — produces a pipelock YAML config -containing the expected top-level keys and per-bottle entries. We -don't fully parse YAML; we grep for content shape.""" - -import os -import tempfile -import unittest -from pathlib import Path - -from claude_bottle.backend.docker.pipelock import DockerPipelockProxy -from claude_bottle.manifest import Manifest -from tests.fixtures import fixture_minimal, fixture_with_ssh - - -class TestPipelockProxyPrepare(unittest.TestCase): - def setUp(self): - self.out_dir = Path(tempfile.mkdtemp()) - self.proxy = DockerPipelockProxy() - - def tearDown(self): - import shutil - shutil.rmtree(self.out_dir, ignore_errors=True) - - def test_minimal(self): - yaml_path = self.out_dir / "min.yaml" - self.proxy.prepare(fixture_minimal().bottles["dev"], "demo", yaml_path) - content = yaml_path.read_text() - self.assertIn("mode: strict", content) - self.assertIn("enforce: true", content) - self.assertIn("api_allowlist:", content) - self.assertIn("api.anthropic.com", content) - self.assertIn("raw.githubusercontent.com", content) - self.assertIn("forward_proxy:", content) - self.assertIn("enabled: true", content) - self.assertIn("dlp:", content) - self.assertIn("include_defaults: true", content) - self.assertIn("scan_env: true", content) - # No ssh entries → no trusted_domains nor ssrf block. - self.assertNotIn("trusted_domains:", content) - self.assertNotIn("ssrf:", content) - - def test_ssh_blocks(self): - yaml_path = self.out_dir / "ssh.yaml" - self.proxy.prepare(fixture_with_ssh().bottles["dev"], "demo", yaml_path) - content = yaml_path.read_text() - self.assertIn("trusted_domains:", content) - self.assertIn("github.com", content) - self.assertIn("ssrf:", content) - self.assertIn("ip_allowlist:", content) - self.assertIn("100.78.141.42/32", content) - # ipv4 host should also be in api_allowlist (strict mode requires both). - self.assertIn("100.78.141.42", content) - - def test_secret_hygiene(self): - manifest = Manifest.from_json_obj({ - "bottles": { - "dev": { - "env": { - "MY_SECRET": "literal-value-should-not-appear", - "ANOTHER": "?prompt-message", - }, - "egress": {"allowlist": ["github.com"]}, - } - }, - "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, - }) - yaml_path = self.out_dir / "secret.yaml" - self.proxy.prepare(manifest.bottles["dev"], "demo", yaml_path) - content = yaml_path.read_text() - self.assertNotIn("literal-value-should-not-appear", content) - self.assertNotIn("MY_SECRET", content) - self.assertNotIn("prompt-message", content) - - def test_file_mode_is_600(self): - yaml_path = self.out_dir / "min.yaml" - self.proxy.prepare(fixture_minimal().bottles["dev"], "demo", yaml_path) - mode = os.stat(yaml_path).st_mode & 0o777 - self.assertEqual(0o600, mode) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_cli_start_format.py b/tests/unit/test_cli_start_format.py new file mode 100644 index 0000000..e88f8d8 --- /dev/null +++ b/tests/unit/test_cli_start_format.py @@ -0,0 +1,53 @@ +"""Unit: argparse-level CLI checks for `start --format`. + +Lives in tests/unit/ because nothing here touches Docker — the CLI +exits at argument-validation time before any backend code runs. +""" + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent + + +class TestStartFormatFlag(unittest.TestCase): + def test_json_format_requires_dry_run(self): + """Emitting JSON in a real run would race the y/N prompt; the + CLI must reject the combination with a message that names the + offending flag.""" + work_dir = Path(tempfile.mkdtemp()) + try: + (work_dir / "claude-bottle.json").write_text(json.dumps({ + "bottles": {"dev": {}}, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + })) + env = os.environ.copy() + env["HOME"] = str(work_dir) + env.pop("CLAUDE_BOTTLE_DRY_RUN", None) + result = subprocess.run( + [ + sys.executable, str(REPO_ROOT / "cli.py"), + "start", "--format", "json", "demo", + ], + cwd=work_dir, + env=env, + capture_output=True, + text=True, + ) + self.assertNotEqual(0, result.returncode) + self.assertIn( + "--format=json requires --dry-run", result.stderr, + f"expected the flag-conflict message; got stderr={result.stderr!r}", + ) + finally: + import shutil + shutil.rmtree(work_dir, ignore_errors=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_manifest_runtime.py b/tests/unit/test_manifest_runtime.py new file mode 100644 index 0000000..b365862 --- /dev/null +++ b/tests/unit/test_manifest_runtime.py @@ -0,0 +1,39 @@ +"""Unit: bottle 'runtime' field is no longer supported (PRD 0003). + +gVisor is now auto-detected by the Docker factory. A manifest carrying +the legacy 'runtime' field must fail, regardless of value, rather than +silently ignoring.""" + +import unittest + +from claude_bottle.log import Die +from claude_bottle.manifest import Bottle, Manifest + + +def _manifest_with_runtime(value: object) -> dict: + return { + "bottles": {"dev": {"runtime": value}}, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + } + + +class TestManifestRuntimeRemoved(unittest.TestCase): + def test_loads_when_runtime_absent(self): + m = Manifest.from_json_obj({ + "bottles": {"dev": {}}, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + }) + self.assertIn("dev", m.bottles) + + def test_bottle_dataclass_has_no_runtime_attribute(self): + self.assertFalse(hasattr(Bottle(), "runtime")) + + def test_any_runtime_value_is_rejected(self): + for value in ("runsc", "runc", "kata-runtime", "", 42, None): + with self.subTest(value=value): + with self.assertRaises(Die): + Manifest.from_json_obj(_manifest_with_runtime(value)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_pipelock_allowlist.py b/tests/unit/test_pipelock_allowlist.py new file mode 100644 index 0000000..501f364 --- /dev/null +++ b/tests/unit/test_pipelock_allowlist.py @@ -0,0 +1,55 @@ +"""Unit: pipelock_effective_allowlist — the union of baked-in defaults, +bottle.egress.allowlist, and bottle.ssh[].Hostname. Plus a small check +that IPv4 hostnames pick up the /32 suffix when classified as CIDRs. + +The lower-level one-line helpers (pipelock_bottle_allowlist, +pipelock_bottle_ssh_hostnames, pipelock_bottle_ssh_trusted_domains) +are exercised end-to-end by test_union_and_dedup, so they don't get +their own tests.""" + +import unittest + +from claude_bottle.manifest import Manifest +from claude_bottle.pipelock import ( + pipelock_bottle_ssh_ip_cidrs, + pipelock_effective_allowlist, +) +from tests.fixtures import fixture_with_ssh + + +class TestEffectiveAllowlist(unittest.TestCase): + def test_union_and_dedup(self): + manifest = Manifest.from_json_obj({ + "bottles": { + "dev": { + "egress": {"allowlist": ["registry.npmjs.org"]}, + "ssh": [ + {"Host": "ts", "IdentityFile": "/dev/null", + "Hostname": "100.78.141.42", "User": "git", "Port": 30009}, + {"Host": "gh", "IdentityFile": "/dev/null", + "Hostname": "github.com", "User": "git", "Port": 22}, + ], + } + }, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + }) + eff = pipelock_effective_allowlist(manifest.bottles["dev"]) + self.assertIn("api.anthropic.com", eff, "baked default present") + self.assertIn("registry.npmjs.org", eff, "egress.allowlist present") + self.assertIn("100.78.141.42", eff, "ssh ipv4 hostname present") + self.assertIn("github.com", eff, "ssh hostname present") + self.assertEqual(len(eff), len(set(eff)), "deduplicated") + self.assertEqual(eff, sorted(eff), "sorted") + + +class TestSSHIPCidrs(unittest.TestCase): + def test_ipv4_hostname_gets_32_suffix(self): + cidrs = pipelock_bottle_ssh_ip_cidrs(fixture_with_ssh().bottles["dev"]) + self.assertIn("100.78.141.42/32", cidrs) + # Hostname-typed entries don't end up here. + self.assertNotIn("github.com", cidrs) + self.assertNotIn("github.com/32", cidrs) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pipelock_classify.py b/tests/unit/test_pipelock_classify.py similarity index 100% rename from tests/test_pipelock_classify.py rename to tests/unit/test_pipelock_classify.py diff --git a/tests/unit/test_pipelock_yaml.py b/tests/unit/test_pipelock_yaml.py new file mode 100644 index 0000000..4618e62 --- /dev/null +++ b/tests/unit/test_pipelock_yaml.py @@ -0,0 +1,97 @@ +"""Unit: pipelock config building and YAML rendering. + +`pipelock_build_config` produces the structured config dict pipelock +will load; tests assert on that dict so they don't break on cosmetic +YAML changes. A small set of tests still hit the rendered output for +properties that only make sense on disk (file mode, no-secret-leakage). +""" + +import os +import tempfile +import unittest +from pathlib import Path + +from claude_bottle.backend.docker.pipelock import DockerPipelockProxy +from claude_bottle.manifest import Manifest +from claude_bottle.pipelock import pipelock_build_config, pipelock_render_yaml +from tests.fixtures import fixture_minimal, fixture_with_ssh + + +class TestBuildConfig(unittest.TestCase): + def test_minimal_shape(self): + cfg = pipelock_build_config(fixture_minimal().bottles["dev"]) + self.assertEqual("strict", cfg["mode"]) + self.assertEqual(True, cfg["enforce"]) + self.assertEqual({"enabled": True}, cfg["forward_proxy"]) + self.assertEqual( + {"include_defaults": True, "scan_env": True}, cfg["dlp"] + ) + # Baked defaults always present. + self.assertIn("api.anthropic.com", cfg["api_allowlist"]) + self.assertIn("raw.githubusercontent.com", cfg["api_allowlist"]) + # No SSH entries → no trusted_domains, no ssrf. + self.assertNotIn("trusted_domains", cfg) + self.assertNotIn("ssrf", cfg) + + def test_ssh_shape(self): + cfg = pipelock_build_config(fixture_with_ssh().bottles["dev"]) + self.assertIn("github.com", cfg["trusted_domains"]) + self.assertNotIn("100.78.141.42", cfg["trusted_domains"]) + self.assertIn("100.78.141.42/32", cfg["ssrf"]["ip_allowlist"]) + # Strict mode: IPv4 host is also in the api_allowlist union. + self.assertIn("100.78.141.42", cfg["api_allowlist"]) + + +class TestRenderAndWrite(unittest.TestCase): + def setUp(self): + self.out_dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + import shutil + shutil.rmtree(self.out_dir, ignore_errors=True) + + def test_render_emits_required_top_level_keys(self): + """One render-level smoke check: the serialized YAML is plausibly + the shape pipelock expects. We don't grep every key here — that's + what TestBuildConfig is for.""" + cfg = pipelock_build_config(fixture_with_ssh().bottles["dev"]) + text = pipelock_render_yaml(cfg) + for required in ( + "api_allowlist:", + "forward_proxy:", + "trusted_domains:", + "ssrf:", + "dlp:", + ): + self.assertIn(required, text) + + def test_prepare_writes_file_at_mode_600(self): + plan = DockerPipelockProxy().prepare( + fixture_minimal().bottles["dev"], "demo", self.out_dir + ) + self.assertEqual(0o600, os.stat(plan.yaml_path).st_mode & 0o777) + + def test_prepare_does_not_leak_env_names_or_values(self): + manifest = Manifest.from_json_obj({ + "bottles": { + "dev": { + "env": { + "MY_SECRET": "literal-value-should-not-appear", + "ANOTHER": "?prompt-message", + }, + "egress": {"allowlist": ["github.com"]}, + } + }, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + }) + plan = DockerPipelockProxy().prepare( + manifest.bottles["dev"], "demo", self.out_dir + ) + content = plan.yaml_path.read_text() + self.assertNotIn("literal-value-should-not-appear", content) + self.assertNotIn("MY_SECRET", content) + self.assertNotIn("prompt-message", content) + + +if __name__ == "__main__": + unittest.main()