fix(test): resolve the remaining review findings on the control-plane auth test

Addresses the 5 lower-priority findings left as follow-up in the earlier
review, now that each has a concrete answer:

- Add gateway_name: str = GATEWAY_NAME to OrchestratorService.__init__
  (mirrors the existing orchestrator_name param) and thread it through
  _gateway(). Deletes the test's _IsolatedOrchestratorService subclass,
  which existed only to override a private method for this one kwarg —
  any caller needing gateway-name isolation can now use the public
  constructor. Backward compatible: every existing caller constructs
  OrchestratorService with keyword args and a sensible default is kept.

- Give the test its own fixed image tags (bot-bottle-orchestrator:itest,
  bot-bottle-gateway:itest) instead of the production :latest ones.
  _running_image_is_current() keys gateway staleness off the image tag's
  ID, not per-instance identity, so rebuilding the shared :latest tag from
  whatever's on disk during a test run could make a real host's running
  production gateway look stale and get force-recreated. Fixed tags (not
  per-run-suffixed, so they don't accumulate) fully decouple the two.

- setUp -> setUpClass/tearDownClass: all 5 tests are read-only checks
  against the same running control plane, so one shared container
  lifecycle replaces 5 (each of which paid its own container-start +
  image-build + health-poll cycle). Cuts the file's wall-clock roughly
  4x (11.5s -> 2.9-4.3s) and, combined with the network-rm cleanup from
  the previous commit, means one cleanup instead of five.

- Reuse OrchestratorClient (bot_bottle/orchestrator/client.py) instead of
  a hand-rolled urllib helper — the test now exercises the same
  request/response code path the real host CLI uses, rather than a
  private copy that could silently drift from it.

- Add the chown workaround test_multitenant_isolation.py already needed
  for this exact bind-mount: the orchestrator container has no USER
  directive, so it writes the registry DB as root into the throwaway
  host_root; chown it back before tempdir cleanup so that doesn't raise
  PermissionError on native Linux Docker (no UID remap, unlike Docker
  Desktop's macOS VM).

Verified: ran the suite twice in a row (idempotency — fixed image tags
don't accumulate, 5/5 pass both times, 2.99-4.33s each), the real
~/.bot-bottle/control-plane-token is untouched, zero leaked networks or
containers after either run, exactly 2 :itest images (not growing), the
full orchestrator unit suite (93 tests) and the sibling docker
gateway/broker integration tests still pass. pyright clean, pylint
10.00/10 on both changed files.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

bot_bottle/orchestrator/lifecycle.py

@@ -26,7 +26,7 @@ from pathlib import Path
 from .. import log
 from ..docker_cmd import run_docker
 from ..paths import CONTROL_PLANE_TOKEN_ENV, bot_bottle_root, host_control_plane_token
-from .gateway import GATEWAY_IMAGE, GATEWAY_NETWORK, DockerGateway, GatewayError
+from .gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK, DockerGateway, GatewayError
 
 DEFAULT_PORT = 8099
 ORCHESTRATOR_NAME = "bot-bottle-orchestrator"
@@ -83,8 +83,11 @@ class OrchestratorService:
     `orchestrator_name` / `orchestrator_label` let backends run independent
     orchestrators on the same host without name collisions (e.g. the
     Firecracker backend uses `bot-bottle-fc-orchestrator` alongside the Docker
-    backend's `bot-bottle-orchestrator`). Subclass and override `_gateway()`
-    to supply a backend-specific gateway variant."""
+    backend's `bot-bottle-orchestrator`); `gateway_name` gives the paired
+    gateway container the same treatment (e.g. isolated integration tests
+    that can't share the production `GATEWAY_NAME` singleton). Subclass and
+    override `_gateway()` for anything `_gateway_image`/`gateway_name` can't
+    express (a genuinely backend-specific gateway variant)."""
 
     def __init__(
         self,
@@ -93,6 +96,7 @@ class OrchestratorService:
         network: str = GATEWAY_NETWORK,
         image: str = ORCHESTRATOR_IMAGE,
         gateway_image: str = GATEWAY_IMAGE,
+        gateway_name: str = GATEWAY_NAME,
         repo_root: Path = _REPO_ROOT,
         host_root: Path | None = None,
         orchestrator_name: str = ORCHESTRATOR_NAME,
@@ -106,6 +110,7 @@ class OrchestratorService:
         # were one conflated image before the split.
         self.image = image
         self._gateway_image = gateway_image
+        self._gateway_name = gateway_name
         self._repo_root = repo_root
         self._host_root = host_root or bot_bottle_root()
         self._orchestrator_name = orchestrator_name
@@ -175,7 +180,10 @@ class OrchestratorService:
 
     def _gateway(self) -> DockerGateway:
         return DockerGateway(
-            self._gateway_image, network=self.network, orchestrator_url=self.internal_url
+            self._gateway_image,
+            name=self._gateway_name,
+            network=self.network,
+            orchestrator_url=self.internal_url,
         )
 
     def _ensure_orchestrator_image(self) -> None:

tests/integration/test_orchestrator_docker_control_plane_auth.py

@@ -6,46 +6,34 @@ actual orchestrator + gateway as Docker containers and drives the real HTTP
 server over its published loopback port, the same path an agent sharing the
 gateway network — or the trusted host CLI — would use.
 
-Gated on a reachable Docker daemon. Uses unique container/network names and
-a throwaway `BOT_BOTTLE_ROOT` so it never collides with a real per-host
-orchestrator or gateway.
+Gated on a reachable Docker daemon. Uses unique container/network names,
+test-only image tags (never the production `:latest` ones, so a rebuild
+here can't make `_running_image_is_current()` see a real host's running
+gateway as stale and force-recreate it), and a throwaway `BOT_BOTTLE_ROOT`
+so a run never touches or collides with a real per-host orchestrator or
+gateway. The whole stack is brought up once for the class (`setUpClass`),
+not per test method — every test here is a read-only check against the
+same running control plane.
 """
 
 from __future__ import annotations
 
-import json
 import os
 import secrets
 import subprocess
 import tempfile
 import unittest
-import urllib.error
-import urllib.request
 from pathlib import Path
 
-from bot_bottle.orchestrator.control_plane import CONTROL_AUTH_HEADER
-from bot_bottle.orchestrator.gateway import DockerGateway
+from bot_bottle.orchestrator.client import OrchestratorClient
 from bot_bottle.orchestrator.lifecycle import OrchestratorService
 from bot_bottle.paths import host_control_plane_token
 from tests._docker import skip_unless_docker
 
-
-class _IsolatedOrchestratorService(OrchestratorService):
-    """`OrchestratorService`, but with a uniquely-named gateway too (the base
-    class only parameterizes the orchestrator's own name/network/port) so a
-    test run can never touch a real host's orchestrator or gateway."""
-
-    def __init__(self, *, gateway_name: str, **kwargs: object) -> None:
-        super().__init__(**kwargs)  # type: ignore[arg-type]
-        self._test_gateway_name = gateway_name
-
-    def _gateway(self) -> DockerGateway:
-        return DockerGateway(
-            self._gateway_image,
-            name=self._test_gateway_name,
-            network=self.network,
-            orchestrator_url=self.internal_url,
-        )
+# Fixed (not per-run-suffixed) so repeated runs reuse the same layer-cached
+# image instead of leaking a new dangling tag on every invocation.
+_TEST_ORCHESTRATOR_IMAGE = "bot-bottle-orchestrator:itest"
+_TEST_GATEWAY_IMAGE = "bot-bottle-gateway:itest"
 
 
 @skip_unless_docker()
@@ -57,18 +45,19 @@ class _IsolatedOrchestratorService(OrchestratorService):
     "bottle-bringup integration tests",
 )
 class TestDockerControlPlaneAuthIntegration(unittest.TestCase):
-    def setUp(self) -> None:
+    @classmethod
+    def setUpClass(cls) -> None:
         suffix = secrets.token_hex(4)
-        self._tmp = tempfile.TemporaryDirectory()  # pylint: disable=consider-using-with
-        self.addCleanup(self._tmp.cleanup)
+        cls._tmp = tempfile.TemporaryDirectory()  # pylint: disable=consider-using-with
+        cls.addClassCleanup(cls._tmp.cleanup)
 
-        # host_control_plane_token() — both the test's own call below and the
-        # one OrchestratorService makes internally to inject the container's
-        # env var — resolves its path via the *ambient* BOT_BOTTLE_ROOT env
-        # var, not the host_root kwarg passed to the constructor (that kwarg
-        # only controls the DB bind-mount destination). Without pointing the
-        # env var at the same throwaway dir, the "isolated" test reads/writes
-        # the developer's real ~/.bot-bottle/control-plane-token.
+        # host_control_plane_token() — both the token read below and the one
+        # OrchestratorService injects into the container's env — resolves its
+        # path via the *ambient* BOT_BOTTLE_ROOT env var, not the host_root
+        # kwarg passed to the constructor (that kwarg only controls the DB
+        # bind-mount destination). Without pointing the env var at the same
+        # throwaway dir, this "isolated" test would read/write the developer's
+        # real ~/.bot-bottle/control-plane-token.
         previous_root = os.environ.get("BOT_BOTTLE_ROOT")
 
         def _restore_root() -> None:
@@ -77,43 +66,61 @@ class TestDockerControlPlaneAuthIntegration(unittest.TestCase):
             else:
                 os.environ["BOT_BOTTLE_ROOT"] = previous_root
 
-        os.environ["BOT_BOTTLE_ROOT"] = self._tmp.name
-        self.addCleanup(_restore_root)
+        os.environ["BOT_BOTTLE_ROOT"] = cls._tmp.name
+        cls.addClassCleanup(_restore_root)
 
-        # ensure_running()/DockerGateway create this network but never remove
-        # it — stop() only removes containers — so every run would otherwise
-        # leak one bridge network permanently.
+        orchestrator_name = f"bot-bottle-orch-itest-{suffix}"
+        gateway_name = f"bot-bottle-gw-itest-{suffix}"
         network = f"bot-bottle-net-itest-{suffix}"
-        self.addCleanup(
-            lambda: subprocess.run(
-                ["docker", "network", "rm", network],
-                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
-            )
+        host_root = Path(cls._tmp.name)
+        cls.addClassCleanup(
+            cls._teardown_docker, orchestrator_name, gateway_name, network, host_root
         )
 
-        self.svc = _IsolatedOrchestratorService(
-            orchestrator_name=f"bot-bottle-orch-itest-{suffix}",
-            gateway_name=f"bot-bottle-gw-itest-{suffix}",
+        cls.svc = OrchestratorService(
+            orchestrator_name=orchestrator_name,
+            gateway_name=gateway_name,
             network=network,
+            image=_TEST_ORCHESTRATOR_IMAGE,
+            gateway_image=_TEST_GATEWAY_IMAGE,
             port=20000 + secrets.randbelow(10000),
-            host_root=Path(self._tmp.name),
+            host_root=host_root,
+        )
+        cls.svc.ensure_running()
+        cls.token = host_control_plane_token()
+
+    @staticmethod
+    def _teardown_docker(
+        orchestrator_name: str, gateway_name: str, network: str, host_root: Path
+    ) -> None:
+        subprocess.run(
+            ["docker", "rm", "--force", orchestrator_name, gateway_name],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
+        )
+        subprocess.run(
+            ["docker", "network", "rm", network],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
+        )
+        # The orchestrator container (no USER directive) wrote the registry
+        # DB as root into the throwaway host_root; chown it back so the
+        # (non-root) tempdir cleanup can remove it. Same workaround
+        # test_multitenant_isolation.py uses for the identical bind mount.
+        subprocess.run(
+            ["docker", "run", "--rm", "-v", f"{host_root}:/r",
+             "--entrypoint", "chown", _TEST_GATEWAY_IMAGE, "-R",
+             f"{os.getuid()}:{os.getgid()}", "/r"],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
         )
-        self.addCleanup(self.svc.stop)
-        self.svc.ensure_running()
-        self.token = host_control_plane_token()
 
     def _request(
-        self, method: str, path: str, *, token: str | None = None
+        self, method: str, path: str, *, token: str = ""
     ) -> tuple[int, dict[str, object]]:
-        headers = {CONTROL_AUTH_HEADER: token} if token is not None else {}
-        req = urllib.request.Request(
-            self.svc.url + path, method=method, headers=headers
-        )
-        try:
-            with urllib.request.urlopen(req, timeout=5) as resp:
-                return resp.status, json.loads(resp.read())
-        except urllib.error.HTTPError as e:
-            return e.code, json.loads(e.read())
+        # Reuses the real host-side client's request/response handling rather
+        # than hand-rolling urllib here; _request (not one of the named
+        # wrapper methods) is what exposes raw status codes for arbitrary
+        # paths/tokens, which is exactly what these auth-boundary tests need.
+        client = OrchestratorClient(self.svc.url, auth_token=token)
+        return client._request(method, path)  # pylint: disable=protected-access
 
     def test_health_is_open_without_a_token(self) -> None:
         status, payload = self._request("GET", "/health")