test(pipelock): drive sidecar smoke through production prepare/start

The old smoke test hand-rolled the docker create/cp/start sequence in
parallel with what DockerPipelockProxy.start already does, so any
divergence in production code wouldn't trip it. Rewritten to call
.prepare and .start directly and probe /health from a sibling curl
container on the same internal network — same access topology the
agent container uses in production.

In-network probing means the test no longer depends on a published
port, so it can run under act_runner (where host-loopback port
publishing isn't reachable from the job container).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

tests/integration/test_pipelock_sidecar_smoke.py

@@ -1,103 +1,113 @@
-"""Integration: full sidecar smoke test. Boots a pipelock container the
+"""Integration: drive the production pipelock-sidecar bring-up
-same way cli.py does (docker create + docker cp YAML + docker start),
+(`DockerPipelockProxy.prepare` → `.start`) and probe /health from a
-then probes /health."""
+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 re
 import shutil
 import subprocess
 import tempfile
-import time
 import unittest
-import urllib.request
 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_IMAGE,
+    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):
-    def setUp(self):
+    @classmethod
-        self.name = f"cb-test-pipelock-smoke-{os.getpid()}"
+    def setUpClass(cls):
-        self.work_dir = Path(tempfile.mkdtemp())
+        # Pre-pull curlimages/curl so the per-test retry loop isn't
-
+        # racing the registry. Skip cleanly if the pull fails (the
-    def tearDown(self):
+        # canary suite will surface a real registry outage separately).
-        subprocess.run(
+        result = subprocess.run(
-            ["docker", "rm", "-f", self.name],
+            ["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(
+    def test_prepare_and_start_yield_healthy_sidecar(self):
-        os.environ.get("GITEA_ACTIONS") == "true",
+        proxy = DockerPipelockProxy()
-        "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)
+        prep = proxy.prepare(fixture_minimal().bottles["dev"], self.slug, yaml_path)
 
-        create = subprocess.run(
+        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", "create",
+                "docker", "run", "--rm",
-                "--name", self.name,
+                "--network", self.internal_net,
-                "-p", "0:8888",
+                CURL_IMAGE,
-                PIPELOCK_IMAGE,
+                "-sf", "--max-time", "2",
-                "run", "--config", "/etc/pipelock.yaml",
+                "--retry", "15",
-                "--listen", "0.0.0.0:8888",
+                "--retry-delay", "1",
+                "--retry-connrefused",
+                f"http://{self.sidecar_name}:{PIPELOCK_PORT}/health",
             ],
-            stdout=subprocess.DEVNULL,
+            capture_output=True,
-            stderr=subprocess.PIPE,
             text=True,
+            timeout=60,
         )
-        self.assertEqual(0, create.returncode, f"docker create failed: {create.stderr}")
+        self.assertEqual(
-
+            0, probe.returncode,
-        # Guard against /etc/pipelock/ regressions: the path must be
+            f"health probe failed: stdout={probe.stdout!r} stderr={probe.stderr!r}",
-        # /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}")
+        body = probe.stdout
-
+        self.assertIn('"status":"healthy"', body)
-        start = subprocess.run(
+        self.assertRegex(body, r'"version":"[0-9]+\.[0-9]+\.[0-9]+"')
-            ["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__":