didericis/bot-bottle
1
0
Fork 0
Code Issues 5 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Merge pull request 'refactor(sidecars): drop vestigial start/stop methods (PRD 0024 chunk 3)' (#57) from prd-0024-chunk-3-backend-python-trim into main
test / unit (push) Successful in 20s
Details
test / integration (push) Successful in 44s
Details

Browse Source
This commit was merged in pull request #57.
This commit is contained in:
didericis
2026-05-27 01:03:11 -04:00
parent c37344608b 539234f29e
commit fff0391d1b
18 changed files with 80 additions and 1758 deletions
Download Patch File Download Diff File Expand all files Collapse all files
claude_bottle/backend/docker/backend.py
+2 -26
View File
@@ -23,15 +23,11 @@ from . import prepare as _prepare
from .bottle import DockerBottle from .bottle import DockerBottle
from .bottle_cleanup_plan import DockerBottleCleanupPlan from .bottle_cleanup_plan import DockerBottleCleanupPlan
from .bottle_plan import DockerBottlePlan from .bottle_plan import DockerBottlePlan
from .egress import DockerEgress
from .git_gate import DockerGitGate
from .pipelock import DockerPipelockProxy
from .provision import ca as _ca from .provision import ca as _ca
from .provision import git as _git from .provision import git as _git
from .provision import prompt as _prompt from .provision import prompt as _prompt
from .provision import skills as _skills from .provision import skills as _skills
from .provision import supervise as _supervise_prov from .provision import supervise as _supervise_prov
from .supervise import DockerSupervise
class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanupPlan"]): class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanupPlan"]):
@@ -40,32 +36,12 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
name = "docker" name = "docker"
def __init__(self) -> None:
self._proxy = DockerPipelockProxy()
self._git_gate = DockerGitGate()
self._egress = DockerEgress()
self._supervise = DockerSupervise()
def _resolve_plan(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan: def _resolve_plan(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan:
return _prepare.resolve_plan( return _prepare.resolve_plan(spec, stage_dir=stage_dir)
spec,
stage_dir=stage_dir,
proxy=self._proxy,
git_gate=self._git_gate,
egress=self._egress,
supervise=self._supervise,
)
@contextmanager @contextmanager
def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]: def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
with _launch.launch( with _launch.launch(plan, provision=self.provision) as bottle:
plan,
proxy=self._proxy,
git_gate=self._git_gate,
egress=self._egress,
supervise=self._supervise,
provision=self.provision,
) as bottle:
yield bottle yield bottle
def provision_ca(self, plan: DockerBottlePlan, target: str) -> None: def provision_ca(self, plan: DockerBottlePlan, target: str) -> None:
claude_bottle/backend/docker/egress.py
+5 -219
View File
@@ -13,14 +13,8 @@ import os
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from ...egress import ( from ...egress import Egress
EGRESS_HOSTNAME, from ...log import die
EGRESS_ROUTES_IN_CONTAINER,
Egress,
EgressPlan,
egress_resolve_token_values,
)
from ...log import die, info, warn
from . import util as docker_mod from . import util as docker_mod
@@ -166,214 +160,6 @@ def egress_tls_init(stage_dir: Path) -> tuple[Path, Path]:
class DockerEgress(Egress): class DockerEgress(Egress):
"""Brings the egress sidecar up and down via Docker.""" """Docker-flavored Egress: inherits `.prepare()` from the base.
Container lifecycle is owned by compose; per-container
def start(self, plan: EgressPlan) -> str: `.start()` / `.stop()` were removed in PRD 0024 chunk 3."""
"""Boot the egress sidecar:
1. Resolve every host TokenRef env var into a concrete
value. Fails early if any are unset.
2. Build the egress image (no-op when cache is hot).
3. `docker create` on the internal network with
`--network-alias egress`, the `HTTPS_PROXY=pipelock`
env (so the upstream leg traverses pipelock), the
`EGRESS_UPSTREAM_CA` env pointing at the in-container
pipelock-CA path (so mitmproxy trusts pipelock's MITM),
and one `-e EGRESS_TOKEN_N` flag per token slot.
Secret values arrive via subprocess env, never argv.
4. `docker cp` the routes.yaml, mitmproxy CA (cert+key
concat), and pipelock CA (cert only) into the container.
5. Attach to the per-agent egress network so the proxy can
reach pipelock.
6. `docker start`.
Returns the container name (the target passed to `.stop`)."""
if not plan.routes:
die("DockerEgress.start called with no routes; caller should skip")
if not plan.internal_network or not plan.egress_network:
die(
"DockerEgress.start: internal_network / egress_network must be "
"populated on the plan before start"
)
if not plan.routes_path.is_file():
die(
f"egress routes file missing at {plan.routes_path}; "
f"Egress.prepare must run first"
)
if plan.mitmproxy_ca_host_path == Path() or not plan.mitmproxy_ca_host_path.is_file():
die(
f"DockerEgress.start: mitmproxy CA missing at "
f"{plan.mitmproxy_ca_host_path}; egress_tls_init must run first"
)
# pipelock CA + upstream proxy URL: both must be present (we
# use HTTPS_PROXY=pipelock with pipelock's own MITM CA on the
# upstream leg) or both absent (egress goes direct, for
# standalone integration tests that don't bring pipelock up).
route_via_pipelock = bool(plan.pipelock_proxy_url) or plan.pipelock_ca_host_path != Path()
if route_via_pipelock:
if not plan.pipelock_proxy_url:
die(
"DockerEgress.start: pipelock_ca_host_path is set but "
"pipelock_proxy_url is empty; populate both or neither."
)
if not plan.pipelock_ca_host_path.is_file():
die(
f"DockerEgress.start: pipelock CA missing at "
f"{plan.pipelock_ca_host_path}; pipelock_tls_init must run first"
)
# Resolve host env vars into concrete values. Must happen at
# start time (not prepare) — the values flow into the sidecar's
# environ via subprocess env. The plan never holds them.
token_values = egress_resolve_token_values(
plan.token_env_map, dict(os.environ),
)
build_egress_image()
name = egress_container_name(plan.slug)
info(f"starting egress sidecar {name} on network {plan.internal_network}")
create_args = [
"docker", "create",
"--name", name,
"--network", plan.internal_network,
"--network-alias", EGRESS_HOSTNAME,
]
if route_via_pipelock:
# Route egress's outbound traffic through pipelock
# so the egress allowlist + DLP body scanner apply to
# the egress → upstream leg. Pipelock MITMs each
# handshake with its per-bottle CA, which is docker-cp'd
# in below and pointed to via the EGRESS_UPSTREAM_CA
# env (entrypoint conditionally adds the matching --set
# flag).
#
# EGRESS_UPSTREAM_PROXY is the mechanism: mitmproxy
# does NOT honor HTTPS_PROXY env vars on its outbound
# side (it's a proxy server, not a client). The
# entrypoint reads this env and switches mitmdump to
# `--mode upstream:<URL>` so all post-MITM traffic
# CONNECTs to pipelock instead of going direct. The
# HTTPS/HTTP_PROXY env vars below are kept for any
# bundled client libraries (mitmproxy plugin requests,
# etc.) that might honor them — harmless if ignored.
create_args.extend([
"-e", f"EGRESS_UPSTREAM_PROXY={plan.pipelock_proxy_url}",
"-e", f"HTTPS_PROXY={plan.pipelock_proxy_url}",
"-e", f"HTTP_PROXY={plan.pipelock_proxy_url}",
"-e", "NO_PROXY=localhost,127.0.0.1",
"-e", f"EGRESS_UPSTREAM_CA={EGRESS_PIPELOCK_CA_IN_CONTAINER}",
])
# One -e flag per token slot; values arrive via subprocess env.
# docker create with `-e NAME` (no =VALUE) reads NAME from the
# current process env at create time. We pass `env=child_env`
# to subprocess.run so the value comes from token_values, not
# the host's os.environ directly — keeps the resolver in one
# place and lets egress_resolve_token_values surface
# missing-env errors with a clear hint.
for token_env in sorted(plan.token_env_map.keys()):
create_args.extend(["-e", token_env])
create_args.append(EGRESS_IMAGE)
child_env: dict[str, str] = {**os.environ, **token_values}
create_result = subprocess.run(
create_args, capture_output=True, text=True, env=child_env, check=False,
)
if create_result.returncode != 0:
die(
f"failed to create egress sidecar {name}: "
f"{create_result.stderr.strip()}"
)
# routes.yaml also lands inside the container; bump to 644
# for the same reason as the CAs — mitmproxy user (uid 1000)
# has to read it. Host stage_dir is mode 700 so the file
# isn't actually exposed to other host users.
plan.routes_path.chmod(0o644)
# Pipelock CA: pipelock itself runs as root so its in-pipelock
# copy doesn't care about mode, but egress's mitmproxy
# user does. Bump on the host so docker cp into egress
# carries world-readable.
if route_via_pipelock:
plan.pipelock_ca_host_path.chmod(0o644)
cps: list[tuple[Path, str, str]] = [
(plan.routes_path, EGRESS_ROUTES_IN_CONTAINER, "routes.yaml"),
(plan.mitmproxy_ca_host_path, EGRESS_CA_IN_CONTAINER, "mitmproxy CA"),
]
if route_via_pipelock:
cps.append((
plan.pipelock_ca_host_path,
EGRESS_PIPELOCK_CA_IN_CONTAINER,
"pipelock CA",
))
for src, dst, label in cps:
cp_result = subprocess.run(
["docker", "cp", str(src), f"{name}:{dst}"],
capture_output=True,
text=True,
check=False,
)
if cp_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to copy {label} into {name}: "
f"{cp_result.stderr.strip()}"
)
connect_result = subprocess.run(
["docker", "network", "connect", plan.egress_network, name],
capture_output=True, text=True, check=False,
)
if connect_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to attach egress sidecar {name} to egress network "
f"{plan.egress_network}: {connect_result.stderr.strip()}"
)
start_result = subprocess.run(
["docker", "start", name], capture_output=True, text=True, check=False,
)
if start_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to start egress sidecar {name}: "
f"{start_result.stderr.strip()}"
)
return name
def stop(self, target: str) -> None:
"""Idempotent: missing container is success. `target` is the
container name returned by `.start`."""
if subprocess.run(
["docker", "inspect", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode == 0:
if subprocess.run(
["docker", "rm", "-f", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
warn(
f"failed to remove egress sidecar {target}; "
f"clean up with 'docker rm -f {target}'"
)
claude_bottle/backend/docker/git_gate.py
+4 -169
View File
@@ -5,17 +5,9 @@ step (upstream lift + entrypoint/hook render) from `GitGate`."""
from __future__ import annotations from __future__ import annotations
import os import os
import subprocess
from pathlib import Path from pathlib import Path
from ...git_gate import ( from ...git_gate import GitGate
GitGate,
GitGatePlan,
git_gate_aggregate_extra_hosts,
git_gate_known_hosts_line,
)
from ...log import die, info, warn
from ...util import expand_tilde
from . import util as docker_mod from . import util as docker_mod
@@ -59,163 +51,6 @@ def build_git_gate_image() -> None:
class DockerGitGate(GitGate): class DockerGitGate(GitGate):
"""Brings the git-gate sidecar up and down via Docker.""" """Docker-flavored GitGate: inherits `.prepare()` from the base.
Container lifecycle is owned by compose; per-container
def start(self, plan: GitGatePlan) -> str: `.start()` / `.stop()` were removed in PRD 0024 chunk 3."""
"""Boot the gate sidecar:
1. Build the gate image (no-op when cache is hot).
2. `docker create` on the internal network with the canonical
name; the image's ENTRYPOINT runs the cp'd entrypoint
script at start time.
3. `docker cp` the entrypoint, the shared pre-receive hook,
and each upstream's identity + known_hosts into the
container.
4. Attach to the per-agent egress network so the gate can
reach the real upstream.
5. `docker start`.
Returns the container name (the target passed to `.stop`)."""
if not plan.upstreams:
die("DockerGitGate.start called with no upstreams; caller should skip")
if not plan.internal_network or not plan.egress_network:
die(
"DockerGitGate.start: internal_network / egress_network must be "
"populated on the plan before start"
)
if not plan.entrypoint_script.is_file():
die(
f"git-gate entrypoint missing at {plan.entrypoint_script}; "
f"GitGate.prepare must run first"
)
if not plan.hook_script.is_file():
die(
f"git-gate hook missing at {plan.hook_script}; "
f"GitGate.prepare must run first"
)
if not plan.access_hook_script.is_file():
die(
f"git-gate access-hook missing at {plan.access_hook_script}; "
f"GitGate.prepare must run first"
)
build_git_gate_image()
name = git_gate_container_name(plan.slug)
info(f"starting git-gate sidecar {name} on network {plan.internal_network}")
create_args = [
"docker", "create",
"--name", name,
"--network", plan.internal_network,
]
for host, ip in git_gate_aggregate_extra_hosts(plan.upstreams).items():
create_args.extend(["--add-host", f"{host}:{ip}"])
create_args.append(GIT_GATE_IMAGE)
create_result = subprocess.run(
create_args, capture_output=True, text=True, check=False,
)
if create_result.returncode != 0:
die(
f"failed to create git-gate sidecar {name}: "
f"{create_result.stderr.strip()}"
)
# Order matters: entrypoint + hook first so they're present
# when docker start fires. Per-upstream creds afterwards.
stage_dir = plan.entrypoint_script.parent
cps: list[tuple[str, str, str]] = [
(str(plan.entrypoint_script), GIT_GATE_ENTRYPOINT_IN_CONTAINER, "entrypoint"),
(str(plan.hook_script), GIT_GATE_HOOK_IN_CONTAINER, "pre-receive hook"),
(str(plan.access_hook_script), GIT_GATE_ACCESS_HOOK_IN_CONTAINER, "access-hook"),
]
for u in plan.upstreams:
keypath = expand_tilde(u.identity_file)
cps.append((
keypath,
f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{u.name}-key",
f"upstream key for '{u.name}'",
))
if u.known_host_key:
hosts_path = stage_dir / f"git_gate_known_hosts_{u.name}"
hosts_path.write_text(
git_gate_known_hosts_line(
u.upstream_host, u.upstream_port, u.known_host_key
)
)
hosts_path.chmod(0o600)
cps.append((
str(hosts_path),
f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{u.name}-known_hosts",
f"upstream known_hosts for '{u.name}'",
))
for src, dst, label in cps:
cp_result = subprocess.run(
["docker", "cp", src, f"{name}:{dst}"],
capture_output=True,
text=True,
check=False,
)
if cp_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to copy {label} into {name}: "
f"{cp_result.stderr.strip()}"
)
connect_result = subprocess.run(
["docker", "network", "connect", plan.egress_network, name],
capture_output=True, text=True, check=False,
)
if connect_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to attach git-gate sidecar {name} to egress network "
f"{plan.egress_network}: {connect_result.stderr.strip()}"
)
start_result = subprocess.run(
["docker", "start", name], capture_output=True, text=True, check=False,
)
if start_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to start git-gate sidecar {name}: "
f"{start_result.stderr.strip()}"
)
return name
def stop(self, target: str) -> None:
"""Idempotent: missing container is success. `target` is the
container name returned by `.start`."""
if subprocess.run(
["docker", "inspect", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode == 0:
if subprocess.run(
["docker", "rm", "-f", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
warn(
f"failed to remove git-gate sidecar {target}; "
f"clean up with 'docker rm -f {target}'"
)
claude_bottle/backend/docker/launch.py
+2 -16
View File
@@ -63,14 +63,11 @@ from .compose import (
compose_up, compose_up,
write_compose_file, write_compose_file,
) )
from .egress import DockerEgress, egress_tls_init from .egress import egress_tls_init
from .git_gate import DockerGitGate
from .pipelock import ( from .pipelock import (
DockerPipelockProxy,
pipelock_proxy_url, pipelock_proxy_url,
pipelock_tls_init, pipelock_tls_init,
) )
from .supervise import DockerSupervise
# Where the repo root lives, for `docker build` context. Computed once. # Where the repo root lives, for `docker build` context. Computed once.
@@ -81,21 +78,10 @@ _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
def launch( def launch(
plan: DockerBottlePlan, plan: DockerBottlePlan,
*, *,
proxy: DockerPipelockProxy,
git_gate: DockerGitGate,
egress: DockerEgress,
supervise: DockerSupervise,
provision: Callable[[DockerBottlePlan, str], str | None], provision: Callable[[DockerBottlePlan, str], str | None],
) -> Generator[DockerBottle, None, None]: ) -> Generator[DockerBottle, None, None]:
"""Build, launch, and provision a Docker bottle via compose. """Build, launch, and provision a Docker bottle via compose.
Teardown on exit. The per-sidecar `proxy / git_gate / egress / Teardown on exit."""
supervise` parameters are vestigial from the pre-compose flow —
kept for backwards-compat with backend.py's call site; the
`start()`/`stop()` methods on those classes are no longer
invoked (chunk 3 collapsed them into the compose service spec).
They'll be removed entirely in a follow-up cleanup."""
del proxy, git_gate, egress, supervise # not invoked in compose flow
stack = ExitStack() stack = ExitStack()
def teardown() -> None: def teardown() -> None:
claude_bottle/backend/docker/pipelock.py
+22 -121
View File
@@ -1,6 +1,14 @@
"""DockerPipelockProxy — the Docker-specific implementation of the """DockerPipelockProxy — the Docker-specific implementation of the
sidecar's start/stop lifecycle. Inherits the platform-agnostic sidecar's `.prepare()` step + in-container CA path constants.
YAML-config generation from PipelockProxy.""" Inherits the platform-agnostic YAML-config generation from
PipelockProxy.
The per-container `.start()` / `.stop()` lifecycle was deleted in
PRD 0024 chunk 3 — compose-up owns the container lifecycle (PRD
0018) and the bundle path (PRD 0024) collapses pipelock + egress
+ git-gate + supervise into one container. What remains here is
the prepare-time YAML rendering + the CA path constants the
compose renderer reads."""
from __future__ import annotations from __future__ import annotations
@@ -8,8 +16,8 @@ import os
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from ...log import die, info, warn from ...log import die
from ...pipelock import PipelockProxy, PipelockProxyPlan from ...pipelock import PipelockProxy
# Pipelock image, pinned by digest. The digest is the multi-arch image # Pipelock image, pinned by digest. The digest is the multi-arch image
@@ -22,9 +30,9 @@ PIPELOCK_IMAGE = os.environ.get(
# Listening port for pipelock's forward proxy. # Listening port for pipelock's forward proxy.
PIPELOCK_PORT = os.environ.get("CLAUDE_BOTTLE_PIPELOCK_PORT", "8888") PIPELOCK_PORT = os.environ.get("CLAUDE_BOTTLE_PIPELOCK_PORT", "8888")
# In-container paths where the per-bottle CA cert + key land after # In-container paths where the per-bottle CA cert + key land via
# `docker cp` in `DockerPipelockProxy.start`. Pipelock's rendered # the compose renderer's bind-mounts. Pipelock's rendered YAML
# YAML references these paths under `tls_interception`. # references these paths under `tls_interception`.
PIPELOCK_CA_CERT_IN_CONTAINER = "/etc/pipelock-ca.pem" PIPELOCK_CA_CERT_IN_CONTAINER = "/etc/pipelock-ca.pem"
PIPELOCK_CA_KEY_IN_CONTAINER = "/etc/pipelock-ca-key.pem" PIPELOCK_CA_KEY_IN_CONTAINER = "/etc/pipelock-ca-key.pem"
@@ -46,10 +54,10 @@ def pipelock_tls_init(stage_dir: Path) -> tuple[Path, Path]:
The image is pinned (same digest the running sidecar uses) so the The image is pinned (same digest the running sidecar uses) so the
generated CA matches what the sidecar expects. Output is owned by generated CA matches what the sidecar expects. Output is owned by
whatever UID the one-shot ran as; `DockerPipelockProxy.start` whatever UID the one-shot ran as; the compose renderer's
`docker cp`s the files into the sidecar's filesystem layer, so bind-mounts pin the files in place at runtime, so ownership
runtime ownership inside the sidecar (root in pipelock's inside the running sidecar (root in pipelock's distroless image)
distroless image) is independent.""" is independent."""
work = stage_dir / "pipelock-ca" work = stage_dir / "pipelock-ca"
work.mkdir(exist_ok=True) work.mkdir(exist_ok=True)
result = subprocess.run( result = subprocess.run(
@@ -77,117 +85,10 @@ def pipelock_tls_init(stage_dir: Path) -> tuple[Path, Path]:
class DockerPipelockProxy(PipelockProxy): class DockerPipelockProxy(PipelockProxy):
"""Brings the pipelock sidecar up and down via Docker.""" """Docker-flavored PipelockProxy: inherits `.prepare()` from the
base, exposes the in-container CA paths the renderer reads.
Container lifecycle is owned by compose."""
CA_CERT_IN_CONTAINER = PIPELOCK_CA_CERT_IN_CONTAINER CA_CERT_IN_CONTAINER = PIPELOCK_CA_CERT_IN_CONTAINER
CA_KEY_IN_CONTAINER = PIPELOCK_CA_KEY_IN_CONTAINER CA_KEY_IN_CONTAINER = PIPELOCK_CA_KEY_IN_CONTAINER
def start(self, plan: PipelockProxyPlan) -> str:
"""Boot the pipelock sidecar:
1. `docker create` on the internal network with the canonical
name and argv `run --config /etc/pipelock.yaml --listen
0.0.0.0:<port>`.
2. `docker cp` the YAML config to /etc/pipelock.yaml.
3. `docker cp` the CA cert + key to /etc/pipelock-ca.pem
and /etc/pipelock-ca-key.pem (pipelock runs as root in
its distroless image, so no chown is needed).
4. Attach to the per-agent egress network.
5. `docker start`.
Returns the container name (the proxy_target passed to .stop)."""
name = pipelock_container_name(plan.slug)
if not plan.yaml_path.is_file():
die(
f"pipelock yaml not found at {plan.yaml_path}; "
f"PipelockProxy.prepare must run first"
)
if not plan.ca_cert_host_path.is_file() or not plan.ca_key_host_path.is_file():
die(
f"pipelock CA missing at {plan.ca_cert_host_path} / "
f"{plan.ca_key_host_path}; pipelock_tls_init must run first"
)
info(f"starting pipelock sidecar {name} on network {plan.internal_network}")
create_args = [
"docker", "create",
"--name", name,
"--network", plan.internal_network,
PIPELOCK_IMAGE,
"run", "--config", "/etc/pipelock.yaml",
"--listen", f"0.0.0.0:{PIPELOCK_PORT}",
]
create_result = subprocess.run(
create_args, capture_output=True, text=True, check=False,
)
if create_result.returncode != 0:
die(
f"failed to create pipelock sidecar {name}: "
f"{create_result.stderr.strip()}"
)
for src, dst, label in (
(plan.yaml_path, "/etc/pipelock.yaml", "yaml"),
(plan.ca_cert_host_path, PIPELOCK_CA_CERT_IN_CONTAINER, "ca cert"),
(plan.ca_key_host_path, PIPELOCK_CA_KEY_IN_CONTAINER, "ca key"),
):
cp_result = subprocess.run(
["docker", "cp", str(src), f"{name}:{dst}"],
capture_output=True,
text=True,
check=False,
)
if cp_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
die(f"failed to copy pipelock {label} into {name}: {cp_result.stderr.strip()}")
connect_result = subprocess.run(
["docker", "network", "connect", plan.egress_network, name],
capture_output=True, text=True, check=False,
)
if connect_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
die(
f"failed to attach pipelock sidecar {name} to egress network "
f"{plan.egress_network}: {connect_result.stderr.strip()}"
)
start_result = subprocess.run(
["docker", "start", name], capture_output=True, text=True, check=False,
)
if start_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
die(
f"failed to start pipelock sidecar {name}: "
f"{start_result.stderr.strip()}"
)
return name
def stop(self, proxy_target: str) -> None:
"""Idempotent: missing container is success. `proxy_target` is
the container name returned by .start."""
if subprocess.run(
["docker", "inspect", proxy_target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode == 0:
if subprocess.run(
["docker", "rm", "-f", proxy_target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
warn(
f"failed to remove pipelock sidecar {proxy_target}; "
f"clean up with 'docker rm -f {proxy_target}'"
)
claude_bottle/backend/docker/prepare.py
+5 -4
View File
@@ -43,16 +43,17 @@ def resolve_plan(
spec: BottleSpec, spec: BottleSpec,
*, *,
stage_dir: Path, stage_dir: Path,
proxy: DockerPipelockProxy,
git_gate: DockerGitGate,
egress: DockerEgress,
supervise: DockerSupervise,
) -> DockerBottlePlan: ) -> DockerBottlePlan:
"""Resolve Docker-specific names and write scratch files. Trusts """Resolve Docker-specific names and write scratch files. Trusts
that the agent and its skills/git-gate keys are present — that the agent and its skills/git-gate keys are present —
validation already ran in the base class.""" validation already ran in the base class."""
docker_mod.require_docker() docker_mod.require_docker()
proxy = DockerPipelockProxy()
git_gate = DockerGitGate()
egress = DockerEgress()
supervise = DockerSupervise()
manifest = spec.manifest manifest = spec.manifest
agent = manifest.agents[spec.agent_name] agent = manifest.agents[spec.agent_name]
bottle = manifest.bottle_for(spec.agent_name) bottle = manifest.bottle_for(spec.agent_name)
claude_bottle/backend/docker/supervise.py
+3 -85
View File
@@ -5,16 +5,12 @@ step (queue dir + current-config staging) from `Supervise`."""
from __future__ import annotations from __future__ import annotations
import os import os
import subprocess
from pathlib import Path from pathlib import Path
from ...log import die, info, warn
from ...supervise import ( from ...supervise import (
QUEUE_DIR_IN_CONTAINER,
SUPERVISE_HOSTNAME, SUPERVISE_HOSTNAME,
SUPERVISE_PORT, SUPERVISE_PORT,
Supervise, Supervise,
SupervisePlan,
) )
from . import util as docker_mod from . import util as docker_mod
@@ -48,84 +44,6 @@ def build_supervise_image() -> None:
class DockerSupervise(Supervise): class DockerSupervise(Supervise):
"""Brings the supervise sidecar up and down via Docker.""" """Docker-flavored Supervise: inherits `.prepare()` from the base.
Container lifecycle is owned by compose; per-container
def start(self, plan: SupervisePlan) -> str: `.start()` / `.stop()` were removed in PRD 0024 chunk 3."""
"""Boot the supervise sidecar:
1. Build the supervise image (no-op when cache is hot).
2. `docker create` on the internal network with
`--network-alias supervise` and SUPERVISE_BOTTLE_SLUG in
the environ.
3. Bind-mount the host queue dir at /run/supervise/queue.
4. `docker start`.
No egress network — the supervise sidecar does not make
outbound calls. Returns the container name."""
if not plan.internal_network:
die("DockerSupervise.start: plan.internal_network must be set before start")
if not plan.queue_dir.is_dir():
die(
f"DockerSupervise.start: queue dir missing at {plan.queue_dir}; "
f"Supervise.prepare must run first"
)
build_supervise_image()
name = supervise_container_name(plan.slug)
info(f"starting supervise sidecar {name} on network {plan.internal_network}")
create_args = [
"docker", "create",
"--name", name,
"--network", plan.internal_network,
"--network-alias", SUPERVISE_HOSTNAME,
"-e", f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
"-e", f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
"-e", f"SUPERVISE_PORT={SUPERVISE_PORT}",
"-v", f"{plan.queue_dir}:{QUEUE_DIR_IN_CONTAINER}",
SUPERVISE_IMAGE,
]
create_result = subprocess.run(
create_args, capture_output=True, text=True, check=False,
)
if create_result.returncode != 0:
die(
f"failed to create supervise sidecar {name}: "
f"{create_result.stderr.strip()}"
)
start_result = subprocess.run(
["docker", "start", name], capture_output=True, text=True, check=False,
)
if start_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to start supervise sidecar {name}: "
f"{start_result.stderr.strip()}"
)
return name
def stop(self, target: str) -> None:
"""Idempotent: missing container is success."""
if subprocess.run(
["docker", "inspect", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode == 0:
if subprocess.run(
["docker", "rm", "-f", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
warn(
f"failed to remove supervise sidecar {target}; "
f"clean up with 'docker rm -f {target}'"
)
claude_bottle/egress.py
-13
View File
@@ -326,19 +326,6 @@ class Egress(ABC):
token_env_map=egress_token_env_map(routes), token_env_map=egress_token_env_map(routes),
) )
@abstractmethod
def start(self, plan: EgressPlan) -> str:
"""Bring up the egress sidecar according to `plan`.
Returns the target string identifying the running instance —
the same value to pass to `.stop`. Backend-specific."""
@abstractmethod
def stop(self, target: str) -> None:
"""Tear down the egress sidecar identified by `target`
(the value `.start` returned). Idempotent: a missing target
is success. Backend-specific."""
__all__ = [ __all__ = [
"DEFAULT_ALLOWLIST", "DEFAULT_ALLOWLIST",
"EGRESS_HOSTNAME", "EGRESS_HOSTNAME",
claude_bottle/git_gate.py
-11
View File
@@ -371,14 +371,3 @@ class GitGate(ABC):
upstreams=upstreams, upstreams=upstreams,
) )
@abstractmethod
def start(self, plan: GitGatePlan) -> str:
"""Bring up the gate sidecar according to `plan`. Returns the
target string identifying the running instance — the same
value to pass to `.stop`. Backend-specific."""
@abstractmethod
def stop(self, target: str) -> None:
"""Tear down the gate sidecar identified by `target` (the
value `.start` returned). Idempotent: a missing target is
success. Backend-specific."""
claude_bottle/pipelock.py
-11
View File
@@ -344,14 +344,3 @@ class PipelockProxy(ABC):
yaml_path.chmod(0o600) yaml_path.chmod(0o600)
return PipelockProxyPlan(yaml_path=yaml_path, slug=slug) return PipelockProxyPlan(yaml_path=yaml_path, slug=slug)
@abstractmethod
def start(self, plan: PipelockProxyPlan) -> str:
"""Bring up the pipelock sidecar according to `plan`. Returns
the proxy_target string identifying the running instance — the
same value to pass to `.stop`. Backend-specific."""
@abstractmethod
def stop(self, proxy_target: str) -> None:
"""Tear down the pipelock sidecar identified by `proxy_target`
(the value `.start` returned). Idempotent: a missing target is
success. Backend-specific."""
claude_bottle/supervise.py
-12
View File
@@ -494,18 +494,6 @@ class Supervise(ABC):
current_config_dir=current_config_dir, current_config_dir=current_config_dir,
) )
@abstractmethod
def start(self, plan: SupervisePlan) -> str:
"""Bring up the supervise sidecar according to `plan`. Returns
the target string identifying the running instance — the same
value to pass to `.stop`. Backend-specific."""
@abstractmethod
def stop(self, target: str) -> None:
"""Tear down the supervise sidecar identified by `target`.
Idempotent: a missing target is success."""
# --- Helpers --------------------------------------------------------------- # --- Helpers ---------------------------------------------------------------
docs/prds/0024-consolidate-sidecar-bundle.md
+22 -8
View File
@@ -376,16 +376,30 @@ rewrite.
`bottle_plan_to_compose` to emit two services. Feature flag `bottle_plan_to_compose` to emit two services. Feature flag
it via env var. Update unit tests to assert on both shapes it via env var. Update unit tests to assert on both shapes
(flag on vs off) during the migration window. (flag on vs off) during the migration window.
3. **Backend Python collapse.** Trim the four docker 3. **Backend Python collapse.** Drop the vestigial per-container
sidecar modules, consolidate container-name helpers, update `.start()` / `.stop()` methods from `DockerPipelockProxy`,
orphan-cleanup logic to look for the bundle by name. Delete `DockerEgress`, `DockerGitGate`, `DockerSupervise` (and from
old Dockerfiles. the ABCs in `claude_bottle/{pipelock,egress,git_gate,supervise}.py`).
4. **Integration test sweep.** Bring every integration test These were already documented as vestigial in PRD 0018 ch3.
that probes a four-container shape (`pipelock_container_name`, Strip vestigial sidecar-instance parameters from
`egress_container_name`, etc.) onto the bundle. Confirm `launch.launch()` and `prepare.resolve_plan()`. Delete the
integration tests that exclusively exercised those methods
(`test_pipelock_sidecar_smoke`, `test_supervise_sidecar`,
`test_git_gate_sidecar`, `test_git_gate_mirror`). Skip
`test_pipelock_apply` pending chunk 4 bringup rewrite.
Orphan cleanup already uses a prefix scan and catches the
bundle for free; no change needed. Dockerfile deletion is
deferred to chunk 5 — until the flag flips, the legacy path
still needs `Dockerfile.{egress,git-gate,supervise}` for
compose `build:`.
4. **Integration test sweep.** Rewrite `test_pipelock_apply`'s
bringup with direct `docker run` so the
`apply_allowlist_change` hot-reload retains coverage. Add
any bundle-specific integration smoke as needed. Confirm
PRD 0022 stays green. PRD 0022 stays green.
5. **Docs + flag removal.** Flip the default, remove the 5. **Docs + flag removal.** Flip the default, remove the
feature flag, update README + CLAUDE.md. feature flag, delete `Dockerfile.{egress,git-gate,supervise}`,
update README + CLAUDE.md.
## Open questions ## Open questions
tests/integration/test_git_gate_mirror.py
-391
View File
@@ -1,391 +0,0 @@
"""Integration: the git-gate is a bidirectional mirror of its
upstream (PRD 0008 v1.1).
Three round-trip assertions against a real Docker daemon plus a
sibling sshd container playing the role of "real upstream":
1. clone-through-gate returns whatever the upstream has at the
moment of clone (refs + content).
2. After a second commit lands on the upstream out-of-band, a
fetch through the gate picks it up — the access-hook is
refreshing before each upload-pack.
3. A push through the gate (clean commit) lands on the upstream's
bare repo — the pre-receive hook's forward phase works.
These are the user-facing semantics: every operation against the
gate is observably equivalent to the same operation against the
real upstream.
"""
import dataclasses
import os
import shutil
import subprocess
import tempfile
import textwrap
import unittest
from pathlib import Path
from claude_bottle.backend.docker.git_gate import (
DockerGitGate,
build_git_gate_image,
)
from claude_bottle.backend.docker.network import (
network_create_egress,
network_create_internal,
network_remove,
)
from claude_bottle.manifest import Manifest
from tests._docker import skip_unless_docker
# Same image used by test_git_gate_sidecar — alpine + git + gitleaks.
CLIENT_IMAGE = "zricethezav/gitleaks@sha256:c00b6bd0aeb3071cbcb79009cb16a60dd9e0a7c60e2be9ab65d25e6bc8abbb7f"
# Built once in setUpClass via `docker build -` from the inline
# Dockerfile below. Carries openssh-server, a `git` user, baked-in
# host keys, and a bare repo at /git/foo.git seeded with one commit.
UPSTREAM_IMAGE = "claude-bottle-test-upstream:latest"
UPSTREAM_DOCKERFILE = textwrap.dedent("""
FROM alpine:3.20
RUN apk add --no-cache openssh-server git
RUN adduser -D -s /usr/bin/git-shell git && \\
passwd -u git && \\
mkdir -p /home/git/.ssh && \\
chown git:git /home/git/.ssh && \\
chmod 700 /home/git/.ssh && \\
mkdir -p /git && \\
chown git:git /git
# Bake host keys into the image so the test can pin the
# KnownHostKey value before the container starts. Re-running
# ssh-keygen -A at boot would invalidate that pinning.
RUN ssh-keygen -A
USER git
RUN git config --global init.defaultBranch main && \\
git config --global user.email upstream@example && \\
git config --global user.name upstream && \\
git init --bare /git/foo.git && \\
git clone /git/foo.git /tmp/w && \\
cd /tmp/w && \\
echo "initial upstream content" > README.md && \\
git add README.md && \\
git commit -q -m "initial commit" && \\
git push -q origin main && \\
rm -rf /tmp/w
USER root
RUN echo "PermitRootLogin no" >> /etc/ssh/sshd_config && \\
echo "PasswordAuthentication no" >> /etc/ssh/sshd_config && \\
echo "AuthorizedKeysFile /home/git/.ssh/authorized_keys" >> /etc/ssh/sshd_config
CMD ["/usr/sbin/sshd", "-D", "-e"]
""").strip()
@skip_unless_docker()
class TestGitGateBidirectionalMirror(unittest.TestCase):
@classmethod
def setUpClass(cls):
# Pull the client image first (other suites do the same — keeps
# registry races contained to setUpClass).
if subprocess.run(
["docker", "pull", CLIENT_IMAGE],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
).returncode != 0:
raise unittest.SkipTest(f"could not pull {CLIENT_IMAGE}")
# Build the upstream sshd image from stdin (no build context
# needed — Dockerfile has no COPY/ADD).
build_result = subprocess.run(
["docker", "build", "-t", UPSTREAM_IMAGE, "-"],
input=UPSTREAM_DOCKERFILE,
text=True,
capture_output=True,
check=False,
)
if build_result.returncode != 0:
raise unittest.SkipTest(
f"could not build upstream image: {build_result.stderr}"
)
# Pull the upstream's baked-in ed25519 host pubkey out of the
# image so we can pin it as KnownHostKey on the gate's manifest
# entry. Reading from a transient container ensures we get the
# same key the running sshd will present.
pub_result = subprocess.run(
["docker", "run", "--rm", "--entrypoint", "cat",
UPSTREAM_IMAGE, "/etc/ssh/ssh_host_ed25519_key.pub"],
capture_output=True, text=True, check=True,
)
parts = pub_result.stdout.strip().split()
# Format: "ssh-ed25519 <base64-pubkey> <comment>" — drop comment.
cls.upstream_host_key = f"{parts[0]} {parts[1]}"
# Build the gate image (uses build cache after the first run).
build_git_gate_image()
def setUp(self):
suffix = self.id().rsplit('.', 1)[-1].replace('_', '-')[-12:]
self.slug = f"t{os.getpid()}-{suffix}"
self.gate_name = ""
self.upstream_name = f"claude-bottle-test-upstream-{self.slug}"
self.internal_net = ""
self.egress_net = ""
self.work_dir = Path(tempfile.mkdtemp())
# Per-test SSH auth keypair. The host gets the private key
# path on disk (manifest IdentityFile); the upstream's
# authorized_keys gets the public key, docker-cp'd in just
# before sshd starts.
self.auth_key = self.work_dir / "auth_key"
subprocess.run(
["ssh-keygen", "-t", "ed25519", "-N", "", "-f", str(self.auth_key),
"-C", "git-gate-test"],
check=True, stdout=subprocess.DEVNULL,
)
self.auth_pub = self.work_dir / "auth_key.pub"
# Networks first so the upstream can attach to the egress
# network at create time.
self.internal_net = network_create_internal(self.slug)
self.egress_net = network_create_egress(self.slug)
# Start the upstream sshd container, attached to the egress
# network (which the gate also lives on). Container name doubles
# as its DNS-resolvable hostname.
subprocess.run(
["docker", "create",
"--name", self.upstream_name,
"--network", self.egress_net,
UPSTREAM_IMAGE],
check=True, stdout=subprocess.DEVNULL,
)
# docker cp the per-test pubkey into the upstream as
# /home/git/.ssh/authorized_keys (right user, right path).
subprocess.run(
["docker", "cp", str(self.auth_pub),
f"{self.upstream_name}:/home/git/.ssh/authorized_keys"],
check=True, stdout=subprocess.DEVNULL,
)
# chown / chmod the authorized_keys before sshd refuses to
# use it.
for argv in (
["chown", "git:git", "/home/git/.ssh/authorized_keys"],
["chmod", "600", "/home/git/.ssh/authorized_keys"],
):
subprocess.run(
["docker", "exec", "-u", "0", self.upstream_name, *argv],
check=False, stdout=subprocess.DEVNULL,
)
# The exec-then-start ordering is unusual — exec on a stopped
# container is OK on modern docker but if it errors we just
# do the chown after start instead. Retry post-start to be
# safe.
subprocess.run(
["docker", "start", self.upstream_name],
check=True, stdout=subprocess.DEVNULL,
)
for argv in (
["chown", "git:git", "/home/git/.ssh/authorized_keys"],
["chmod", "600", "/home/git/.ssh/authorized_keys"],
):
subprocess.run(
["docker", "exec", "-u", "0", self.upstream_name, *argv],
check=False, stdout=subprocess.DEVNULL,
)
# Wait for sshd to bind; a short retry against TCP 22 is enough.
ready = False
for _ in range(30):
probe = subprocess.run(
["docker", "exec", self.upstream_name,
"sh", "-c", "nc -z 127.0.0.1 22"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
if probe.returncode == 0:
ready = True
break
subprocess.run(["sleep", "0.2"], check=False)
if not ready:
self.fail("upstream sshd never bound port 22")
# Build the gate plan + start it. Upstream URL points at the
# upstream container's hostname (Docker DNS resolves it on the
# egress network) on port 22, user `git`.
manifest = Manifest.from_json_obj({
"bottles": {
"dev": {
"git": [{
"Name": "foo",
"Upstream": f"ssh://git@{self.upstream_name}/git/foo.git",
"IdentityFile": str(self.auth_key),
"KnownHostKey": self.upstream_host_key,
}],
},
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
bottle = manifest.bottles["dev"]
gate = DockerGitGate()
prep = gate.prepare(bottle, self.slug, self.work_dir)
plan = dataclasses.replace(
prep,
internal_network=self.internal_net,
egress_network=self.egress_net,
)
self.gate_name = gate.start(plan)
def tearDown(self):
if self.gate_name:
DockerGitGate().stop(self.gate_name)
if self.upstream_name:
subprocess.run(
["docker", "rm", "-f", self.upstream_name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
for n in (self.internal_net, self.egress_net):
if n:
network_remove(n)
shutil.rmtree(self.work_dir, ignore_errors=True)
def _upstream_main_sha(self) -> str:
"""Read upstream's current refs/heads/main sha by exec'ing
directly into the upstream container's bare repo."""
out = subprocess.run(
["docker", "exec", "-u", "git", self.upstream_name,
"git", "-C", "/git/foo.git", "rev-parse", "refs/heads/main"],
capture_output=True, text=True, check=True,
)
return out.stdout.strip()
def _push_to_upstream_oob(self, message: str) -> str:
"""Make a new commit directly on the upstream's bare repo
(out-of-band, not through the gate). Returns the new sha."""
script = textwrap.dedent(f"""
set -e
cd /tmp
rm -rf w
git clone /git/foo.git w
cd w
git config user.email upstream@example
git config user.name upstream
echo "$RANDOM-$$" >> README.md
git add README.md
git commit -q -m "{message}"
git push -q origin main
git rev-parse HEAD
""").strip()
out = subprocess.run(
["docker", "exec", "-u", "git", self.upstream_name,
"sh", "-c", script],
capture_output=True, text=True, check=True,
)
return out.stdout.strip().splitlines()[-1]
@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_clone_and_refetch_reflect_upstream(self):
"""Clone via gate returns upstream's commit. After a second
commit lands on the upstream out-of-band, a re-fetch through
the gate picks it up — the access-hook is refreshing before
each upload-pack."""
initial_sha = self._upstream_main_sha()
# Clone via gate.
clone_script = (
f"set -e\n"
f"cd /tmp && git clone -q git://{self.gate_name}/foo.git r\n"
f"git -C r rev-parse refs/remotes/origin/main\n"
f"cat r/README.md\n"
)
clone = subprocess.run(
["docker", "run", "--rm",
"--network", self.internal_net,
"--entrypoint", "sh",
CLIENT_IMAGE,
"-c", clone_script],
capture_output=True, text=True, timeout=60, check=False,
)
self.assertEqual(
0, clone.returncode,
f"clone via gate failed: stdout={clone.stdout!r} "
f"stderr={clone.stderr!r}",
)
cloned_sha = clone.stdout.strip().splitlines()[0]
self.assertEqual(
initial_sha, cloned_sha,
"clone via gate must return the upstream's current sha",
)
self.assertIn("initial upstream content", clone.stdout)
# Out-of-band commit on the upstream.
new_sha = self._push_to_upstream_oob("second commit")
self.assertNotEqual(initial_sha, new_sha)
# ls-remote via gate (re-fetch should pick up the new sha).
ls = subprocess.run(
["docker", "run", "--rm",
"--network", self.internal_net,
"--entrypoint", "sh",
CLIENT_IMAGE,
"-c", f"git ls-remote git://{self.gate_name}/foo.git refs/heads/main"],
capture_output=True, text=True, timeout=60, check=False,
)
self.assertEqual(0, ls.returncode, f"ls-remote failed: {ls.stderr!r}")
gate_sha = ls.stdout.split()[0]
self.assertEqual(
new_sha, gate_sha,
"ls-remote via gate must reflect the upstream's out-of-band update; "
"if this assertion fails, the access-hook is not refreshing on every "
"upload-pack and the gate is serving stale data",
)
@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_push_through_gate_lands_on_upstream(self):
"""A clean (no-gitleaks-hit) push through the gate lands on
the upstream's bare repo — pre-receive phase 2 forwards
the accepted refs."""
# Make a commit through the gate. The script clones via gate
# (so the commit will be a child of upstream's current main).
push_script = textwrap.dedent(f"""
set -e
cd /tmp
git clone -q git://{self.gate_name}/foo.git r
cd r
git config user.email client@example
git config user.name client
echo "client-side commit" > NEW.md
git add NEW.md
git commit -q -m "client commit"
git rev-parse HEAD
git push origin main 2>&1
""").strip()
push = subprocess.run(
["docker", "run", "--rm",
"--network", self.internal_net,
"--entrypoint", "sh",
CLIENT_IMAGE,
"-c", push_script],
capture_output=True, text=True, timeout=120, check=False,
)
self.assertEqual(
0, push.returncode,
f"push via gate failed: stdout={push.stdout!r} "
f"stderr={push.stderr!r}",
)
client_sha = push.stdout.splitlines()[0].strip()
self.assertEqual(
client_sha, self._upstream_main_sha(),
"push via gate must land on upstream's bare repo; "
"if this fails the pre-receive forward phase is broken or the "
"upstream credential is misconfigured",
)
if __name__ == "__main__":
unittest.main()
tests/integration/test_git_gate_sidecar.py
-224
View File
@@ -1,224 +0,0 @@
"""Integration: per-agent git-gate sidecar (PRD 0008).
Two tests against a real Docker daemon:
1. ls-remote against a gate whose upstream is unreachable fails
with the access-hook's fail-closed rejection. Proves the
daemon is bound to its port AND the access-hook is wired:
a working ls-remote against the gate is necessarily a working
ls-remote against the upstream (PRD 0008's transparent-mirror
contract).
2. A push containing a gitleaks-detectable secret is rejected
by the pre-receive hook with a non-zero exit on the agent
side and a gitleaks-rejection line in the response. The PRD's
primary success criterion.
A successful round-trip (clone through gate reflects upstream)
needs a reachable upstream SSH host; deferred to a follow-up.
"""
import dataclasses
import os
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
from claude_bottle.backend.docker.git_gate import (
DockerGitGate,
build_git_gate_image,
)
from claude_bottle.backend.docker.network import (
network_create_egress,
network_create_internal,
network_remove,
)
from claude_bottle.manifest import Manifest
from tests._docker import skip_unless_docker
# The official gitleaks image already has git + alpine; reusing it
# for the client side too saves a separate image pull.
CLIENT_IMAGE = "zricethezav/gitleaks@sha256:c00b6bd0aeb3071cbcb79009cb16a60dd9e0a7c60e2be9ab65d25e6bc8abbb7f"
# Synthetic high-entropy AKIA-shaped string; gitleaks's aws-access-token
# rule fires on this with the default config. AWS's own example
# ("AKIAIOSFODNN7EXAMPLE") is NOT flagged by gitleaks v8.x — entropy
# filter rejects it — so we use a distinct random-looking value.
FAKE_AWS_KEY = "AKIAQRJHK7N5ZPM2VXTL"
@skip_unless_docker()
class TestGitGateSidecar(unittest.TestCase):
@classmethod
def setUpClass(cls):
# Pre-pull the client/gitleaks base so per-test runs aren't
# racing the registry. Skip cleanly on pull failure (a real
# outage is out of scope here).
result = subprocess.run(
["docker", "pull", CLIENT_IMAGE],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
if result.returncode != 0:
raise unittest.SkipTest(f"could not pull {CLIENT_IMAGE}")
# Build the gate image once for the class. Layer cache makes
# repeated runs cheap.
build_git_gate_image()
def setUp(self):
# DNS hostnames on user-defined Docker networks max out at 63
# chars per label (RFC 1035). The full container name is
# `claude-bottle-git-gate-<slug>` = 23 + len(slug), so the slug
# has to stay under ~40 to be resolvable. Keep it short.
suffix = self.id().rsplit('.', 1)[-1].replace('_', '-')[-12:]
self.slug = f"t{os.getpid()}-{suffix}"
self.gate_name = ""
self.internal_net = ""
self.egress_net = ""
self.work_dir = Path(tempfile.mkdtemp())
def tearDown(self):
if self.gate_name:
DockerGitGate().stop(self.gate_name)
for n in (self.internal_net, self.egress_net):
if n:
network_remove(n)
shutil.rmtree(self.work_dir, ignore_errors=True)
def _start_gate(self, name: str = "foo") -> str:
"""Build a one-upstream gate and bring it up. Returns the
container name (== git-gate hostname on the internal net)."""
# Contents of the fake key don't matter for these tests — the
# rejection-path hook never reaches phase 2 where it would be
# used, and ls-remote doesn't push.
fake_key = self.work_dir / "fake-key"
fake_key.write_text("not-a-real-key\n")
manifest = Manifest.from_json_obj({
"bottles": {
"dev": {
"git": [{
"Name": name,
"Upstream": "ssh://git@upstream.invalid/path.git",
"IdentityFile": str(fake_key),
"KnownHostKey": "ssh-ed25519 AAAAEXAMPLE",
}],
},
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
bottle = manifest.bottles["dev"]
gate = DockerGitGate()
prep = gate.prepare(bottle, 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.gate_name = gate.start(plan)
return self.gate_name
@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_ls_remote_fails_closed_when_upstream_unreachable(self):
"""The gate's access-hook runs `git fetch origin --prune` before
every upload-pack. With the fixture's deliberately unreachable
`ssh://git@upstream.invalid/...`, that fetch fails and the
hook exits 1; the daemon reports access-denied. Asserting
non-zero here is what proves the access-hook is wired: under
the v1 (push-only) design ls-remote against a fresh gate
returned exit 0 with no refs."""
gate = self._start_gate("foo")
# Daemon still has to bind first; retry the TCP connect a few
# times. The expected end state is a non-zero exit from the
# daemon's access-denied response — not a connection refused.
probe = subprocess.run(
["docker", "run", "--rm",
"--network", self.internal_net,
"--entrypoint", "sh",
CLIENT_IMAGE,
"-c",
f"for i in $(seq 1 15); do "
f" out=$(git ls-remote git://{gate}/foo.git 2>&1) && exit 99;"
f" case \"$out\" in *'access denied'*|*'not exported'*) "
f" echo \"$out\"; exit 1;; esac;"
f" sleep 1;"
f"done;"
f"echo TIMEOUT; exit 2"],
capture_output=True, text=True, timeout=60, check=False,
)
# exit 1: daemon access-denied as expected. exit 99 would mean
# ls-remote actually succeeded against the unreachable upstream
# (impossible — would indicate stale-data serving, the very
# thing the access-hook is meant to prevent).
self.assertEqual(
1, probe.returncode,
f"expected fail-closed access-denied; got "
f"exit={probe.returncode} stdout={probe.stdout!r} "
f"stderr={probe.stderr!r}",
)
@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_push_with_secret_is_rejected(self):
"""The PRD 0008 success criterion: a push containing a
gitleaks-detectable secret is rejected; the hook's "gitleaks
rejected" line appears in the response, and git push exits
non-zero on the client side."""
gate = self._start_gate("foo")
push_script = (
"set -e\n"
"cd /tmp\n"
# Wait for git daemon to bind. Under the v1.1 design,
# ls-remote never returns 0 against an unreachable
# upstream (access-hook fail-closed), so we wait for *any*
# response (the daemon's access-denied line) as the
# readiness signal.
f"for i in $(seq 1 15); do "
f" out=$(git ls-remote git://{gate}/foo.git 2>&1) || true;"
f" case \"$out\" in *'remote error'*|*'access denied'*) break;; esac;"
f" sleep 1;"
f"done\n"
"git init -q -b main repo\n"
"cd repo\n"
"git config user.email test@example.com\n"
"git config user.name test\n"
f"echo '{FAKE_AWS_KEY}' > leak.txt\n"
"git add leak.txt\n"
"git commit -q -m leak\n"
f"git push git://{gate}/foo.git main 2>&1\n"
)
result = subprocess.run(
["docker", "run", "--rm",
"--network", self.internal_net,
"--entrypoint", "sh",
CLIENT_IMAGE,
"-c", push_script],
capture_output=True, text=True, timeout=120, check=False,
)
combined = result.stdout + result.stderr
self.assertNotEqual(
0, result.returncode,
f"expected push to fail; output={combined!r}",
)
# Hook's stderr is delivered to the client via the `remote:`
# prefix during a git push. Either token is enough to prove
# the pre-receive hook ran and rejected the push.
self.assertTrue(
"gitleaks rejected" in combined or "leaks found" in combined,
f"expected a gitleaks rejection in the response; got: {combined!r}",
)
if __name__ == "__main__":
unittest.main()
tests/integration/test_orphan_cleanup.py
+10 -13
View File
@@ -1,8 +1,13 @@
"""Integration: the cleanup primitives the start-flow trap depends on """Integration: the network-cleanup primitives the start-flow trap
are idempotent. The original orphan-network bug was a trap-ordering depends on are idempotent. The original orphan-network bug was a
issue; the fix moved the install earlier. The trap is only safe if trap-ordering issue; the fix moved the install earlier. The trap
network_remove and PipelockProxy.stop are no-ops against missing is only safe if network_remove is a no-op against missing
resources.""" resources.
The PipelockProxy.stop idempotency case that used to live here was
removed in PRD 0024 chunk 3 when the per-container .stop method
went away — sidecar teardown is now compose's responsibility, and
`compose down` already no-ops on missing containers."""
import os import os
import subprocess import subprocess
@@ -13,10 +18,6 @@ from claude_bottle.backend.docker.network import (
network_create_internal, network_create_internal,
network_remove, network_remove,
) )
from claude_bottle.backend.docker.pipelock import (
DockerPipelockProxy,
pipelock_container_name,
)
from tests._docker import skip_unless_docker from tests._docker import skip_unless_docker
@@ -71,10 +72,6 @@ class TestOrphanCleanup(unittest.TestCase):
self.assertTrue(network_remove(self.internal_name)) self.assertTrue(network_remove(self.internal_name))
self.assertTrue(network_remove(self.egress_name)) self.assertTrue(network_remove(self.egress_name))
def test_pipelock_stop_missing_sidecar(self):
# Should not raise.
DockerPipelockProxy().stop(pipelock_container_name(f"missing-{self.slug}"))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
tests/integration/test_pipelock_apply.py
+5
View File
@@ -50,6 +50,11 @@ from tests.fixtures import fixture_minimal
"skipped under act_runner: pipelock_tls_init uses a host bind mount " "skipped under act_runner: pipelock_tls_init uses a host bind mount "
"that doesn't share fs with the runner container", "that doesn't share fs with the runner container",
) )
@unittest.skip(
"PRD 0024 chunk 3: the .start-based bringup helper this test used was "
"deleted. Chunk 4 rewrites the bringup with a direct `docker run` so "
"the apply_allowlist_change hot-reload retains integration coverage."
)
class TestPipelockApply(unittest.TestCase): class TestPipelockApply(unittest.TestCase):
def setUp(self): def setUp(self):
self.slug = f"cb-test-pla-{os.getpid()}-{int(time.time())}" self.slug = f"cb-test-pla-{os.getpid()}-{int(time.time())}"
tests/integration/test_pipelock_sidecar_smoke.py
-128
View File
@@ -1,128 +0,0 @@
"""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,
pipelock_tls_init,
)
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,
check=False,
)
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)
# PRD 0006: pipelock's tls_interception block in the rendered
# YAML references in-container CA paths; .start docker-cp's
# those files in. The full launch flow generates the CA via
# `pipelock_tls_init`; this smoke test calls it directly.
ca_cert_host, ca_key_host = pipelock_tls_init(self.work_dir)
plan = dataclasses.replace(
prep,
internal_network=self.internal_net,
egress_network=self.egress_net,
ca_cert_host_path=ca_cert_host,
ca_key_host_path=ca_key_host,
)
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,
check=False,
)
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/integration/test_supervise_sidecar.py
-307
View File
@@ -1,307 +0,0 @@
"""Integration: drive `DockerSupervise.start` against the supervise
sidecar and round-trip an MCP tool call through the queue (PRD 0013).
Topology mirrors production minimally: a per-bottle internal docker
network for the agent ↔ supervise leg, no egress network (supervise
doesn't make outbound calls). The "agent" is a curl container on the
internal net; the supervisor lives on the host (this test process)
and uses claude_bottle.cli.dashboard helpers to write Response files.
Verifies:
1. `tools/list` returns the three PRD 0013 tool names over real MCP
wire format.
2. A `tools/call` from the in-container agent blocks until the host
writes a Response to the queue; once written, the agent receives
the approval payload.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import tempfile
import threading
import time
import unittest
from pathlib import Path
from claude_bottle import supervise as _sv
from claude_bottle.backend.docker.network import (
network_create_internal,
network_remove,
)
from claude_bottle.backend.docker.supervise import (
DockerSupervise,
build_supervise_image,
supervise_container_name,
)
from claude_bottle.cli import dashboard
from claude_bottle.supervise import SupervisePlan, list_pending_proposals
from tests._docker import skip_unless_docker
CURL_IMAGE = "curlimages/curl:latest"
@skip_unless_docker()
class TestSuperviseSidecar(unittest.TestCase):
@classmethod
def setUpClass(cls):
r = subprocess.run(
["docker", "pull", CURL_IMAGE],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
if r.returncode != 0:
raise unittest.SkipTest(f"could not pull {CURL_IMAGE}")
build_supervise_image()
def setUp(self):
self.slug = f"cb-test-sv-{os.getpid()}-{int(time.time())}"
self.sidecar_name = ""
self.internal_net = ""
self.work_dir = Path(tempfile.mkdtemp(prefix="supervise-int."))
self.queue_dir = self.work_dir / "queue"
self.queue_dir.mkdir()
def tearDown(self):
if self.sidecar_name:
subprocess.run(
["docker", "rm", "-f", self.sidecar_name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
if self.internal_net:
network_remove(self.internal_net)
shutil.rmtree(self.work_dir, ignore_errors=True)
def _require_bind_mount_sharing(self) -> None:
"""Skip if `docker run -v <host-path>:<container-path>` doesn't
share the filesystem between the test process and the spawned
container. In docker-in-docker CI (Gitea Actions runner with
host socket forwarded), bind-mount paths are resolved against
the outer host's fs, not the runner container's — so the
sidecar writes proposals to a dir the test process can't see.
Cached on the class so the probe runs once per test session."""
cached = getattr(type(self), "_bind_mount_ok", None)
if cached is True:
return
if cached is False:
self.skipTest(
"docker bind mounts don't share fs with this test process "
"(likely docker-in-docker); the supervise queue round-trip "
"requires real host fs sharing"
)
probe_dir = Path(tempfile.mkdtemp(prefix="supervise-bind-probe."))
try:
(probe_dir / "from-host").write_text("x")
r = subprocess.run(
[
"docker", "run", "--rm",
"-v", f"{probe_dir}:/probe",
"--entrypoint", "sh",
CURL_IMAGE,
"-c", "test -f /probe/from-host && touch /probe/from-container",
],
capture_output=True,
check=False,
)
ok = (
r.returncode == 0
and (probe_dir / "from-container").exists()
)
finally:
shutil.rmtree(probe_dir, ignore_errors=True)
type(self)._bind_mount_ok = ok
if not ok:
self.skipTest(
"docker bind mounts don't share fs with this test process "
"(likely docker-in-docker); the supervise queue round-trip "
"requires real host fs sharing"
)
def _bring_up_sidecar(self) -> None:
self.internal_net = network_create_internal(self.slug)
plan = SupervisePlan(
slug=self.slug,
queue_dir=self.queue_dir,
current_config_dir=self.work_dir / "current-config",
internal_network=self.internal_net,
)
# current_config_dir isn't bind-mounted into the sidecar, only
# the queue dir is. Create it for symmetry with production.
plan.current_config_dir.mkdir()
self.sidecar_name = DockerSupervise().start(plan)
# Block until the server is ready to answer (the container
# `docker start` returns immediately; python is still
# binding to the port).
deadline = time.monotonic() + 10.0
while time.monotonic() < deadline:
rc = subprocess.run(
[
"docker", "run", "--rm",
"--network", self.internal_net,
CURL_IMAGE,
"-fsS", "-o", "/dev/null",
"--max-time", "2",
f"http://{_sv.SUPERVISE_HOSTNAME}:{_sv.SUPERVISE_PORT}/health",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode
if rc == 0:
return
time.sleep(0.25)
raise AssertionError("supervise sidecar /health never came up")
def _curl_jsonrpc(self, body: dict[str, object]) -> dict[str, object]:
"""Invoke curl on the internal network to POST a JSON-RPC
request to the supervise sidecar and parse the response."""
payload = json.dumps(body)
result = subprocess.run(
[
"docker", "run", "--rm",
"--network", self.internal_net,
CURL_IMAGE,
"-sS", "--max-time", "30",
"-H", "Content-Type: application/json",
"-X", "POST",
"--data", payload,
f"http://{_sv.SUPERVISE_HOSTNAME}:{_sv.SUPERVISE_PORT}/",
],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
raise AssertionError(
f"curl to supervise failed: {result.stderr}\n"
f"stdout: {result.stdout}"
)
return json.loads(result.stdout)
def test_tools_list_over_mcp(self):
self._bring_up_sidecar()
result = self._curl_jsonrpc(
{"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
)
self.assertEqual(1, result["id"])
names = {t["name"] for t in result["result"]["tools"]}
self.assertEqual(
{
_sv.TOOL_EGRESS_BLOCK,
_sv.TOOL_PIPELOCK_BLOCK,
_sv.TOOL_CAPABILITY_BLOCK,
_sv.TOOL_LIST_EGRESS_ROUTES,
},
names,
)
def test_tools_call_round_trips_through_queue(self):
"""End-to-end: agent in the bottle calls egress-block;
the call blocks on the queue; the host approves via the
dashboard helpers; the agent receives the approval.
This test focuses on the supervise sidecar's queue + response
plumbing, not the egress apply path itself. The apply
function is stubbed so we don't need to bring up a real
egress sidecar (its docker lifecycle has its own
integration coverage)."""
self._require_bind_mount_sharing()
self._bring_up_sidecar()
# Stub the apply step. The dashboard's approve() calls
# add_route to docker-exec into the egress sidecar;
# this test isn't exercising the real sidecar, so patch it
# to a no-op that returns plausible before/after strings
# the audit-log writer can render.
from claude_bottle.cli import dashboard as _dash
original_apply = _dash.add_route
_dash.add_route = (
lambda slug, new: ("(stubbed before)", new)
)
captured: dict[str, object] = {}
def caller() -> None:
captured["response"] = self._curl_jsonrpc({
"jsonrpc": "2.0", "id": 7, "method": "tools/call",
"params": {
"name": _sv.TOOL_EGRESS_BLOCK,
"arguments": {
"host": "api.example.com",
"justification": "integration test",
},
},
})
t = threading.Thread(target=caller)
t.start()
try:
# Wait for the proposal to appear in the queue (the
# sidecar writes it before blocking on wait_for_response).
deadline = time.monotonic() + 10.0
qp = None
while time.monotonic() < deadline:
pending = list_pending_proposals(self.queue_dir)
if pending:
qp = dashboard.QueuedProposal(
proposal=pending[0], queue_dir=self.queue_dir,
)
break
time.sleep(0.1)
self.assertIsNotNone(qp, "proposal never appeared in queue")
assert qp is not None # type-narrowing
self.assertEqual(
_sv.TOOL_EGRESS_BLOCK, qp.proposal.tool,
)
self.assertEqual("integration test", qp.proposal.justification)
# Approve via the dashboard helper. The apply step (now
# stubbed) would docker-exec into the egress sidecar
# and SIGHUP it. The supervise sidecar sees the response
# file and returns to the curl caller.
dashboard.approve(qp, notes="lgtm from integration test")
finally:
_dash.add_route = original_apply
t.join(timeout=20)
response = captured.get("response")
self.assertIsNotNone(response, "curl thread never produced a response")
assert isinstance(response, dict) # type-narrowing
self.assertEqual(7, response["id"])
result = response["result"]
assert isinstance(result, dict)
self.assertFalse(result.get("isError"))
text = result["content"][0]["text"]
self.assertIn("status: approved", text)
self.assertIn("notes: lgtm from integration test", text)
def test_orphan_sidecar_name_collision_recovered(self):
"""An orphan supervise sidecar from a previous run blocks
the next .start with a duplicate-name error. Documents the
observed behavior so a future change that adds auto-cleanup
can flip the assertion."""
self._bring_up_sidecar()
self.assertEqual(supervise_container_name(self.slug), self.sidecar_name)
# Second .start should fail because the container name is
# taken. cleanup is handled by the orphan probe in prepare.py
# (tested separately in test_orphan_cleanup).
with self.assertRaises(SystemExit):
DockerSupervise().start(SupervisePlan(
slug=self.slug,
queue_dir=self.queue_dir,
current_config_dir=self.work_dir / "current-config",
internal_network=self.internal_net,
))
if __name__ == "__main__":
unittest.main()