FabledSteward/tests/integration/test_docker.py

"""Integration: per-host Docker schema + host-scoped persistence.

Validates the docker_002 migration (host_id columns + composite key) applied and
that persist_host_docker scopes containers/metrics by host so same-named
containers on different hosts no longer collide. Requires STEWARD_DATABASE_URL.
"""
from __future__ import annotations

import asyncio
import json
import os
import uuid
from datetime import datetime, timedelta, timezone

import pytest

pytestmark = pytest.mark.integration

_NEEDS_DB = pytest.mark.skipif(
    not os.environ.get("STEWARD_DATABASE_URL"),
    reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
)


@pytest.fixture
def app():
    if not os.environ.get("STEWARD_DATABASE_URL"):
        pytest.skip("needs Postgres")
    from steward.app import create_app
    return create_app(testing=False)


@_NEEDS_DB
def test_docker_tables_are_host_scoped(app):
    from sqlalchemy import text

    async def _go():
        async with app.db_sessionmaker() as s:
            # docker_containers carries a host_id column after docker_002.
            for tbl in ("docker_containers",):
                cols = {r[0] for r in (await s.execute(text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_name = :t"), {"t": tbl})).all()}
                assert "host_id" in cols, f"{tbl} missing host_id"
            # docker_containers is keyed by (host_id, name), not name alone.
            pk_cols = {r[0] for r in (await s.execute(text(
                "SELECT a.attname FROM pg_index i "
                "JOIN pg_attribute a ON a.attrelid = i.indrelid "
                " AND a.attnum = ANY(i.indkey) "
                "WHERE i.indrelid = 'docker_containers'::regclass AND i.indisprimary"
            ))).all()}
            assert pk_cols == {"host_id", "name"}

    asyncio.run(_go())


@_NEEDS_DB
def test_events_and_swarm_tables_exist(app):
    """docker_004 created the lifecycle + swarm topology tables with host_id FKs."""
    from sqlalchemy import text

    async def _go():
        async with app.db_sessionmaker() as s:
            for tbl in ("docker_events", "docker_swarm_services", "docker_swarm_nodes"):
                cols = {r[0] for r in (await s.execute(text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_name = :t"), {"t": tbl})).all()}
                assert cols, f"{tbl} missing entirely"
                assert "host_id" in cols, f"{tbl} missing host_id"
            # docker_events carries the lifecycle columns the ingest derivation writes.
            ev_cols = {r[0] for r in (await s.execute(text(
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_name = 'docker_events'"))).all()}
            assert {"event", "container_name", "at", "detail"} <= ev_cols

    asyncio.run(_go())


@_NEEDS_DB
def test_docker_logs_table_shape(app):
    """docker_009 created docker_logs: host-scoped, with the twin indexes the
    viewer (host, container, ts) and the age-cutoff prune (ts) rely on."""
    from sqlalchemy import text

    async def _go():
        async with app.db_sessionmaker() as s:
            cols = {r[0] for r in (await s.execute(text(
                "SELECT column_name FROM information_schema.columns "
                "WHERE table_name = 'docker_logs'"))).all()}
            assert cols, "docker_logs missing entirely"
            assert {"id", "host_id", "container_name", "ts", "stream", "line"} <= cols
            idx = {r[0] for r in (await s.execute(text(
                "SELECT indexname FROM pg_indexes "
                "WHERE tablename = 'docker_logs'"))).all()}
            assert "ix_docker_logs_host_container_time" in idx
            assert "ix_docker_logs_ts" in idx

    asyncio.run(_go())


def _persist_fn(app):
    """Resolve persist_host_docker via the registered capability if the docker
    plugin is loaded, else import it directly (the import is safe only when the
    plugin is NOT loaded — otherwise its models are already mapped)."""
    from steward.core.capabilities import has_capability, get_capability
    if has_capability("docker.persist_host_samples"):
        return get_capability("docker.persist_host_samples").fn
    from plugins.docker.ingest import persist_host_docker
    return persist_host_docker


def _retention_fn(app):
    """Resolve run_docker_retention via capability (or direct import if unloaded)."""
    from steward.core.capabilities import has_capability, get_capability
    if has_capability("docker.run_retention"):
        return get_capability("docker.run_retention").fn
    from plugins.docker.retention import run_docker_retention
    return run_docker_retention


def _prune_logs_fn(app):
    """Resolve prune_docker_logs via capability (or direct import if unloaded)."""
    from steward.core.capabilities import has_capability, get_capability
    if has_capability("docker.prune_logs"):
        return get_capability("docker.prune_logs").fn
    from plugins.docker.retention import prune_docker_logs
    return prune_docker_logs


@_NEEDS_DB
def test_persist_scopes_containers_by_host(app):
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    now = datetime.now(timezone.utc)
    snapshot = [(now, [{
        "name": "web", "container_id": "abc123", "image": "nginx",
        "status": "running", "cpu_pct": 12.5, "mem_pct": 30.0,
        "mem_usage_bytes": 100, "mem_limit_bytes": 200, "ports": [], "started_at": None,
        # enrichment (agent ≥ 1.4.0)
        "health": "healthy", "restart_count": 2, "exit_code": 0, "oom_killed": False,
        "compose_project": "stack", "service_name": "web",
        "net_rx_bytes": 1000, "net_tx_bytes": 2000,
        "blk_read_bytes": 4096, "blk_write_bytes": 8192,
    }])]

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_containers"))
                await s.execute(text(
                    "DELETE FROM plugin_metrics WHERE source_module = 'docker'"))
                h1 = Host(id=str(uuid.uuid4()), name="alpha", address="10.0.0.1")
                h2 = Host(id=str(uuid.uuid4()), name="beta", address="10.0.0.2")
                s.add_all([h1, h2])
                await s.flush()
                # Same container name on two hosts — must not collide now.
                await persist(s, h1, snapshot)
                await persist(s, h2, snapshot)

            web_rows = (await s.execute(text(
                "SELECT COUNT(*) FROM docker_containers WHERE name = 'web'"))).scalar()
            # The series went into the one metric store, host-scoped (#5752),
            # at the snapshot's time.
            metric_hosts = (await s.execute(text(
                "SELECT COUNT(DISTINCT resource_name) FROM plugin_metrics "
                "WHERE source_module = 'docker' AND metric_name = 'mem_usage_bytes' "
                "AND recorded_at = :at"), {"at": now})).scalar()
            resources = {r[0] for r in (await s.execute(text(
                "SELECT DISTINCT resource_name FROM plugin_metrics "
                "WHERE source_module = 'docker'"))).all()}
            # Enrichment columns persisted (check one host's row).
            enrich = (await s.execute(text(
                "SELECT health, restart_count, service_name, net_rx_bytes "
                "FROM docker_containers WHERE name = 'web' LIMIT 1"))).first()
            return web_rows, metric_hosts, resources, enrich

    web_rows, metric_hosts, resources, enrich = asyncio.run(_go())
    assert web_rows == 2          # one 'web' per host, keyed by (host_id, name)
    assert metric_hosts == 2      # time-series rows scoped per host
    assert resources == {"alpha/web", "beta/web"}
    assert enrich == ("healthy", 2, "web", 1000)   # enrichment round-trips


@_NEEDS_DB
def test_persist_logs_stores_lines(app):
    """Pushed container logs land in docker_logs, host-scoped, one row per line,
    ordered by their own ts; the agent's _steward truncation marker is dropped."""
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    now = datetime.now(timezone.utc)
    log_batches = [(now, [
        {"container": "web", "stream": "stdout",
         "ts": "2026-01-01T00:00:01+00:00", "line": "started"},
        {"container": "web", "stream": "stderr",
         "ts": "2026-01-01T00:00:02+00:00", "line": "warn: x"},
        {"container": "_steward", "stream": "stderr", "ts": None, "line": "truncated"},
    ])]

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_logs"))
                h = Host(id=str(uuid.uuid4()), name="loghost", address="10.0.0.9")
                s.add(h)
                await s.flush()
                hid = h.id
                # snapshots empty; logs passed as the 6th positional arg.
                await persist(s, h, [], None, None, log_batches)
            rows = (await s.execute(text(
                "SELECT container_name, stream, line FROM docker_logs "
                "WHERE host_id = :h ORDER BY ts"), {"h": hid})).all()
            return [tuple(r) for r in rows]

    assert asyncio.run(_go()) == [
        ("web", "stdout", "started"),
        ("web", "stderr", "warn: x"),
    ]


@_NEEDS_DB
def test_persist_logs_respects_toggle_and_exclude(app):
    """Server-side controls (Settings): an excluded container's lines are dropped
    while others persist; the global toggle off stores nothing new."""
    from types import SimpleNamespace
    from sqlalchemy import text
    from steward.models.hosts import Host
    from steward.core.settings import set_setting

    persist = _persist_fn(app)
    now = datetime.now(timezone.utc)
    batch = [(now, [
        {"container": "web", "stream": "stdout", "ts": None, "line": "keep-me"},
        {"container": "noisy", "stream": "stdout", "ts": None, "line": "drop-me"},
    ])]

    hid = str(uuid.uuid4())
    # Stand-in host: the logs-only persist path reads only host.id, so a plain
    # object sidesteps ORM attribute-expiry across the commit boundaries below.
    host = SimpleNamespace(id=hid, name="loghost3")

    async def _go():
        async with app.db_sessionmaker() as s:
            # Back-to-back begin blocks with NO read in between — a SELECT between
            # them would autobegin a txn and make the next begin() collide.
            async with s.begin():
                await s.execute(text("DELETE FROM docker_logs"))
                await set_setting(s, "docker.logs.enabled", True)
                await set_setting(s, "docker.logs.exclude", ["noisy"])
                s.add(Host(id=hid, name="loghost3", address="10.7.7.10"))
                await s.flush()   # host row before its log rows (FK order)
                await persist(s, host, [], None, None, batch)   # web kept, noisy excluded
            async with s.begin():          # global kill-switch off → nothing stored
                await set_setting(s, "docker.logs.enabled", False)
                await persist(s, host, [], None, None, batch)
            async with s.begin():          # restore defaults for other tests
                await set_setting(s, "docker.logs.enabled", True)
                await set_setting(s, "docker.logs.exclude", [])
            # Reads only after the last begin block (they autobegin, but nothing
            # opens a begin() after them). Toggle-off added nothing, so the state
            # here still reflects the exclude phase.
            kept = {r[0] for r in (await s.execute(text(
                "SELECT DISTINCT container_name FROM docker_logs WHERE host_id=:h"),
                {"h": hid})).all()}
            after_off = (await s.execute(text(
                "SELECT COUNT(*) FROM docker_logs WHERE host_id=:h"), {"h": hid})).scalar()
            return kept, after_off

    kept, after_off = asyncio.run(_go())
    assert kept == {"web"}      # excluded 'noisy' dropped on ingest
    assert after_off == 1       # toggle off added nothing → still just the one 'web' line


@_NEEDS_DB
def test_large_memory_values_persist_as_bigint(app):
    """A container using >2^31 bytes of RAM must persist. Regression: mem_usage_bytes
    / mem_limit_bytes were int4 and overflowed (asyncpg 'value out of int32 range'),
    failing the whole ingest batch for any host with a >2.1 GB container."""
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    now = datetime.now(timezone.utc)
    big_usage = 8_185_077_760    # ~8.18 GB — well past int4 max (2_147_483_647)
    big_limit = 17_179_869_184   # 16 GB
    snapshot = [(now, [{
        "name": "bigmem", "container_id": "deadbeef", "image": "postgres",
        "status": "running", "cpu_pct": 1.0, "mem_pct": 50.0,
        "mem_usage_bytes": big_usage, "mem_limit_bytes": big_limit,
        "ports": [], "started_at": None,
    }])]

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_containers"))
                h = Host(id=str(uuid.uuid4()), name="bigmemhost", address="10.0.0.9")
                s.add(h)
                await s.flush()
                await persist(s, h, snapshot)
            row = (await s.execute(text(
                "SELECT mem_usage_bytes, mem_limit_bytes FROM docker_containers "
                "WHERE name = 'bigmem'"))).first()
            metric = (await s.execute(text(
                "SELECT value FROM plugin_metrics WHERE source_module = 'docker' "
                "AND resource_name = 'bigmemhost/bigmem' "
                "AND metric_name = 'mem_usage_bytes'"))).scalar()
            return row, metric

    row, metric = asyncio.run(_go())
    assert row == (big_usage, big_limit)   # current-state row round-trips
    assert metric == big_usage             # time-series row round-trips


def test_lifecycle_events_derived_across_snapshots(app):
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    t1 = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
    t2 = datetime(2026, 6, 19, 12, 0, 30, tzinfo=timezone.utc)
    running = {"name": "web", "container_id": "abc", "image": "nginx",
               "status": "running", "cpu_pct": 5.0, "mem_pct": 10.0,
               "restart_count": 0, "exit_code": None, "oom_killed": False,
               "health": "healthy"}
    died = {**running, "status": "exited", "cpu_pct": None,
            "exit_code": 137, "oom_killed": True, "health": None}

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_events"))
                await s.execute(text("DELETE FROM docker_containers"))
                h = Host(id=str(uuid.uuid4()), name="evt", address="10.9.9.9")
                s.add(h)
                await s.flush()
                # Baseline snapshot establishes state — must NOT emit a start.
                await persist(s, h, [(t1, [running])])
                # Second snapshot: the container OOM-died.
                await persist(s, h, [(t2, [died])])
                hid = h.id
            rows = (await s.execute(text(
                "SELECT event, detail FROM docker_events WHERE host_id = :h "
                "ORDER BY event"), {"h": hid})).all()
            return [(r[0], r[1]) for r in rows]

    events = asyncio.run(_go())
    # baseline emitted nothing; the death emitted both die (exit 137) and oom.
    assert ("die", "exit 137") in events
    assert ("oom", None) in events
    assert not any(e[0] == "start" for e in events)


@_NEEDS_DB
def test_vanished_container_is_reaped(app):
    """A container absent from the latest snapshot is deleted, not left behind as
    a permanent stopped row. Regression: removed one-shot containers (CI jobs,
    buildkit builders) accumulated forever and inflated the dashboard counts."""
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    t1 = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
    t2 = datetime(2026, 6, 19, 12, 0, 30, tzinfo=timezone.utc)
    base = {"status": "running", "cpu_pct": 1.0, "mem_pct": 1.0,
            "restart_count": 0, "exit_code": None, "oom_killed": False, "health": None}
    keep = {**base, "name": "keep"}
    gone = {**base, "name": "ci-job-123"}

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_containers"))
                h = Host(id=str(uuid.uuid4()), name="reap", address="10.9.9.8")
                s.add(h)
                await s.flush()
                await persist(s, h, [(t1, [keep, gone])])   # both present
                before = [r[0] for r in (await s.execute(text(
                    "SELECT name FROM docker_containers WHERE host_id = :h ORDER BY name"),
                    {"h": h.id})).all()]
                await persist(s, h, [(t2, [keep])])         # ci-job-123 vanished
                after = [r[0] for r in (await s.execute(text(
                    "SELECT name FROM docker_containers WHERE host_id = :h ORDER BY name"),
                    {"h": h.id})).all()]
                return before, after

    before, after = asyncio.run(_go())
    assert before == ["ci-job-123", "keep"]
    assert after == ["keep"]   # the vanished container was reaped


@_NEEDS_DB
def test_persist_marks_each_part_for_the_slow_ingest_line(app):
    """Each part of the docker persist is marked, so the slow-ingest line
    names the part its time went to (#5181)."""
    from steward.models.hosts import Host

    class _Marks:
        def __init__(self):
            self.names = []

        def mark(self, name):
            self.names.append(name)

    persist = _persist_fn(app)
    at = datetime(2026, 10, 6, 23, 0, 0, tzinfo=timezone.utc)
    c = {"name": "web", "status": "running", "cpu_pct": 1.0, "mem_pct": 1.0,
         "restart_count": 0, "exit_code": None, "oom_killed": False, "health": None}
    logs = [(at, [{"container": "web", "stream": "stdout", "ts": None, "line": "hello"}])]
    marks = _Marks()

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                h = Host(id=str(uuid.uuid4()), name="marks", address="10.9.9.7")
                s.add(h)
                await s.flush()
                await persist(s, h, [(at, [c])], None, None, logs,
                              {"status": "ok", "detail": None, "at": at, "swarm_manager": None},
                              phases=marks)

    asyncio.run(_go())
    assert marks.names == ["docker.status", "docker.logs", "docker.metrics", "docker.events",
                           "docker.containers", "docker.alerts", "docker.reap"]


@_NEEDS_DB
def test_an_empty_snapshot_from_a_docker_that_answered_clears_the_host(app):
    """Docker answering with no containers removes the host's rows (#5152); a
    sample with no Docker answer at all removes nothing."""
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    t1 = datetime(2026, 10, 6, 12, 0, 0, tzinfo=timezone.utc)
    t2 = datetime(2026, 10, 6, 12, 0, 30, tzinfo=timezone.utc)
    web = {"name": "web", "status": "running", "cpu_pct": 1.0, "mem_pct": 1.0,
           "restart_count": 0, "exit_code": None, "oom_killed": False, "health": None}

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                h = Host(id=str(uuid.uuid4()), name=f"empty-{uuid.uuid4().hex[:6]}", address="10.9.9.7")
                s.add(h)
                await s.flush()

                async def names():
                    return [r[0] for r in (await s.execute(text(
                        "SELECT name FROM docker_containers WHERE host_id = :h"), {"h": h.id})).all()]

                await persist(s, h, [(t1, [web])], status={"status": "ok", "detail": None, "at": t1})
                first = await names()
                await persist(s, h, [], status={"status": "error", "detail": "timed out", "at": t2})
                after_error = await names()
                await persist(s, h, [(t2, [])], status={"status": "ok", "detail": None, "at": t2})
                return first, after_error, await names()

    first, after_error, after_empty = asyncio.run(_go())
    assert first == ["web"]
    assert after_error == ["web"]   # couldn't ask: keep what Docker last said
    assert after_empty == []        # asked, and there are none


@_NEEDS_DB
def test_docker_status_is_recorded_once_a_host_has_had_docker(app):
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    t1 = datetime(2026, 10, 6, 12, 0, 0, tzinfo=timezone.utc)
    t2 = t1 + timedelta(seconds=30)
    t3 = t1 + timedelta(seconds=60)

    async def _go():
        seen = []
        async with app.db_sessionmaker() as s:
            async with s.begin():
                never = Host(id=str(uuid.uuid4()), name=f"plain-{uuid.uuid4().hex[:6]}", address="10.9.9.5")
                had = Host(id=str(uuid.uuid4()), name=f"had-{uuid.uuid4().hex[:6]}", address="10.9.9.6")
                s.add_all([never, had])
                await s.flush()
                gone = {"status": "unavailable", "detail": "no Docker socket at /var/run/docker.sock"}
                await persist(s, never, [], status=dict(gone, at=t1))
                await s.flush()

                async def row(host):
                    return (await s.execute(text(
                        "SELECT status, detail, since, reported_at FROM docker_host_status "
                        "WHERE host_id = :h"), {"h": host.id})).first()

                seen.append(await row(never))
                for status, at in (({"status": "ok", "detail": None}, t1),
                                   (gone, t2), (gone, t3),
                                   ({"status": "error", "detail": "permission denied on /var/run/docker.sock"}, t3)):
                    await persist(s, had, [], status=dict(status, at=at))
                    await s.flush()
                    seen.append(tuple(await row(had)))
        return seen

    never, ok, gone1, gone2, err = asyncio.run(_go())
    assert never is None                 # never had Docker: nothing to say
    assert ok == ("ok", None, t1, t1)
    assert gone1 == ("unavailable", "no Docker socket at /var/run/docker.sock", t2, t2)
    assert gone2[2] == t2 and gone2[3] == t3   # still gone: since holds, the report moves on
    assert err[:3] == ("error", "permission denied on /var/run/docker.sock", t3)


@_NEEDS_DB
def test_a_node_that_is_no_longer_a_manager_loses_its_swarm_rows(app):
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    t1 = datetime(2026, 10, 6, 12, 0, 0, tzinfo=timezone.utc)
    swarm = {"services": [{"service_name": "web", "mode": "replicated", "desired": 1, "running": 1}],
             "nodes": [{"node_id": "n1", "hostname": "mgr", "role": "manager"}]}

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                h = Host(id=str(uuid.uuid4()), name=f"demoted-{uuid.uuid4().hex[:6]}", address="10.9.9.4")
                s.add(h)
                await s.flush()

                async def counts():
                    out = []
                    for t in ("docker_swarm_services", "docker_swarm_nodes"):
                        out.append((await s.execute(text(
                            f"SELECT count(*) FROM {t} WHERE host_id = :h"), {"h": h.id})).scalar())
                    return tuple(out)

                await persist(s, h, [], swarm, status={"status": "ok", "detail": None, "at": t1})
                manager = await counts()
                # /info failed: not knowing is not "no longer a manager".
                await persist(s, h, [], status={"status": "error", "detail": "x", "at": t1,
                                                "swarm_manager": None})
                unknown = await counts()
                await persist(s, h, [], status={"status": "ok", "detail": None, "at": t1,
                                                "swarm_manager": False})
                return manager, unknown, await counts()

    manager, unknown, demoted = asyncio.run(_go())
    assert manager == (1, 1)
    assert unknown == (1, 1)
    assert demoted == (0, 0)


@_NEEDS_DB
def test_swarm_topology_persisted(app):
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    swarm = {
        "services": [{"service_name": "web", "mode": "replicated",
                      "desired": 3, "running": 2, "image": "nginx",
                      "placement": [{"node_id": "n1", "running": 2}],
                      "tasks": [{"task_id": "t1", "slot": 3, "node_id": "n1",
                                 "desired_state": "shutdown", "state": "failed",
                                 "message": "started", "error": "task: non-zero exit (1)",
                                 "exit_code": 1, "at": "2026-10-06T13:19:51Z"}]}],
        "nodes": [{"node_id": "n1", "hostname": "mgr", "role": "manager",
                   "availability": "active", "status": "ready", "leader": True}],
    }

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_swarm_services"))
                await s.execute(text("DELETE FROM docker_swarm_nodes"))
                h = Host(id=str(uuid.uuid4()), name="mgr", address="10.8.8.8")
                s.add(h)
                await s.flush()
                # No containers in this sample — swarm must still persist.
                from steward.core.plugin_manager import loaded_plugin_submodule
                loaded_plugin_submodule("docker", "swarm_health")._last_recorded = None
                since = datetime.now(timezone.utc)
                await persist(s, h, [], swarm)
                hid = h.id
            # The swarm-wide figures alert rules watch (#5788): web is 2 of 3.
            figures = dict((await s.execute(text(
                "SELECT metric_name, value FROM plugin_metrics WHERE source_module = 'docker' "
                "AND resource_name = 'swarm' AND recorded_at >= :t"), {"t": since})).all())
            svc = (await s.execute(text(
                "SELECT mode, desired, running, image, placement_json, tasks_json "
                "FROM docker_swarm_services WHERE host_id = :h"), {"h": hid})).first()
            node = (await s.execute(text(
                "SELECT role, availability, status, leader "
                "FROM docker_swarm_nodes WHERE host_id = :h"), {"h": hid})).first()
            return svc, node, figures

    svc, node, figures = asyncio.run(_go())
    assert figures == {"swarm_services_short": 1.0, "swarm_nodes_down": 0.0,
                       "swarm_managers_unreachable": 0.0}
    assert svc[0] == "replicated" and svc[1] == 3 and svc[2] == 2 and svc[3] == "nginx"
    assert "n1" in svc[4]  # placement JSON carries the node id
    (task,) = json.loads(svc[5])  # docker_010: the manager's task records survive
    assert task["error"] == "task: non-zero exit (1)" and task["slot"] == 3
    assert node[0] == "manager" and node[2] == "ready" and node[3] is True


@_NEEDS_DB
def test_disk_usage_persisted(app):
    """persist_host_docker stores the /system/df summary + image rows, and
    replaces the image set on the next report (stale images pruned)."""
    from sqlalchemy import text
    from steward.models.hosts import Host

    persist = _persist_fn(app)
    disk1 = {
        "layers_size": 1000, "images_total": 2, "images_active": 1,
        "images_size": 1000, "images_reclaimable": 700,
        "containers_count": 2, "containers_size": 100,
        "volumes_count": 1, "volumes_size": 40, "build_cache_size": 15,
        "images": [
            {"image_id": "aaaaaaaaaaaa", "repo_tag": "app:1", "size": 300,
             "shared_size": 50, "containers": 1},
            {"image_id": "bbbbbbbbbbbb", "repo_tag": "<none>", "size": 700,
             "shared_size": 0, "containers": 0},
        ],
    }
    disk2 = {**disk1, "images_reclaimable": 0, "images": [disk1["images"][0]]}

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_images"))
                await s.execute(text("DELETE FROM docker_disk_usage"))
                hid = str(uuid.uuid4())
                h = Host(id=hid, name=f"dsk-{hid[:8]}", address="10.6.6.6")
                s.add(h)
                await s.flush()
                hid = h.id
                await persist(s, h, [], None, disk1)
            summary = (await s.execute(text(
                "SELECT images_reclaimable, images_total, build_cache_size "
                "FROM docker_disk_usage WHERE host_id=:h"), {"h": hid})).first()
            img_count = (await s.execute(text(
                "SELECT COUNT(*) FROM docker_images WHERE host_id=:h"), {"h": hid})).scalar()
            # End the autobegun read transaction before opening the next write one.
            await s.rollback()
            # Re-report with one image dropped — the stale row must be pruned.
            async with s.begin():
                await persist(s, await s.get(Host, hid), [], None, disk2)
            after = (await s.execute(text(
                "SELECT image_id FROM docker_images WHERE host_id=:h"), {"h": hid})).all()
            reclaimable2 = (await s.execute(text(
                "SELECT images_reclaimable FROM docker_disk_usage WHERE host_id=:h"),
                {"h": hid})).scalar()
            # Each report is recorded as metrics too, resource = the host (#5304).
            metrics = (await s.execute(text(
                "SELECT metric_name, value FROM plugin_metrics WHERE source_module = 'docker' "
                "AND resource_name = :r"), {"r": f"dsk-{hid[:8]}"})).all()
            return summary, img_count, [r[0] for r in after], reclaimable2, metrics

    summary, img_count, after_ids, reclaimable2, metrics = asyncio.run(_go())
    assert {v for m, v in metrics if m == "disk_total_bytes"} == {1155.0}       # 1000 + 100 + 40 + 15
    assert sorted(v for m, v in metrics if m == "disk_reclaimable_bytes") == [0.0, 700.0]
    assert {m for m, _ in metrics} == {"disk_images_bytes", "disk_containers_bytes", "disk_volumes_bytes",
                                       "disk_build_cache_bytes", "disk_total_bytes", "disk_reclaimable_bytes"}
    assert summary[0] == 700 and summary[1] == 2 and summary[2] == 15
    assert img_count == 2
    assert after_ids == ["aaaaaaaaaaaa"]   # the unreferenced image pruned on re-report
    assert reclaimable2 == 0               # summary upserted in place


@_NEEDS_DB
def test_retention_prunes_events_past_their_window(app):
    """Docker retention prunes lifecycle events past the events window; the
    container series roll up with the one metric store (#5752)."""
    from sqlalchemy import text
    from steward.models.hosts import Host

    run_retention = _retention_fn(app)
    now = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
    old_event = datetime(2026, 5, 1, 0, 0, 0, tzinfo=timezone.utc)  # > events window
    new_event = datetime(2026, 6, 18, 0, 0, 0, tzinfo=timezone.utc)  # inside events window

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_events"))
                h = Host(id=str(uuid.uuid4()), name="ret", address="10.7.7.7")
                s.add(h)
                await s.flush()
                hid = h.id
                for ev_at, ev in [(old_event, "stop"), (new_event, "start")]:
                    await s.execute(text(
                        "INSERT INTO docker_events (id, host_id, container_name, event, at) "
                        "VALUES (:id,:h,'web',:ev,:at)"),
                        {"id": str(uuid.uuid4()), "h": hid, "ev": ev, "at": ev_at})
            async with s.begin():
                counts = await run_retention(s, events_days=30, now=now)
            events_left = (await s.execute(text(
                "SELECT COUNT(*) FROM docker_events WHERE host_id=:h"), {"h": hid})).scalar()
            tables = {r[0] for r in (await s.execute(text(
                "SELECT table_name FROM information_schema.tables "
                "WHERE table_name LIKE 'docker_metrics%'"))).all()}
            return counts, events_left, tables

    counts, events_left, tables = asyncio.run(_go())
    assert events_left == 1                    # only the in-window event survives
    assert counts == {"events_pruned": 1}
    assert tables == set()                     # docker_012 dropped both tables


@_NEEDS_DB
def test_retention_logs_ring(app):
    """docker_logs ring (m79): lines past the age window are pruned; within it,
    each container keeps only the newest ~cap bytes; containers are independent,
    including a same-named container on another host (#5449 ranks only the
    containers over the cap, joined on host and name)."""
    from datetime import timedelta
    from sqlalchemy import text
    from steward.models.hosts import Host

    prune_logs = _prune_logs_fn(app)
    now = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc)
    line40 = "x" * 40   # 40 chars/line; cap=100 keeps 3 lines (excl-prefix 0/40/80)

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                await s.execute(text("DELETE FROM docker_logs"))
                h = Host(id=str(uuid.uuid4()), name="loghost2", address="10.7.7.8")
                h2 = Host(id=str(uuid.uuid4()), name="loghost3", address="10.7.7.9")
                s.add_all([h, h2])
                await s.flush()
                hid, hid2 = h.id, h2.id

                def ins(cn, ts, line, host=None):
                    return s.execute(text(
                        "INSERT INTO docker_logs "
                        "(id, host_id, container_name, ts, stream, line) "
                        "VALUES (:id,:h,:cn,:ts,'stdout',:line)"),
                        {"id": str(uuid.uuid4()), "h": host or hid, "cn": cn,
                         "ts": ts, "line": line})

                # Past the 3-day age window → age-pruned.
                await ins("old", now - timedelta(days=10), "z")
                # 5 recent 'web' lines (40 bytes each = 200 > cap 100) → keep 3.
                for i in range(1, 6):
                    await ins("web", now - timedelta(minutes=6 - i), line40)
                # 2 recent 'db' lines (80 bytes ≤ cap) → both survive (isolation).
                await ins("db", now - timedelta(minutes=2), line40)
                await ins("db", now - timedelta(minutes=1), line40)
                # 'web' on the other host: 2 lines, under the cap → both survive.
                await ins("web", now - timedelta(minutes=2), line40, host=hid2)
                await ins("web", now - timedelta(minutes=1), line40, host=hid2)

            async with s.begin():
                counts = await prune_logs(
                    s, retention_days=3, max_bytes_per_container=100, now=now)

            def n(cn, host=None):
                return s.execute(text(
                    "SELECT COUNT(*) FROM docker_logs WHERE host_id=:h AND container_name=:cn"),
                    {"h": host or hid, "cn": cn})
            web = (await n("web")).scalar()
            db = (await n("db")).scalar()
            old = (await n("old")).scalar()
            web2 = (await n("web", hid2)).scalar()
            return counts, web, db, old, web2

    counts, web, db, old, web2 = asyncio.run(_go())
    assert old == 0                       # age window
    assert web == 3                       # newest ~100 bytes kept, oldest 2 dropped
    assert db == 2                        # under cap → untouched (per-container)
    assert web2 == 2                      # same name, other host → its own ring
    assert counts["logs_age_pruned"] == 1
    assert counts["logs_size_pruned"] == 2


def test_widget_dedup_collapses_cross_manager_duplicates():
    """The same swarm task is reported by every manager (identical container_id);
    the dashboard widget must count it once. Older agents send no container_id,
    so those rows can't be matched and are all kept."""
    from types import SimpleNamespace
    from plugins.docker.dedup import dedup_by_container_id

    def c(cid, name, status="running"):
        return SimpleNamespace(container_id=cid, name=name, status=status)

    rows = [c("abc", "web@m1"), c("abc", "web@m2"),  # same task, two managers
            c("def", "db"), c("", "noid-1"), c("", "noid-2")]
    out = dedup_by_container_id(rows)

    assert len(out) == 4                                       # one "abc" dropped
    abc = [r for r in out if r.container_id == "abc"]
    assert len(abc) == 1 and abc[0].name == "web@m1"           # first occurrence wins
    assert sum(1 for r in out if r.container_id == "") == 2    # id-less rows all kept