FabledSteward/tests/plugins/docker/test_ingest_events.py

"""Unit tests for the docker plugin's lifecycle-event derivation.

_derive_events is a pure function (no DB), and ingest.py imports its models
lazily inside functions, so importing it here doesn't register ORM tables —
safe for the no-DB unit lane.
"""
import pytest

from plugins.docker.ingest import _derive_events


def _c(name, status, **kw):
    return {"name": name, "status": status, **kw}


def test_start_on_running_after_stopped():
    old = {"web": {"status": "exited", "health": None, "oom_killed": False, "exit_code": 0}}
    events = _derive_events(old, [_c("web", "running", image="nginx")])
    assert events == [("web", "start", "nginx")]


def test_start_for_new_container_when_state_exists():
    # old_state is non-empty (host already seen), a fresh container appears running.
    old = {"other": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
    events = _derive_events(old, [
        _c("other", "running"),
        _c("new", "running", image="redis"),
    ])
    assert ("new", "start", "redis") in events
    # the unchanged 'other' produces nothing
    assert all(e[0] != "other" for e in events)


def test_stop_on_clean_exit():
    old = {"job": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
    events = _derive_events(old, [_c("job", "exited", exit_code=0)])
    assert events == [("job", "stop", None)]


def test_die_on_nonzero_exit():
    old = {"job": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
    events = _derive_events(old, [_c("job", "exited", exit_code=137)])
    assert events == [("job", "die", "exit 137")]


def test_oom_on_flip_to_true():
    old = {"hog": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
    # OOM-killed → also no longer running with a non-zero exit, so die + oom both fire.
    events = _derive_events(old, [_c("hog", "exited", exit_code=137, oom_killed=True)])
    assert ("hog", "oom", None) in events
    assert ("hog", "die", "exit 137") in events


def test_health_change_emitted():
    old = {"svc": {"status": "running", "health": "healthy", "oom_killed": False, "exit_code": None}}
    events = _derive_events(old, [_c("svc", "running", health="unhealthy")])
    assert events == [("svc", "health_change", "healthy→unhealthy")]


def test_removed_running_container_stops():
    old = {"gone": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
    events = _derive_events(old, [])  # vanished from the listing entirely
    assert events == [("gone", "stop", "removed")]


def test_no_event_when_unchanged():
    old = {"web": {"status": "running", "health": "healthy", "oom_killed": False, "exit_code": None}}
    events = _derive_events(old, [_c("web", "running", health="healthy")])
    assert events == []


# ── container-log row shaping (m79, pure) ─────────────────────────────────────

@pytest.mark.parametrize("old_status, old_n, new, events", [
    # Docker restarted it between samples: a restart, not silence (#5153).
    ("running", 2, _c("web", "running", restart_count=3), [("web", "restart", "restart #3")]),
    # Caught restarting, now back up: a restart, said instead of a start.
    ("restarting", 2, _c("web", "running", restart_count=3), [("web", "restart", "restart #3")]),
    ("running", 2, _c("web", "running", restart_count=5), [("web", "restart", "3 restarts, now #5")]),
    # Restarted, then died for good: both happened.
    ("running", 3, _c("web", "exited", restart_count=4, exit_code=1),
     [("web", "restart", "restart #4"), ("web", "die", "exit 1")]),
    # Started by hand (the count did not move): a start.
    ("exited", 2, _c("web", "running", restart_count=2, image="nginx"), [("web", "start", "nginx")]),
    # Recreated under the same name, so the count reset: no restart.
    ("running", 5, _c("web", "running", restart_count=0), []),
    ("running", 2, _c("web", "running", restart_count=2), []),
])
def test_restart_events(old_status, old_n, new, events):
    old = {"web": {"status": old_status, "health": None, "oom_killed": False, "exit_code": None,
                   "restart_count": old_n}}
    assert _derive_events(old, [new]) == events


def test_log_rows_shapes_and_filters():
    from datetime import datetime, timezone
    from plugins.docker.ingest import _log_rows

    rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
    batches = [(rec_at, [
        {"container": "web", "stream": "stdout",
         "ts": "2026-01-01T00:00:01+00:00", "line": "hello"},
        {"container": "web", "stream": "stderr", "ts": None, "line": "no-ts"},
        {"container": "_steward", "stream": "stderr",
         "ts": None, "line": "truncated"},          # advisory marker → dropped
        {"container": "web", "stream": "weird", "ts": None, "line": "bad-stream"},
        {"container": "", "line": "x"},              # no name → dropped
        {"container": "web", "ts": None, "line": None},   # no line → dropped
        "not-a-dict",                                # malformed → dropped
    ])]
    rows = list(_log_rows(batches, "host-1"))
    assert [r["line"] for r in rows] == ["hello", "no-ts", "bad-stream"]
    assert rows[0]["host_id"] == "host-1" and rows[0]["container_name"] == "web"
    assert rows[1]["ts"] == rec_at            # None ts → recorded_at fallback
    assert rows[2]["stream"] == "stdout"      # unknown stream normalised


def test_log_rows_skips_non_list_records():
    from datetime import datetime, timezone
    from plugins.docker.ingest import _log_rows
    rec_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
    assert list(_log_rows([(rec_at, None)], "h")) == []