test(anaf): assert monitor_v2 emits GSTACK-CRON marker as last stdout line

Four checks:
- The script file exists at the expected path.
- The source contains the marker print statement (fast regression guard).
- Running the script against an empty config produces a matching marker
  (^GSTACK-CRON: changes=\d+$) with changes=0.
- The marker is the last non-empty line of stdout so tailers can parse it.

The runtime test copies the script into a tmp cwd so that the script's
SCRIPT_DIR-relative state files (hashes.json, versions.json, snapshots/,
monitor.log) don't pollute the repo.
This commit is contained in:
2026-04-21 07:05:46 +00:00
parent c82dbc5654
commit af444d7066

100
tests/test_anaf_marker.py Normal file
View File

@@ -0,0 +1,100 @@
"""Regression guard: tools/anaf-monitor/monitor_v2.py must emit the
GSTACK-CRON marker so the Echo-Core scheduler (report_on="changes") can
decide whether to forward stdout to a channel.
Two checks:
1. Static — the marker print statement is present in the script source.
2. Runtime — running the script via subprocess in an isolated cwd (with
config empty and network disabled via a stubbed urlopen) produces
a trailing line matching ^GSTACK-CRON: changes=\\d+$.
"""
import json
import re
import subprocess
import sys
from pathlib import Path
import pytest
SCRIPT = (
Path(__file__).resolve().parent.parent
/ "tools"
/ "anaf-monitor"
/ "monitor_v2.py"
)
MARKER_RE = re.compile(r"^GSTACK-CRON:\s+changes=(\d+)\s*$", re.MULTILINE)
def test_script_exists():
assert SCRIPT.is_file(), f"Expected ANAF monitor at {SCRIPT}"
def test_anaf_monitor_source_contains_marker_print():
"""Weak but fast regression guard: the marker print must exist in source."""
src = SCRIPT.read_text(encoding="utf-8")
assert 'print(f"GSTACK-CRON: changes=' in src, (
"monitor_v2.py must emit GSTACK-CRON marker on its own line — "
"contract with the Echo-Core shell scheduler (report_on='changes')."
)
def test_anaf_monitor_emits_gstack_marker(tmp_path):
"""Run the script end-to-end with an empty config and assert the marker."""
work_dir = tmp_path / "anaf"
work_dir.mkdir()
# Empty config → zero pages processed → num_changes = 0
(work_dir / "config.json").write_text(json.dumps({"pages": []}))
# Copy the script to the isolated dir so SCRIPT_DIR-relative paths
# (hashes.json, versions.json, snapshots/, monitor.log) don't pollute
# the repo.
local_script = work_dir / "monitor_v2.py"
local_script.write_text(SCRIPT.read_text(encoding="utf-8"))
proc = subprocess.run(
[sys.executable, str(local_script)],
capture_output=True,
text=True,
timeout=30,
cwd=str(work_dir),
)
assert proc.returncode == 0, (
f"Script exited {proc.returncode}, stderr: {proc.stderr}"
)
match = MARKER_RE.search(proc.stdout)
assert match is not None, (
"monitor_v2.py did not emit a GSTACK-CRON marker. "
f"stdout was:\n{proc.stdout}"
)
# With empty config there are zero changes.
assert int(match.group(1)) == 0
def test_anaf_monitor_marker_is_last_line(tmp_path):
"""Marker should be the final meaningful line so log-tailers can parse."""
work_dir = tmp_path / "anaf"
work_dir.mkdir()
(work_dir / "config.json").write_text(json.dumps({"pages": []}))
local_script = work_dir / "monitor_v2.py"
local_script.write_text(SCRIPT.read_text(encoding="utf-8"))
proc = subprocess.run(
[sys.executable, str(local_script)],
capture_output=True,
text=True,
timeout=30,
cwd=str(work_dir),
)
assert proc.returncode == 0
# Strip trailing whitespace/newlines, grab last non-empty line.
lines = [ln for ln in proc.stdout.splitlines() if ln.strip()]
assert lines, "Script produced no stdout lines"
assert MARKER_RE.match(lines[-1]), (
f"Last stdout line is not the GSTACK-CRON marker. Got: {lines[-1]!r}"
)