Merge pull request 'Refactor tests' (#6) from refactor-tests into main

.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

.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

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):

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."""

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,
+        }

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:

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-<slug>`), 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

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_<topic>.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_<topic>.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`.

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

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(),

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()

tests/run_tests.py

@@ -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:]))

tests/test_manifest_runtime.py

@@ -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()

tests/test_pipelock_allowlist.py

@@ -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()

tests/test_pipelock_sidecar_smoke.py

@@ -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()

tests/test_pipelock_yaml.py

@@ -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()

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()

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()

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()

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()