jpmschweitzerandClaude
7b80e30691
Non-interactive equivalent of system-admin-toj's prune-docker.sh, which prompts per stage and so cannot run from cron. Only the stages that discard regenerable data run by default: build cache and dangling images. Unused images and volumes are opt-in, because docker volume prune removes volumes belonging to merely-stopped containers rather than only orphaned ones, which on this host is a plausible way to lose a database. A failing stage is reported and the remaining stages still run, since a partial reclaim beats none, but the task still ends up failed so the error is not swallowed. Co-Authored-By: Claude <noreply@anthropic.com>
136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
"""
|
|
Tests for the docker prune executor.
|
|
|
|
The important property is which stages run. `volumes` removes volumes belonging
|
|
to merely-stopped containers, so it must never be enabled by accident, and the
|
|
safe stages must stay on by default.
|
|
"""
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.config import Settings
|
|
from src.executors import docker_prune_executor as prune
|
|
|
|
|
|
def _runner(reclaimed="Total reclaimed space: 1.5GB", rc=0):
|
|
"""Fake _run returning docker-shaped output for every invocation."""
|
|
async def run(args):
|
|
if args[:3] == ["docker", "system", "df"]:
|
|
return 0, "TYPE TOTAL ACTIVE SIZE RECLAIMABLE", ""
|
|
return rc, reclaimed, "" if rc == 0 else "boom"
|
|
return run
|
|
|
|
|
|
@pytest.mark.executor
|
|
@pytest.mark.unit
|
|
class TestStageSelection:
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_defaults_run_only_the_safe_stages(self, test_settings: Settings):
|
|
calls = []
|
|
|
|
async def run(args):
|
|
calls.append(args)
|
|
if args[:3] == ["docker", "system", "df"]:
|
|
return 0, "df output", ""
|
|
return 0, "Total reclaimed space: 0B", ""
|
|
|
|
with patch.object(prune, "_run", run):
|
|
await prune.execute({}, test_settings)
|
|
|
|
joined = [" ".join(c) for c in calls]
|
|
assert any("builder prune" in c for c in joined)
|
|
assert any("image prune -f" in c for c in joined)
|
|
# The destructive ones must not appear without being asked for.
|
|
assert not any("volume prune" in c for c in joined)
|
|
assert not any("image prune -a" in c for c in joined)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_volumes_only_when_explicitly_enabled(self, test_settings: Settings):
|
|
calls = []
|
|
|
|
async def run(args):
|
|
calls.append(args)
|
|
if args[:3] == ["docker", "system", "df"]:
|
|
return 0, "df output", ""
|
|
return 0, "Total reclaimed space: 2GB", ""
|
|
|
|
with patch.object(prune, "_run", run):
|
|
await prune.execute({"volumes": True}, test_settings)
|
|
|
|
assert any("volume prune" in " ".join(c) for c in calls)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_stages_disabled_is_a_no_op(self, test_settings: Settings):
|
|
with patch.object(prune, "_run", AsyncMock()) as run:
|
|
result = await prune.execute(
|
|
{"build_cache": False, "dangling_images": False}, test_settings
|
|
)
|
|
assert "nothing to do" in result
|
|
run.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dry_run_executes_no_prune(self, test_settings: Settings):
|
|
calls = []
|
|
|
|
async def run(args):
|
|
calls.append(args)
|
|
return 0, "df output", ""
|
|
|
|
with patch.object(prune, "_run", run):
|
|
result = await prune.execute({"dry_run": True}, test_settings)
|
|
|
|
assert "dry run" in result
|
|
assert all("prune" not in " ".join(c) for c in calls)
|
|
|
|
|
|
@pytest.mark.executor
|
|
@pytest.mark.unit
|
|
class TestFailureHandling:
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_docker_unavailable_raises(self, test_settings: Settings):
|
|
async def run(args):
|
|
return 1, "", "Cannot connect to the Docker daemon"
|
|
|
|
with patch.object(prune, "_run", run):
|
|
with pytest.raises(Exception, match="docker unavailable"):
|
|
await prune.execute({}, test_settings)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_stage_surfaces_but_others_still_run(self, test_settings: Settings):
|
|
attempted = []
|
|
|
|
async def run(args):
|
|
if args[:3] == ["docker", "system", "df"]:
|
|
return 0, "df output", ""
|
|
attempted.append(" ".join(args))
|
|
if "builder" in args:
|
|
return 1, "", "builder exploded"
|
|
return 0, "Total reclaimed space: 3MB", ""
|
|
|
|
with patch.object(prune, "_run", run):
|
|
with pytest.raises(Exception, match="one or more prune stages failed"):
|
|
await prune.execute({}, test_settings)
|
|
|
|
# The image stage must still have been attempted after builder failed.
|
|
assert any("image prune" in a for a in attempted)
|
|
|
|
|
|
@pytest.mark.executor
|
|
@pytest.mark.unit
|
|
class TestOutputParsing:
|
|
|
|
@pytest.mark.parametrize(
|
|
"output,expected",
|
|
[
|
|
("Total reclaimed space: 1.5GB", "1.5GB"),
|
|
("deleted: sha256:abc\nTotal reclaimed space: 0B", "0B"),
|
|
("no such line", "0B"),
|
|
("", "0B"),
|
|
],
|
|
)
|
|
def test_reclaimed_parsing(self, output, expected):
|
|
assert prune._reclaimed(output) == expected
|