stackpilot/backend/tests/test_network_prune.py

"""Network prune with visible progress.

Same mechanism as the image and volume prunes (services/prune_job.py). Docker
has no -a for networks, and nothing to measure: networks hold no data.

Pinned here: the built-in and swarm networks are never touched, and neither is
a network a *stopped* container is attached to. That is stricter than
``docker network prune``, which goes by active endpoints and so leaves a
stopped container unable to start ("network not found").
"""
from __future__ import annotations

import threading

import pytest
from docker.errors import APIError


class FakeApi:
    def __init__(self):
        self._networks = [
            {"Id": "n-bridge", "Name": "bridge", "Scope": "local"},
            {"Id": "n-host", "Name": "host", "Scope": "local"},
            {"Id": "n-none", "Name": "none", "Scope": "local"},
            {"Id": "n-ingress", "Name": "ingress", "Scope": "swarm"},
            {"Id": "n-gw", "Name": "docker_gwbridge", "Scope": "local"},
            {"Id": "n-overlay", "Name": "shared_overlay", "Scope": "swarm"},
            {"Id": "n-run", "Name": "media_default", "Scope": "local"},
            {"Id": "n-stopped", "Name": "wiki_default", "Scope": "local"},
            {"Id": "n-old1", "Name": "oldstack_default", "Scope": "local"},
            {"Id": "n-old2", "Name": "scratch", "Scope": "local"},
        ]
        self._containers = [
            # running
            {"State": "running", "NetworkSettings": {"Networks": {
                "media_default": {"NetworkID": "n-run"}}}},
            # stopped: still attached, must keep its network
            {"State": "exited", "NetworkSettings": {"Networks": {
                "wiki_default": {"NetworkID": "n-stopped"}}}},
        ]
        self.removed: list[str] = []
        self.refuse: set[str] = set()

    def containers(self, all=False):
        assert all is True
        return self._containers

    def networks(self):
        return list(self._networks)

    def remove_network(self, net_id):
        if net_id in self.refuse:
            raise APIError(f"error while removing network: {net_id} has active endpoints")
        self.removed.append(net_id)


class FakeClient:
    def __init__(self, api):
        self.api = api


@pytest.fixture
def svc(monkeypatch):
    from services import network_service

    api = FakeApi()

    def short():
        raise AssertionError("prune must not use the 60s client")

    monkeypatch.setattr(network_service, "get_client", short)
    monkeypatch.setattr(network_service, "get_long_client", lambda: FakeClient(api))
    yield network_service, api
    if network_service._prune._lock.locked():
        network_service._prune._lock.release()


def test_only_networks_no_container_is_attached_to(svc):
    network_service, api = svc

    result = network_service.prune_networks()

    assert set(api.removed) == {"n-old1", "n-old2"}
    assert set(result["NetworksDeleted"]) == {"oldstack_default", "scratch"}
    status = network_service.prune_status()
    assert (status["total"], status["done"], status["deleted"]) == (2, 2, 2)
    assert status["reclaimed"] is None


def test_refused_network_does_not_stop_the_rest(svc):
    network_service, api = svc
    api.refuse.add("n-old1")

    result = network_service.prune_networks()

    assert result["NetworksDeleted"] == ["scratch"]
    status = network_service.prune_status()
    assert status["failed"] == 1 and status["errors"][0].startswith("oldstack_default: ")


def test_endpoint_background_then_poll(svc, as_admin):
    r = as_admin.post("/api/networks/prune", params={"background": True})
    assert r.status_code == 202 and r.json()["total"] == 2

    for _ in range(100):
        s = as_admin.get("/api/networks/prune-status").json()
        if not s["active"]:
            break
        threading.Event().wait(0.02)
    assert s["deleted"] == 2


def test_endpoint_waits_by_default_and_is_audited(svc, as_admin):
    from sqlmodel import Session, select

    from database import engine
    from models.audit import AuditLog

    r = as_admin.post("/api/networks/prune")
    assert r.status_code == 200
    assert r.json()["NetworksDeleted"] == ["oldstack_default", "scratch"]

    with Session(engine) as session:
        row = session.exec(
            select(AuditLog).where(AuditLog.action == "network.prune").order_by(AuditLog.id.desc())
        ).first()
    assert row.detail == "['oldstack_default', 'scratch']"


def test_endpoint_rejects_a_second_prune(svc, as_admin):
    network_service, _ = svc
    network_service._prune._lock.acquire()
    try:
        assert as_admin.post("/api/networks/prune").status_code == 409
    finally:
        network_service._prune._lock.release()


def test_prune_is_admin_only(svc, as_user):
    assert as_user.post("/api/networks/prune").status_code == 403
    assert as_user.get("/api/networks/prune-status").status_code == 403