Compare commits

..

3 Commits

Author SHA1 Message Date
didericis ea0c070cfe feat(orchestrator): slice 13b — consolidated registration inputs (egress policy)
lint / lint (push) Successful in 2m10s
test / unit (pull_request) Successful in 1m12s
test / integration (pull_request) Successful in 29s
test / coverage (pull_request) Successful in 1m20s
Bridges the per-bottle `prepare` output to the consolidated registry:
`registration_inputs(plan)` turns a prepared egress plan into the
backend-neutral inputs `Orchestrator.launch_bottle` takes.

- egress_policy(plan): the policy blob = the exact routes YAML the
  per-bottle sidecar read from a file; the multi-tenant gateway's
  PolicyResolver now fetches it from the registry per request instead.
  Same egress_render_routes, so a bottle's allow-list is byte-identical
  moving onto the shared gateway.
- RegistrationInputs bundles policy + metadata; metadata carries the human
  slug so the shared registry can map a minted bottle id back to a name
  (console / supervise display).

Host-side glue (imports bot_bottle.egress) used by the launch path — not
the lean orchestrator control-plane process. Not yet wired into launch
(that's 13c/13d), consistent with the slice-6/10 pattern of landing the
primitive before the cut-over.

Key test: the policy round-trips through load_config back to the same
routes + log level — the resolver-served policy reconstructs the per-bottle
egress config exactly.

pyright 0 errors; pylint 9.83/10; unit suite green (1718 tests; the 13
test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-13 20:40:38 -04:00
didericis 24df322c31 feat(orchestrator): slice 13a — orchestrator process lifecycle (idempotent singleton)
lint / lint (push) Successful in 2m9s
test / unit (pull_request) Successful in 1m4s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m14s
First sub-slice of docker launch integration: the foundation the CLI needs
to ensure exactly one orchestrator control plane + shared gateway is up
before registering/launching bottles. Does NOT touch the real launch path
yet — it's the lifecycle primitive the cut-over slices build on.

- OrchestratorProcess.ensure_running(): idempotent singleton — returns the
  control-plane URL if a healthy one already answers /health, else spawns
  `python -m bot_bottle.orchestrator --broker docker --gateway` detached
  (start_new_session, output tee'd to <root>/orchestrator.log) and polls
  /health until healthy or timeout (OrchestratorStartError).
- The control-plane port is the singleton key: a second orchestrator can't
  bind it, so a stray double-start fails fast rather than forking a rival.
- Host-process (not container) per the PRD's dev-harness sequencing — it
  already has the host user's docker access to broker launches; the
  data-plane gateway it manages is the container.

pyright 0 errors; pylint 9.83/10; unit suite green (1714 tests; the 13
test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-13 20:33:14 -04:00
didericis 80d75de08a feat(supervise+orchestrator): slice 12 — multi-tenant git-gate hook + supervise server
lint / lint (push) Successful in 2m2s
test / unit (pull_request) Successful in 1m0s
test / integration (pull_request) Successful in 22s
test / coverage (pull_request) Successful in 1m11s
Finishes the supervise data-plane writers for the shared gateway. Both
still keyed off a single SUPERVISE_BOTTLE_SLUG env; now each proposal is
attributed to the calling bottle, keeping slices 10/11's per-bottle model.

- git_http_backend: in consolidated mode stamp SUPERVISE_BOTTLE_SLUG=
  <bottle_id> into the CGI env. The gitleaks-allow pre-receive hook runs as
  a child of the `git http-backend` we spawn, so it inherits the stamp and
  queues under the right bottle — no change to the (fragile) embedded hook.
  The bottle id is the namespaced root's final component (slice 10), the
  same key egress uses. Single-tenant leaves the container-stamped slug.
- supervise_server: resolve the proposal slug per request by source IP when
  BOT_BOTTLE_ORCHESTRATOR_URL is set (`_attributed_config`), fail-closed —
  an unattributed / unreachable source raises rather than queue under the
  wrong or empty slug. `serve`/`main` build + attach the resolver and make
  the env slug optional in consolidated mode. Single-tenant unchanged.

Tests: a real git push proving the slug reaches the hook's env; the
supervise attribution matrix (single-tenant slug kept, source-IP bottle
bound, unattributed + resolver-error fail closed).

Remaining supervise gap (noted): websocket DLP still self.config-only.

pyright 0 errors; pylint 9.83/10; unit suite green (1705 tests; the 13
test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-13 20:04:26 -04:00
8 changed files with 519 additions and 5 deletions
+8
View File
@@ -188,6 +188,14 @@ class GitHttpHandler(BaseHTTPRequestHandler):
"SERVER_PORT": str(self.server.server_port), # type: ignore
"SERVER_PROTOCOL": self.request_version,
})
# Consolidated mode: attribute the gitleaks-allow supervise proposal
# (written by receive-pack's pre-receive hook, a child of the CGI we
# spawn below) to the calling bottle. The namespaced root is
# `<base>/<bottle_id>`, so its final component is the bottle id — the
# same per-bottle key egress uses. Single-tenant leaves the hook's
# container-stamped SUPERVISE_BOTTLE_SLUG untouched.
if getattr(self.server, "policy_resolver", None) is not None:
env["SUPERVISE_BOTTLE_SLUG"] = sandbox_root.name
for header, variable in (
("accept", "HTTP_ACCEPT"),
("content-encoding", "HTTP_CONTENT_ENCODING"),
+141
View File
@@ -0,0 +1,141 @@
"""Orchestrator process lifecycle (PRD 0070, docker slice).
Before the CLI can register or launch bottles against the consolidated
model, exactly one orchestrator control plane — and the single per-host
gateway it manages — must be running. This starts the orchestrator
dev-harness (`python -m bot_bottle.orchestrator`) as a background host
process and health-checks it.
It is an **idempotent singleton**: `ensure_running` returns immediately if a
healthy control plane already answers on the port, and otherwise spawns one
and waits for it to come up. The control-plane port is the singleton key —
a second orchestrator can't bind it, so a stray double-start fails fast
rather than forking a rival.
Host-process (not container) on purpose: the PRD sequences the orchestrator
as a plain-process dev-harness first (fast iteration, and it already has the
host user's docker access to broker launches), while the data-plane
*gateway* it manages runs as a container. Wrapping the orchestrator itself
in a backend-native unit is a later step.
"""
from __future__ import annotations
import subprocess
import sys
import time
import urllib.error
import urllib.request
from .. import log
from ..paths import bot_bottle_root
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8080
# Poll cadence + default ceiling while waiting for a freshly-spawned control
# plane to answer /health (the first start also builds/boots the gateway).
_HEALTH_POLL_SECONDS = 0.25
DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
_HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
class OrchestratorStartError(RuntimeError):
"""The orchestrator process did not become healthy within the timeout."""
class OrchestratorProcess:
"""Manages the local orchestrator control-plane process for the docker
backend. Backend-neutral callers only need `ensure_running()` + `url`."""
def __init__(
self,
host: str = DEFAULT_HOST,
port: int = DEFAULT_PORT,
*,
broker: str = "docker",
gateway: bool = True,
) -> None:
self.host = host
self.port = port
self._broker = broker
self._gateway = gateway
@property
def url(self) -> str:
"""The control-plane base URL — also what the data plane's
BOT_BOTTLE_ORCHESTRATOR_URL points at."""
return f"http://{self.host}:{self.port}"
def is_healthy(self, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS) -> bool:
"""True iff a control plane answers `GET /health` with 200 — the
singleton liveness check."""
try:
with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp:
return resp.status == 200
except (urllib.error.URLError, TimeoutError, OSError):
return False
def ensure_running(
self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
) -> str:
"""Return the control-plane URL, starting the orchestrator first if it
isn't already healthy. Idempotent — a healthy control plane is left
untouched. Raises `OrchestratorStartError` if a freshly-spawned one
doesn't answer within `startup_timeout`."""
if self.is_healthy():
return self.url
log.info("starting orchestrator", context={"url": self.url, "broker": self._broker})
self._spawn()
deadline = time.monotonic() + startup_timeout
while time.monotonic() < deadline:
if self.is_healthy():
log.info("orchestrator healthy", context={"url": self.url})
return self.url
time.sleep(_HEALTH_POLL_SECONDS)
raise OrchestratorStartError(
f"orchestrator at {self.url} did not become healthy within "
f"{startup_timeout:g}s"
)
def _argv(self) -> list[str]:
"""`python -m bot_bottle.orchestrator ...` — static flags only."""
argv = [
sys.executable, "-m", "bot_bottle.orchestrator",
"--host", self.host, "--port", str(self.port),
"--broker", self._broker,
]
if self._gateway:
argv.append("--gateway")
return argv
def _log_path(self) -> str:
"""Where the detached orchestrator's stdout/stderr goes so a failed
start is diagnosable after the CLI has moved on."""
return str(bot_bottle_root() / "orchestrator.log")
def _spawn(self) -> None:
"""Launch the orchestrator detached so it outlives this CLI process,
with its output tee'd to a log file under the bot-bottle root."""
root = bot_bottle_root()
root.mkdir(parents=True, exist_ok=True)
logfile = open(self._log_path(), "a", encoding="utf-8") # noqa: SIM115 # pylint: disable=consider-using-with
try:
subprocess.Popen( # noqa: S603 # pylint: disable=consider-using-with
self._argv(),
stdout=logfile,
stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL,
start_new_session=True,
)
finally:
# The child inherits its own dup'd fd; this handle is ours to drop.
logfile.close()
__all__ = [
"OrchestratorProcess",
"OrchestratorStartError",
"DEFAULT_HOST",
"DEFAULT_PORT",
"DEFAULT_STARTUP_TIMEOUT_SECONDS",
]
+57
View File
@@ -0,0 +1,57 @@
"""Consolidated registration inputs (PRD 0070, docker slice).
Bridges the existing per-bottle `prepare` output to the consolidated
registry: turns a prepared bottle's egress plan into the backend-neutral
inputs `Orchestrator.launch_bottle` takes — the egress **policy** blob and
launch **metadata**.
The policy blob is the exact routes YAML the per-bottle egress sidecar used
to read from a file; in the consolidated model the multi-tenant gateway's
`PolicyResolver` fetches it from the registry per request (keyed by source
IP) instead. Same render, so consolidated and single-tenant egress apply
byte-identical policy — a bottle's allow-list doesn't change when it moves
onto the shared gateway.
Host-side glue (imports `bot_bottle.egress`), used by the launch path — not
by the lean orchestrator control-plane process itself.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from ..egress import EgressPlan, egress_render_routes
@dataclass(frozen=True)
class RegistrationInputs:
"""What `Orchestrator.launch_bottle` needs to register a bottle, derived
from its prepared plan. `policy` is served verbatim by the gateway's
`/resolve`; `metadata` is opaque forward-compat state — it carries the
human slug so the console / supervise can show a name, not just the
minted bottle id."""
policy: str
metadata: str
def egress_policy(plan: EgressPlan) -> str:
"""The bottle's egress policy blob: the routes YAML the gateway serves
and the addon parses with `load_config`. Identical to the per-bottle
`routes.yaml` render, so the consolidated path applies the same
allow-list."""
return egress_render_routes(plan.routes, log=plan.log)
def registration_inputs(plan: EgressPlan) -> RegistrationInputs:
"""Assemble the orchestrator registration inputs from a prepared egress
plan. `metadata` records the slug so the shared registry can map a minted
bottle id back to its human name."""
return RegistrationInputs(
policy=egress_policy(plan),
metadata=json.dumps({"slug": plan.slug}),
)
__all__ = ["RegistrationInputs", "egress_policy", "registration_inputs"]
+50 -5
View File
@@ -15,6 +15,12 @@ The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at
container creation by the backend's start step). SUPERVISE_DB_PATH
points at the bind-mounted host database.
Consolidated (PRD 0070): when BOT_BOTTLE_ORCHESTRATOR_URL is set, one
shared server fronts every bottle and attributes each proposal to the
calling bottle by source IP (resolved from the orchestrator) instead of a
fixed slug — an unattributed source fails closed. Unset → the legacy
per-bottle single-tenant server, unchanged.
Speaks MCP over HTTP+JSON-RPC. Methods handled:
* `initialize` — handshake; returns server info + caps.
@@ -40,16 +46,18 @@ import time
import typing
import urllib.error
import urllib.request
from dataclasses import dataclass
from dataclasses import dataclass, replace
try:
# Same-directory imports inside the bundle container; these files are
# COPYed flat under /app by Dockerfile.sidecars.
from egress_addon_core import LOG_OFF, load_config
from policy_resolver import PolicyResolveError, PolicyResolver
import supervise as _sv
except ModuleNotFoundError:
# Package imports for host-side tests and tooling.
from .egress_addon_core import LOG_OFF, load_config
from .policy_resolver import PolicyResolveError, PolicyResolver
from . import supervise as _sv
@@ -73,6 +81,12 @@ DEFAULT_RESPONSE_TIMEOUT_SECONDS = 30.0
MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.05
EGRESS_LIST_TIMEOUT_SECONDS = 5.0
# Consolidated (multi-tenant) mode: when set, one shared supervise server
# fronts every bottle and attributes each proposal to the calling bottle by
# source IP (resolved from the orchestrator), instead of a single
# SUPERVISE_BOTTLE_SLUG env. Unset → legacy per-bottle single-tenant.
ORCHESTRATOR_URL_ENV = "BOT_BOTTLE_ORCHESTRATOR_URL"
@dataclass(frozen=True)
class JsonRpcRequest:
@@ -511,9 +525,30 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
if method == "tools/list":
return handle_tools_list(req.params)
if method == "tools/call":
return handle_tools_call(req.params, config)
# Attribute the proposal to the calling bottle. Single-tenant → the
# env slug on `config`; consolidated → the source-IP-resolved
# bottle id, so one shared server queues each bottle's proposal
# under its own slug.
return handle_tools_call(req.params, self._attributed_config(config))
raise _RpcClientError(ERR_METHOD_NOT_FOUND, f"method not found: {method}")
def _attributed_config(self, config: ServerConfig) -> ServerConfig:
"""The ServerConfig with `bottle_slug` bound to *this request's* bottle.
Single-tenant (no resolver): unchanged. Consolidated: the bottle id
attributed from the source IP — **fail-closed**, an unattributed or
unreachable source raises so no proposal is queued under the wrong (or
empty) slug."""
resolver = getattr(self.server, "policy_resolver", None)
if resolver is None:
return config
try:
bottle_id = resolver.resolve_bottle_id(self.client_address[0])
except PolicyResolveError as e:
raise _RpcInternalError(f"orchestrator unreachable, cannot attribute: {e}") from e
if not bottle_id:
raise _RpcInternalError("request source is not attributed to a bottle")
return replace(config, bottle_slug=bottle_id)
def _write_jsonrpc(self, body: bytes) -> None:
self.send_response(200)
self.send_header("Content-Type", "application/json")
@@ -537,6 +572,9 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
allow_reuse_address = True
daemon_threads = True
config: ServerConfig = ServerConfig(bottle_slug="")
# None → single-tenant (proposals use config.bottle_slug); set → consolidated
# (each proposal attributed to the source-IP-resolved bottle).
policy_resolver: "PolicyResolver | None" = None
# --- Entry point -----------------------------------------------------------
@@ -548,15 +586,17 @@ def serve(
port: int = _sv.SUPERVISE_PORT,
bind: str = "0.0.0.0",
response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS,
resolver: "PolicyResolver | None" = None,
) -> typing.NoReturn:
server = MCPServer((bind, port), MCPHandler)
server.config = ServerConfig(
bottle_slug=bottle_slug,
response_timeout_seconds=response_timeout_seconds,
)
server.policy_resolver = resolver
mode = "multi-tenant" if resolver else f"slug={bottle_slug!r}"
sys.stderr.write(
f"supervise listening on {bind}:{port}; "
f"slug={bottle_slug!r}; "
f"supervise listening on {bind}:{port}; {mode}; "
f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n" # type: ignore[arg-type]
)
sys.stderr.flush()
@@ -571,8 +611,12 @@ def serve(
def main(argv: list[str]) -> int:
del argv # config is env-only, no CLI flags
orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
resolver = PolicyResolver(orch_url) if orch_url else None
bottle_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "")
if not bottle_slug:
# Consolidated mode resolves the slug per request, so the env slug is
# optional there; single-tenant still requires it.
if not bottle_slug and resolver is None:
sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n")
return 2
port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT)))
@@ -587,6 +631,7 @@ def main(argv: list[str]) -> int:
port=port,
bind=bind,
response_timeout_seconds=response_timeout_seconds,
resolver=resolver,
)
return 0 # serve() does not return
+67
View File
@@ -13,6 +13,17 @@ from bot_bottle.git_gate import GIT_GATE_TIMEOUT_SECS
from bot_bottle.git_http_backend import GitHttpHandler, MAX_BODY_BYTES
class _FixedResolver:
"""Maps every source IP to one bottle id (consolidated-mode stub)."""
def __init__(self, bottle_id: str) -> None:
self._bottle_id = bottle_id
def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str:
del source_ip, identity_token
return self._bottle_id
class TestGitHttpBackend(unittest.TestCase):
def test_real_git_push_reaches_bare_repo(self):
from http.server import ThreadingHTTPServer
@@ -94,6 +105,62 @@ class TestGitHttpBackend(unittest.TestCase):
).strip()
self.assertEqual(head, cloned)
def test_consolidated_push_stamps_bottle_slug_for_the_hook(self):
# In consolidated mode the backend attributes the push by source IP and
# stamps SUPERVISE_BOTTLE_SLUG=<bottle_id> into the CGI env, so the
# gitleaks-allow pre-receive hook queues its proposal under the right
# bottle. The hook here just records what it received.
from http.server import ThreadingHTTPServer
bottle_id = "bottleab12"
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
bare = root / bottle_id / "repo.git" # namespaced per bottle (slice 10)
bare.parent.mkdir(parents=True)
subprocess.run(["git", "init", "--bare", str(bare)],
check=True, capture_output=True, text=True)
subprocess.run(
["git", "-C", str(bare), "config", "http.receivepack", "true"],
check=True,
)
capture = root / "slug-capture"
hook = bare / "hooks" / "pre-receive"
hook.write_text(
f"#!/bin/sh\nprintf '%s' \"${{SUPERVISE_BOTTLE_SLUG:-UNSET}}\" > "
f"{capture}\ncat >/dev/null\nexit 0\n"
)
hook.chmod(0o755)
old_root = os.environ.get("GIT_PROJECT_ROOT")
os.environ["GIT_PROJECT_ROOT"] = str(root) # base; backend nests per bottle
self.addCleanup(self._restore_env, old_root)
server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
server.policy_resolver = _FixedResolver(bottle_id) # type: ignore[attr-defined]
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
self.addCleanup(server.shutdown)
self.addCleanup(server.server_close)
work = root / "work"
work.mkdir()
subprocess.run(["git", "init"], cwd=work, check=True,
capture_output=True, text=True)
subprocess.run(["git", "config", "user.name", "test"], cwd=work, check=True)
subprocess.run(["git", "config", "user.email", "t@example.invalid"],
cwd=work, check=True)
(work / "README.md").write_text("test\n")
subprocess.run(["git", "add", "README.md"], cwd=work, check=True)
subprocess.run(["git", "commit", "-m", "init"], cwd=work,
check=True, capture_output=True, text=True)
url = f"http://127.0.0.1:{server.server_port}/repo.git"
subprocess.run(
["git", "push", url, "HEAD:refs/heads/main"],
cwd=work, check=True, capture_output=True, text=True, timeout=5,
)
self.assertEqual(bottle_id, capture.read_text())
def test_post_forwards_git_cgi_headers(self):
from http.server import ThreadingHTTPServer
+87
View File
@@ -0,0 +1,87 @@
"""Unit: orchestrator process lifecycle — idempotent singleton (PRD 0070)."""
from __future__ import annotations
import tempfile
import unittest
import urllib.error
from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle.orchestrator.lifecycle import (
OrchestratorProcess,
OrchestratorStartError,
)
from tests.unit import use_bottle_root
_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen"
_POPEN = "bot_bottle.orchestrator.lifecycle.subprocess.Popen"
_SLEEP = "bot_bottle.orchestrator.lifecycle.time.sleep"
_MONOTONIC = "bot_bottle.orchestrator.lifecycle.time.monotonic"
def _health(status: int) -> MagicMock:
"""A urlopen() context-manager whose `.status` is `status`."""
m = MagicMock()
m.__enter__.return_value.status = status
return m
class TestOrchestratorProcess(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.addCleanup(use_bottle_root(Path(self._tmp.name)))
self.p = OrchestratorProcess(port=8099)
def test_url(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.p.url)
def test_is_healthy_true_on_200(self) -> None:
with patch(_URLOPEN, return_value=_health(200)):
self.assertTrue(self.p.is_healthy())
def test_is_healthy_false_on_error(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
self.assertFalse(self.p.is_healthy())
def test_ensure_running_noop_when_already_healthy(self) -> None:
with patch(_URLOPEN, return_value=_health(200)), patch(_POPEN) as popen:
self.assertEqual(self.p.url, self.p.ensure_running())
popen.assert_not_called() # a live control plane is left untouched
def test_ensure_running_spawns_then_waits_for_health(self) -> None:
# First check (before spawn) fails; after spawn the poll succeeds.
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_POPEN) as popen, patch(_SLEEP):
url = self.p.ensure_running()
self.assertEqual(self.p.url, url)
popen.assert_called_once()
def test_ensure_running_raises_on_startup_timeout(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_POPEN), patch(_SLEEP), \
patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
with self.assertRaises(OrchestratorStartError):
self.p.ensure_running(startup_timeout=1.0)
def test_argv_includes_gateway_and_broker(self) -> None:
argv = OrchestratorProcess(port=8099, broker="docker", gateway=True)._argv()
self.assertIn("--gateway", argv)
self.assertIn("bot_bottle.orchestrator", argv)
self.assertEqual("docker", argv[argv.index("--broker") + 1])
def test_argv_omits_gateway_when_disabled(self) -> None:
self.assertNotIn("--gateway", OrchestratorProcess(gateway=False)._argv())
def test_spawn_launches_detached_and_logs(self) -> None:
with patch(_POPEN) as popen:
self.p._spawn()
popen.assert_called_once()
kwargs = popen.call_args.kwargs
self.assertTrue(kwargs["start_new_session"]) # outlives the CLI
self.assertTrue((Path(self._tmp.name) / "orchestrator.log").exists())
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,56 @@
"""Unit: consolidated registration inputs — egress policy round-trip (PRD 0070)."""
from __future__ import annotations
import json
import unittest
from pathlib import Path
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.egress_addon_core import LOG_BLOCKS, load_config
from bot_bottle.orchestrator.registration import (
RegistrationInputs,
egress_policy,
registration_inputs,
)
def _plan(routes: tuple[EgressRoute, ...], *, slug: str = "demo", log: int = 0) -> EgressPlan:
return EgressPlan(
slug=slug,
routes_path=Path("/unused/routes.yaml"),
routes=routes,
token_env_map={},
log=log,
)
class TestEgressPolicy(unittest.TestCase):
def test_policy_round_trips_through_load_config(self) -> None:
# The policy the gateway serves must parse back to the same allow-list
# the per-bottle sidecar applied — moving onto the shared gateway must
# not change a bottle's egress.
routes = (EgressRoute(host="api.example.com"), EgressRoute(host="pypi.org"))
cfg = load_config(egress_policy(_plan(routes)))
self.assertEqual(("api.example.com", "pypi.org"), tuple(r.host for r in cfg.routes))
def test_policy_preserves_log_level(self) -> None:
plan = _plan((EgressRoute(host="x.example.com"),), log=LOG_BLOCKS)
self.assertEqual(LOG_BLOCKS, load_config(egress_policy(plan)).log)
def test_empty_routes_yield_deny_all(self) -> None:
cfg = load_config(egress_policy(_plan(())))
self.assertEqual((), cfg.routes) # no routes → default-deny
class TestRegistrationInputs(unittest.TestCase):
def test_bundles_policy_and_slug_metadata(self) -> None:
plan = _plan((EgressRoute(host="api.example.com"),), slug="my-bot")
inputs = registration_inputs(plan)
self.assertIsInstance(inputs, RegistrationInputs)
self.assertEqual("my-bot", json.loads(inputs.metadata)["slug"])
self.assertEqual(egress_policy(plan), inputs.policy) # same blob egress_policy renders
if __name__ == "__main__":
unittest.main()
+53
View File
@@ -6,6 +6,7 @@ import sys
import tempfile
import threading
import time
import types
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -629,5 +630,57 @@ class TestHttpEndToEnd(unittest.TestCase):
conn.close()
class _FakeResolver:
def __init__(self, bottle_id: str | None = None, raises: bool = False) -> None:
self._bottle_id = bottle_id
self._raises = raises
self.calls: list[str] = []
def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str | None:
del identity_token
self.calls.append(source_ip)
if self._raises:
# Raise the exact class supervise_server catches (it imports
# policy_resolver flat inside the bundle, package-side in tests).
raise supervise_server.PolicyResolveError("orchestrator down")
return self._bottle_id
def _handler(resolver: object) -> MCPHandler:
"""A bare MCPHandler wired with a server (carrying the resolver) and a
client address, enough to exercise `_attributed_config` off-socket."""
h: MCPHandler = MCPHandler.__new__(MCPHandler)
h.server = types.SimpleNamespace(policy_resolver=resolver) # type: ignore[assignment]
h.client_address = ("10.0.0.7", 4321)
return h
class TestAttributedConfig(unittest.TestCase):
"""Consolidated supervise: each proposal is attributed to the calling
bottle by source IP; single-tenant keeps the env slug (PRD 0070)."""
def test_single_tenant_keeps_env_slug(self) -> None:
cfg = _handler(None)._attributed_config(ServerConfig(bottle_slug="dev"))
self.assertEqual("dev", cfg.bottle_slug)
def test_consolidated_binds_source_ip_bottle(self) -> None:
r = _FakeResolver(bottle_id="bottle-x")
cfg = _handler(r)._attributed_config(ServerConfig(bottle_slug="ignored"))
self.assertEqual("bottle-x", cfg.bottle_slug) # resolved slug wins
self.assertEqual(["10.0.0.7"], r.calls)
def test_unattributed_source_fails_closed(self) -> None:
with self.assertRaises(_RpcInternalError):
_handler(_FakeResolver(bottle_id=None))._attributed_config(
ServerConfig(bottle_slug="x")
)
def test_resolver_error_fails_closed(self) -> None:
with self.assertRaises(_RpcInternalError):
_handler(_FakeResolver(raises=True))._attributed_config(
ServerConfig(bottle_slug="x")
)
if __name__ == "__main__":
unittest.main()