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 os
import uuid
from datetime import datetime, 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:
            # Both tables carry a host_id column after docker_002.
            for tbl in ("docker_containers", "docker_metrics"):
                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())


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


@_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_metrics"))
                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()
            metric_hosts = (await s.execute(text(
                "SELECT COUNT(DISTINCT host_id) FROM docker_metrics "
                "WHERE container_name = 'web'"))).scalar()
            # Alert pipeline received host-scoped resource names.
            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_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_metrics"))
                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 mem_usage_bytes FROM docker_metrics "
                "WHERE container_name = 'bigmem'"))).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_metrics"))
                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_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}]}],
        "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.
                await persist(s, h, [], swarm)
                hid = h.id
            svc = (await s.execute(text(
                "SELECT mode, desired, running, image, placement_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

    svc, node = asyncio.run(_go())
    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
    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"))
                h = Host(id=str(uuid.uuid4()), name="dsk", 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()
            return summary, img_count, [r[0] for r in after], reclaimable2

    summary, img_count, after_ids, reclaimable2 = asyncio.run(_go())
    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_rollup_and_prune(app):
    """Old raw metrics roll up to hourly averages then delete; stale rollup +
    events prune; recent rows survive."""
    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)
    # One old hour (10 days back) with three samples → one rolled-up bucket.
    old_hour = datetime(2026, 6, 9, 9, 0, 0, tzinfo=timezone.utc)
    recent = datetime(2026, 6, 19, 11, 0, 0, tzinfo=timezone.utc)  # inside raw window
    ancient_bucket = datetime(2026, 3, 1, 0, 0, 0, tzinfo=timezone.utc)  # > rollup window
    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_metrics_hourly"))
                await s.execute(text("DELETE FROM docker_metrics"))
                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
                # Three raw samples in the old hour: cpu 10/20/30, mem 40/50/60.
                for i, (ts, cpu, mem, usage) in enumerate([
                    (old_hour, 10.0, 40.0, 100),
                    (old_hour.replace(second=30), 20.0, 50.0, 200),
                    (old_hour.replace(minute=1), 30.0, 60.0, 300),
                ]):
                    await s.execute(text(
                        "INSERT INTO docker_metrics "
                        "(id, host_id, container_name, scraped_at, cpu_pct, mem_pct, mem_usage_bytes) "
                        "VALUES (:id,:h,'web',:ts,:cpu,:mem,:usage)"),
                        {"id": str(uuid.uuid4()), "h": hid, "ts": ts,
                         "cpu": cpu, "mem": mem, "usage": usage})
                # A recent sample (within the 7-day raw window) — must survive raw.
                await s.execute(text(
                    "INSERT INTO docker_metrics "
                    "(id, host_id, container_name, scraped_at, cpu_pct, mem_pct, mem_usage_bytes) "
                    "VALUES (:id,:h,'web',:ts,5.0,5.0,50)"),
                    {"id": str(uuid.uuid4()), "h": hid, "ts": recent})
                # A pre-existing rollup row older than the 90-day rollup window.
                await s.execute(text(
                    "INSERT INTO docker_metrics_hourly "
                    "(id, host_id, container_name, bucket, cpu_pct, mem_pct, mem_usage_bytes, sample_count) "
                    "VALUES (:id,:h,'ancient',:b,1,1,1,1)"),
                    {"id": str(uuid.uuid4()), "h": hid, "b": ancient_bucket})
                # Events either side of the 30-day events window.
                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, metrics_raw_days=7,
                    metrics_rollup_days=90, now=now,
                )

            raw_left = (await s.execute(text(
                "SELECT COUNT(*) FROM docker_metrics WHERE host_id=:h"), {"h": hid})).scalar()
            bucket = (await s.execute(text(
                "SELECT cpu_pct, mem_pct, mem_usage_bytes, sample_count "
                "FROM docker_metrics_hourly WHERE host_id=:h AND container_name='web'"),
                {"h": hid})).first()
            ancient_left = (await s.execute(text(
                "SELECT COUNT(*) FROM docker_metrics_hourly "
                "WHERE host_id=:h AND container_name='ancient'"), {"h": hid})).scalar()
            events_left = (await s.execute(text(
                "SELECT COUNT(*) FROM docker_events WHERE host_id=:h"), {"h": hid})).scalar()
            return counts, raw_left, bucket, ancient_left, events_left

    counts, raw_left, bucket, ancient_left, events_left = asyncio.run(_go())
    assert raw_left == 1                       # only the recent sample survives raw
    assert bucket is not None
    assert bucket[0] == 20.0 and bucket[1] == 50.0   # avg cpu / mem over the 3 samples
    assert bucket[2] == 200 and bucket[3] == 3       # avg usage + sample_count
    assert ancient_left == 0                   # stale rollup pruned
    assert events_left == 1                    # only the in-window event survives
    assert counts["buckets_rolled"] == 1 and counts["raw_rows_rolled"] == 3
    assert counts["events_pruned"] == 1 and counts["rollup_pruned"] == 1