feat(cleanup): walk every backend, reap smolmachines orphans too
test / unit (pull_request) Successful in 26s
test / integration (pull_request) Successful in 41s

`./cli.py cleanup` previously called only the env-var-selected
backend's `prepare_cleanup` / `cleanup` — so a leftover smolvm
machine + bundle container + bundle network from a crashed
smolmachines bottle would survive a default `docker`-mode cleanup
indefinitely.

Smolmachines now has a real `cleanup` module (alongside
`enumerate.py` from issue #77) that walks:

  - smolvm machines named `claude-bottle-*` (via
    `smolvm machine ls --json`)
  - bundle containers `claude-bottle-sidecars-*`
  - bundle networks `claude-bottle-bundle-*`

Cleanup runs stop+delete on the machines, force-rm on the
containers, network rm on the networks. Each step is best-effort
so a failed rm doesn't block the rest.

`cli.py cleanup` walks every backend in `known_backend_names()`
and runs each backend's `cleanup` after a single y/N prompt that
shows a combined plan.

State dirs (`~/.claude-bottle/state/<slug>/`) are shared layout
with the docker backend, which still owns the orphan-state-dir
bucket. It now consults `enumerate_active_bottles()` for the
cross-backend live identity set so a running smolmachines
bottle's state dir isn't reaped during a cleanup.

Tests: smolmachines cleanup (prepare + cleanup ordering + failure
handling); cross-backend orphan protection on the docker
state-dir check; CLI cmd_cleanup walks both backends, short-
circuits on all-empty, aborts on N. 617 unit tests pass.

End-to-end verified on this host:
  $ smolvm machine ls --json | jq '.[].name'
  "claude-bottle-researcher-m3hxd"
  $ ./cli.py cleanup
  --- smolmachines backend ---
  smolvm machine:  claude-bottle-researcher-m3hxd
  remove all of the above? [y/N]

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-27 18:38:00 -04:00
parent 5e0130b56f
commit 346367ca32
8 changed files with 538 additions and 37 deletions
+23 -4
View File
@@ -83,11 +83,18 @@ def _list_prefixed_networks() -> list[str]:
return sorted(set(out))
def _list_orphan_state_dirs(live_projects: set[str]) -> list[str]:
def _list_orphan_state_dirs(
live_projects: set[str], protected_identities: set[str],
) -> list[str]:
"""State identities whose compose project isn't running and
that don't have a `.preserve` marker. `.preserve` means the
user (or an auto-preserve-on-crash) wants the state kept for
`resume`."""
`resume`.
`protected_identities` is the set of slugs that are live in
ANY backend — used so this docker-side check doesn't reap a
running smolmachines bottle's state dir (the layout is shared
across both backends)."""
state_root = _supervise.claude_bottle_root() / "state"
if not state_root.is_dir():
return []
@@ -99,6 +106,8 @@ def _list_orphan_state_dirs(live_projects: set[str]) -> list[str]:
project = f"{COMPOSE_PROJECT_PREFIX}{identity}"
if project in live_projects:
continue
if identity in protected_identities:
continue
if is_preserved(identity):
continue
orphans.append(identity)
@@ -106,15 +115,25 @@ def _list_orphan_state_dirs(live_projects: set[str]) -> list[str]:
def prepare_cleanup() -> DockerBottleCleanupPlan:
"""Enumerate everything cleanup will touch. No removals."""
"""Enumerate everything cleanup will touch. No removals.
Pulls the union of live identities across backends via
`enumerate_active_agents()` so the orphan-state-dir bucket
doesn't include slugs whose smolmachines VM is still up."""
docker_mod.require_docker()
projects = list_compose_projects()
project_set = set(projects)
# Late import to avoid a circular at module-load time —
# the backend package's __init__ imports this module.
from .. import enumerate_active_agents
protected = {a.slug for a in enumerate_active_agents()}
return DockerBottleCleanupPlan(
projects=tuple(projects),
stray_containers=tuple(_list_prefixed_containers()),
stray_networks=tuple(_list_prefixed_networks()),
orphan_state_dirs=tuple(_list_orphan_state_dirs(project_set)),
orphan_state_dirs=tuple(
_list_orphan_state_dirs(project_set, protected),
),
)
@@ -8,6 +8,7 @@ from pathlib import Path
from typing import Generator, Sequence
from .. import ActiveAgent, BottleBackend, BottleSpec
from . import cleanup as _cleanup
from . import enumerate as _enumerate
from . import launch as _launch
from . import prepare as _prepare
@@ -76,12 +77,10 @@ class SmolmachinesBottleBackend(
_supervise.provision_supervise(plan, target)
def prepare_cleanup(self) -> SmolmachinesBottleCleanupPlan:
return SmolmachinesBottleCleanupPlan()
return _cleanup.prepare_cleanup()
def cleanup(self, plan: SmolmachinesBottleCleanupPlan) -> None:
del plan
# Nothing to clean in chunks 1-3 — see
# SmolmachinesBottleCleanupPlan docstring.
_cleanup.cleanup(plan)
def enumerate_active(self) -> Sequence[ActiveAgent]:
return _enumerate.enumerate_active()
@@ -1,13 +1,29 @@
"""SmolmachinesBottleCleanupPlan — concrete BottleCleanupPlan stub
(PRD 0023 chunk 1).
"""SmolmachinesBottleCleanupPlan — concrete BottleCleanupPlan (issue #77).
Chunk 1 always reports nothing-to-clean. Real enumeration —
orphaned smolvm machines, stranded gvproxy sockets, leftover
sidecar bundle containers — lands in chunk 4 alongside the
integration-test sweep that exercises teardown."""
Tracks the resources `SmolmachinesBottleBackend.cleanup` will
remove:
- machines: smolvm machines whose name starts with
`claude-bottle-` (running or stopped). Stopped +
deleted via `smolvm machine stop` + `machine delete -f`.
- bundles: docker containers `claude-bottle-sidecars-<slug>`
left over from a smolmachines bottle (the bundle's
port-forwards stay published on lo0 aliases until
the container is gone). Removed via `docker rm -f`.
- networks: docker networks `claude-bottle-bundle-<slug>`
attached to the bundles. Removed via
`docker network rm`.
Smolmachines state dirs live under the same `~/.claude-bottle/state/`
path the docker backend uses; the docker backend's
`prepare_cleanup` already enumerates orphan state dirs and is the
single source of truth for that bucket (consults
`enumerate_active_bottles()` so it doesn't reap a live
smolmachines bottle's dir)."""
from __future__ import annotations
import sys
from dataclasses import dataclass
from ...log import info
@@ -16,10 +32,24 @@ from .. import BottleCleanupPlan
@dataclass(frozen=True)
class SmolmachinesBottleCleanupPlan(BottleCleanupPlan):
def print(self) -> None:
info("smolmachines cleanup: nothing to remove (chunk 4 will "
"enumerate orphan machines + gvproxy sockets)")
"""Resources SmolmachinesBottleBackend.cleanup will remove.
Produced by `prepare_cleanup`; sorted so the y/N output is
stable."""
machines: tuple[str, ...] = ()
bundles: tuple[str, ...] = ()
networks: tuple[str, ...] = ()
@property
def empty(self) -> bool:
return True
return not self.machines and not self.bundles and not self.networks
def print(self) -> None:
print(file=sys.stderr)
for name in self.machines:
info(f"smolvm machine: {name}")
for name in self.bundles:
info(f"bundle container:{name}")
for name in self.networks:
info(f"bundle network: {name}")
print(file=sys.stderr)
@@ -0,0 +1,159 @@
"""Cleanup + active-listing for the smolmachines backend (issue #77).
`prepare_cleanup` enumerates leftover smolmachines resources:
- smolvm machines (`smolvm machine ls --json`) whose name starts
with `claude-bottle-`.
- bundle docker containers (`claude-bottle-sidecars-<slug>`).
- bundle docker networks (`claude-bottle-bundle-<slug>`).
State dirs live under `~/.claude-bottle/state/<identity>/` —
shared layout with the docker backend, which has the single
orphan-state-dir enumerator (it already consults
`enumerate_active_agents()` so a live smolmachines bottle's dir
is preserved).
`cleanup` removes everything in the plan: stop + delete each VM,
force-rm each container, rm each network. Each step is
best-effort — a failure on one resource doesn't block the others."""
from __future__ import annotations
import json
import subprocess
from ...log import info, warn
from . import sidecar_bundle as _bundle
from . import smolvm as _smolvm
from .bottle_cleanup_plan import SmolmachinesBottleCleanupPlan
# Both names start with the same prefix the launcher uses.
_VM_PREFIX = "claude-bottle-"
_BUNDLE_PREFIX = _bundle.bundle_container_name("") # `claude-bottle-sidecars-`
_NETWORK_PREFIX = _bundle.bundle_network_name("") # `claude-bottle-bundle-`
def prepare_cleanup() -> SmolmachinesBottleCleanupPlan:
"""Enumerate every smolmachines-owned resource on the host.
No side effects. Returns an empty plan when smolvm isn't on
PATH (no machines to reap) — `cleanup` is a no-op in that
case too."""
machines = _list_claude_bottle_machines()
bundles = _list_bundle_containers()
networks = _list_bundle_networks()
return SmolmachinesBottleCleanupPlan(
machines=tuple(sorted(machines)),
bundles=tuple(sorted(bundles)),
networks=tuple(sorted(networks)),
)
def cleanup(plan: SmolmachinesBottleCleanupPlan) -> None:
"""Remove everything in the plan. Order matters: stop VMs
first (they hold ports on lo0 aliases via libkrun), then the
bundle containers (which hold the host port-forwards), then
the networks (which docker won't reap until the containers
are gone)."""
for name in plan.machines:
info(f"stopping smolvm machine {name}")
subprocess.run(
["smolvm", "machine", "stop", "--name", name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
check=False,
)
info(f"deleting smolvm machine {name}")
r = subprocess.run(
["smolvm", "machine", "delete", "-f", name],
capture_output=True, text=True, check=False,
)
if r.returncode != 0:
warn(
f"smolvm machine delete -f {name} failed: "
f"{(r.stderr or '').strip()}"
)
for name in plan.bundles:
info(f"removing bundle container {name}")
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
check=False,
)
for name in plan.networks:
info(f"removing bundle network {name}")
r = subprocess.run(
["docker", "network", "rm", name],
capture_output=True, text=True, check=False,
)
if r.returncode != 0 and "no such network" not in (r.stderr or "").lower():
warn(
f"docker network rm {name} failed: "
f"{(r.stderr or '').strip()}"
)
def _list_claude_bottle_machines() -> list[str]:
"""All smolvm machines named `claude-bottle-*`, regardless of
state (running / stopped / created). Empty when smolvm isn't
installed."""
if not _smolvm.is_available():
return []
r = subprocess.run(
["smolvm", "machine", "ls", "--json"],
capture_output=True, text=True, check=False,
)
if r.returncode != 0:
return []
try:
machines = json.loads(r.stdout or "[]")
except json.JSONDecodeError:
return []
return [
m["name"] for m in machines
if isinstance(m, dict)
and m.get("name", "").startswith(_VM_PREFIX)
]
def _list_bundle_containers() -> list[str]:
"""All docker containers named `claude-bottle-sidecars-*`,
running or stopped. Empty when docker isn't installed."""
# Late import: `backend/__init__` imports this module
# transitively via the smolmachines backend.
from .. import has_backend
if not has_backend("docker"):
return []
r = subprocess.run(
["docker", "ps", "-a",
"--filter", f"name=^{_BUNDLE_PREFIX}",
"--format", "{{.Names}}"],
capture_output=True, text=True, check=False,
)
if r.returncode != 0:
return []
return [
line for line in (r.stdout or "").splitlines()
if line and line.startswith(_BUNDLE_PREFIX)
]
def _list_bundle_networks() -> list[str]:
"""All docker networks named `claude-bottle-bundle-*`. Empty
when docker isn't installed."""
from .. import has_backend
if not has_backend("docker"):
return []
r = subprocess.run(
["docker", "network", "ls",
"--filter", f"name={_NETWORK_PREFIX}",
"--format", "{{.Name}}"],
capture_output=True, text=True, check=False,
)
if r.returncode != 0:
return []
return [
line for line in (r.stdout or "").splitlines()
if line and line.startswith(_NETWORK_PREFIX)
]
+29 -12
View File
@@ -1,11 +1,16 @@
"""cleanup: stop and remove all orphaned claude-bottle resources.
PRD 0018 chunk 4: backend's prepare_cleanup carries everything in
one plan — live compose projects (whose `compose down` removes
containers + networks atomically), legacy stray containers/networks
that aren't in any project, and orphan state dirs (per-bottle
state with no live project AND no `.preserve` marker). One prompt,
one cleanup call.
Walks every registered backend (docker + smolmachines) so a single
`./cli.py cleanup` reaps both backends' leftovers — orphaned
smolvm machines won't survive a docker-only cleanup pass (issue
addressed alongside #77).
Each backend's `prepare_cleanup` enumerates its own resources;
docker's `_list_orphan_state_dirs` consults
`enumerate_active_agents()` for the union of live identities so
state dirs of running smolmachines bottles aren't reaped. State
dirs are shared layout, so docker is the single owner of that
bucket.
State dirs with `.preserve` are intentionally never touched — they
hold capability-block rebuilds or crash snapshots the operator may
@@ -17,25 +22,37 @@ from __future__ import annotations
import sys
from ..backend import get_bottle_backend
from ..backend import get_bottle_backend, known_backend_names
from ..log import info
from ._common import read_tty_line
def cmd_cleanup(_argv: list[str]) -> int:
backend = get_bottle_backend()
plan = backend.prepare_cleanup()
# Order: stable backend iteration so the y/N output is
# deterministic across runs.
plans = [
(name, get_bottle_backend(name)) for name in known_backend_names()
]
prepared = [(name, b, b.prepare_cleanup()) for name, b in plans]
if plan.empty:
if all(p.empty for _, _, p in prepared):
info("no claude-bottle resources to clean up")
return 0
plan.print()
for name, _, plan in prepared:
if plan.empty:
continue
info(f"--- {name} backend ---")
plan.print()
if not _prompt_yes("remove all of the above?"):
info("cleanup: skipped")
return 0
backend.cleanup(plan)
for name, backend, plan in prepared:
if plan.empty:
continue
backend.cleanup(plan)
info("cleanup: done")
return 0
@@ -0,0 +1,105 @@
"""Unit: `cli.py cleanup` walks every backend (issue follow-up).
Asserts cmd_cleanup queries each backend's `prepare_cleanup`,
combines the y/N output, and runs each backend's `cleanup` when
the operator confirms. Mocks the backends and stdin."""
from __future__ import annotations
import sys
import unittest
from unittest.mock import patch, MagicMock
from claude_bottle.cli import cleanup as cmd
def _make_backend(empty: bool = True):
backend = MagicMock()
plan = MagicMock(empty=empty)
backend.prepare_cleanup.return_value = plan
backend.cleanup = MagicMock()
return backend, plan
class TestCmdCleanup(unittest.TestCase):
def test_iterates_every_backend(self):
docker, docker_plan = _make_backend(empty=False)
smol, smol_plan = _make_backend(empty=False)
backends_by_name = {"docker": docker, "smolmachines": smol}
with patch.object(
cmd, "known_backend_names",
return_value=("docker", "smolmachines"),
), patch.object(
cmd, "get_bottle_backend",
side_effect=lambda name: backends_by_name[name],
), patch.object(
cmd, "_prompt_yes", return_value=True,
):
self.assertEqual(0, cmd.cmd_cleanup([]))
docker.prepare_cleanup.assert_called_once()
smol.prepare_cleanup.assert_called_once()
docker.cleanup.assert_called_once_with(docker_plan)
smol.cleanup.assert_called_once_with(smol_plan)
def test_short_circuits_when_all_empty(self):
docker, _ = _make_backend(empty=True)
smol, _ = _make_backend(empty=True)
backends_by_name = {"docker": docker, "smolmachines": smol}
with patch.object(
cmd, "known_backend_names",
return_value=("docker", "smolmachines"),
), patch.object(
cmd, "get_bottle_backend",
side_effect=lambda name: backends_by_name[name],
), patch.object(
cmd, "_prompt_yes",
) as prompt:
self.assertEqual(0, cmd.cmd_cleanup([]))
prompt.assert_not_called()
docker.cleanup.assert_not_called()
smol.cleanup.assert_not_called()
def test_abort_at_prompt_runs_nothing(self):
docker, _ = _make_backend(empty=False)
smol, _ = _make_backend(empty=True)
backends_by_name = {"docker": docker, "smolmachines": smol}
with patch.object(
cmd, "known_backend_names",
return_value=("docker", "smolmachines"),
), patch.object(
cmd, "get_bottle_backend",
side_effect=lambda name: backends_by_name[name],
), patch.object(
cmd, "_prompt_yes", return_value=False,
):
self.assertEqual(0, cmd.cmd_cleanup([]))
docker.cleanup.assert_not_called()
smol.cleanup.assert_not_called()
def test_skips_empty_plans_when_others_have_work(self):
# docker has work, smolmachines doesn't — only docker.cleanup
# is called.
docker, docker_plan = _make_backend(empty=False)
smol, _ = _make_backend(empty=True)
backends_by_name = {"docker": docker, "smolmachines": smol}
with patch.object(
cmd, "known_backend_names",
return_value=("docker", "smolmachines"),
), patch.object(
cmd, "get_bottle_backend",
side_effect=lambda name: backends_by_name[name],
), patch.object(
cmd, "_prompt_yes", return_value=True,
):
cmd.cmd_cleanup([])
docker.cleanup.assert_called_once_with(docker_plan)
smol.cleanup.assert_not_called()
if __name__ == "__main__":
unittest.main()
+31 -7
View File
@@ -44,7 +44,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
self._teardown_fake_home()
def test_no_state_root_returns_empty(self):
self.assertEqual([], _list_orphan_state_dirs(set()))
self.assertEqual([], _list_orphan_state_dirs(set(), set()))
def test_state_dir_with_no_live_no_preserve_is_orphan(self):
# Just touch the dir; no metadata, no preserve marker — the
@@ -52,7 +52,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
bottle_state.write_per_bottle_dockerfile("solo-aaa", "FROM x\n")
self.assertEqual(
["solo-aaa"],
_list_orphan_state_dirs(set()),
_list_orphan_state_dirs(set(), set()),
)
def test_live_project_skips_dir(self):
@@ -61,7 +61,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
bottle_state.write_per_bottle_dockerfile("live-bbb", "FROM x\n")
self.assertEqual(
[],
_list_orphan_state_dirs({"claude-bottle-live-bbb"}),
_list_orphan_state_dirs({"claude-bottle-live-bbb"}, set()),
)
def test_preserve_marker_skips_dir(self):
@@ -71,14 +71,14 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
bottle_state.mark_preserved("kept-ccc")
self.assertEqual(
[],
_list_orphan_state_dirs(set()),
_list_orphan_state_dirs(set(), set()),
)
def test_preserve_overrides_no_live_project(self):
# Even without a live project, a preserve marker keeps it.
bottle_state.write_per_bottle_dockerfile("kept-ddd", "FROM x\n")
bottle_state.mark_preserved("kept-ddd")
self.assertEqual([], _list_orphan_state_dirs(set()))
self.assertEqual([], _list_orphan_state_dirs(set(), set()))
def test_mixed_set_categorized_correctly(self):
bottle_state.write_per_bottle_dockerfile("orphan-eee", "FROM x\n")
@@ -86,7 +86,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
bottle_state.write_per_bottle_dockerfile("kept-ggg", "FROM z\n")
bottle_state.mark_preserved("kept-ggg")
result = _list_orphan_state_dirs({"claude-bottle-live-fff"})
result = _list_orphan_state_dirs({"claude-bottle-live-fff"}, set())
self.assertEqual(["orphan-eee"], result)
def test_sorted_output(self):
@@ -94,7 +94,31 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
bottle_state.write_per_bottle_dockerfile(name, "FROM x\n")
self.assertEqual(
["aaa-1", "mmm-1", "zzz-1"],
_list_orphan_state_dirs(set()),
_list_orphan_state_dirs(set(), set()),
)
def test_protected_identity_skips_dir(self):
# `protected_identities` carries slugs that are live in
# any backend (smolmachines included). docker's orphan
# detection respects them so a running smolmachines
# bottle's state dir isn't reaped while the VM is up.
bottle_state.write_per_bottle_dockerfile("smol-hhh", "FROM x\n")
self.assertEqual(
[],
_list_orphan_state_dirs(set(), {"smol-hhh"}),
)
def test_protected_overrides_no_live_project(self):
# A smolmachines bottle has no docker compose project but
# IS in the protected set; the absence of a project
# shouldn't cause a reap.
bottle_state.write_per_bottle_dockerfile("smol-iii", "FROM x\n")
self.assertEqual(
[],
_list_orphan_state_dirs(
{"claude-bottle-something-else"}, # different project up
{"smol-iii"}, # but smol-iii is live
),
)
+148
View File
@@ -0,0 +1,148 @@
"""Unit: smolmachines backend cleanup (`cleanup.py` +
`bottle_cleanup_plan.py`).
Tests mock `subprocess.run` + `has_backend` so they execute
without docker / smolvm on PATH. Each cleanup step verifies argv
shape; teardown verifies order (machines → bundles → networks)."""
from __future__ import annotations
import subprocess
import unittest
from unittest.mock import patch
from claude_bottle import backend as backend_mod
from claude_bottle.backend.smolmachines import cleanup
from claude_bottle.backend.smolmachines.bottle_cleanup_plan import (
SmolmachinesBottleCleanupPlan,
)
def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
return subprocess.CompletedProcess(
args=[], returncode=0, stdout=stdout, stderr=stderr,
)
class TestPrepareCleanup(unittest.TestCase):
def test_empty_when_nothing_running(self):
with patch.object(cleanup, "_smolvm") as smolvm, \
patch.object(cleanup.subprocess, "run") as run, \
patch.object(backend_mod, "has_backend", return_value=True):
smolvm.is_available.return_value = True
run.return_value = _ok(stdout="[]")
plan = cleanup.prepare_cleanup()
self.assertTrue(plan.empty)
def test_lists_machines_bundles_networks(self):
def fake_run(argv, *a, **kw):
if argv[:3] == ["smolvm", "machine", "ls"]:
return _ok(stdout=(
'[{"name":"claude-bottle-a-1","state":"running"},'
' {"name":"claude-bottle-b-2","state":"created"},'
' {"name":"unrelated","state":"running"}]'
))
if argv[:2] == ["docker", "ps"]:
return _ok(stdout=(
"claude-bottle-sidecars-a-1\n"
"claude-bottle-sidecars-b-2\n"
))
if argv[:3] == ["docker", "network", "ls"]:
return _ok(stdout=(
"claude-bottle-bundle-a-1\n"
"claude-bottle-bundle-b-2\n"
))
return _ok()
with patch.object(cleanup, "_smolvm") as smolvm, \
patch.object(cleanup.subprocess, "run", side_effect=fake_run), \
patch.object(backend_mod, "has_backend", return_value=True):
smolvm.is_available.return_value = True
plan = cleanup.prepare_cleanup()
# `unrelated` filtered out (no claude-bottle- prefix).
self.assertEqual(
("claude-bottle-a-1", "claude-bottle-b-2"),
plan.machines,
)
self.assertEqual(
("claude-bottle-sidecars-a-1", "claude-bottle-sidecars-b-2"),
plan.bundles,
)
self.assertEqual(
("claude-bottle-bundle-a-1", "claude-bottle-bundle-b-2"),
plan.networks,
)
def test_no_smolvm_means_no_machines(self):
with patch.object(cleanup, "_smolvm") as smolvm, \
patch.object(cleanup.subprocess, "run", return_value=_ok()), \
patch.object(backend_mod, "has_backend", return_value=True):
smolvm.is_available.return_value = False
plan = cleanup.prepare_cleanup()
self.assertEqual((), plan.machines)
class TestCleanup(unittest.TestCase):
def test_machines_stopped_then_deleted_then_bundles_then_networks(self):
plan = SmolmachinesBottleCleanupPlan(
machines=("claude-bottle-a-1",),
bundles=("claude-bottle-sidecars-a-1",),
networks=("claude-bottle-bundle-a-1",),
)
calls: list[list[str]] = []
def fake_run(argv, *a, **kw):
calls.append(list(argv[:4]))
return _ok()
with patch.object(cleanup.subprocess, "run", side_effect=fake_run):
cleanup.cleanup(plan)
# Stop precedes delete precedes bundle rm precedes network rm.
self.assertEqual(
["smolvm", "machine", "stop", "--name"], calls[0],
)
self.assertEqual(
["smolvm", "machine", "delete", "-f"], calls[1],
)
self.assertEqual(
["docker", "rm", "-f", "claude-bottle-sidecars-a-1"], calls[2],
)
self.assertEqual(
["docker", "network", "rm", "claude-bottle-bundle-a-1"], calls[3],
)
def test_failures_are_warnings_not_fatal(self):
# smolvm machine delete -f returning non-zero should warn
# but continue with bundles + networks. The cleanup is
# idempotent on success and tries to remove every resource.
plan = SmolmachinesBottleCleanupPlan(
machines=("claude-bottle-a-1",),
bundles=("claude-bottle-sidecars-a-1",),
networks=(),
)
results = iter([
_ok(), # stop succeeds
subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="boom"), # delete fails
_ok(), # bundle rm succeeds
])
def fake_run(argv, *a, **kw):
return next(results)
with patch.object(cleanup.subprocess, "run", side_effect=fake_run), \
patch.object(cleanup, "warn") as warn:
cleanup.cleanup(plan)
# warn called once for the delete failure.
warn.assert_called_once()
def test_empty_plan_is_noop(self):
plan = SmolmachinesBottleCleanupPlan()
with patch.object(cleanup.subprocess, "run") as run:
cleanup.cleanup(plan)
run.assert_not_called()
if __name__ == "__main__":
unittest.main()