test: add cross-backend print parity tests (PRD 0044)
Shared fixtures build DockerBottlePlan and SmolmachinesBottlePlan from identical git_gate_plan and egress_plan inputs and assert that both backends render the same git gate lines (name → host:port) and egress lines (host [auth:scheme] when authenticated, host alone otherwise).
This commit is contained in:
@@ -0,0 +1,246 @@
|
||||
"""Unit: BottlePlan.print parity across Docker and smolmachines (PRD 0044).
|
||||
|
||||
Both backends inherit a single concrete print() from BottlePlan. These
|
||||
tests verify that identical git_gate_plan and egress_plan inputs produce
|
||||
identical preflight output regardless of backend-specific fields.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from bot_bottle.agent_provider import AgentProvisionPlan
|
||||
from bot_bottle.backend import BottleSpec
|
||||
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
|
||||
from bot_bottle.backend.smolmachines.bottle_plan import SmolmachinesBottlePlan
|
||||
from bot_bottle.egress import EgressPlan, EgressRoute
|
||||
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
|
||||
from bot_bottle.manifest import Manifest
|
||||
from bot_bottle.pipelock import PipelockProxyPlan
|
||||
|
||||
|
||||
def _manifest() -> Manifest:
|
||||
return Manifest.from_json_obj({
|
||||
"bottles": {"dev": {}},
|
||||
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
||||
})
|
||||
|
||||
|
||||
def _spec(manifest: Manifest, tmp: str) -> BottleSpec:
|
||||
return BottleSpec(
|
||||
manifest=manifest,
|
||||
agent_name="demo",
|
||||
copy_cwd=False,
|
||||
user_cwd=tmp,
|
||||
identity="test-00001",
|
||||
)
|
||||
|
||||
|
||||
def _git_gate_plan(tmp: str) -> GitGatePlan:
|
||||
stage = Path(tmp)
|
||||
return GitGatePlan(
|
||||
slug="test-00001",
|
||||
entrypoint_script=stage / "entrypoint.sh",
|
||||
hook_script=stage / "hook.sh",
|
||||
access_hook_script=stage / "access-hook.sh",
|
||||
upstreams=(
|
||||
GitGateUpstream(
|
||||
name="myrepo",
|
||||
upstream_url="ssh://git@gitea.example.com:30009/org/myrepo.git",
|
||||
upstream_host="gitea.example.com",
|
||||
upstream_port="30009",
|
||||
identity_file="/dev/null",
|
||||
known_host_key="ssh-ed25519 AAAA...",
|
||||
extra_hosts={},
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _egress_plan(tmp: str) -> EgressPlan:
|
||||
return EgressPlan(
|
||||
slug="test-00001",
|
||||
routes_path=Path(tmp) / "egress.yaml",
|
||||
routes=(
|
||||
EgressRoute(
|
||||
host="api.example.com",
|
||||
path_allowlist=("/v1/",),
|
||||
auth_scheme="bearer",
|
||||
token_env="EGRESS_TOKEN_0",
|
||||
token_ref="TOKEN",
|
||||
),
|
||||
EgressRoute(
|
||||
host="static.example.com",
|
||||
path_allowlist=("/",),
|
||||
),
|
||||
),
|
||||
token_env_map={"EGRESS_TOKEN_0": "TOKEN"},
|
||||
)
|
||||
|
||||
|
||||
def _agent_provision() -> AgentProvisionPlan:
|
||||
return AgentProvisionPlan(
|
||||
template="claude",
|
||||
command="claude",
|
||||
prompt_mode="append_file",
|
||||
image="",
|
||||
dockerfile="",
|
||||
guest_env={"HTTPS_PROXY": "http://127.0.0.1:9999"},
|
||||
)
|
||||
|
||||
|
||||
def _proxy_plan(tmp: str) -> PipelockProxyPlan:
|
||||
return PipelockProxyPlan(
|
||||
yaml_path=Path(tmp) / "pipelock.yaml",
|
||||
slug="test-00001",
|
||||
)
|
||||
|
||||
|
||||
def _docker_plan(spec: BottleSpec, tmp: str) -> DockerBottlePlan:
|
||||
stage = Path(tmp)
|
||||
return DockerBottlePlan(
|
||||
spec=spec,
|
||||
stage_dir=stage,
|
||||
git_gate_plan=_git_gate_plan(tmp),
|
||||
egress_plan=_egress_plan(tmp),
|
||||
supervise_plan=None,
|
||||
agent_provision=_agent_provision(),
|
||||
slug="test-00001",
|
||||
container_name="bot-bottle-test-00001",
|
||||
container_name_pinned=False,
|
||||
image="bot-bottle-claude:latest",
|
||||
derived_image="",
|
||||
runtime_image="bot-bottle-claude:latest",
|
||||
dockerfile_path="",
|
||||
env_file=stage / "env",
|
||||
forwarded_env={},
|
||||
prompt_file=stage / "prompt.txt",
|
||||
proxy_plan=_proxy_plan(tmp),
|
||||
use_runsc=False,
|
||||
)
|
||||
|
||||
|
||||
def _smolmachines_plan(spec: BottleSpec, tmp: str) -> SmolmachinesBottlePlan:
|
||||
stage = Path(tmp)
|
||||
return SmolmachinesBottlePlan(
|
||||
spec=spec,
|
||||
stage_dir=stage,
|
||||
git_gate_plan=_git_gate_plan(tmp),
|
||||
egress_plan=_egress_plan(tmp),
|
||||
supervise_plan=None,
|
||||
agent_provision=_agent_provision(),
|
||||
slug="test-00001",
|
||||
bundle_subnet="10.99.0.0/24",
|
||||
bundle_gateway="10.99.0.1",
|
||||
bundle_ip="10.99.0.2",
|
||||
machine_name="bot-bottle-test-00001",
|
||||
agent_image_ref="bot-bottle-claude:latest",
|
||||
guest_env={"HTTPS_PROXY": "http://127.0.0.1:9999"},
|
||||
prompt_file=stage / "prompt.txt",
|
||||
proxy_plan=_proxy_plan(tmp),
|
||||
)
|
||||
|
||||
|
||||
def _capture_print(plan: DockerBottlePlan | SmolmachinesBottlePlan) -> list[str]:
|
||||
buf = io.StringIO()
|
||||
orig = sys.stderr
|
||||
sys.stderr = buf
|
||||
try:
|
||||
plan.print(remote_control=False)
|
||||
finally:
|
||||
sys.stderr = orig
|
||||
return buf.getvalue().splitlines()
|
||||
|
||||
|
||||
class TestGitGatePrintParity(unittest.TestCase):
|
||||
"""Both backends render git gate entries as 'name → host:port'."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.mkdtemp(prefix="plan-print-parity-")
|
||||
manifest = _manifest()
|
||||
spec = _spec(manifest, self._tmp)
|
||||
self._docker_lines = _capture_print(_docker_plan(spec, self._tmp))
|
||||
self._smol_lines = _capture_print(_smolmachines_plan(spec, self._tmp))
|
||||
|
||||
def _git_gate_lines(self, lines: list[str]) -> list[str]:
|
||||
return [ln for ln in lines if "git gate" in ln]
|
||||
|
||||
def test_docker_renders_name_arrow_host_port(self) -> None:
|
||||
git_lines = self._git_gate_lines(self._docker_lines)
|
||||
self.assertEqual(1, len(git_lines))
|
||||
self.assertIn("myrepo → gitea.example.com:30009", git_lines[0])
|
||||
|
||||
def test_smolmachines_renders_name_arrow_host_port(self) -> None:
|
||||
git_lines = self._git_gate_lines(self._smol_lines)
|
||||
self.assertEqual(1, len(git_lines))
|
||||
self.assertIn("myrepo → gitea.example.com:30009", git_lines[0])
|
||||
|
||||
def test_git_gate_lines_match_across_backends(self) -> None:
|
||||
self.assertEqual(
|
||||
self._git_gate_lines(self._docker_lines),
|
||||
self._git_gate_lines(self._smol_lines),
|
||||
)
|
||||
|
||||
|
||||
class TestEgressPrintParity(unittest.TestCase):
|
||||
"""Both backends render egress with auth annotation where present."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.mkdtemp(prefix="plan-print-parity-")
|
||||
manifest = _manifest()
|
||||
spec = _spec(manifest, self._tmp)
|
||||
self._docker_lines = _capture_print(_docker_plan(spec, self._tmp))
|
||||
self._smol_lines = _capture_print(_smolmachines_plan(spec, self._tmp))
|
||||
|
||||
def _egress_section(self, lines: list[str]) -> list[str]:
|
||||
"""Return lines from the egress label through the last route entry.
|
||||
|
||||
print_multi renders the first route on the label line and
|
||||
aligns additional routes as indented continuation lines
|
||||
(no repeated label). Collect the label line plus every
|
||||
non-blank, non-labelled line that follows before the next
|
||||
top-level section begins."""
|
||||
result: list[str] = []
|
||||
collecting = False
|
||||
indent_prefix = None
|
||||
for ln in lines:
|
||||
stripped = ln.lstrip()
|
||||
if "egress" in stripped and ":" in stripped:
|
||||
collecting = True
|
||||
# Determine the continuation indent from this line's prefix.
|
||||
idx = ln.index("egress")
|
||||
indent_prefix = ln[:idx]
|
||||
result.append(ln)
|
||||
elif collecting:
|
||||
if ln.startswith(indent_prefix) and "egress" not in ln and ":" not in ln.lstrip()[:20]:
|
||||
result.append(ln)
|
||||
else:
|
||||
break
|
||||
return result
|
||||
|
||||
def test_docker_includes_auth_annotation(self) -> None:
|
||||
combined = "\n".join(self._egress_section(self._docker_lines))
|
||||
self.assertIn("api.example.com [auth:bearer]", combined)
|
||||
|
||||
def test_smolmachines_includes_auth_annotation(self) -> None:
|
||||
combined = "\n".join(self._egress_section(self._smol_lines))
|
||||
self.assertIn("api.example.com [auth:bearer]", combined)
|
||||
|
||||
def test_unauthenticated_route_has_no_annotation(self) -> None:
|
||||
full = "\n".join(self._docker_lines)
|
||||
self.assertIn("static.example.com", full)
|
||||
self.assertNotIn("static.example.com [auth:", full)
|
||||
|
||||
def test_egress_lines_match_across_backends(self) -> None:
|
||||
self.assertEqual(
|
||||
self._egress_section(self._docker_lines),
|
||||
self._egress_section(self._smol_lines),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user