FabledSteward/tests/core/test_swarm_prune.py

"""The swarm prune keeps every image a swarm service is pinned to on every
node, removes the rest no container uses, and removes nothing when it can't
see the swarm."""
import importlib.util
import json
from pathlib import Path

import pytest
import yaml

from steward.ansible.sources import discover_playbook_meta, discover_playbooks, get_sources

BUILTIN = Path(get_sources({"sources": []})[0]["path"])


@pytest.fixture
def prune():
    spec = importlib.util.spec_from_file_location(
        "swarm_prune", BUILTIN / "maintenance" / "swarm_prune" / "files" / "swarm_prune.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _image(id_, tags=(), digests=(), size=100):
    return {"Id": id_, "RepoTags": list(tags), "RepoDigests": list(digests), "Size": size}


def test_the_playbook_is_listed_and_asks_before_it_runs():
    assert "maintenance/swarm_prune/swarm_prune.yml" in discover_playbooks(str(BUILTIN))
    content = (BUILTIN / "maintenance" / "swarm_prune" / "swarm_prune.yml").read_text()
    meta = discover_playbook_meta(content)
    assert meta["confirm"] is True and meta["category"] == "maintenance" and meta["description"]
    play = yaml.safe_load(content)[0]
    script = next(t for t in play["tasks"] if "ansible.builtin.script" in t)
    assert script["check_mode"] is False and "--dry-run" in script["ansible.builtin.script"]["cmd"]
    # No docker prune touches images: an image swarm pulled by digest has no
    # tag, and Docker calls that dangling (on 2026-10-25 a `system prune`
    # took 29 a node was keeping for the other's services).
    commands = yaml.safe_dump([t for t in play["tasks"] if "ansible.builtin.command" in t])
    assert "'system'" not in commands and "'image'" not in commands
    assert "'container', 'prune'" in commands and "'builder', 'prune'" in commands


def test_services_are_matched_by_digest_or_else_by_tag(prune):
    digests, refs = prune.wanted([
        "registry.example/team/app:dev@sha256:aaa",
        "docker.io/library/redis:7",
        "registry.example:5000/tool",
        None,
    ])
    assert digests == {"sha256:aaa"}
    assert refs == {"redis:7", "registry.example:5000/tool:latest"}


def test_an_image_a_service_or_a_container_needs_stays(prune):
    images = [
        _image("sha256:1", ["registry.example/team/app:dev"], ["registry.example/team/app@sha256:aaa"]),
        _image("sha256:2", ["registry.example/team/app:dev-old"], ["registry.example/team/app@sha256:old"]),
        _image("sha256:3", ["redis:7"], ["redis@sha256:rrr"]),
        _image("sha256:4", ["tool:1"]),
        _image("sha256:5", [], ["registry.example/team/app@sha256:bbb"]),
    ]
    gone = prune.removable(images, {"sha256:aaa"}, {"redis:7"}, used={"sha256:4"})
    assert [im["Id"] for im in gone] == ["sha256:2", "sha256:5"]


def _fake_docker(monkeypatch, prune, answers, removed):
    def docker(*args):
        if args[:2] == ("image", "rm"):
            removed.append(args[-1])
            return ""
        key = " ".join(args[:2])
        if key not in answers:
            raise RuntimeError(f"docker {key}: unexpected")
        answer = answers[key]
        if isinstance(answer, Exception):
            raise answer
        return answer
    monkeypatch.setattr(prune, "docker", docker)
    monkeypatch.setattr(prune, "api", lambda path: answers[path])


def _answers():
    return {
        "service ls": "s1 s2\n",
        "service inspect": json.dumps([
            {"Spec": {"TaskTemplate": {"ContainerSpec": {"Image": "registry.example/app:dev@sha256:aaa"}}}},
            {"Spec": {"TaskTemplate": {"ContainerSpec": {"Image": "proxy:3@sha256:ppp"}}}}]),
        "ps -aq": "c1\n",
        "container inspect": json.dumps([{"Image": "sha256:9", "Config": {"Image": "local:1"}}]),
        "/images/json": ([
            _image("sha256:1", ["registry.example/app:dev"], ["registry.example/app@sha256:aaa"]),
            _image("sha256:2", ["proxy:3"], ["proxy@sha256:ppp"]),
            _image("sha256:3", ["registry.example/app:old"], ["registry.example/app@sha256:old"], size=500),
            _image("sha256:9", ["local:1"]),
        ]),
    }


def test_a_run_removes_only_what_nothing_needs_and_says_so(prune, monkeypatch, capsys):
    removed: list = []
    _fake_docker(monkeypatch, prune, _answers(), removed)
    assert prune.main([]) == 0
    out = json.loads(capsys.readouterr().out)
    assert removed == ["sha256:3"]
    assert out == {"kept": 3, "removed": ["registry.example/app:old"], "failed": [],
                   "freed_bytes": 500, "dry_run": False}


def test_a_dry_run_removes_nothing(prune, monkeypatch, capsys):
    removed: list = []
    _fake_docker(monkeypatch, prune, _answers(), removed)
    assert prune.main(["--dry-run"]) == 0
    assert removed == [] and json.loads(capsys.readouterr().out)["removed"] == ["registry.example/app:old"]


def test_a_node_that_cant_list_the_swarm_removes_nothing(prune, monkeypatch, capsys):
    removed: list = []
    answers = _answers()
    answers["service ls"] = RuntimeError("docker service ls: This node is not a swarm manager.")
    _fake_docker(monkeypatch, prune, answers, removed)
    assert prune.main([]) == 2
    assert removed == [] and "nothing removed" in json.loads(capsys.readouterr().out)["error"]