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)