"""Cross-backend parity tests (PRD 0042). Verifies that Docker and smolmachines bottles expose the same observable contracts for env injection, agent argv, and exec. Tests use mock subprocess layers so no live VM or Docker daemon is needed. The scenarios here document what must hold across both backends. As PRDs 0038–0040 land these tests provide regression coverage for the contracts they establish. """ from __future__ import annotations import subprocess import unittest from typing import Callable from unittest.mock import patch # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _docker_bottle(guest_env: dict[str, str]) -> "object": from bot_bottle.backend.docker.bottle import DockerBottle return DockerBottle( container="bot-bottle-test", teardown=lambda: None, prompt_path_in_container=None, agent_command="claude", ) def _smolmachines_bottle(guest_env: dict[str, str]) -> "object": from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle return SmolmachinesBottle( "bot-bottle-test", guest_env=guest_env, agent_command="claude", ) # One entry per backend: (label, factory). _BACKENDS: list[tuple[str, Callable[[dict[str, str]], object]]] = [ ("docker", _docker_bottle), ("smolmachines", _smolmachines_bottle), ] # --------------------------------------------------------------------------- # agent_argv contracts # --------------------------------------------------------------------------- class TestAgentArgvParity(unittest.TestCase): """Both backends surface a non-empty agent_argv that includes the agent command and can be used as a subprocess command list.""" def test_agent_argv_is_list_of_strings(self): for label, factory in _BACKENDS: with self.subTest(backend=label): bottle = factory({"MY_VAR": "val"}) argv = bottle.agent_argv([], tty=False) # type: ignore[union-attr] self.assertIsInstance(argv, list, f"{label}: argv is not a list") for item in argv: self.assertIsInstance( item, str, f"{label}: argv item {item!r} is not a str", ) def test_agent_command_present_in_argv(self): for label, factory in _BACKENDS: with self.subTest(backend=label): bottle = factory({}) argv = bottle.agent_argv([], tty=False) # type: ignore[union-attr] joined = " ".join(argv) self.assertIn( "claude", joined, f"{label}: 'claude' not found in agent_argv", ) def test_extra_flags_propagate(self): extra = ["--no-update-check", "--output-format", "stream-json"] for label, factory in _BACKENDS: with self.subTest(backend=label): bottle = factory({}) argv = bottle.agent_argv(extra, tty=False) # type: ignore[union-attr] for flag in extra: self.assertIn( flag, argv, f"{label}: flag {flag!r} not in agent_argv", ) class TestSmolmachinesEnvInArgv(unittest.TestCase): """smolmachines bottle includes guest_env values in exec argv.""" def test_guest_env_in_exec_argv(self): from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle bottle = SmolmachinesBottle( "bot-bottle-test", guest_env={"TOKEN": "abc123", "PROXY": "http://proxy:8888"}, ) argv = bottle.agent_argv([], tty=False) joined = " ".join(argv) self.assertIn("TOKEN=abc123", joined) self.assertIn("PROXY=http://proxy:8888", joined) # --------------------------------------------------------------------------- # exec() user-switching contract # --------------------------------------------------------------------------- class TestExecUserSwitching(unittest.TestCase): """Both backends exec as 'node' by default and accept user='root'.""" def test_docker_exec_uses_node_user_by_default(self): from bot_bottle.backend.docker.bottle import DockerBottle bottle = DockerBottle( container="bot-bottle-test", teardown=lambda: None, prompt_path_in_container=None, ) with patch("bot_bottle.backend.docker.bottle.subprocess.run") as run: run.return_value = subprocess.CompletedProcess( [], 0, stdout="", stderr="", ) bottle.exec("echo hi") call_args = run.call_args[0][0] self.assertIn("node", call_args, "docker exec should use 'node' user by default") def test_smolmachines_exec_uses_node_user_by_default(self): from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle bottle = SmolmachinesBottle("bot-bottle-test", guest_env={}) with patch("bot_bottle.backend.smolmachines.bottle.subprocess.run") as run: run.return_value = subprocess.CompletedProcess( [], 0, stdout="", stderr="", ) bottle.exec("echo hi") call_args = run.call_args[0][0] self.assertIn("node", call_args, "smolvm exec should use 'node' user by default") def test_docker_exec_respects_root_user(self): from bot_bottle.backend.docker.bottle import DockerBottle bottle = DockerBottle( container="bot-bottle-test", teardown=lambda: None, prompt_path_in_container=None, ) with patch("bot_bottle.backend.docker.bottle.subprocess.run") as run: run.return_value = subprocess.CompletedProcess( [], 0, stdout="", stderr="", ) bottle.exec("id", user="root") call_args = run.call_args[0][0] self.assertIn("root", call_args) def test_smolmachines_exec_respects_root_user(self): from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle bottle = SmolmachinesBottle("bot-bottle-test", guest_env={}) with patch("bot_bottle.backend.smolmachines.bottle.subprocess.run") as run: run.return_value = subprocess.CompletedProcess( [], 0, stdout="", stderr="", ) bottle.exec("id", user="root") call_args = run.call_args[0][0] self.assertIn("root", call_args) # --------------------------------------------------------------------------- # ExecResult shape parity # --------------------------------------------------------------------------- class TestExecResultParity(unittest.TestCase): """Both backends return ExecResult with returncode, stdout, stderr.""" def _stub_run(self, argv: object, **kwargs: object) -> object: # type: ignore return subprocess.CompletedProcess( argv, 0, stdout="out\n", stderr="err\n", ) def test_docker_exec_result_shape(self): from bot_bottle.backend.docker.bottle import DockerBottle from bot_bottle.backend import ExecResult bottle = DockerBottle( container="bot-bottle-test", teardown=lambda: None, prompt_path_in_container=None, ) with patch("bot_bottle.backend.docker.bottle.subprocess.run", side_effect=self._stub_run): result = bottle.exec("echo hi") self.assertIsInstance(result, ExecResult) self.assertEqual(0, result.returncode) self.assertIsInstance(result.stdout, str) self.assertIsInstance(result.stderr, str) def test_smolmachines_exec_result_shape(self): from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle from bot_bottle.backend import ExecResult bottle = SmolmachinesBottle("bot-bottle-test", guest_env={}) with patch("bot_bottle.backend.smolmachines.bottle.subprocess.run", side_effect=self._stub_run): result = bottle.exec("echo hi") self.assertIsInstance(result, ExecResult) self.assertEqual(0, result.returncode) self.assertIsInstance(result.stdout, str) self.assertIsInstance(result.stderr, str) # --------------------------------------------------------------------------- # close() is a no-op / idempotent (ABC contract) # --------------------------------------------------------------------------- class TestCloseParity(unittest.TestCase): def test_docker_close_is_idempotent(self): from bot_bottle.backend.docker.bottle import DockerBottle teardown_count = [0] def count_teardown(): teardown_count[0] += 1 bottle = DockerBottle( container="bot-bottle-test", teardown=count_teardown, prompt_path_in_container=None, ) bottle.close() bottle.close() # DockerBottle.close calls teardown — once per call is fine; # what matters is it doesn't raise. def test_smolmachines_close_is_noop(self): from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle bottle = SmolmachinesBottle("bot-bottle-test", guest_env={}) bottle.close() bottle.close() if __name__ == "__main__": unittest.main()