"""Unit tests for the docker plugin's shared query helpers (no DB)."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS

from plugins.docker import service as svc
from plugins.docker import swarm_view as sv

NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)


def test_escape_like_makes_wildcards_literal():
    # A log search for "100%" or "user_id" must match those characters, not
    # treat them as LIKE wildcards.
    assert svc.escape_like("100%") == "100\\%"
    assert svc.escape_like("user_id") == "user\\_id"
    assert svc.escape_like("C:\\temp") == "C:\\\\temp"
    assert svc.escape_like("plain text") == "plain text"


def test_bucket_keeps_short_series_and_averages_long_ones():
    assert svc.bucket([1.0, 2.0, 3.0]) == [1.0, 2.0, 3.0]
    out = svc.bucket([float(i) for i in range(400)], target=40)
    assert len(out) == 40
    assert out[0] == sum(range(10)) / 10


def test_prune_extra_vars_mapping():
    """The prune buttons drive one playbook via prune_target; only 'images'
    widens to ALL unused images (docker image prune -a), system stays -f."""
    assert svc.prune_extra_vars("containers") == {"prune_target": "containers"}
    assert svc.prune_extra_vars("images") == {
        "prune_target": "images", "prune_all_images": True,
    }
    # System prune stays conservative — no prune_all_images key.
    assert svc.prune_extra_vars("system") == {"prune_target": "system"}


def test_a_stop_is_not_a_failure():
    # docker stop ends a container with 143 (SIGTERM) or 137 (SIGKILL).
    assert not svc.exit_is_failure(None) and not svc.exit_is_failure(0)
    assert not svc.exit_is_failure(137) and not svc.exit_is_failure(143)
    assert svc.exit_is_failure(1) and svc.exit_is_failure(2)
    assert svc.exit_is_failure(137, oom_killed=True)


def _ev(name, event, detail="", mins=5, host="h1"):
    return NS(container_name=name, host_id=host, event=event, detail=detail, at=NOW - timedelta(minutes=mins))


def test_event_is_failure_reads_the_exit_from_the_die_detail():
    assert svc.event_is_failure(_ev("a", "die", "exit 1"))
    assert not svc.event_is_failure(_ev("a", "die", "exit 143"))
    assert not svc.event_is_failure(_ev("a", "die", "exit 0"))
    assert svc.event_is_failure(_ev("a", "oom"))
    assert not svc.event_is_failure(_ev("a", "start"))


def _service(name, *, known=True, failures=0, error=None):
    return {"name": name, "tasks_known": known, "failures_24h": failures if known else None,
            "last_error": error}


def test_failures_24h_counts_services_once_and_skips_stops():
    events = [_ev("backup", "die", "exit 2", mins=30), _ev("backup", "die", "exit 2", mins=90),
              _ev("web", "die", "exit 143", mins=10),                         # someone stopped it
              _ev("app.1.abc", "die", "exit 1", mins=3),                      # Swarm counts its tasks
              _ev("old.1.x", "die", "exit 1", mins=4), _ev("old.2.y", "die", "exit 1", mins=40),
              _ev("old.1.x", "die", "exit 1", mins=4, host="h2")]               # a second manager's copy
    services = [_service("app", failures=3, error={"state": "failed", "error": "boom", "node": "nas",
                                                   "at": (NOW - timedelta(minutes=1)).isoformat()}),
                _service("old", known=False)]                                 # agent before 1.8.0
    rows = svc.failures_24h(events, services, {"h1": "nas"})
    got = {(r["kind"], r["name"]): r for r in rows}
    assert set(got) == {("container", "backup"), ("service", "app"), ("service", "old")}
    assert got[("container", "backup")]["count"] == 2 and got[("container", "backup")]["host"] == "nas"
    assert got[("service", "app")]["count"] == 3 and got[("service", "app")]["detail"] == "boom"
    assert got[("service", "old")]["count"] == 2                              # its tasks' deaths
    assert [r["name"] for r in rows] == ["app", "old", "backup"]              # newest first


def test_task_label_names_the_service_and_slot():
    assert svc.task_label(NS(name="nginx", service_name=None)) == "nginx"
    assert svc.task_label(NS(name="codex_db.1.37ma", service_name="codex_db")) == "codex_db · 1"
    agent = NS(name="agent.n0d3id.t4sk", service_name="agent", node_id="n0d3id")
    assert svc.task_label(agent, {"n0d3id": "node-3"}) == "agent · node-3"
    assert svc.task_label(agent) == "agent"


def test_usage_reads_cpu_in_cores_and_memory_against_a_real_limit():
    gib = 1024 ** 3
    c = NS(cpu_pct=150.0, mem_usage_bytes=gib, mem_limit_bytes=16 * gib, mem_pct=6.25)
    u = svc.usage(c, 4, 16 * gib)
    assert u["cores"] == 1.5 and u["cpu_share"] == 37.5
    # A limit of the host's whole memory is Docker saying "no limit".
    assert u["mem_limit"] is None and u["mem_of"] == "the host's memory"
    limited = svc.usage(NS(cpu_pct=None, mem_usage_bytes=gib, mem_limit_bytes=2 * gib, mem_pct=50.0), 4, 16 * gib)
    assert limited["cores"] is None and limited["mem_limit"] == 2 * gib and limited["mem_of"] == "its limit"
    # Unknown host size: cores still read, no share to grade against.
    assert svc.usage(c, None, None)["cpu_share"] is None
    assert svc.usage(c, 4, 16 * gib, cpu_pct=50.0)["cores"] == 0.5             # an average in place


def test_services_in_trouble_come_first_worst_first():
    def s(name, state, loop=False, f24=0):
        return {"name": name, "state": state, "crash_looping": loop, "failures_24h": f24}
    services = [s("alpha", "healthy"), s("beta", "healthy", f24=2), s("gamma", "degraded"),
                s("delta", "down"), s("eps", "down", loop=True)]
    assert [x["name"] for x in sv.services_problems_first(services)] == ["eps", "delta", "gamma", "beta", "alpha"]
    # A recovered service's failed tasks are history: not in Needs attention.
    assert [x["name"] for x in sv.failing_services(services)] == ["eps", "delta", "gamma"]