#!/usr/bin/env python3 """ Test suite for deadman.py. Plain asserts, no pytest needed: python3 test_deadman.py # prints "deadman rules: N passed, M failed, K assertions" python3 -m pytest test_deadman.py -q # works too, if pytest is installed Every test drives the real decision path (deadman.main) through a fake HTTP layer, so nothing here opens a socket. The fake answers with the same JSON shape n8n's public API returns, and can raise the same urllib errors, which is how the probe-fault tests get to be honest tests rather than pretend ones. There is no `|| true` anywhere, no bare `except: pass`, and no assertion whose negation is equally true: each one names the value it wanted and the value it got. The suite was proved to be able to fail by planting three breaks into deadman.py (see README, "How this suite was proved to bite"). """ import contextlib import io import json import os import shutil import sys import tempfile import time import urllib.error import urllib.parse import deadman HERE = os.path.dirname(os.path.abspath(__file__)) WORK = tempfile.mkdtemp(prefix="deadman-tests-") PASSED = [] FAILED = [] ASSERTIONS = 0 # The moment the demo fixture and the documented demo command are built around. DEMO_NOW = "2026-11-13T12:00:00Z" # -------------------------------------------------------------------------- # tiny assertion helpers: each one counts itself, each one can fail # -------------------------------------------------------------------------- def ok(condition, label): global ASSERTIONS ASSERTIONS += 1 if not condition: raise AssertionError("expected %s" % label) def eq(actual, expected, label): global ASSERTIONS ASSERTIONS += 1 if actual != expected: raise AssertionError("%s: expected %r, got %r" % (label, expected, actual)) def contains(haystack, needle, label): global ASSERTIONS ASSERTIONS += 1 if needle not in haystack: raise AssertionError("%s: %r not found in output %r" % (label, needle, haystack)) def absent(haystack, needle, label): global ASSERTIONS ASSERTIONS += 1 if needle in haystack: raise AssertionError("%s: %r should NOT be in output %r" % (label, needle, haystack)) # -------------------------------------------------------------------------- # config and fixture builders # -------------------------------------------------------------------------- def write(name, text): path = os.path.join(WORK, name) with open(path, "w", encoding="utf-8") as handle: handle.write(text) return path def make_cfg(case, instance=True, cooldown=3600, timeout=5, probe_extra=(), workflows=(("wf_hourly", "Hourly sync", "1h", "10m"),), alert="", base_url="https://n8n.test.invalid"): lines = ["[probe]", 'state_file = "%s"' % os.path.join(WORK, case + ".state.json"), "cooldown_s = %d" % cooldown, "timeout_s = %d" % timeout] lines.extend(probe_extra) if instance: lines.extend(["", "[instance]"]) if base_url: lines.append('base_url = "%s"' % base_url) lines.append('api_key = "test-key-not-real"') for workflow_id, name, every, grace in workflows: lines.extend(["", "[[workflow]]", 'id = "%s"' % workflow_id, 'name = "%s"' % name, 'every = "%s"' % every, 'grace = "%s"' % grace]) if alert: lines.append(alert) return write(case + ".toml", "\n".join(lines) + "\n") def state_path(case): return os.path.join(WORK, case + ".state.json") def run_case(case, now, opener=None, simulate=None, cfg_path=None): """One poll of the real program. Returns (exit code, stdout, stderr).""" argv = ["--config", cfg_path or make_cfg(case), "--state", state_path(case)] if now: argv += ["--now", now] if simulate: argv += ["--simulate", simulate] out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stderr(err): code = deadman.main(argv, stdout=out, opener=opener) return code, out.getvalue(), err.getvalue() def read_state(case): with open(state_path(case), "r", encoding="utf-8") as handle: return json.load(handle) def row(execution_id, workflow_id="wf_hourly", status="success", started="2026-11-13T09:55:00Z"): """A row shaped like one item of n8n's `data` array (fields verified in deadman.py).""" return {"id": str(execution_id), "workflowId": workflow_id, "status": status, "startedAt": started, "stoppedAt": None, "finished": status in ("success", "error"), "mode": "trigger", "retryOf": None, "retrySuccessId": None, "waitTill": None} class FakeResponse: def __init__(self, body, status=200): self._body = body if isinstance(body, bytes) else body.encode("utf-8") self.status = status def read(self): return self._body def __enter__(self): return self def __exit__(self, *exc): return False class FakeHTTP: """Stands in for urllib.request.urlopen: answers GETs from `pages`, records POSTs. `fault` is an exception instance raised instead of answering, which is how a dead instance and a 401 are tested without a network. """ def __init__(self, pages=None, fault=None, first_cursor=None, post_status=200): self.pages = pages or {} self.fault = fault self.first_cursor = first_cursor self.post_status = post_status self.gets = [] self.posts = [] def __call__(self, request, timeout=None): if self.fault is not None: raise self.fault method = request.get_method() url = request.full_url headers = {} for key, value in request.header_items(): headers[str(key).lower()] = value parsed = urllib.parse.urlsplit(url) if method == "POST": self.posts.append((url, json.loads(request.data.decode("utf-8")), headers)) return FakeResponse(b'{"ok":true}', self.post_status) query = urllib.parse.parse_qs(parsed.query) self.gets.append((parsed.path, query, headers, url)) cursor = self.first_cursor if (self.first_cursor and "cursor" not in query) else None body = json.dumps({"data": self.pages.get(query.get("workflowId", [None])[0], []), "nextCursor": cursor}) return FakeResponse(body) def http_error(code): return urllib.error.HTTPError("https://n8n.test.invalid/api/v1/executions", code, "HTTP %d" % code, {}, None) def down_error(): return urllib.error.URLError("[Errno 111] Connection refused") def make_fixture(name, executions, fault=None): return write(name + ".fixture.json", json.dumps({"fault": fault, "executions": executions}, indent=2)) # ========================================================================== # 1. the healthy path # ========================================================================== def test_healthy_run_exits_zero_and_says_healthy(): http = FakeHTTP({"wf_hourly": [row(100, status="success", started="2026-11-13T09:55:00Z")]}) code, out, err = run_case("healthy", "2026-11-13T10:00:00Z", opener=http) eq(code, 0, "healthy exit code") contains(out, "healthy", "healthy verdict line") absent(out, "ALERT", "no alert on a healthy run") def test_healthy_run_records_state_the_readme_promises(): case = "healthy" http = FakeHTTP({case and "wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]}) code, out, err = run_case(case, "2026-11-13T10:00:00Z", opener=http) state = read_state(case) eq(state["last_check_ok"], "2026-11-13T10:00:00Z", "last successful check time") eq(state["open_alerts"], [], "no alerts open") memory = state["workflows"]["wf_hourly"] eq(memory["last_seen_id"], "100", "last-seen execution id per workflow") eq(memory["last_status"], "success", "last-seen status") eq(memory["last_seen_at"], "2026-11-13T09:55:00Z", "last-seen execution time") def test_api_call_is_the_shape_n8n_documents(): http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]}) run_case("shape", "2026-11-13T10:00:00Z", opener=http) eq(len(http.gets), 1, "exactly one API call per workflow per poll") path, query, headers, url = http.gets[0] eq(path, "/api/v1/executions", "the endpoint path") eq(query["workflowId"], ["wf_hourly"], "workflowId query param") eq(query["limit"], ["250"], "page size at n8n's documented maximum") eq(query["startedAfter"], ["2026-11-12T10:00:00Z"], "startedAfter is the EARLIER of the silence window (08:50) and the 24h error lookback") ok(query["startedAfter"][0].endswith("Z"), "startedAfter carries Z, not a local zone") absent(query, "includeData", "execution payloads are not asked for") eq(headers.get("x-n8n-api-key"), "test-key-not-real", "X-N8N-API-KEY header is sent") absent(url, "test-key-not-real", "the key never appears in the URL") def test_undeclared_workflows_are_never_asked_about(): http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")], "wf_secret": [row(7, status="error", started="2026-11-13T09:59:00Z")]}) code, out, err = run_case("scope", "2026-11-13T10:00:00Z", opener=http) eq(code, 0, "exit code when only an undeclared workflow is broken") absent(out, "wf_secret", "a workflow nobody declared is not reported") absent(str([g[0] for g in http.gets]) + str([g[1].get("workflowId") for g in http.gets]), "wf_secret", "deadman never queried the undeclared workflow") def test_pagination_follows_n8ns_own_cursor(): http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]}, first_cursor="CURSOR_PAGE_2") code, out, err = run_case("paging", "2026-11-13T10:00:00Z", opener=http) eq(len(http.gets), 2, "a second page is fetched when nextCursor is not null") ok("cursor" in http.gets[1][1], "the second request carries the cursor param") eq(http.gets[1][1]["cursor"], ["CURSOR_PAGE_2"], "the cursor value is the one n8n returned") eq(code, 0, "paging does not change the verdict") # ========================================================================== # 2. SILENCE — the class an n8n error workflow cannot see # ========================================================================== def test_first_silence_alerts_once(): http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]}) code, out, err = run_case("silence", "2026-11-13T10:00:00Z", opener=http) eq(code, 1, "silence exit code") contains(out, "ALERT silence:wf_hourly", "the silence alert") contains(out, "no execution in 4200s", "the window is stated in seconds") contains(out, "expected every 3600s + 600s grace", "cadence and grace are named") def test_same_silence_does_not_realert_on_the_next_poll(): case = "silence_dedupe" rows = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} code1, out1, _ = run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows)) code2, out2, _ = run_case(case, "2026-11-13T10:05:00Z", opener=FakeHTTP(rows)) eq(code1, 1, "first poll alerts") eq(code2, 1, "second poll still reports the alert as open") contains(out1, "ALERT", "first poll sent the alert") absent(out2, "ALERT", "second poll sent nothing new") contains(out2, "unchanged", "second poll says the alert is still open") absent(out2, "healthy", "a poll with an open alert is never called healthy") def test_realert_happens_exactly_at_the_cooldown_boundary(): case = "silence_cooldown" rows = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows)) code, out, _ = run_case(case, "2026-11-13T11:00:00Z", opener=FakeHTTP(rows)) eq(code, 1, "still alerting") contains(out, "RE-ALERT silence:wf_hourly", "re-alert at exactly cooldown_s (3600s)") def test_realert_does_not_happen_one_second_before_the_cooldown(): case = "silence_cooldown_early" rows = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows)) code, out, _ = run_case(case, "2026-11-13T10:59:59Z", opener=FakeHTTP(rows)) eq(code, 1, "still open") absent(out, "RE-ALERT", "no re-alert 1s before the cooldown elapsed") def test_cooldown_zero_never_nags(): case = "cooldown_zero" rows = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} cfg_path = make_cfg(case, cooldown=0) run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows), cfg_path=cfg_path) code, out, _ = run_case(case, "2026-11-14T10:00:00Z", opener=FakeHTTP(rows), cfg_path=cfg_path) eq(code, 1, "the alert is still open") absent(out, "RE-ALERT", "cooldown_s = 0 means alert once per episode") def test_recovery_notice_exactly_once_when_it_starts_running_again(): case = "silence_recovery" dead = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} alive = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z"), row(101, started="2026-11-13T10:30:00Z")]} run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(dead)) code, out, _ = run_case(case, "2026-11-13T10:35:00Z", opener=FakeHTTP(alive)) eq(code, 0, "healthy again") contains(out, "RECOVERED silence:wf_hourly", "the recovery notice") code2, out2, _ = run_case(case, "2026-11-13T10:40:00Z", opener=FakeHTTP(alive)) eq(code2, 0, "still healthy") absent(out2, "RECOVERED", "recovery is said once, never again") def test_a_queued_execution_that_never_started_is_not_a_run(): # n8n documents startedAt as null until the execution leaves the queue. http = FakeHTTP({"wf_hourly": [row(100, status="new", started=None)]}) code, out, err = run_case("queued", "2026-11-13T10:00:00Z", opener=http) eq(code, 1, "a workflow that never left the queue is silent") contains(out, "ALERT silence:wf_hourly", "silence alert for a queued-but-unstarted run") def test_silence_window_boundary_is_inclusive_then_exclusive(): # every 20m + 5m grace = a 1500s window, so with now at 00:05:00 the edge is # 23:40:00 on the PREVIOUS UTC day exactly. The window has to be stated in # the test, otherwise the assertion passes for reasons other than the one it # names: with the 1h+10m default, a 23:40 row is healthy no matter what. short = (("wf_hourly", "Twenty min", "20m", "5m"),) case = "silence_boundary" cfg = make_cfg(case, workflows=short) on_line = {"wf_hourly": [row(100, started="2026-11-01T23:40:00Z")]} code, out, _ = run_case(case, "2026-11-02T00:05:00Z", opener=FakeHTTP(on_line), cfg_path=cfg) eq(code, 0, "an execution exactly at the window edge counts as having run") absent(out, "ALERT", "no alert at the boundary") case2 = "silence_boundary_over" cfg2 = make_cfg(case2, workflows=short) over = {"wf_hourly": [row(100, started="2026-11-01T23:39:59Z")]} code2, out2, _ = run_case(case2, "2026-11-02T00:05:00Z", opener=FakeHTTP(over), cfg_path=cfg2) eq(code2, 1, "one second past the window is silence") contains(out2, "ALERT silence:wf_hourly", "the alert 1s over the edge") def test_utc_day_rollover_does_not_confuse_the_window(): # window crosses midnight UTC backwards: 00:05 minus 25 minutes = previous day 23:40. http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-01T23:50:00Z")]}) code, out, _ = run_case("rollover", "2026-11-02T00:05:00Z", opener=http) eq(code, 0, "a run at 23:50 satisfies a window that starts the previous day") absent(out, "ALERT", "no false alert across the UTC day boundary") def test_cadence_math_is_right_across_the_fall_back_dst_boundary(): """US autumn 2026: local clocks go BACK at 05:00Z on Nov 1. 05:30Z and 06:30Z are the SAME local wall-clock minute (01:30 EDT, then 01:30 EST). An implementation that subtracted local wall-clock times would see zero elapsed and report this healthy workflow as fine. Real elapsed time is 3600s against a 2100s window, so the honest answer is SILENT. """ case = "dst_fall" previous_tz = os.environ.get("TZ") os.environ["TZ"] = "America/Toronto" time.tzset() try: http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-01T05:30:00Z")]}) cfg_path = make_cfg(case, workflows=(("wf_hourly", "Half hourly", "30m", "5m"),)) code, out, _ = run_case(case, "2026-11-01T06:30:00Z", opener=http, cfg_path=cfg_path) eq(code, 1, "3600s elapsed against a 2100s window is silent, DST or not") contains(out, "ALERT silence:wf_hourly", "the real silence is not hidden by the clock change") # A week later both instants are on the same offset; the verdict must not move. code2, out2, _ = run_case("dst_fall_control", "2026-11-08T06:30:00Z", opener=FakeHTTP({"wf_hourly": [row(101, started="2026-11-08T05:30:00Z")]}), cfg_path=make_cfg("dst_fall_control", workflows=(("wf_hourly", "Half hourly", "30m", "5m"),))) eq(code2, code, "the same 60-minute gap gives the same verdict outside the transition") finally: if previous_tz is None: os.environ.pop("TZ", None) else: os.environ["TZ"] = previous_tz time.tzset() def test_cadence_math_invents_no_silence_across_the_spring_forward_boundary(): """US spring 2026: local clocks jump FORWARD at 07:00Z on Mar 8. 05:59Z is 00:59 EST and 07:01Z is 03:01 EDT, so local wall-clock subtraction reads 122 minutes and would raise a FALSE alert against a 90-minute window. True elapsed is 3720s, which is inside the window: the honest answer is healthy. """ previous_tz = os.environ.get("TZ") os.environ["TZ"] = "America/Toronto" time.tzset() try: case = "dst_spring" cfg_path = make_cfg(case, workflows=(("wf_hourly", "Ninety min", "70m", "20m"),)) http = FakeHTTP({"wf_hourly": [row(100, started="2026-03-08T05:59:00Z")]}) code, out, _ = run_case(case, "2026-03-08T07:01:00Z", opener=http, cfg_path=cfg_path) eq(code, 0, "3720s against a 5400s window is healthy, DST jump or not") absent(out, "ALERT", "no false alert invented by the clock jump") finally: if previous_tz is None: os.environ.pop("TZ", None) else: os.environ["TZ"] = previous_tz time.tzset() # ========================================================================== # 3. ERROR — a new failed execution # ========================================================================== def test_new_failed_execution_alerts(): http = FakeHTTP({"wf_hourly": [row(100, status="error", started="2026-11-13T09:55:00Z")]}) code, out, _ = run_case("error_new", "2026-11-13T10:00:00Z", opener=http) eq(code, 1, "a failure alerts") contains(out, "ALERT error:wf_hourly", "the error alert") contains(out, "1 new failed execution(s): 100", "the offending execution id is in the alert") def test_crashed_counts_as_a_failure_and_canceled_does_not(): crashed = FakeHTTP({"wf_hourly": [row(100, status="crashed", started="2026-11-13T09:55:00Z")]}) code, out, _ = run_case("err_crashed", "2026-11-13T10:00:00Z", opener=crashed) eq(code, 1, "'crashed' is a failure") contains(out, "ALERT error:wf_hourly", "the crash is alerted") canceled = FakeHTTP({"wf_hourly": [row(101, status="canceled", started="2026-11-13T09:55:00Z")]}) code2, out2, _ = run_case("err_canceled", "2026-11-13T10:00:00Z", opener=canceled) eq(code2, 0, "a deliberately canceled run is not a failure") absent(out2, "ALERT", "no error alert for a cancellation") def test_fail_statuses_override_is_honoured(): case = "fail_statuses" cfg_path = make_cfg(case, probe_extra=['fail_statuses = ["error"]']) http = FakeHTTP({"wf_hourly": [row(100, status="crashed", started="2026-11-13T09:55:00Z")]}) code, out, _ = run_case(case, "2026-11-13T10:00:00Z", opener=http, cfg_path=cfg_path) eq(code, 0, "with fail_statuses = [error] a crash is not alerted") absent(out, "ALERT", "no alert") def test_the_same_failure_is_not_alerted_twice(): case = "error_dedupe" rows = {"wf_hourly": [row(100, status="error", started="2026-11-13T09:55:00Z")]} code1, out1, _ = run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows)) code2, out2, _ = run_case(case, "2026-11-13T10:05:00Z", opener=FakeHTTP(rows)) eq(code1, 1, "first poll alerts") contains(out1, "ALERT error:wf_hourly", "the alert went out once") eq(code2, 1, "the episode is still open") absent(out2, "ALERT", "the same failed id is not re-reported") eq(read_state(case)["workflows"]["wf_hourly"]["seen_error_ids"], ["100"], "the failed id is remembered") def test_no_false_recovery_while_the_last_run_is_still_the_failure(): """The lie this test exists to prevent. Nothing NEW failed since the last poll, but nothing SUCCEEDED either: the workflow's newest execution is still the failure. Reporting that as clear would tell someone their pipeline is fine when it is not. """ case = "error_no_lie" rows = {"wf_hourly": [row(100, status="error", started="2026-11-13T09:55:00Z")]} run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows)) code, out, _ = run_case(case, "2026-11-13T10:10:00Z", opener=FakeHTTP(rows)) eq(code, 1, "still broken, so still exit 1") absent(out, "RECOVERED", "no recovery notice without a newer success") contains(out, "unchanged", "it says the alert is unchanged instead") def test_error_recovers_only_after_a_newer_success_and_then_silently(): case = "error_recovery" broken = {"wf_hourly": [row(100, status="error", started="2026-11-13T09:55:00Z")]} fixed = {"wf_hourly": [row(100, status="error", started="2026-11-13T09:55:00Z"), row(101, status="success", started="2026-11-13T10:15:00Z")]} run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(broken)) code, out, _ = run_case(case, "2026-11-13T10:20:00Z", opener=FakeHTTP(fixed)) eq(code, 0, "a newer success clears it") contains(out, "RECOVERED error:wf_hourly", "the recovery notice") code2, out2, _ = run_case(case, "2026-11-13T10:25:00Z", opener=FakeHTTP(fixed)) eq(code2, 0, "still healthy") absent(out2, "RECOVERED", "recovery is said exactly once") def test_failure_older_than_the_lookback_is_not_reported_as_new(): case = "error_lookback" cfg_path = make_cfg(case, probe_extra=["error_lookback_s = 600"]) # 09:20 sits INSIDE the 1h+10m silence window, so silence is not what this # test is proving; it is OUTSIDE the 600s error lookback, so exactly one # rule is under test here. rows = {"wf_hourly": [row(100, status="error", started="2026-11-13T09:20:00Z")]} code, out, _ = run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows), cfg_path=cfg_path) eq(code, 0, "a failure beyond the lookback is not reported as new") absent(out, "ALERT error", "no error alert for it") absent(out, "ALERT silence", "and it is not silent either: it ran 40 minutes ago") def test_the_lookback_reaches_back_past_a_short_silence_window(): """The bug this test exists to keep dead. The query n8n is given must start at the EARLIER of the silence window and the error lookback. Asking from the later one made a 24h lookback invisible past a 1h cadence, because the request never reached that far back. """ case = "lookback_reaches" cfg_path = make_cfg(case, probe_extra=["error_lookback_s = 86400"]) http = FakeHTTP({"wf_hourly": [row(200, status="success", started="2026-11-13T09:55:00Z"), row(199, status="error", started="2026-11-13T02:00:00Z")]}) code, out, _ = run_case(case, "2026-11-13T10:00:00Z", opener=http, cfg_path=cfg_path) eq(http.gets[0][1]["startedAfter"], ["2026-11-12T10:00:00Z"], "the request reaches back the full 24h lookback, not just the 1h window") eq(code, 1, "the 8-hour-old failure is inside the lookback, so it is reported") contains(out, "ALERT error:wf_hourly", "the old failure is alerted") contains(out, "199", "and its execution id is named") # ========================================================================== # 4. PROBE-FAULT — the class that must never masquerade as silence # ========================================================================== def test_instance_down_is_probe_fault_and_not_silence(): http = FakeHTTP(fault=down_error()) code, out, err = run_case("down", "2026-11-13T10:00:00Z", opener=http) eq(code, 2, "unreachable instance is exit 2, not 1") contains(out, "PROBE-FAULT", "it is named as a probe fault") contains(err, "workflow verdicts are NOT computed", "the refusal to judge is said out loud") absent(out, "silence", "an unreachable instance is never reported as silent workflows") def test_401_is_probe_fault_and_not_silence(): http = FakeHTTP(fault=http_error(401)) code, out, err = run_case("auth401", "2026-11-13T10:00:00Z", opener=http) eq(code, 2, "a refused API key is exit 2") contains(out, "auth", "it is classified as auth") absent(out, "silence", "a rotated key is never reported as 47 dead workflows") contains(out, "ALERT probe", "the key problem itself is alerted once") def test_403_is_probe_fault_too(): code, out, err = run_case("auth403", "2026-11-13T10:00:00Z", opener=FakeHTTP(fault=http_error(403))) eq(code, 2, "403 is exit 2") contains(out, "auth", "403 is the auth class") def test_500_and_non_json_are_probe_faults(): code, out, err = run_case("http500", "2026-11-13T10:00:00Z", opener=FakeHTTP(fault=http_error(500))) eq(code, 2, "a server error is exit 2") contains(out, "HTTP 500", "the status is quoted") class Garbage: def __call__(self, request, timeout=None): return FakeResponse(b"not json at all") code2, out2, err2 = run_case("garbage", "2026-11-13T10:00:00Z", opener=Garbage()) eq(code2, 2, "a non-JSON reply is exit 2") contains(out2, "malformed", "it is named as a malformed reply") def test_probe_fault_leaves_every_workflow_verdict_untouched(): case = "fault_untouched" cfg_path = make_cfg(case, workflows=(("wf_a", "A", "1h", "10m"), ("wf_b", "B", "1h", "10m"))) code, out, err = run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(fault=down_error()), cfg_path=cfg_path) eq(code, 2, "exit 2") state = read_state(case) eq(state["open_alerts"], ["probe"], "only the probe episode is open") eq(sorted(state["workflows"]), ["wf_a", "wf_b"], "both declared workflows are recorded") for workflow_id in ("wf_a", "wf_b"): eq(state["workflows"][workflow_id]["last_seen_id"], None, "%s has no verdict recorded" % workflow_id) eq(state["last_fault"]["kind"], "unreachable", "the fault is stored in state") def test_probe_fault_does_not_resolve_alerts_that_were_already_open(): case = "fault_over_alert" rows = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} code1, _, _ = run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows)) eq(code1, 1, "a silence alert is open first") code2, out2, _ = run_case(case, "2026-11-13T10:05:00Z", opener=FakeHTTP(fault=down_error())) eq(code2, 2, "the probe fault takes precedence in the exit code") absent(out2, "RECOVERED", "losing sight of the instance does not 'recover' anything") eq(read_state(case)["open_alerts"], ["probe", "silence:wf_hourly"], "both episodes are still open") def test_probe_fault_recovers_exactly_once_when_the_key_works_again(): case = "fault_recovery" run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(fault=http_error(401))) code, out, _ = run_case(case, "2026-11-13T10:05:00Z", opener=FakeHTTP({"wf_hourly": [row(101, started="2026-11-13T10:04:00Z")]})) eq(code, 0, "healthy again") contains(out, "RECOVERED probe", "the probe recovery notice") code2, out2, _ = run_case(case, "2026-11-13T10:10:00Z", opener=FakeHTTP({"wf_hourly": [row(102, started="2026-11-13T10:09:00Z")]})) absent(out2, "RECOVERED", "said once, never again") def test_probe_fault_realerts_after_the_cooldown(): case = "fault_cooldown" run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(fault=down_error())) _, out_mid, _ = run_case(case, "2026-11-13T10:59:00Z", opener=FakeHTTP(fault=down_error())) absent(out_mid, "RE-ALERT", "no nag before the cooldown") _, out_after, _ = run_case(case, "2026-11-13T11:00:00Z", opener=FakeHTTP(fault=down_error())) contains(out_after, "RE-ALERT probe", "the still-dead instance is re-reported at the cooldown") # ========================================================================== # 5. config: every bad field gets its own clear message # ========================================================================== def test_each_bad_config_field_has_its_own_distinct_message(): # Validation runs root -> probe -> instance -> workflow -> alert, so a case # that tests the workflow or alert section needs the sections above it to be # valid, or the earlier error is what comes back and the rule under test is # never reached. `base` is that valid preamble. base = ('[probe]\nstate_file = "/tmp/x.json"\n' '[instance]\nbase_url = "https://a.test"\napi_key = "k"\n') # `probe` is the same preamble minus the instance, for the cases that test # the instance section itself. probe = '[probe]\nstate_file = "/tmp/x.json"\n' # `bw` is a whole valid document minus the alert section, for the cases that # test the alert section itself (workflow is validated before alert). bw = base + '[[workflow]]\nid = "a"\nevery = "1h"\n' cases = [ ("no file at all", None, "config file not found"), ("bad toml", "x = = 1\n", "not valid TOML"), ("unknown root section", "[bogus]\nk = 1\n", "unknown field 'bogus'"), ("no state file", "[[workflow]]\nid = \"a\"\nevery = \"1h\"\n", "'state_file' is required"), ("no workflows", "[probe]\nstate_file = \"/tmp/x.json\"\n" "[instance]\nbase_url = \"https://a.test\"\napi_key = \"k\"\n", "at least one [[workflow]]"), ("workflow without id", base + "[[workflow]]\nevery = \"1h\"\n", "is required"), ("workflow without every", base + "[[workflow]]\nid = \"a\"\n", "'every' is required"), ("unreadable cadence", base + "[[workflow]]\nid=\"a\"\nevery=\"1x\"\n", "unreadable duration"), ("empty cadence", base + "[[workflow]]\nid=\"a\"\nevery=\"\"\n", "duration is empty"), ("unreadable grace", base + "[[workflow]]\nid=\"a\"\nevery=\"1h\"\ngrace=\"soon\"\n", "'grace'"), ("zero cadence", base + "[[workflow]]\nid=\"a\"\nevery=0\n", "positive duration"), ("duplicate workflow id", base + "[[workflow]]\nid=\"dup\"\nevery=\"1h\"\n[[workflow]]\nid=\"dup\"\nevery=\"1h\"\n", "duplicate workflow id"), ("unknown workflow field", base + "[[workflow]]\nid=\"a\"\nevery=\"1h\"\ncronish=\"* * *\"\n", "unknown field 'cronish'"), ("base_url without scheme", probe + "[instance]\nbase_url = \"n8n.example.com\"\napi_key = \"k\"\n", "starting with http:// or https://"), ("missing base_url", probe + "[instance]\napi_key = \"k\"\n", "'base_url' is required"), ("missing api key", probe + "[instance]\nbase_url = \"https://a.test\"\n", "'api_key' is required"), ("both key forms set", probe + "[instance]\nbase_url = \"https://a.test\"\napi_key = \"k\"\napi_key_env = \"NOPE\"\n", "not both"), ("api_key_env not in environment", probe + "[instance]\nbase_url = \"https://a.test\"\napi_key_env = \"DEADMAN_ABSENT_KEY\"\n", "which is not set in this environment"), ("negative cooldown", "[probe]\nstate_file=\"/tmp/x\"\ncooldown_s = -5\n", "'cooldown_s' must be >= 0"), ("zero timeout", "[probe]\nstate_file=\"/tmp/x\"\ntimeout_s = 0\n", "'timeout_s' must be a positive"), ("bad max_pages", "[probe]\nstate_file=\"/tmp/x\"\nmax_pages = 0\n", "'max_pages' must be an integer"), ("fail_status not in n8n enum", "[probe]\nstate_file=\"/tmp/x\"\nfail_statuses = [\"exploded\"]\n", "is not an n8n execution status"), ("webhook without url", bw + "[[alert.webhook]]\nname = \"x\"\n", "'url' is required"), ("email without host", bw + "[[alert.email]]\nfrom = \"a@b.c\"\nto = [\"d@e.f\"]\n", "'host' is required"), ("email with empty to", bw + "[[alert.email]]\nhost=\"h\"\nfrom=\"a@b.c\"\nto=[]\n", "'to' must be a non-empty"), ("email with a non-address", bw + "[[alert.email]]\nhost=\"h\"\nfrom=\"a@b.c\"\nto=[\"not an address\"]\n", "which is not an address"), ("email with bad security", bw + "[[alert.email]]\nhost=\"h\"\nfrom=\"a@b.c\"\nto=[\"d@e.f\"]\nsecurity=\"magic\"\n", "'security' must be 'starttls'"), ("email with bad port", bw + "[[alert.email]]\nhost=\"h\"\nfrom=\"a@b.c\"\nto=[\"d@e.f\"]\nport=99999\n", "'port' must be an integer"), ] messages = [] for label, body, expected in cases: path = os.path.join(WORK, "bad-%s.toml" % abs(hash(label))) if body is not None: with open(path, "w", encoding="utf-8") as handle: handle.write(body) else: path = os.path.join(WORK, "definitely-not-here.toml") out, err = io.StringIO(), io.StringIO() argv = ["--config", path, "--state", os.path.join(WORK, "bad.state.json")] with contextlib.redirect_stderr(err): code = deadman.main(argv, stdout=out) eq(code, 2, "a bad config is exit 2 (%s)" % label) message = err.getvalue().strip() ok(message.startswith("deadman:"), "the message is prefixed (%s) got %r" % (label, message)) contains(message, expected, "the message for %r" % label) messages.append(message) eq(len(set(messages)), len(messages), "every failure produced its OWN message, none duplicated") def test_a_good_config_does_not_raise_and_names_its_fields(): case = "good_cfg" cfg_path = make_cfg(case, probe_extra=['error_lookback_s = 7200', 'max_pages = 5'], alert='[[alert.webhook]]\nurl = "https://hooks.test.invalid/x"\n') http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]}) code, out, err = run_case(case, "2026-11-13T10:00:00Z", opener=http, cfg_path=cfg_path) eq(code, 0, "a valid config runs") absent(err, "deadman:", "no error output") def test_naive_now_is_refused_rather_than_guessed(): out, err = io.StringIO(), io.StringIO() argv = ["--config", make_cfg("naive"), "--now", "2026-11-13T10:00:00", "--state", state_path("naive")] with contextlib.redirect_stderr(err): code = deadman.main(argv, stdout=out) eq(code, 2, "a zone-less --now is refused") contains(err.getvalue(), "has no UTC offset", "the refusal says why") def test_now_with_an_offset_is_converted_to_utc_before_it_is_used(): # 06:00-04:00 IS 10:00Z: the same instant must give the same verdict. rows = {"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]} code_z, out_z, _ = run_case("off_z", "2026-11-13T10:00:00Z", opener=FakeHTTP(rows), cfg_path=make_cfg("off_z")) code_off, out_off, _ = run_case("off_offset", "2026-11-13T06:00:00-04:00", opener=FakeHTTP(rows), cfg_path=make_cfg("off_offset")) eq(code_z, 0, "the UTC form is healthy") eq(code_off, 0, "the offset form is healthy") contains(out_off, "healthy", "same instant, same verdict, whatever the written form") def test_duration_forms_the_readme_promises_all_parse(): eq(deadman.parse_duration("45s", "t"), 45.0, "45s") eq(deadman.parse_duration("90m", "t"), 5400.0, "90m") eq(deadman.parse_duration("6h", "t"), 21600.0, "6h") eq(deadman.parse_duration("2d", "t"), 172800.0, "2d") eq(deadman.parse_duration("1h30m", "t"), 5400.0, "1h30m") eq(deadman.parse_duration("3600", "t"), 3600.0, "a bare number is seconds") eq(deadman.parse_duration(7200, "t"), 7200.0, "a bare TOML integer is seconds") # ========================================================================== # 6. the state file must never be able to stop the monitoring # ========================================================================== def test_a_corrupt_state_file_is_set_aside_and_the_check_still_runs(): case = "corrupt" path = state_path(case) with open(path, "w", encoding="utf-8") as handle: handle.write("{ this is not json at all,,,") cfg_path = make_cfg(case) http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]}) code, out, err = run_case(case, "2026-11-13T10:00:00Z", opener=http, cfg_path=cfg_path) eq(code, 0, "a corrupt state file does not stop the check") contains(err, "not valid JSON", "the corruption is named") contains(err, ".corrupt", "the operator is told where the bad copy went") ok(os.path.exists(path + ".corrupt"), "the unreadable copy is actually kept") ok(os.path.exists(path), "a fresh readable state file was written") def test_a_state_file_of_the_wrong_shape_is_recovered_from(): case = "wrong_shape" with open(state_path(case), "w", encoding="utf-8") as handle: handle.write("[1, 2, 3]") code, out, err = run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]})) eq(code, 0, "a list where an object belongs does not crash it") contains(err, "not the shape", "the shape problem is named") def test_a_state_file_from_another_version_is_not_trusted_silently(): case = "wrong_version" with open(state_path(case), "w", encoding="utf-8") as handle: json.dump({"version": 97, "alerts": {}, "workflows": {}}, handle) code, out, err = run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T09:55:00Z")]})) eq(code, 0, "an unknown state version still produces a verdict") contains(err, "version", "the version mismatch is said out loud") def test_state_survives_a_repeated_run_without_losing_what_is_open(): case = "state_persist" rows = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} run_case(case, "2026-11-13T10:00:00Z", opener=FakeHTTP(rows)) state = read_state(case) eq(state["open_alerts"], ["silence:wf_hourly"], "the open alert is on disk") eq(state["alerts"]["silence:wf_hourly"]["alert_count"], 1, "one notice sent for this episode") run_case(case, "2026-11-13T10:10:00Z", opener=FakeHTTP(rows)) eq(read_state(case)["alerts"]["silence:wf_hourly"]["alert_count"], 1, "the count did not move on a quiet poll") # ========================================================================== # 7. alert channels, both optional # ========================================================================== def test_it_runs_with_no_alert_channel_at_all(): http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]}) code, out, err = run_case("no_channel", "2026-11-13T10:00:00Z", opener=http) eq(code, 1, "the exit code carries the verdict with no channel configured") contains(out, "ALERT silence:wf_hourly", "stdout still says it") absent(err, "channel", "no channel is complained about when none is configured") def test_webhook_receives_the_notice_as_json(): case = "webhook" cfg_path = make_cfg(case, alert='[[alert.webhook]]\nurl = "https://hooks.test.invalid/slack"\n') http = FakeHTTP({"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]}) code, out, err = run_case(case, "2026-11-13T10:00:00Z", opener=http, cfg_path=cfg_path) eq(code, 1, "the alert is still open") eq(len(http.posts), 1, "exactly one webhook POST") url, body, headers = http.posts[0] eq(url, "https://hooks.test.invalid/slack", "posted to the configured url") eq(headers.get("content-type"), "application/json", "JSON content type") contains(body["text"], "ALERT silence:wf_hourly", "the text field a Slack webhook shows") eq(body["action"], "alert", "the machine-readable action") eq(body["key"], "silence:wf_hourly", "the machine-readable key") def test_a_broken_webhook_is_named_every_poll_and_does_not_spam_alerts(): case = "webhook_broken" cfg_path = make_cfg(case, alert='[[alert.webhook]]\nurl = "https://hooks.test.invalid/dead"\n') rows = {"wf_hourly": [row(100, started="2026-11-13T05:00:00Z")]} http = FakeHTTP(rows, post_status=500) code1, out1, err1 = run_case(case, "2026-11-13T10:00:00Z", opener=http, cfg_path=cfg_path) eq(code1, 1, "the alert is open regardless of the channel") contains(err1, "alert channel", "the broken channel is named") code2, out2, err2 = run_case(case, "2026-11-13T10:10:00Z", opener=http, cfg_path=cfg_path) eq(code2, 1, "still open") absent(out2, "RE-ALERT", "a dead webhook does not turn the alert into a per-poll loop") contains(err2, "alert channel", "but the broken channel keeps being reported") ok(read_state(case)["alerts"]["silence:wf_hourly"]["delivery_failed"] is True, "the episode records that delivery failed") def test_email_channel_is_asked_to_send_the_notice(): case = "email" cfg_path = make_cfg(case, alert=( '[[alert.email]]\nhost = "smtp.test.invalid"\nport = 587\n' 'from = "deadman@test.invalid"\nto = ["ops@test.invalid"]\nsecurity = "none"\n')) calls = [] def fake_mailer(*args): calls.append(args) config = deadman.load_config(cfg_path) config["probe"]["state_file"] = state_path(case) channel = deadman.EmailChannel(config["emails"][0], mailer=fake_mailer) notice = {"key": "silence:wf_hourly", "action": "alert", "at": "2026-11-13T10:00:00Z", "open_since": "2026-11-13T10:00:00Z", "detail": {"summary": "no execution in 4200s", "workflow": "Hourly sync"}} channel.send(notice) eq(len(calls), 1, "the mailer was called once") host, port, message, sender, recipients = calls[0][:5] eq(host, "smtp.test.invalid", "the configured host") eq(port, 587, "the configured port") eq(recipients, ["ops@test.invalid"], "the configured recipient") contains(message, "Subject: ALERT silence:wf_hourly", "the subject carries the alert") contains(message, "no execution in 4200s", "the body carries the reason") # ========================================================================== # 8. --simulate: the demo a stranger can run with nothing of their own # ========================================================================== def test_simulate_needs_no_instance_and_no_key_in_the_environment(): case = "sim" cfg_path = make_cfg(case, instance=False) fixture = make_fixture("sim", [row(100, started="2026-11-13T05:00:00Z")]) code, out, err = run_case(case, "2026-11-13T10:00:00Z", simulate=fixture, cfg_path=cfg_path) eq(code, 1, "the fixture's silence is alerted") contains(out, "ALERT silence:wf_hourly", "the alert came from the fixture, not an instance") def test_simulate_injects_a_probe_fault_without_any_real_instance(): case = "sim_fault" cfg_path = make_cfg(case, instance=False) fixture = make_fixture("sim_fault", [row(100)], fault={"kind": "auth", "detail": "HTTP 401"}) code, out, _ = run_case(case, "2026-11-13T10:00:00Z", simulate=fixture, cfg_path=cfg_path) eq(code, 2, "an injected auth fault is exit 2") contains(out, "HTTP 401", "the fixture's own detail is quoted") absent(out, "silence", "no workflow is judged while the probe is blind") def test_a_fixture_missing_a_required_n8n_field_is_refused_by_name(): case = "sim_bad" cfg_path = make_cfg(case, instance=False) fixture = write("sim_bad.json", json.dumps({"executions": [{"id": "1", "status": "error"}]})) code, out, err = run_case(case, "2026-11-13T10:00:00Z", simulate=fixture, cfg_path=cfg_path) eq(code, 2, "an unusable fixture is exit 2") contains(err, "workflowId", "the missing field is named") contains(err, "executions[0]", "and its position in the file") def test_the_shipped_demo_works_and_shows_two_failure_classes(): """The exact command on the public page, against the files that ship.""" case = "demo" state = os.path.join(WORK, "demo.state.json") argv = ["--config", os.path.join(HERE, "config.example.toml"), "--simulate", os.path.join(HERE, "demo_fixture.json"), "--now", DEMO_NOW, "--state", state] http = FakeHTTP({}) out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stderr(err): code = deadman.main(argv, stdout=out, opener=http) text = out.getvalue() eq(code, 1, "the demo exits 1: it is showing broken things") contains(text, "ALERT error:PLACEHOLDER_WORKFLOW_ID", "the ERROR class is demonstrated") contains(text, "ALERT silence:PLACEHOLDER_SECOND_WORKFLOW_ID", "the SILENCE class is demonstrated") absent(text, "NOT_DECLARED", "the fixture's undeclared workflow is left alone") out2, err2 = io.StringIO(), io.StringIO() argv2 = list(argv) argv2[argv2.index("--now") + 1] = "2026-11-13T12:05:00Z" with contextlib.redirect_stderr(err2): code2 = deadman.main(argv2, stdout=out2, opener=FakeHTTP({})) eq(code2, 1, "still open") absent(out2.getvalue(), "ALERT", "the second run sends nothing new") contains(out2.getvalue(), "unchanged", "and says what is still open instead") def test_the_shipped_example_config_is_valid(): # simulate=True because the example ships api_key_env pointing at a variable # that is, rightly, not set in a test environment. config = deadman.load_config(os.path.join(HERE, "config.example.toml"), simulate=True) eq(len(config["workflows"]), 2, "the example declares two workflows") eq(config["workflows"][0]["window_s"], 86400.0 + 1800.0, "1d + 30m grace") eq(config["workflows"][1]["window_s"], 900.0 + 300.0, "15m + 5m grace") eq(config["probe"]["fail_statuses"], ["error", "crashed"], "the documented default failure set") def test_the_help_text_documents_every_exit_code(): out = io.StringIO() with contextlib.redirect_stdout(out): try: deadman.main(["--help"]) except SystemExit as exc: eq(exc.code, 0, "--help exits 0") text = out.getvalue() for line in ("0 healthy", "1 one or more alerts open", "2 the probe could not do its job"): contains(text, line, "the exit-code documentation") # -------------------------------------------------------------------------- # runner: plain-python, and importable by pytest without running anything # -------------------------------------------------------------------------- def _run_all(): names = sorted(name for name in list(globals()) if name.startswith("test_")) for name in names: try: globals()[name]() PASSED.append(name) except Exception as exc: # recorded, never swallowed FAILED.append("%s -> %s: %s" % (name, type(exc).__name__, exc)) for line in FAILED: print("FAILED %s" % line) print("deadman rules: %d passed, %d failed, %d assertions" % (len(PASSED), len(FAILED), ASSERTIONS)) return 1 if FAILED else 0 if __name__ == "__main__": try: sys.exit(_run_all()) finally: shutil.rmtree(WORK, ignore_errors=True)