Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 29 additions & 6 deletions cli/localci/cli/images.py
Original file line number Diff line number Diff line change
Expand Up @@ -226,20 +226,45 @@ def images_build(
# ---------------------------------------------------------------------------


def _cleanup_targets(installed_tags: list[str], registry_tags: set[str]) -> list[str]:
Comment thread
clean6378-max-it marked this conversation as resolved.
"""Installed repo:tags that are also in the registry, sorted."""
return sorted(tag for tag in installed_tags if tag in registry_tags)


@images.command("clean")
@click.option("--all", "clean_all", is_flag=True, help="Remove all localci images.")
@click.option("--dry-run", is_flag=True, help="Preview without removing.")
@click.option(
"--registry",
"-r",
"registry_path",
type=click.Path(path_type=Path, exists=True),
default=None,
help="Path to image-registry.yml.",
)
@click.pass_context
def images_clean(
ctx: click.Context,
clean_all: bool,
dry_run: bool,
registry_path: Path | None,
) -> None:
"""Clean up Docker images."""
if not clean_all:
print_info("Nothing to clean. Use --all to remove localci images.")
return

try:
registry = _get_registry(registry_path)
Comment thread
clean6378-max-it marked this conversation as resolved.
except FileNotFoundError as exc:
print_error(str(exc))
ctx.exit(1)
return
except (OSError, yaml.YAMLError, TypeError, ValueError) as exc:
print_error(f"Could not load image registry: {exc}")
ctx.exit(1)
return

try:
dm = DockerManager()
except DockerNotAvailableError as exc:
Expand All @@ -253,13 +278,11 @@ def images_clean(
ctx.exit(result.returncode)
return

targets = [
line.strip()
for line in result.stdout.splitlines()
if line.strip().startswith("capy-ubuntu-")
]
registry_tags = {e.docker_tag for e in registry.entries if e.docker_tag}
installed = [line.strip() for line in result.stdout.splitlines() if line.strip()]
targets = _cleanup_targets(installed, registry_tags)
if not targets:
print_info("No localci capy images found.")
print_info("No localci images found.")
return

if dry_run:
Expand Down
293 changes: 293 additions & 0 deletions cli/tests/test_images_paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import pytest
from click.testing import CliRunner

from localci.cli.images import _cleanup_targets
from localci.cli.main import cli
from localci.utils.paths import (
IMAGES_CAPY_REL,
Expand Down Expand Up @@ -250,3 +251,295 @@ def fake_run(cmd: list[str]) -> MagicMock:

assert result.exit_code == 0
assert captured[0][1] == str(images_dir / "build-all.sh")


def _docker_run_side_effect(
installed_tags: list[str],
*,
captured_rmi: list[list[str]] | None = None,
):
"""Stub _run for clean tests: fake image ls output, record rmi argv."""

def fake_run(cmd: list[str]) -> MagicMock:
if "image" in cmd and "ls" in cmd:
stdout = "\n".join(installed_tags)
if stdout:
stdout += "\n"
return MagicMock(returncode=0, stdout=stdout, stderr="")
if "rmi" in cmd:
if captured_rmi is not None:
captured_rmi.append(cmd)
return MagicMock(returncode=0, stdout="", stderr="")
return MagicMock(returncode=0, stdout="", stderr="")

return fake_run


class TestCleanupTargets:
def test_intersection_sorted(self) -> None:
installed = ["b:tag", "a:tag", "other:tag"]
registry = {"a:tag", "b:tag", "registry-only:tag"}
assert _cleanup_targets(installed, registry) == ["a:tag", "b:tag"]

def test_empty_when_no_match(self) -> None:
assert _cleanup_targets(["other:tag"], {"ubuntu-latest-gcc15:latest"}) == []


class TestImagesClean:
def test_clean_generic_tag_removes_registry_match_only(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
registry = _write_registry_with_image(
repo, docker_tag="ubuntu-latest-gcc15:latest"
)
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(outside)

captured_rmi: list[list[str]] = []
fake_dm = MagicMock()
fake_dm.build_cmd.side_effect = lambda *args: ["docker", *args]

with (
patch("localci.cli.images.DockerManager", return_value=fake_dm),
patch(
"localci.cli.images._run",
side_effect=_docker_run_side_effect(
["ubuntu-latest-gcc15:latest", "other:tag"],
captured_rmi=captured_rmi,
),
),
):
result = runner.invoke(
cli,
["images", "clean", "--all", "--registry", str(registry)],
)

assert result.exit_code == 0
assert len(captured_rmi) == 1
assert captured_rmi[0][-1] == "ubuntu-latest-gcc15:latest"
assert "other:tag" not in captured_rmi[0]

def test_clean_capy_tag_regression(
Comment thread
clean6378-max-it marked this conversation as resolved.
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
registry = _write_registry_with_image(
repo, docker_tag="capy-ubuntu-25.04-gcc15:latest"
)
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(outside)

captured_rmi: list[list[str]] = []
fake_dm = MagicMock()
fake_dm.build_cmd.side_effect = lambda *args: ["docker", *args]

with (
patch("localci.cli.images.DockerManager", return_value=fake_dm),
patch(
"localci.cli.images._run",
side_effect=_docker_run_side_effect(
[
"capy-ubuntu-25.04-gcc15:latest",
"capy-ubuntu-orphan:latest",
"unrelated:tag",
],
captured_rmi=captured_rmi,
),
),
):
result = runner.invoke(
cli,
["images", "clean", "--all", "--registry", str(registry)],
)

assert result.exit_code == 0
assert len(captured_rmi) == 1
assert captured_rmi[0][-1] == "capy-ubuntu-25.04-gcc15:latest"
assert "capy-ubuntu-orphan:latest" not in captured_rmi[0]
assert "unrelated:tag" not in captured_rmi[0]

def test_clean_dry_run_lists_targets_without_rmi(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
registry = _write_registry_with_image(
repo, docker_tag="ubuntu-latest-gcc15:latest"
)
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(outside)

captured_rmi: list[list[str]] = []
fake_dm = MagicMock()
fake_dm.build_cmd.side_effect = lambda *args: ["docker", *args]

with (
patch("localci.cli.images.DockerManager", return_value=fake_dm),
patch(
"localci.cli.images._run",
side_effect=_docker_run_side_effect(
["ubuntu-latest-gcc15:latest", "other:tag"],
captured_rmi=captured_rmi,
),
),
):
result = runner.invoke(
cli,
[
"images",
"clean",
"--all",
"--dry-run",
"--registry",
str(registry),
],
)

assert result.exit_code == 0
assert "ubuntu-latest-gcc15:latest" in result.output
assert "other:tag" not in result.output
assert "Dry-run" in result.output
assert captured_rmi == []

def test_clean_without_all_does_not_invoke_docker(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(outside)

with patch("localci.cli.images._run") as mock_run:
result = runner.invoke(cli, ["images", "clean"])

mock_run.assert_not_called()
assert result.exit_code == 0
assert "Nothing to clean" in result.output

def test_clean_empty_intersection_reports_no_images(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
registry = _write_registry_with_image(
repo, docker_tag="ubuntu-latest-gcc15:latest"
)
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(outside)

captured_rmi: list[list[str]] = []
fake_dm = MagicMock()
fake_dm.build_cmd.side_effect = lambda *args: ["docker", *args]

with (
patch("localci.cli.images.DockerManager", return_value=fake_dm),
patch(
"localci.cli.images._run",
side_effect=_docker_run_side_effect(
["other:tag"],
captured_rmi=captured_rmi,
),
),
):
result = runner.invoke(
cli,
["images", "clean", "--all", "--registry", str(registry)],
)

assert result.exit_code == 0
assert "No localci images found" in result.output
assert captured_rmi == []

def test_clean_uses_explicit_registry(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
registry = _write_registry_with_image(repo, docker_tag="explicit:tag")
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(outside)

captured_rmi: list[list[str]] = []
fake_dm = MagicMock()
fake_dm.build_cmd.side_effect = lambda *args: ["docker", *args]

with (
patch("localci.cli.images.DockerManager", return_value=fake_dm),
patch(
"localci.cli.images._run",
side_effect=_docker_run_side_effect(
["explicit:tag"],
captured_rmi=captured_rmi,
),
),
):
result = runner.invoke(
cli,
["images", "clean", "--all", "--registry", str(registry)],
)

assert result.exit_code == 0
assert captured_rmi[0][-1] == "explicit:tag"

def test_clean_discovers_registry_from_cwd(
Comment thread
clean6378-max-it marked this conversation as resolved.
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
repo = tmp_path / "repo"
nested = repo / "examples" / "validation-project"
nested.mkdir(parents=True)
_write_registry_with_image(repo, docker_tag="ubuntu-latest-gcc15:latest")
monkeypatch.chdir(nested)

captured_rmi: list[list[str]] = []
fake_dm = MagicMock()
fake_dm.build_cmd.side_effect = lambda *args: ["docker", *args]

with (
patch("localci.cli.images.DockerManager", return_value=fake_dm),
patch(
"localci.cli.images._run",
side_effect=_docker_run_side_effect(
["ubuntu-latest-gcc15:latest", "other:tag"],
captured_rmi=captured_rmi,
),
),
):
result = runner.invoke(cli, ["images", "clean", "--all"])

assert result.exit_code == 0
assert len(captured_rmi) == 1
assert captured_rmi[0][-1] == "ubuntu-latest-gcc15:latest"
assert "other:tag" not in captured_rmi[0]

def test_clean_all_without_discoverable_registry_fails_before_docker(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
site_module = (
tmp_path
/ "venv"
/ "lib"
/ "python3.10"
/ "site-packages"
/ "localci"
/ "cli"
/ "images.py"
)
site_module.parent.mkdir(parents=True, exist_ok=True)
site_module.write_text("# installed copy\n", encoding="utf-8")
outside = tmp_path / "outside"
outside.mkdir()
monkeypatch.chdir(outside)

with (
patch("localci.cli.images._IMAGES_MODULE", site_module),
patch("localci.cli.images.DockerManager") as mock_dm,
patch("localci.cli.images._run") as mock_run,
):
result = runner.invoke(cli, ["images", "clean", "--all"])

assert result.exit_code == 1
assert REGISTRY_FILENAME in result.output
mock_dm.assert_not_called()
mock_run.assert_not_called()
Loading