diff --git a/tests/unit/test_cli_start_stale.py b/tests/unit/test_cli_start_stale.py new file mode 100644 index 0000000..0e0d7dc --- /dev/null +++ b/tests/unit/test_cli_start_stale.py @@ -0,0 +1,168 @@ +"""Unit: _launch_bottle StaleImageError handling. + +Exercises the while-True / try-except loop around backend.launch: + - headless mode → die on stale + - interactive mode, user declines → stop without launching + - interactive mode, user confirms → retry with skip_stale=True +""" + +from __future__ import annotations + +import contextlib +import io +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import MagicMock, call, patch + +from bot_bottle.image_cache import StaleImageError +from bot_bottle.log import Die + + +def _fake_plan() -> Any: + provision = SimpleNamespace(startup_args=()) + return cast(Any, SimpleNamespace( + agent_provision=provision, + agent_provider_template="claude", + slug="dev-abc", + )) + + +def _stale_cm() -> MagicMock: + """Return a context-manager mock whose __enter__ raises StaleImageError.""" + cm = MagicMock() + cm.__enter__ = MagicMock(side_effect=StaleImageError("image is 5 day(s) old")) + cm.__exit__ = MagicMock(return_value=False) + return cm + + +def _ok_cm(bottle: Any) -> MagicMock: + """Return a context-manager mock that yields `bottle`.""" + cm = MagicMock() + cm.__enter__ = MagicMock(return_value=bottle) + cm.__exit__ = MagicMock(return_value=False) + return cm + + +class TestLaunchBottleStaleHandling(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.mkdtemp(prefix="cli-stale-test.") + + def _spec(self): + from bot_bottle.backend import BottleSpec + from bot_bottle.manifest import ManifestIndex + idx = ManifestIndex.from_json_obj({ + "bottles": {"dev": {}}, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + }) + return BottleSpec( + manifest=idx, + agent_name="demo", + copy_cwd=False, + user_cwd=self._tmp, + identity="dev-abc", + ) + + def _run_launch(self, **patch_kwargs) -> int: + import bot_bottle.cli.start as start_mod + spec = self._spec() + with patch.object(start_mod, "prepare_with_preflight", + return_value=(_fake_plan(), "dev-abc")), \ + patch.object(start_mod, "settle_state"), \ + patch.object(start_mod, "info"): + return start_mod._launch_bottle( + spec, + dry_run=False, + backend_name="docker", + **patch_kwargs, + ) + + def test_headless_stale_calls_die(self): + """In headless mode (assume_yes=True), a StaleImageError must call die().""" + import bot_bottle.cli.start as start_mod + + backend_mock = MagicMock() + backend_mock.launch.return_value = _stale_cm() + + with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \ + patch.object(start_mod, "die", side_effect=Die("stale")): + with self.assertRaises(Die): + self._run_launch(assume_yes=True) + + def test_interactive_user_declines_stops_loop(self): + """Interactive user answering 'n' → loop exits without a second launch.""" + import bot_bottle.cli.start as start_mod + + backend_mock = MagicMock() + backend_mock.launch.return_value = _stale_cm() + + with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \ + patch.object(start_mod, "read_tty_line", return_value="n"), \ + patch("sys.stderr", new_callable=io.StringIO): + rc = self._run_launch(assume_yes=False) + + self.assertEqual(0, rc) + # launch was called once (stale), then the user declined → no second call. + backend_mock.launch.assert_called_once() + + def test_interactive_user_confirms_retries_with_skip_stale(self): + """Interactive user answering 'y' → second launch with skip_stale=True.""" + import bot_bottle.cli.start as start_mod + + bottle_mock = MagicMock() + bottle_mock.name = "dev-abc" + + stale_calls = [0] + + def launch_side_effect(plan, *, skip_stale=False): + if not skip_stale: + return _stale_cm() + return _ok_cm(bottle_mock) + + backend_mock = MagicMock() + backend_mock.launch.side_effect = launch_side_effect + + with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \ + patch.object(start_mod, "read_tty_line", return_value="y"), \ + patch.object(start_mod, "attach_agent", return_value=0), \ + patch.object(start_mod, "capture_claude_session_state"), \ + patch("sys.stderr", new_callable=io.StringIO): + rc = self._run_launch(assume_yes=False) + + self.assertEqual(0, rc) + # First call: skip_stale=False → stale. Second call: skip_stale=True → ok. + self.assertEqual(2, backend_mock.launch.call_count) + calls = backend_mock.launch.call_args_list + self.assertFalse(calls[0].kwargs.get("skip_stale", False)) + self.assertTrue(calls[1].kwargs.get("skip_stale", False)) + + def test_interactive_yes_uppercase_also_accepted(self): + """'Y' or 'YES' should also be accepted as confirmation.""" + import bot_bottle.cli.start as start_mod + + bottle_mock = MagicMock() + bottle_mock.name = "dev-abc" + + def launch_side_effect(plan, *, skip_stale=False): + if not skip_stale: + return _stale_cm() + return _ok_cm(bottle_mock) + + backend_mock = MagicMock() + backend_mock.launch.side_effect = launch_side_effect + + with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \ + patch.object(start_mod, "read_tty_line", return_value="YES"), \ + patch.object(start_mod, "attach_agent", return_value=0), \ + patch.object(start_mod, "capture_claude_session_state"), \ + patch("sys.stderr", new_callable=io.StringIO): + rc = self._run_launch(assume_yes=False) + + self.assertEqual(0, rc) + self.assertEqual(2, backend_mock.launch.call_count) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_config_store.py b/tests/unit/test_config_store.py index df8b717..5987ae0 100644 --- a/tests/unit/test_config_store.py +++ b/tests/unit/test_config_store.py @@ -54,6 +54,33 @@ class TestConfigStore(unittest.TestCase): manager.migrate() self.assertTrue(manager.is_migrated()) + def test_cached_image_warning_days_returns_default_when_db_missing(self) -> None: + # When the db file doesn't exist yet (parent exists, file doesn't), + # the store returns the default without touching the file. + with tempfile.TemporaryDirectory(prefix="config-store.") as tmp: + store = ConfigStore(Path(tmp) / "missing.db") + self.assertEqual( + DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS, + store.cached_image_stale_warning_days(), + ) + + def test_cached_image_warning_days_returns_default_on_null_value(self) -> None: + # If the row exists but the value is NULL (or not castable to int), + # the store falls back to the default. + with tempfile.TemporaryDirectory(prefix="config-store.") as tmp: + db_path = Path(tmp) / "bot-bottle.db" + store = ConfigStore(db_path) + store.migrate() + # Write a NULL value directly. + with sqlite3.connect(db_path) as conn: + conn.execute( + "UPDATE bot_bottle_config SET cached_image_stale_warning_days = NULL WHERE id = 1" + ) + self.assertEqual( + DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS, + store.cached_image_stale_warning_days(), + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_docker_util_image.py b/tests/unit/test_docker_util_image.py index 35b1499..ca38bd2 100644 --- a/tests/unit/test_docker_util_image.py +++ b/tests/unit/test_docker_util_image.py @@ -70,6 +70,35 @@ class TestImageCreatedAt(unittest.TestCase): run.call_args.args[0], ) + def test_dies_on_inspect_failure(self): + with patch.object( + docker_mod.subprocess, "run", return_value=_fail("No such image"), + ), patch.object( + docker_mod, "die", side_effect=SystemExit("die"), + ) as die: + with self.assertRaises(SystemExit): + docker_mod.image_created_at("missing:tag") + die.assert_called_once() + self.assertIn("missing:tag", die.call_args.args[0]) + + def test_dies_on_invalid_timestamp(self): + with patch.object( + docker_mod.subprocess, "run", + return_value=_ok(stdout="not-a-timestamp\n"), + ), patch.object( + docker_mod, "die", side_effect=SystemExit("die"), + ) as die: + with self.assertRaises(SystemExit): + docker_mod.image_created_at("some:tag") + die.assert_called_once() + self.assertIn("some:tag", die.call_args.args[0]) + + def test_parse_docker_timestamp_no_tzinfo_defaults_to_utc(self): + # A bare datetime with no tz offset should be treated as UTC. + dt = docker_mod._parse_docker_timestamp("2024-05-01T10:00:00.000000") + self.assertIsNotNone(dt.tzinfo) + self.assertEqual(timezone.utc, dt.tzinfo) + class TestSave(unittest.TestCase): def test_save_runs_docker_save(self): diff --git a/tests/unit/test_image_cache.py b/tests/unit/test_image_cache.py new file mode 100644 index 0000000..904a02f --- /dev/null +++ b/tests/unit/test_image_cache.py @@ -0,0 +1,85 @@ +"""Unit: image_cache.py — check_stale / check_stale_path.""" + +from __future__ import annotations + +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +from bot_bottle.image_cache import StaleImageError, check_stale, check_stale_path + + +def _utc(**kwargs) -> datetime: + return datetime.now(tz=timezone.utc) - timedelta(**kwargs) + + +class TestCheckStale(unittest.TestCase): + def _run(self, threshold: int, age_days: float) -> None: + created = datetime.now(tz=timezone.utc) - timedelta(days=age_days) + with patch("bot_bottle.image_cache.ConfigStore") as cs: + cs.return_value.cached_image_stale_warning_days.return_value = threshold + check_stale("test image", created) + + def test_negative_threshold_never_raises(self): + # Threshold < 0 means the check is disabled — always passes. + self._run(threshold=-1, age_days=9999) + + def test_zero_threshold_raises_immediately(self): + # threshold=0 means any image is stale the moment it exists. + with self.assertRaises(StaleImageError): + self._run(threshold=0, age_days=0.1) + + def test_within_threshold_does_not_raise(self): + # Age well under threshold — should pass silently. + self._run(threshold=7, age_days=2) + + def test_at_threshold_does_not_raise(self): + # Exactly at the boundary is fine (<=, not <). + self._run(threshold=1, age_days=0.9999) + + def test_exceeds_threshold_raises(self): + created = datetime.now(tz=timezone.utc) - timedelta(days=3) + with patch("bot_bottle.image_cache.ConfigStore") as cs: + cs.return_value.cached_image_stale_warning_days.return_value = 1 + with self.assertRaises(StaleImageError) as ctx: + check_stale("agent image 'bot-bottle:latest'", created) + self.assertIn("agent image", str(ctx.exception)) + self.assertIn("day(s) old", str(ctx.exception)) + + def test_naive_datetime_treated_as_utc(self): + # check_stale calls .astimezone(utc) on the input; naive datetimes + # that would be interpreted as local time should still work. + # We can't control the local tz in a unit test, so just ensure + # no exception is thrown for a very recent naive datetime. + naive_now = datetime.utcnow() + with patch("bot_bottle.image_cache.ConfigStore") as cs: + cs.return_value.cached_image_stale_warning_days.return_value = 1 + # Should not raise — the image is brand new. + check_stale("test image", naive_now) + + +class TestCheckStalePath(unittest.TestCase): + def test_delegates_to_check_stale_with_mtime(self): + with tempfile.NamedTemporaryFile() as f: + path = Path(f.name) + with patch("bot_bottle.image_cache.check_stale") as mock_check: + check_stale_path("some artifact", path) + mock_check.assert_called_once() + label, dt = mock_check.call_args.args + self.assertEqual("some artifact", label) + self.assertIsInstance(dt, datetime) + self.assertIsNotNone(dt.tzinfo) + + def test_raises_stale_for_old_file(self): + with tempfile.NamedTemporaryFile() as f: + path = Path(f.name) + with patch("bot_bottle.image_cache.ConfigStore") as cs: + cs.return_value.cached_image_stale_warning_days.return_value = 0 + with self.assertRaises(StaleImageError): + check_stale_path("cached artifact", path) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_macos_container_launch.py b/tests/unit/test_macos_container_launch.py index d6ee03d..0803d21 100644 --- a/tests/unit/test_macos_container_launch.py +++ b/tests/unit/test_macos_container_launch.py @@ -365,5 +365,103 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase): self.assertEqual("bot-bottle-agent:latest", calls[1][0]) +class TestMacosContainerLaunchCachedImages(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.stage_dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_cached_mode_dies_when_agent_image_missing(self): + plan = _build_plan(self.stage_dir) + from bot_bottle.backend import BottleSpec + import dataclasses + plan = dataclasses.replace( + plan, + spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")), + ) + from bot_bottle.log import Die + + def image_exists(ref: str) -> bool: + return False # Nothing cached. + + with patch.object( + launch, "read_committed_image", return_value="", + ), patch.object( + launch.container_mod, "image_exists", side_effect=image_exists, + ), patch.object( + launch, "die", side_effect=Die("no agent image"), + ): + with self.assertRaises(Die): + launch._build_images(plan) + + def test_cached_mode_dies_when_sidecar_image_missing(self): + plan = _build_plan(self.stage_dir) + from bot_bottle.backend import BottleSpec + import dataclasses + plan = dataclasses.replace( + plan, + spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")), + ) + from bot_bottle.log import Die + + def image_exists(ref: str) -> bool: + # Agent image is present; sidecar is missing. + return ref == plan.image + + with patch.object( + launch, "read_committed_image", return_value="", + ), patch.object( + launch.container_mod, "image_exists", side_effect=image_exists, + ), patch.object( + launch, "die", side_effect=Die("no sidecar image"), + ): + with self.assertRaises(Die): + launch._build_images(plan) + + def test_cached_mode_both_present_returns_unchanged_plan(self): + plan = _build_plan(self.stage_dir) + from bot_bottle.backend import BottleSpec + import dataclasses + plan = dataclasses.replace( + plan, + spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")), + ) + with patch.object( + launch, "read_committed_image", return_value="", + ), patch.object( + launch.container_mod, "image_exists", return_value=True, + ), patch.object( + launch.container_mod, "build_image", + ) as build, patch.object(launch, "info"): + result = launch._build_images(plan) + + build.assert_not_called() + self.assertEqual(plan.image, result.image) + + def test_committed_image_plus_cached_skips_sidecar_build(self): + plan = _build_plan(self.stage_dir) + from bot_bottle.backend import BottleSpec + import dataclasses + plan = dataclasses.replace( + plan, + spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")), + ) + with patch.object( + launch, "read_committed_image", + return_value="bot-bottle-committed-dev-abc:latest", + ), patch.object( + launch.container_mod, "image_exists", return_value=True, + ), patch.object( + launch.container_mod, "build_image", + ) as build, patch.object(launch, "info"): + result = launch._build_images(plan) + + # In cached mode with a committed image, no builds should fire. + build.assert_not_called() + self.assertEqual("bot-bottle-committed-dev-abc:latest", result.image) + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_macos_container_util.py b/tests/unit/test_macos_container_util.py index 2e0f3bd..ab2f1fd 100644 --- a/tests/unit/test_macos_container_util.py +++ b/tests/unit/test_macos_container_util.py @@ -273,5 +273,71 @@ resolver #2 ) +class TestMacosContainerImageCreatedAt(unittest.TestCase): + def _ok(self, stdout: str) -> "util.subprocess.CompletedProcess": # type: ignore + return util.subprocess.CompletedProcess( + args=[], returncode=0, stdout=stdout, stderr="", + ) + + def _fail(self, stderr: str = "no such image") -> "util.subprocess.CompletedProcess": # type: ignore + return util.subprocess.CompletedProcess( + args=[], returncode=1, stdout="", stderr=stderr, + ) + + def test_parses_iso_timestamp_from_dict(self): + payload = '[{"created": "2025-06-01T12:00:00"}]' + with patch.object(util.subprocess, "run", return_value=self._ok(payload)): + dt = util.image_created_at("bot-bottle-agent:latest") + self.assertEqual(2025, dt.year) + self.assertEqual(6, dt.month) + self.assertEqual(1, dt.day) + + def test_accepts_list_or_dict_input(self): + # Container CLI may return a list; we take the first element. + payload = '[{"created": "2024-01-15T08:30:00"}]' + with patch.object(util.subprocess, "run", return_value=self._ok(payload)): + dt = util.image_created_at("some-image:latest") + self.assertEqual(2024, dt.year) + + def test_accepts_uppercase_Created_field(self): + payload = '[{"Created": "2024-03-20T10:00:00"}]' + with patch.object(util.subprocess, "run", return_value=self._ok(payload)): + dt = util.image_created_at("some-image:latest") + self.assertEqual(2024, dt.year) + self.assertEqual(3, dt.month) + + def test_dies_on_nonzero_returncode(self): + with patch.object(util.subprocess, "run", return_value=self._fail("not found")), \ + patch.object(util, "die", side_effect=SystemExit("die")) as die: + with self.assertRaises(SystemExit): + util.image_created_at("missing:tag") + die.assert_called_once() + self.assertIn("missing:tag", die.call_args.args[0]) + + def test_dies_on_malformed_json(self): + with patch.object(util.subprocess, "run", return_value=self._ok("not-json {")), \ + patch.object(util, "die", side_effect=SystemExit("die")) as die: + with self.assertRaises(SystemExit): + util.image_created_at("some:tag") + die.assert_called_once() + + def test_dies_when_no_created_field(self): + payload = '[{"id": "sha256:abc123"}]' + with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \ + patch.object(util, "die", side_effect=SystemExit("die")) as die: + with self.assertRaises(SystemExit): + util.image_created_at("some:tag") + die.assert_called_once() + self.assertIn("creation timestamp", die.call_args.args[0]) + + def test_dies_on_invalid_timestamp_format(self): + payload = '[{"created": "not-a-date"}]' + with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \ + patch.object(util, "die", side_effect=SystemExit("die")) as die: + with self.assertRaises(SystemExit): + util.image_created_at("some:tag") + die.assert_called_once() + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_stale_checks.py b/tests/unit/test_stale_checks.py new file mode 100644 index 0000000..21e5e59 --- /dev/null +++ b/tests/unit/test_stale_checks.py @@ -0,0 +1,334 @@ +"""Unit: stale-image check functions across backends, and the +BottleBackend.launch template method (skip_stale flag). + +No real images or containers are used — all Docker/container/smolmachine +calls are mocked at the module boundary.""" + +from __future__ import annotations + +import contextlib +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast +from unittest.mock import MagicMock, patch + +from bot_bottle.image_cache import StaleImageError + + +# --------------------------------------------------------------------------- +# BottleBackend.launch template — _image_stale_checks + skip_stale +# --------------------------------------------------------------------------- + +class TestBottleBackendLaunchTemplate(unittest.TestCase): + """Verify the concrete launch() method on BottleBackend calls + _image_stale_checks unless skip_stale=True, then delegates to _launch_impl.""" + + def _make_backend(self): + from bot_bottle.backend import BottleBackend + + class _FakeBottle: + name = "fake" + + @contextlib.contextmanager + def _fake_impl(plan): + yield _FakeBottle() + + class _FakeBackend(BottleBackend): + name = "fake" + + @classmethod + def is_available(cls): return True + + def _preflight(self): pass + def _build_guest_env(self, env): return {} + def _resolve_plan(self, spec, **kw): return cast(Any, SimpleNamespace()) + + @contextlib.contextmanager + def _launch_impl(self, plan): + yield _FakeBottle() + + def prepare_cleanup(self): return cast(Any, None) + def cleanup(self, plan): pass + def enumerate_active(self): return [] + + return _FakeBackend() + + def test_stale_checks_called_by_default(self): + backend = self._make_backend() + called = [] + backend._image_stale_checks = lambda plan: called.append(True) # type: ignore[method-assign] + plan = cast(Any, SimpleNamespace()) + with backend.launch(plan): + pass + self.assertEqual([True], called) + + def test_skip_stale_bypasses_stale_checks(self): + backend = self._make_backend() + backend._image_stale_checks = lambda plan: (_ for _ in ()).throw( # type: ignore[method-assign] + AssertionError("should not be called") + ) + plan = cast(Any, SimpleNamespace()) + with backend.launch(plan, skip_stale=True): + pass + + def test_noop_default_image_stale_checks(self): + from bot_bottle.backend import BottleBackend + + class _MinimalBackend(BottleBackend): + name = "minimal" + + @classmethod + def is_available(cls): return True + + def _preflight(self): pass + def _build_guest_env(self, env): return {} + def _resolve_plan(self, spec, **kw): return cast(Any, SimpleNamespace()) + + @contextlib.contextmanager + def _launch_impl(self, plan): + yield cast(Any, SimpleNamespace(name="x")) + + def prepare_cleanup(self): return cast(Any, None) + def cleanup(self, plan): pass + def enumerate_active(self): return [] + + backend = _MinimalBackend() + # The default _image_stale_checks must not raise — it's a no-op. + backend._image_stale_checks(cast(Any, SimpleNamespace())) + + +# --------------------------------------------------------------------------- +# Docker backend stale_checks +# --------------------------------------------------------------------------- + +class TestDockerStaleChecks(unittest.TestCase): + from bot_bottle.backend.docker import launch as _docker_launch_mod + + def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any: + spec = SimpleNamespace(image_policy=policy) + provision = SimpleNamespace(image="bot-bottle-agent:latest") + return cast(Any, SimpleNamespace( + spec=spec, + slug=slug, + image="bot-bottle-agent:latest", + agent_provision=provision, + )) + + def test_fresh_policy_is_noop(self): + from bot_bottle.backend.docker import launch as mod + with patch.object(mod, "read_committed_image") as rci, \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan("fresh")) + rci.assert_not_called() + cs.assert_not_called() + + def test_committed_image_present_checks_only_committed(self): + from bot_bottle.backend.docker import launch as mod + from datetime import datetime, timezone + ts = datetime(2025, 1, 1, tzinfo=timezone.utc) + with patch.object(mod, "read_committed_image", return_value="committed:latest"), \ + patch.object(mod.docker_mod, "image_exists", return_value=True), \ + patch.object(mod.docker_mod, "image_created_at", return_value=ts), \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan()) + cs.assert_called_once() + self.assertIn("committed:latest", cs.call_args.args[0]) + + def test_no_committed_image_checks_agent_and_sidecar(self): + from bot_bottle.backend.docker import launch as mod + from datetime import datetime, timezone + ts = datetime(2025, 1, 1, tzinfo=timezone.utc) + with patch.object(mod, "read_committed_image", return_value=""), \ + patch.object(mod.docker_mod, "image_exists", return_value=True), \ + patch.object(mod.docker_mod, "image_created_at", return_value=ts), \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan()) + # Should have been called for both agent and sidecar images. + self.assertEqual(2, cs.call_count) + + def test_image_not_present_skips_check(self): + from bot_bottle.backend.docker import launch as mod + with patch.object(mod, "read_committed_image", return_value=""), \ + patch.object(mod.docker_mod, "image_exists", return_value=False), \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan()) + cs.assert_not_called() + + def test_backend_image_stale_checks_delegates(self): + from bot_bottle.backend.docker.backend import DockerBottleBackend + from bot_bottle.backend.docker import launch as mod + backend = DockerBottleBackend() + plan = self._plan() + with patch.object(mod, "stale_checks") as sc: + backend._image_stale_checks(plan) + sc.assert_called_once_with(plan) + + def test_no_committed_image_only_sidecar_missing_skips_check(self): + # When no committed image, stale_checks checks each image that EXISTS. + # If an image isn't present, check_stale is not called for it. + from bot_bottle.backend.docker import launch as mod + from datetime import datetime, timezone + ts = datetime(2025, 1, 1, tzinfo=timezone.utc) + plan = self._plan() + + def image_exists(ref: str) -> bool: + return ref == plan.image # Only agent present; sidecar missing. + + with patch.object(mod, "read_committed_image", return_value=""), \ + patch.object(mod.docker_mod, "image_exists", side_effect=image_exists), \ + patch.object(mod.docker_mod, "image_created_at", return_value=ts), \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(plan) + # Only agent was checked (sidecar not present → skipped). + self.assertEqual(1, cs.call_count) + self.assertIn(plan.image, cs.call_args.args[0]) + + +# --------------------------------------------------------------------------- +# smolmachines backend stale_checks +# --------------------------------------------------------------------------- + +class TestSmolmachinesStaleChecks(unittest.TestCase): + def _plan(self, policy: str = "cached") -> Any: + return cast(Any, SimpleNamespace( + spec=SimpleNamespace(image_policy=policy), + slug="dev-abc", + agent_image="bot-bottle-claude:latest", + )) + + def test_fresh_policy_is_noop(self): + from bot_bottle.backend.smolmachines import launch as mod + with patch.object(mod, "read_committed_image") as rci, \ + patch.object(mod, "check_stale_path") as csp: + mod.stale_checks(self._plan("fresh")) + rci.assert_not_called() + csp.assert_not_called() + + def test_committed_artifact_found_checks_it(self): + from bot_bottle.backend.smolmachines import launch as mod + with tempfile.NamedTemporaryFile(suffix=".smolmachine") as f: + committed_path = f.name + with patch.object(mod, "read_committed_image", return_value=committed_path), \ + patch.object(mod.docker_mod, "image_exists", return_value=False), \ + patch.object(mod, "check_stale_path") as csp: + mod.stale_checks(self._plan()) + csp.assert_called_once() + self.assertIn("agent smolmachine", csp.call_args.args[0]) + + def test_no_committed_agent_artifact_in_cache_is_checked(self): + from bot_bottle.backend.smolmachines import launch as mod + digest = "abcdef0123456789" + with tempfile.TemporaryDirectory() as tmp: + artifact = Path(tmp) / f"{digest}.smolmachine.smolmachine" + artifact.write_text("") + with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \ + patch.object(mod, "read_committed_image", return_value=""), \ + patch.object(mod.docker_mod, "image_exists", return_value=True), \ + patch.object(mod.docker_mod, "image_id", + return_value=f"sha256:{digest}ffffffffffffffff"), \ + patch.object(mod, "check_stale_path") as csp: + mod.stale_checks(self._plan()) + # At least the agent artifact check ran. + self.assertGreaterEqual(csp.call_count, 1) + + def test_sidecar_artifact_in_cache_is_checked(self): + from bot_bottle.backend.smolmachines import launch as mod + digest = "1234567890abcdef" + sidecar_digest = "fedcba9876543210" + with tempfile.TemporaryDirectory() as tmp: + sidecar_artifact = Path(tmp) / f"{sidecar_digest}.smolmachine.smolmachine" + sidecar_artifact.write_text("") + + def image_exists(ref: str) -> bool: + return True + + def image_id(ref: str) -> str: + if "sidecar" in ref or ref == mod._bundle.SIDECAR_BUNDLE_IMAGE: + return f"sha256:{sidecar_digest}ffffffff" + return f"sha256:{digest}ffffffff" + + with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \ + patch.object(mod, "read_committed_image", return_value=""), \ + patch.object(mod.docker_mod, "image_exists", side_effect=image_exists), \ + patch.object(mod.docker_mod, "image_id", side_effect=image_id), \ + patch.object(mod, "check_stale_path") as csp: + mod.stale_checks(self._plan()) + # sidecar artifact is present, so check_stale_path should fire for it. + sidecar_calls = [c for c in csp.call_args_list if "sidecar" in c.args[0]] + self.assertTrue(len(sidecar_calls) >= 1) + + def test_backend_image_stale_checks_delegates(self): + from bot_bottle.backend.smolmachines.backend import SmolmachinesBottleBackend + from bot_bottle.backend.smolmachines import launch as mod + backend = SmolmachinesBottleBackend() + plan = self._plan() + with patch.object(mod, "stale_checks") as sc: + backend._image_stale_checks(plan) + sc.assert_called_once_with(plan) + + +# --------------------------------------------------------------------------- +# macOS container backend stale_checks +# --------------------------------------------------------------------------- + +class TestMacosContainerStaleChecks(unittest.TestCase): + def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any: + return cast(Any, SimpleNamespace( + spec=SimpleNamespace(image_policy=policy), + slug=slug, + image="bot-bottle-agent:latest", + )) + + def test_fresh_policy_is_noop(self): + from bot_bottle.backend.macos_container import launch as mod + with patch.object(mod, "read_committed_image") as rci, \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan("fresh")) + rci.assert_not_called() + cs.assert_not_called() + + def test_committed_image_present_checks_only_committed(self): + from bot_bottle.backend.macos_container import launch as mod + from datetime import datetime, timezone + ts = datetime(2025, 1, 1, tzinfo=timezone.utc) + with patch.object(mod, "read_committed_image", return_value="committed:latest"), \ + patch.object(mod.container_mod, "image_exists", return_value=True), \ + patch.object(mod.container_mod, "image_created_at", return_value=ts), \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan()) + cs.assert_called_once() + self.assertIn("committed:latest", cs.call_args.args[0]) + + def test_no_committed_image_checks_agent_and_sidecar(self): + from bot_bottle.backend.macos_container import launch as mod + from datetime import datetime, timezone + ts = datetime(2025, 1, 1, tzinfo=timezone.utc) + with patch.object(mod, "read_committed_image", return_value=""), \ + patch.object(mod.container_mod, "image_exists", return_value=True), \ + patch.object(mod.container_mod, "image_created_at", return_value=ts), \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan()) + self.assertEqual(2, cs.call_count) + + def test_image_not_present_skips_check(self): + from bot_bottle.backend.macos_container import launch as mod + with patch.object(mod, "read_committed_image", return_value=""), \ + patch.object(mod.container_mod, "image_exists", return_value=False), \ + patch.object(mod, "check_stale") as cs: + mod.stale_checks(self._plan()) + cs.assert_not_called() + + def test_backend_image_stale_checks_delegates(self): + from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend + from bot_bottle.backend.macos_container import launch as mod + backend = MacosContainerBottleBackend() + plan = self._plan() + with patch.object(mod, "stale_checks") as sc: + backend._image_stale_checks(plan) + sc.assert_called_once_with(plan) + + +if __name__ == "__main__": + unittest.main()