fix(tests): resolve pyright errors in stale-check test files
lint / lint (push) Successful in 2m15s
test / unit (pull_request) Successful in 1m3s
test / integration (pull_request) Successful in 21s
test / coverage (pull_request) Successful in 1m15s

Remove unused imports, add missing type annotations, fix Die()
constructor calls (int code, not str), replace fake backend
subclass with patch.object approach to avoid reportMissingParameterType
and override-incompatibility errors in strict pyright mode.
This commit is contained in:
2026-07-09 19:17:43 +00:00
parent 1a53e07039
commit 60231d9070
4 changed files with 104 additions and 163 deletions
+11 -15
View File
@@ -8,14 +8,12 @@ Exercises the while-True / try-except loop around backend.launch:
from __future__ import annotations from __future__ import annotations
import contextlib
import io import io
import tempfile import tempfile
import unittest import unittest
from pathlib import Path
from types import SimpleNamespace from types import SimpleNamespace
from typing import Any, cast from typing import Any, cast
from unittest.mock import MagicMock, call, patch from unittest.mock import MagicMock, patch
from bot_bottle.image_cache import StaleImageError from bot_bottle.image_cache import StaleImageError
from bot_bottle.log import Die from bot_bottle.log import Die
@@ -47,10 +45,10 @@ def _ok_cm(bottle: Any) -> MagicMock:
class TestLaunchBottleStaleHandling(unittest.TestCase): class TestLaunchBottleStaleHandling(unittest.TestCase):
def setUp(self): def setUp(self) -> None:
self._tmp = tempfile.mkdtemp(prefix="cli-stale-test.") self._tmp = tempfile.mkdtemp(prefix="cli-stale-test.")
def _spec(self): def _spec(self) -> Any:
from bot_bottle.backend import BottleSpec from bot_bottle.backend import BottleSpec
from bot_bottle.manifest import ManifestIndex from bot_bottle.manifest import ManifestIndex
idx = ManifestIndex.from_json_obj({ idx = ManifestIndex.from_json_obj({
@@ -65,7 +63,7 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
identity="dev-abc", identity="dev-abc",
) )
def _run_launch(self, **patch_kwargs) -> int: def _run_launch(self, **patch_kwargs: Any) -> int:
import bot_bottle.cli.start as start_mod import bot_bottle.cli.start as start_mod
spec = self._spec() spec = self._spec()
with patch.object(start_mod, "prepare_with_preflight", with patch.object(start_mod, "prepare_with_preflight",
@@ -79,7 +77,7 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
**patch_kwargs, **patch_kwargs,
) )
def test_headless_stale_calls_die(self): def test_headless_stale_calls_die(self) -> None:
"""In headless mode (assume_yes=True), a StaleImageError must call die().""" """In headless mode (assume_yes=True), a StaleImageError must call die()."""
import bot_bottle.cli.start as start_mod import bot_bottle.cli.start as start_mod
@@ -87,11 +85,11 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
backend_mock.launch.return_value = _stale_cm() backend_mock.launch.return_value = _stale_cm()
with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \ with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
patch.object(start_mod, "die", side_effect=Die("stale")): patch.object(start_mod, "die", side_effect=Die()):
with self.assertRaises(Die): with self.assertRaises(Die):
self._run_launch(assume_yes=True) self._run_launch(assume_yes=True)
def test_interactive_user_declines_stops_loop(self): def test_interactive_user_declines_stops_loop(self) -> None:
"""Interactive user answering 'n' → loop exits without a second launch.""" """Interactive user answering 'n' → loop exits without a second launch."""
import bot_bottle.cli.start as start_mod import bot_bottle.cli.start as start_mod
@@ -107,16 +105,14 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
# launch was called once (stale), then the user declined → no second call. # launch was called once (stale), then the user declined → no second call.
backend_mock.launch.assert_called_once() backend_mock.launch.assert_called_once()
def test_interactive_user_confirms_retries_with_skip_stale(self): def test_interactive_user_confirms_retries_with_skip_stale(self) -> None:
"""Interactive user answering 'y' → second launch with skip_stale=True.""" """Interactive user answering 'y' → second launch with skip_stale=True."""
import bot_bottle.cli.start as start_mod import bot_bottle.cli.start as start_mod
bottle_mock = MagicMock() bottle_mock = MagicMock()
bottle_mock.name = "dev-abc" bottle_mock.name = "dev-abc"
stale_calls = [0] def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
def launch_side_effect(plan, *, skip_stale=False):
if not skip_stale: if not skip_stale:
return _stale_cm() return _stale_cm()
return _ok_cm(bottle_mock) return _ok_cm(bottle_mock)
@@ -138,14 +134,14 @@ class TestLaunchBottleStaleHandling(unittest.TestCase):
self.assertFalse(calls[0].kwargs.get("skip_stale", False)) self.assertFalse(calls[0].kwargs.get("skip_stale", False))
self.assertTrue(calls[1].kwargs.get("skip_stale", False)) self.assertTrue(calls[1].kwargs.get("skip_stale", False))
def test_interactive_yes_uppercase_also_accepted(self): def test_interactive_yes_uppercase_also_accepted(self) -> None:
"""'Y' or 'YES' should also be accepted as confirmation.""" """'Y' or 'YES' should also be accepted as confirmation."""
import bot_bottle.cli.start as start_mod import bot_bottle.cli.start as start_mod
bottle_mock = MagicMock() bottle_mock = MagicMock()
bottle_mock.name = "dev-abc" bottle_mock.name = "dev-abc"
def launch_side_effect(plan, *, skip_stale=False): def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
if not skip_stale: if not skip_stale:
return _stale_cm() return _stale_cm()
return _ok_cm(bottle_mock) return _ok_cm(bottle_mock)
+1 -5
View File
@@ -11,10 +11,6 @@ from unittest.mock import patch
from bot_bottle.image_cache import StaleImageError, check_stale, check_stale_path 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): class TestCheckStale(unittest.TestCase):
def _run(self, threshold: int, age_days: float) -> None: def _run(self, threshold: int, age_days: float) -> None:
created = datetime.now(tz=timezone.utc) - timedelta(days=age_days) created = datetime.now(tz=timezone.utc) - timedelta(days=age_days)
@@ -53,7 +49,7 @@ class TestCheckStale(unittest.TestCase):
# that would be interpreted as local time should still work. # that would be interpreted as local time should still work.
# We can't control the local tz in a unit test, so just ensure # We can't control the local tz in a unit test, so just ensure
# no exception is thrown for a very recent naive datetime. # no exception is thrown for a very recent naive datetime.
naive_now = datetime.utcnow() naive_now = datetime(2099, 1, 1) # far future, always "fresh"
with patch("bot_bottle.image_cache.ConfigStore") as cs: with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = 1 cs.return_value.cached_image_stale_warning_days.return_value = 1
# Should not raise — the image is brand new. # Should not raise — the image is brand new.
+18 -37
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import dataclasses
import unittest import unittest
import tempfile import tempfile
from pathlib import Path from pathlib import Path
@@ -15,6 +16,7 @@ from bot_bottle.backend.macos_container import launch
from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
from bot_bottle.egress import EgressPlan from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan from bot_bottle.git_gate import GitGatePlan
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestIndex from bot_bottle.manifest import ManifestIndex
_MANIFEST = ManifestIndex.from_json_obj({ _MANIFEST = ManifestIndex.from_json_obj({
@@ -373,17 +375,16 @@ class TestMacosContainerLaunchCachedImages(unittest.TestCase):
def tearDown(self): def tearDown(self):
self._tmp.cleanup() self._tmp.cleanup()
def test_cached_mode_dies_when_agent_image_missing(self): def _cached_plan(self) -> MacosContainerBottlePlan:
plan = _build_plan(self.stage_dir) return dataclasses.replace(
from bot_bottle.backend import BottleSpec _build_plan(self.stage_dir),
import dataclasses
plan = dataclasses.replace(
plan,
spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")), spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")),
) )
from bot_bottle.log import Die
def image_exists(ref: str) -> bool: def test_cached_mode_dies_when_agent_image_missing(self) -> None:
plan = self._cached_plan()
def image_exists(ref: str) -> bool: # pylint: disable=unused-argument
return False # Nothing cached. return False # Nothing cached.
with patch.object( with patch.object(
@@ -391,43 +392,29 @@ class TestMacosContainerLaunchCachedImages(unittest.TestCase):
), patch.object( ), patch.object(
launch.container_mod, "image_exists", side_effect=image_exists, launch.container_mod, "image_exists", side_effect=image_exists,
), patch.object( ), patch.object(
launch, "die", side_effect=Die("no agent image"), launch, "die", side_effect=Die(),
): ):
with self.assertRaises(Die): with self.assertRaises(Die):
launch._build_images(plan) launch._build_images(plan)
def test_cached_mode_dies_when_sidecar_image_missing(self): def test_cached_mode_dies_when_sidecar_image_missing(self) -> None:
plan = _build_plan(self.stage_dir) plan = self._cached_plan()
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: def image_exists(ref: str) -> bool:
# Agent image is present; sidecar is missing. return ref == plan.image # Agent present; sidecar missing.
return ref == plan.image
with patch.object( with patch.object(
launch, "read_committed_image", return_value="", launch, "read_committed_image", return_value="",
), patch.object( ), patch.object(
launch.container_mod, "image_exists", side_effect=image_exists, launch.container_mod, "image_exists", side_effect=image_exists,
), patch.object( ), patch.object(
launch, "die", side_effect=Die("no sidecar image"), launch, "die", side_effect=Die(),
): ):
with self.assertRaises(Die): with self.assertRaises(Die):
launch._build_images(plan) launch._build_images(plan)
def test_cached_mode_both_present_returns_unchanged_plan(self): def test_cached_mode_both_present_returns_unchanged_plan(self) -> None:
plan = _build_plan(self.stage_dir) plan = self._cached_plan()
from bot_bottle.backend import BottleSpec
import dataclasses
plan = dataclasses.replace(
plan,
spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")),
)
with patch.object( with patch.object(
launch, "read_committed_image", return_value="", launch, "read_committed_image", return_value="",
), patch.object( ), patch.object(
@@ -440,14 +427,8 @@ class TestMacosContainerLaunchCachedImages(unittest.TestCase):
build.assert_not_called() build.assert_not_called()
self.assertEqual(plan.image, result.image) self.assertEqual(plan.image, result.image)
def test_committed_image_plus_cached_skips_sidecar_build(self): def test_committed_image_plus_cached_skips_sidecar_build(self) -> None:
plan = _build_plan(self.stage_dir) plan = self._cached_plan()
from bot_bottle.backend import BottleSpec
import dataclasses
plan = dataclasses.replace(
plan,
spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")),
)
with patch.object( with patch.object(
launch, "read_committed_image", launch, "read_committed_image",
return_value="bot-bottle-committed-dev-abc:latest", return_value="bot-bottle-committed-dev-abc:latest",
+74 -106
View File
@@ -6,7 +6,6 @@ calls are mocked at the module boundary."""
from __future__ import annotations from __future__ import annotations
import contextlib
import tempfile import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
@@ -14,7 +13,13 @@ from types import SimpleNamespace
from typing import Any, cast from typing import Any, cast
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from bot_bottle.image_cache import StaleImageError
def _bottle_cm(bottle: Any) -> MagicMock:
"""Return a mock context manager that yields `bottle`."""
cm = MagicMock()
cm.__enter__ = MagicMock(return_value=bottle)
cm.__exit__ = MagicMock(return_value=False)
return cm
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -25,78 +30,44 @@ class TestBottleBackendLaunchTemplate(unittest.TestCase):
"""Verify the concrete launch() method on BottleBackend calls """Verify the concrete launch() method on BottleBackend calls
_image_stale_checks unless skip_stale=True, then delegates to _launch_impl.""" _image_stale_checks unless skip_stale=True, then delegates to _launch_impl."""
def _make_backend(self): def _make_backend(self) -> Any:
from bot_bottle.backend import BottleBackend from bot_bottle.backend.docker.backend import DockerBottleBackend
return DockerBottleBackend()
class _FakeBottle: def test_stale_checks_called_by_default(self) -> None:
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() backend = self._make_backend()
called = []
backend._image_stale_checks = lambda plan: called.append(True) # type: ignore[method-assign]
plan = cast(Any, SimpleNamespace()) plan = cast(Any, SimpleNamespace())
with backend.launch(plan): bottle = MagicMock()
pass with patch.object(
self.assertEqual([True], called) backend, "_image_stale_checks",
) as stale_mock, patch.object(
backend, "_launch_impl",
return_value=_bottle_cm(bottle),
):
with backend.launch(plan):
pass
stale_mock.assert_called_once_with(plan)
def test_skip_stale_bypasses_stale_checks(self): def test_skip_stale_bypasses_stale_checks(self) -> None:
backend = self._make_backend() 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()) plan = cast(Any, SimpleNamespace())
with backend.launch(plan, skip_stale=True): bottle = MagicMock()
pass with patch.object(
backend, "_image_stale_checks",
) as stale_mock, patch.object(
backend, "_launch_impl",
return_value=_bottle_cm(bottle),
):
with backend.launch(plan, skip_stale=True):
pass
stale_mock.assert_not_called()
def test_noop_default_image_stale_checks(self): def test_noop_default_image_stale_checks(self) -> None:
from bot_bottle.backend.docker.backend import DockerBottleBackend
from bot_bottle.backend import BottleBackend from bot_bottle.backend import BottleBackend
backend = DockerBottleBackend()
class _MinimalBackend(BottleBackend): # Call the base-class _image_stale_checks directly to verify it's a no-op.
name = "minimal" BottleBackend._image_stale_checks(backend, cast(Any, SimpleNamespace())) # type: ignore[arg-type]
@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()))
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -104,8 +75,6 @@ class TestBottleBackendLaunchTemplate(unittest.TestCase):
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
class TestDockerStaleChecks(unittest.TestCase): 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: def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any:
spec = SimpleNamespace(image_policy=policy) spec = SimpleNamespace(image_policy=policy)
provision = SimpleNamespace(image="bot-bottle-agent:latest") provision = SimpleNamespace(image="bot-bottle-agent:latest")
@@ -116,7 +85,7 @@ class TestDockerStaleChecks(unittest.TestCase):
agent_provision=provision, agent_provision=provision,
)) ))
def test_fresh_policy_is_noop(self): def test_fresh_policy_is_noop(self) -> None:
from bot_bottle.backend.docker import launch as mod from bot_bottle.backend.docker import launch as mod
with patch.object(mod, "read_committed_image") as rci, \ with patch.object(mod, "read_committed_image") as rci, \
patch.object(mod, "check_stale") as cs: patch.object(mod, "check_stale") as cs:
@@ -124,7 +93,7 @@ class TestDockerStaleChecks(unittest.TestCase):
rci.assert_not_called() rci.assert_not_called()
cs.assert_not_called() cs.assert_not_called()
def test_committed_image_present_checks_only_committed(self): def test_committed_image_present_checks_only_committed(self) -> None:
from bot_bottle.backend.docker import launch as mod from bot_bottle.backend.docker import launch as mod
from datetime import datetime, timezone from datetime import datetime, timezone
ts = datetime(2025, 1, 1, tzinfo=timezone.utc) ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
@@ -136,7 +105,7 @@ class TestDockerStaleChecks(unittest.TestCase):
cs.assert_called_once() cs.assert_called_once()
self.assertIn("committed:latest", cs.call_args.args[0]) self.assertIn("committed:latest", cs.call_args.args[0])
def test_no_committed_image_checks_agent_and_sidecar(self): def test_no_committed_image_checks_agent_and_sidecar(self) -> None:
from bot_bottle.backend.docker import launch as mod from bot_bottle.backend.docker import launch as mod
from datetime import datetime, timezone from datetime import datetime, timezone
ts = datetime(2025, 1, 1, tzinfo=timezone.utc) ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
@@ -148,7 +117,7 @@ class TestDockerStaleChecks(unittest.TestCase):
# Should have been called for both agent and sidecar images. # Should have been called for both agent and sidecar images.
self.assertEqual(2, cs.call_count) self.assertEqual(2, cs.call_count)
def test_image_not_present_skips_check(self): def test_image_not_present_skips_check(self) -> None:
from bot_bottle.backend.docker import launch as mod from bot_bottle.backend.docker import launch as mod
with patch.object(mod, "read_committed_image", return_value=""), \ with patch.object(mod, "read_committed_image", return_value=""), \
patch.object(mod.docker_mod, "image_exists", return_value=False), \ patch.object(mod.docker_mod, "image_exists", return_value=False), \
@@ -156,18 +125,7 @@ class TestDockerStaleChecks(unittest.TestCase):
mod.stale_checks(self._plan()) mod.stale_checks(self._plan())
cs.assert_not_called() cs.assert_not_called()
def test_backend_image_stale_checks_delegates(self): def test_only_sidecar_missing_checks_only_agent(self) -> None:
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 bot_bottle.backend.docker import launch as mod
from datetime import datetime, timezone from datetime import datetime, timezone
ts = datetime(2025, 1, 1, tzinfo=timezone.utc) ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
@@ -181,10 +139,18 @@ class TestDockerStaleChecks(unittest.TestCase):
patch.object(mod.docker_mod, "image_created_at", return_value=ts), \ patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
patch.object(mod, "check_stale") as cs: patch.object(mod, "check_stale") as cs:
mod.stale_checks(plan) mod.stale_checks(plan)
# Only agent was checked (sidecar not present → skipped).
self.assertEqual(1, cs.call_count) self.assertEqual(1, cs.call_count)
self.assertIn(plan.image, cs.call_args.args[0]) self.assertIn(plan.image, cs.call_args.args[0])
def test_backend_image_stale_checks_delegates(self) -> None:
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) # type: ignore[arg-type]
sc.assert_called_once_with(plan)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# smolmachines backend stale_checks # smolmachines backend stale_checks
@@ -198,7 +164,7 @@ class TestSmolmachinesStaleChecks(unittest.TestCase):
agent_image="bot-bottle-claude:latest", agent_image="bot-bottle-claude:latest",
)) ))
def test_fresh_policy_is_noop(self): def test_fresh_policy_is_noop(self) -> None:
from bot_bottle.backend.smolmachines import launch as mod from bot_bottle.backend.smolmachines import launch as mod
with patch.object(mod, "read_committed_image") as rci, \ with patch.object(mod, "read_committed_image") as rci, \
patch.object(mod, "check_stale_path") as csp: patch.object(mod, "check_stale_path") as csp:
@@ -206,7 +172,7 @@ class TestSmolmachinesStaleChecks(unittest.TestCase):
rci.assert_not_called() rci.assert_not_called()
csp.assert_not_called() csp.assert_not_called()
def test_committed_artifact_found_checks_it(self): def test_committed_artifact_found_checks_it(self) -> None:
from bot_bottle.backend.smolmachines import launch as mod from bot_bottle.backend.smolmachines import launch as mod
with tempfile.NamedTemporaryFile(suffix=".smolmachine") as f: with tempfile.NamedTemporaryFile(suffix=".smolmachine") as f:
committed_path = f.name committed_path = f.name
@@ -217,7 +183,7 @@ class TestSmolmachinesStaleChecks(unittest.TestCase):
csp.assert_called_once() csp.assert_called_once()
self.assertIn("agent smolmachine", csp.call_args.args[0]) self.assertIn("agent smolmachine", csp.call_args.args[0])
def test_no_committed_agent_artifact_in_cache_is_checked(self): def test_no_committed_agent_artifact_in_cache_is_checked(self) -> None:
from bot_bottle.backend.smolmachines import launch as mod from bot_bottle.backend.smolmachines import launch as mod
digest = "abcdef0123456789" digest = "abcdef0123456789"
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
@@ -226,28 +192,31 @@ class TestSmolmachinesStaleChecks(unittest.TestCase):
with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \ with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
patch.object(mod, "read_committed_image", return_value=""), \ patch.object(mod, "read_committed_image", return_value=""), \
patch.object(mod.docker_mod, "image_exists", return_value=True), \ patch.object(mod.docker_mod, "image_exists", return_value=True), \
patch.object(mod.docker_mod, "image_id", patch.object(
return_value=f"sha256:{digest}ffffffffffffffff"), \ mod.docker_mod, "image_id",
return_value=f"sha256:{digest}ffffffffffffffff",
), \
patch.object(mod, "check_stale_path") as csp: patch.object(mod, "check_stale_path") as csp:
mod.stale_checks(self._plan()) mod.stale_checks(self._plan())
# At least the agent artifact check ran. # At least the agent artifact check ran.
self.assertGreaterEqual(csp.call_count, 1) self.assertGreaterEqual(csp.call_count, 1)
def test_sidecar_artifact_in_cache_is_checked(self): def test_sidecar_artifact_in_cache_is_checked(self) -> None:
from bot_bottle.backend.smolmachines import launch as mod from bot_bottle.backend.smolmachines import launch as mod
digest = "1234567890abcdef"
sidecar_digest = "fedcba9876543210" sidecar_digest = "fedcba9876543210"
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
sidecar_artifact = Path(tmp) / f"{sidecar_digest}.smolmachine.smolmachine" sidecar_artifact = (
Path(tmp) / f"{sidecar_digest}.smolmachine.smolmachine"
)
sidecar_artifact.write_text("") sidecar_artifact.write_text("")
def image_exists(ref: str) -> bool: def image_exists(ref: str) -> bool: # pylint: disable=unused-argument
return True return True
def image_id(ref: str) -> str: def image_id(ref: str) -> str:
if "sidecar" in ref or ref == mod._bundle.SIDECAR_BUNDLE_IMAGE: if ref == mod._bundle.SIDECAR_BUNDLE_IMAGE: # type: ignore[attr-defined]
return f"sha256:{sidecar_digest}ffffffff" return f"sha256:{sidecar_digest}ffffffff"
return f"sha256:{digest}ffffffff" return "sha256:0000000000000000ffffffff"
with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \ with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
patch.object(mod, "read_committed_image", return_value=""), \ patch.object(mod, "read_committed_image", return_value=""), \
@@ -255,17 +224,16 @@ class TestSmolmachinesStaleChecks(unittest.TestCase):
patch.object(mod.docker_mod, "image_id", side_effect=image_id), \ patch.object(mod.docker_mod, "image_id", side_effect=image_id), \
patch.object(mod, "check_stale_path") as csp: patch.object(mod, "check_stale_path") as csp:
mod.stale_checks(self._plan()) 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]] sidecar_calls = [c for c in csp.call_args_list if "sidecar" in c.args[0]]
self.assertTrue(len(sidecar_calls) >= 1) self.assertGreaterEqual(len(sidecar_calls), 1)
def test_backend_image_stale_checks_delegates(self): def test_backend_image_stale_checks_delegates(self) -> None:
from bot_bottle.backend.smolmachines.backend import SmolmachinesBottleBackend from bot_bottle.backend.smolmachines.backend import SmolmachinesBottleBackend
from bot_bottle.backend.smolmachines import launch as mod from bot_bottle.backend.smolmachines import launch as mod
backend = SmolmachinesBottleBackend() backend = SmolmachinesBottleBackend()
plan = self._plan() plan = self._plan()
with patch.object(mod, "stale_checks") as sc: with patch.object(mod, "stale_checks") as sc:
backend._image_stale_checks(plan) backend._image_stale_checks(plan) # type: ignore[arg-type]
sc.assert_called_once_with(plan) sc.assert_called_once_with(plan)
@@ -281,7 +249,7 @@ class TestMacosContainerStaleChecks(unittest.TestCase):
image="bot-bottle-agent:latest", image="bot-bottle-agent:latest",
)) ))
def test_fresh_policy_is_noop(self): def test_fresh_policy_is_noop(self) -> None:
from bot_bottle.backend.macos_container import launch as mod from bot_bottle.backend.macos_container import launch as mod
with patch.object(mod, "read_committed_image") as rci, \ with patch.object(mod, "read_committed_image") as rci, \
patch.object(mod, "check_stale") as cs: patch.object(mod, "check_stale") as cs:
@@ -289,7 +257,7 @@ class TestMacosContainerStaleChecks(unittest.TestCase):
rci.assert_not_called() rci.assert_not_called()
cs.assert_not_called() cs.assert_not_called()
def test_committed_image_present_checks_only_committed(self): def test_committed_image_present_checks_only_committed(self) -> None:
from bot_bottle.backend.macos_container import launch as mod from bot_bottle.backend.macos_container import launch as mod
from datetime import datetime, timezone from datetime import datetime, timezone
ts = datetime(2025, 1, 1, tzinfo=timezone.utc) ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
@@ -301,7 +269,7 @@ class TestMacosContainerStaleChecks(unittest.TestCase):
cs.assert_called_once() cs.assert_called_once()
self.assertIn("committed:latest", cs.call_args.args[0]) self.assertIn("committed:latest", cs.call_args.args[0])
def test_no_committed_image_checks_agent_and_sidecar(self): def test_no_committed_image_checks_agent_and_sidecar(self) -> None:
from bot_bottle.backend.macos_container import launch as mod from bot_bottle.backend.macos_container import launch as mod
from datetime import datetime, timezone from datetime import datetime, timezone
ts = datetime(2025, 1, 1, tzinfo=timezone.utc) ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
@@ -312,7 +280,7 @@ class TestMacosContainerStaleChecks(unittest.TestCase):
mod.stale_checks(self._plan()) mod.stale_checks(self._plan())
self.assertEqual(2, cs.call_count) self.assertEqual(2, cs.call_count)
def test_image_not_present_skips_check(self): def test_image_not_present_skips_check(self) -> None:
from bot_bottle.backend.macos_container import launch as mod from bot_bottle.backend.macos_container import launch as mod
with patch.object(mod, "read_committed_image", return_value=""), \ with patch.object(mod, "read_committed_image", return_value=""), \
patch.object(mod.container_mod, "image_exists", return_value=False), \ patch.object(mod.container_mod, "image_exists", return_value=False), \
@@ -320,13 +288,13 @@ class TestMacosContainerStaleChecks(unittest.TestCase):
mod.stale_checks(self._plan()) mod.stale_checks(self._plan())
cs.assert_not_called() cs.assert_not_called()
def test_backend_image_stale_checks_delegates(self): def test_backend_image_stale_checks_delegates(self) -> None:
from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
from bot_bottle.backend.macos_container import launch as mod from bot_bottle.backend.macos_container import launch as mod
backend = MacosContainerBottleBackend() backend = MacosContainerBottleBackend()
plan = self._plan() plan = self._plan()
with patch.object(mod, "stale_checks") as sc: with patch.object(mod, "stale_checks") as sc:
backend._image_stale_checks(plan) backend._image_stale_checks(plan) # type: ignore[arg-type]
sc.assert_called_once_with(plan) sc.assert_called_once_with(plan)