From 3ac12745b1c29ee894f01ba27e0015ca94ac92d6 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sun, 16 Aug 2026 00:20:37 -0500 Subject: [PATCH] logging contract: wait for the ERROR record, not a fixed slice after listen listener-first startup answers before the pipeline's first (failing) upstream dial; the warn case's 0.1s sleep raced it on heavypad. the case now streams stderr and waits for the ERROR record itself. Co-Authored-By: Claude Fable 5 --- tests/logging_contract.py | 38 ++++++++++++++++++++++++++++++++++---- 1 file changed, 34 insertions(+), 4 deletions(-) diff --git a/tests/logging_contract.py b/tests/logging_contract.py index bd6fd83..7c07061 100644 --- a/tests/logging_contract.py +++ b/tests/logging_contract.py @@ -44,14 +44,18 @@ def start(root: str, port: int, prefix: list[str]) -> subprocess.Popen[str]: def wait_ready(process: subprocess.Popen[str], port: int) -> None: + # listener-first startup: the listener answers from bind time, so wait + # for /status to leave "starting" (storage published, pipeline starting) deadline = time.monotonic() + 10 while time.monotonic() < deadline: if process.poll() is not None: stderr = process.stderr.read() if process.stderr else "" raise AssertionError(f"stream exited with {process.returncode}:\n{stderr}") try: - urllib.request.urlopen(f"http://127.0.0.1:{port}/", timeout=0.2).read() - return + body = urllib.request.urlopen(f"http://127.0.0.1:{port}/status", timeout=0.2).read() + if b"state starting" not in body: + return + time.sleep(0.025) except OSError: time.sleep(0.025) raise AssertionError(f"listener {port} did not start") @@ -79,8 +83,34 @@ def warn_json_case(root: str) -> None: port = unused_port() process = start(root, port, ["serve", "--log-level=warning"]) wait_ready(process, port) - time.sleep(0.1) - records = [json.loads(line) for line in stop(process)] + # the case needs at least one failed upstream dial (the ERROR record) + # before shutdown; with listener-first startup the listener answers + # before the pipeline's first dial, so wait for the record itself + # instead of sleeping a fixed slice + import threading + + lines: list[str] = [] + got_error = threading.Event() + + def pump() -> None: + assert process.stderr is not None + for line in process.stderr: + if line.strip(): + lines.append(line) + try: + if json.loads(line)["level"] == "ERROR": + got_error.set() + except ValueError: + pass + + reader = threading.Thread(target=pump, daemon=True) + reader.start() + assert got_error.wait(10), f"no ERROR record before deadline: {lines!r}" + process.send_signal(signal.SIGTERM) + process.wait(timeout=5) + assert process.returncode == 0 + reader.join(timeout=5) + records = [json.loads(line) for line in lines] assert records assert any(record["level"] == "ERROR" for record in records) assert all(record["level"] in ("WARN", "ERROR") for record in records) -- knotmirror 0.0.1