bot-bottle/tests/integration/test_pipelock_apply.py

"""Integration: drive `apply_allowlist_change` against a real
pipelock sidecar (PRD 0015).

Brings up a real pipelock container via direct `docker run` (the
old `.start()` helper went away in PRD 0024 chunk 3), calls
apply_allowlist_change to swap the api_allowlist, restarts
pipelock, and verifies the running container now serves the new
yaml.

The hot-reload code path under test (apply_allowlist_change,
fetch_current_yaml, fetch_current_allowlist) is unchanged from
PRD 0015 — only the test's bringup helper moved.

Setup uses pipelock_tls_init which bind-mounts a host path into a
one-shot pipelock container — that doesn't work in DinD, so the
test skips under GITEA_ACTIONS.
"""

from __future__ import annotations

import os
import shutil
import subprocess
import tempfile
import time
import unittest
from pathlib import Path

from claude_bottle.backend.docker.bottle_state import pipelock_state_dir
from claude_bottle.backend.docker.network import (
    network_create_egress,
    network_create_internal,
    network_remove,
)
from claude_bottle.backend.docker.pipelock import (
    PIPELOCK_CA_CERT_IN_CONTAINER,
    PIPELOCK_CA_KEY_IN_CONTAINER,
    DockerPipelockProxy,
    pipelock_tls_init,
)
from claude_bottle.backend.docker.pipelock_apply import (
    PipelockApplyError,
    apply_allowlist_change,
    fetch_current_allowlist,
    fetch_current_yaml,
)
from claude_bottle.backend.docker.sidecar_bundle import (
    SIDECAR_BUNDLE_IMAGE,
    sidecar_bundle_container_name,
)
from claude_bottle.yaml_subset import parse_yaml_subset
from tests._docker import skip_unless_docker
from tests.fixtures import fixture_minimal


@skip_unless_docker()
@unittest.skipIf(
    os.environ.get("GITEA_ACTIONS") == "true",
    "skipped under act_runner: pipelock_tls_init uses a host bind mount "
    "that doesn't share fs with the runner container",
)
class TestPipelockApply(unittest.TestCase):
    def setUp(self):
        self.slug = f"cb-test-pla-{os.getpid()}-{int(time.time())}"
        self.sidecar_name = ""
        self.internal_net = ""
        self.egress_net = ""
        self.work_dir = Path(tempfile.mkdtemp(prefix="pipelock-apply."))

    def tearDown(self):
        if self.sidecar_name:
            subprocess.run(
                ["docker", "rm", "-f", self.sidecar_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)
        # Clean up the per-slug state dir under ~/.claude-bottle/state/
        # (apply_allowlist_change writes there; _bring_up calls
        # proxy.prepare with the same path so the bind-mount and the
        # hot-reload write target stay coherent).
        shutil.rmtree(pipelock_state_dir(self.slug), ignore_errors=True)

    def _bring_up(self) -> None:
        """Brings up the bundle image with only the pipelock daemon
        selected. The bundle's Python supervisor is PID 1, which is
        what apply_allowlist_change targets via `docker kill
        --signal USR1` — pipelock alone as PID 1 wouldn't survive
        SIGUSR1 (default disposition = terminate). This shape is
        what runs in production minus the other three daemons.

        The yaml stages into the production-real
        `pipelock_state_dir(slug)` (not a private temp dir) so the
        bind-mount target matches what `apply_allowlist_change`
        writes to — otherwise the hot-reload would write to a
        nowhere-mounted host path and the container would never see
        the updated config."""
        state_dir = pipelock_state_dir(self.slug)
        state_dir.mkdir(parents=True, exist_ok=True)
        prep = DockerPipelockProxy().prepare(
            fixture_minimal().bottles["dev"], self.slug, state_dir,
        )
        self.internal_net = network_create_internal(self.slug)
        self.egress_net = network_create_egress(self.slug)
        ca_cert_host, ca_key_host = pipelock_tls_init(state_dir)

        # Ensure the bundle image is built. compose normally builds
        # this lazily; we go through `docker run` here so we have to
        # do it ourselves. Idempotent — cached layers make repeats
        # fast.
        repo_root = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
        subprocess.run(
            ["docker", "build",
             "-t", SIDECAR_BUNDLE_IMAGE,
             "-f", "Dockerfile.sidecars", "."],
            cwd=repo_root, check=True, capture_output=True,
        )

        self.sidecar_name = sidecar_bundle_container_name(self.slug)
        subprocess.run(
            ["docker", "create",
             "--name", self.sidecar_name,
             "--network", self.internal_net,
             "-e", "CLAUDE_BOTTLE_SIDECAR_DAEMONS=pipelock",
             "-v", f"{prep.yaml_path}:/etc/pipelock.yaml:ro",
             "-v", f"{ca_cert_host}:{PIPELOCK_CA_CERT_IN_CONTAINER}:ro",
             "-v", f"{ca_key_host}:{PIPELOCK_CA_KEY_IN_CONTAINER}:ro",
             SIDECAR_BUNDLE_IMAGE],
            check=True, capture_output=True,
        )
        subprocess.run(
            ["docker", "network", "connect", self.egress_net, self.sidecar_name],
            check=True, capture_output=True,
        )
        subprocess.run(
            ["docker", "start", self.sidecar_name],
            check=True, capture_output=True,
        )
        # Wait until fetch_current_yaml succeeds — it's a docker cp
        # which works on a started-but-not-yet-ready pipelock, so
        # this is more of a "container exists" probe than a
        # readiness one; the hot-reload tests below tolerate
        # pipelock briefly being slow to serve.
        deadline = time.monotonic() + 15.0
        while time.monotonic() < deadline:
            try:
                fetch_current_yaml(self.slug)
                return
            except PipelockApplyError:
                pass
            time.sleep(0.25)
        raise AssertionError("pipelock sidecar never became reachable")

    def _wait_for_yaml(self, contains: str, *, deadline_s: float = 15.0) -> str:
        """Poll docker exec until /etc/pipelock.yaml contains `contains`,
        returning the yaml. Used to bridge the docker-restart window."""
        deadline = time.monotonic() + deadline_s
        while time.monotonic() < deadline:
            try:
                yaml = fetch_current_yaml(self.slug)
                if contains in yaml:
                    return yaml
            except PipelockApplyError:
                pass
            time.sleep(0.25)
        self.fail(f"never saw {contains!r} in /etc/pipelock.yaml")

    def test_apply_swaps_api_allowlist(self):
        self._bring_up()

        initial_yaml = fetch_current_yaml(self.slug)
        # fixture_minimal yields the baked-in DEFAULT_ALLOWLIST in
        # pipelock.py; api.anthropic.com is in there.
        self.assertIn("api.anthropic.com", initial_yaml)

        new_content = "api.anthropic.com\nnew-host.example\n"
        before, after = apply_allowlist_change(self.slug, new_content)
        self.assertIn("api.anthropic.com", before)
        self.assertNotIn("new-host.example", before)
        self.assertIn("new-host.example", after)

        updated = self._wait_for_yaml("new-host.example")
        cfg = parse_yaml_subset(updated)
        self.assertIn("new-host.example", cfg["api_allowlist"])  # type: ignore[operator]
        self.assertIn("api.anthropic.com", cfg["api_allowlist"])  # type: ignore[operator]
        # tls_interception block (set up by the production prepare
        # via pipelock_build_config) is preserved across the swap.
        self.assertIn("tls_interception", cfg)

    def test_apply_with_invalid_host_raises(self):
        self._bring_up()
        with self.assertRaises(PipelockApplyError):
            apply_allowlist_change(self.slug, "host with space.example\n")

    def test_fetch_current_allowlist_renders_one_per_line(self):
        self._bring_up()
        listing = fetch_current_allowlist(self.slug)
        self.assertTrue(listing.endswith("\n"))
        self.assertIn("api.anthropic.com\n", listing)

    def test_apply_against_missing_sidecar_raises(self):
        # Don't bring up — the slug points at nothing.
        with self.assertRaises(PipelockApplyError):
            apply_allowlist_change(self.slug, "x.example\n")


if __name__ == "__main__":
    unittest.main()