bvandeusenandClaude Opus 5.5
326a9cd254
Agent 1.11.0 sends finished_at (State.FinishedAt) and reads a stopped container's logs once per stop: since its cursor if it was being read, or its last 200 lines when first seen already crashed (non-zero exit, stopped in the last 10 min). Cursors and finals are forgotten once a container is no longer listed. Ingest gives a container first seen already exited non-zero a die event, and dates die/stop events by finished_at when the agent sends it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
139 lines
6.5 KiB
Python
139 lines
6.5 KiB
Python
"""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")]
|
|
|
|
|
|
@pytest.mark.parametrize("status,exit_code,events", [
|
|
("exited", 1, [("codex_db.1.x", "die", "exit 1")]), # crashed between samples
|
|
("dead", 137, [("codex_db.1.x", "die", "exit 137")]),
|
|
("exited", 0, []), # a one-shot that finished
|
|
("exited", None, []), # inspect failed: no claim
|
|
("created", 1, []), # never ran
|
|
])
|
|
def test_new_container_first_seen_down(status, exit_code, events):
|
|
# #5049: a task that starts and exits between two samples is first seen
|
|
# already down; a non-zero exit is the death the samples missed.
|
|
old = {"other": {"status": "running", "health": None, "oom_killed": False, "exit_code": None}}
|
|
got = _derive_events(old, [_c("other", "running"),
|
|
_c("codex_db.1.x", status, exit_code=exit_code)])
|
|
assert got == events
|
|
|
|
|
|
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")) == []
|