diff --git a/tests/unit/test_plan_print_parity.py b/tests/unit/test_plan_print_parity.py new file mode 100644 index 0000000..2e76361 --- /dev/null +++ b/tests/unit/test_plan_print_parity.py @@ -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()