mirror of
https://github.com/lexmount/moli.git
synced 2026-10-30 08:07:38 +00:00
test(cdp-smoke): capture browser close timeouts before cleanup
This commit is contained in:
@@ -478,6 +478,11 @@ jobs:
|
||||
uv sync --project moli-cdp-smoke --locked
|
||||
npm ci --prefix moli-cdp-smoke
|
||||
uv run --project moli-cdp-smoke --no-sync playwright install --with-deps chromium
|
||||
- name: Repeat target semantics for shutdown investigation
|
||||
run: |
|
||||
uv run --project moli-cdp-smoke --no-sync moli-cdp-smoke \
|
||||
--browser chromium --group target-semantics --repeat 20 --jobs 4 \
|
||||
--output-dir target/smoke/cdp-chromium-ci/shutdown-repeat
|
||||
- name: Run full CDP smoke on Chromium
|
||||
run: |
|
||||
# Piped core handlers ignore ulimit, and can block Page.crash teardown.
|
||||
|
||||
@@ -29,6 +29,13 @@ observes the parent's return code independently of stdout/stderr EOF: a
|
||||
descendant may retain an inherited output pipe after the browser has exited,
|
||||
which can leave `asyncio.subprocess.Process.wait()` pending. Failures identify
|
||||
the session case and lifecycle stage, including acknowledgment and process exit.
|
||||
Failed cases retain their close response, completed cases, browser log tail, and
|
||||
connection state in the worker's `observed` result before any socket or process
|
||||
cleanup. On Linux they also capture the owned PID's status, thread wait channels
|
||||
and syscalls, and descendant process states. This distinguishes a live or stuck
|
||||
browser from an already reaped browser whose descendants still own output pipes;
|
||||
diagnostic collection does not relax the natural-exit deadline or count forced
|
||||
cleanup as a successful exit.
|
||||
|
||||
Lifecycle artifacts retain total FDs for both engines. Chromium also records
|
||||
FDs for regular files inside its owned profile: delayed database initialization
|
||||
|
||||
@@ -5,7 +5,7 @@ import json
|
||||
from typing import Any
|
||||
|
||||
from ..assertions import SmokeError, assert_equal, wait_until
|
||||
from ..process import wait_for_process_exit
|
||||
from ..process import process_exit_diagnostics, wait_for_process_exit
|
||||
from ..raw_cdp import RawCdpClient, connect_raw_cdp, connect_raw_cdp_websocket
|
||||
from ..serve import browser_identity, start_browser, stop_browser, wait_for_browser_endpoint
|
||||
|
||||
@@ -13,6 +13,12 @@ from ..serve import browser_identity, start_browser, stop_browser, wait_for_brow
|
||||
_OMITTED = object()
|
||||
|
||||
|
||||
class BrowserCloseError(SmokeError):
|
||||
def __init__(self, message: str, observed: dict[str, Any]) -> None:
|
||||
super().__init__(message)
|
||||
self.observed = observed
|
||||
|
||||
|
||||
async def _command(
|
||||
client: RawCdpClient, method: str, params: dict[str, Any] | None = None,
|
||||
*, session_id: str | None = None,
|
||||
@@ -134,6 +140,8 @@ async def browser_close_instance_lifecycle(endpoint: str, fixture: str) -> dict[
|
||||
serving = await start_browser()
|
||||
clients: list[RawCdpClient] = []
|
||||
stage = "startup"
|
||||
child_endpoint = None
|
||||
response = None
|
||||
try:
|
||||
child_endpoint = await wait_for_browser_endpoint(serving)
|
||||
await browser_identity(child_endpoint, identity["engine"])
|
||||
@@ -168,7 +176,32 @@ async def browser_close_instance_lifecycle(endpoint: str, fixture: str) -> dict[
|
||||
assert_equal(code, 0, "natural process exit")
|
||||
observed.append({"case": name, "response": response, "exitCode": code, "rejected": rejected})
|
||||
except Exception as error:
|
||||
raise SmokeError(f"{name} ({stage}): {type(error).__name__}: {error}") from error
|
||||
# Keep the pre-cleanup state: closing the clients or sending SIGTERM
|
||||
# first would conceal whether the browser itself was still alive.
|
||||
diagnostics = process_exit_diagnostics(serving.process)
|
||||
observed_failure = {
|
||||
"browser": identity,
|
||||
"completedCases": observed,
|
||||
"failedCase": {
|
||||
"case": name, "scope": scope, "stage": stage,
|
||||
"endpoint": child_endpoint, "closeResponse": response,
|
||||
"process": diagnostics,
|
||||
"connections": [
|
||||
{"state": client.websocket.state.name,
|
||||
"closeCode": client.websocket.close_code,
|
||||
"closeReason": client.websocket.close_reason}
|
||||
for client in clients
|
||||
],
|
||||
"browserLogTail": serving.logs[-80:],
|
||||
},
|
||||
}
|
||||
process_state = diagnostics.get("process", {})
|
||||
raise BrowserCloseError(
|
||||
f"{name} ({stage}): {type(error).__name__}: {error}; "
|
||||
f"pid={serving.process.pid} returncode={diagnostics['returncode']} "
|
||||
f"state={process_state.get('State')} wchan={process_state.get('wchan')}",
|
||||
observed_failure,
|
||||
) from error
|
||||
finally:
|
||||
await asyncio.gather(
|
||||
*(client.websocket.close() for client in clients), return_exceptions=True,
|
||||
|
||||
@@ -10,7 +10,7 @@ from urllib.parse import quote
|
||||
from ..assertions import SmokeError, assert_equal, record_contract, wait_until
|
||||
from ..progress import await_with_progress
|
||||
from ..raw_cdp import RawCdpClient, RawCdpError, connect_raw_cdp
|
||||
from .browser_close import browser_close_instance_lifecycle
|
||||
from .browser_close import BrowserCloseError, browser_close_instance_lifecycle
|
||||
|
||||
|
||||
RawSemanticScenario = Callable[[str, str], Awaitable[dict[str, Any]]]
|
||||
@@ -37,17 +37,18 @@ async def run_target_semantics_group(
|
||||
item.scenario(endpoint, fixture),
|
||||
)
|
||||
except Exception as error:
|
||||
results.append(
|
||||
{
|
||||
"name": item.name,
|
||||
"ok": False,
|
||||
"contract": item.contract,
|
||||
"source": item.source,
|
||||
"commands": item.commands,
|
||||
"errorType": type(error).__name__,
|
||||
"error": str(error),
|
||||
}
|
||||
)
|
||||
failed = {
|
||||
"name": item.name,
|
||||
"ok": False,
|
||||
"contract": item.contract,
|
||||
"source": item.source,
|
||||
"commands": item.commands,
|
||||
"errorType": type(error).__name__,
|
||||
"error": str(error),
|
||||
}
|
||||
if isinstance(error, BrowserCloseError):
|
||||
failed["observed"] = error.observed
|
||||
results.append(failed)
|
||||
else:
|
||||
record_contract(
|
||||
results,
|
||||
|
||||
@@ -3,9 +3,12 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from os import PathLike
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
_PROCESS_CLEANUP_TIMEOUT_SECONDS = 2.0
|
||||
@@ -52,6 +55,71 @@ async def wait_for_process_exit(
|
||||
return process.returncode
|
||||
|
||||
|
||||
def process_exit_diagnostics(process: asyncio.subprocess.Process) -> dict[str, Any]:
|
||||
"""Snapshot a failed natural exit before cleanup changes the evidence."""
|
||||
result: dict[str, Any] = {"pid": process.pid, "returncode": process.returncode}
|
||||
if not sys.platform.startswith("linux"):
|
||||
result["procfsUnavailable"] = sys.platform
|
||||
return result
|
||||
|
||||
def read(path: Path) -> str:
|
||||
try:
|
||||
return path.read_text(encoding="utf-8", errors="replace").strip()
|
||||
except OSError as error:
|
||||
return f"{type(error).__name__}: {error}"
|
||||
|
||||
def describe(pid: int) -> dict[str, Any]:
|
||||
directory = Path(f"/proc/{pid}")
|
||||
status = read(directory / "status")
|
||||
fields = ("Name", "State", "PPid", "Threads", "VmRSS")
|
||||
description: dict[str, Any] = {"pid": pid, "wchan": read(directory / "wchan")}
|
||||
if directory.exists():
|
||||
description.update(
|
||||
(key, value.strip())
|
||||
for line in status.splitlines()
|
||||
if ":" in line
|
||||
for key, value in [line.split(":", 1)]
|
||||
if key in fields
|
||||
)
|
||||
else:
|
||||
description["procfsError"] = status
|
||||
return description
|
||||
|
||||
result["process"] = describe(process.pid)
|
||||
tasks = Path(f"/proc/{process.pid}/task")
|
||||
try:
|
||||
thread_dirs = sorted(tasks.iterdir(), key=lambda path: int(path.name))
|
||||
except OSError as error:
|
||||
result["threadsError"] = f"{type(error).__name__}: {error}"
|
||||
thread_dirs = []
|
||||
result["threads"] = [
|
||||
{"tid": int(path.name), "name": read(path / "comm"),
|
||||
"wchan": read(path / "wchan"), "syscall": read(path / "syscall")}
|
||||
for path in thread_dirs[:64]
|
||||
]
|
||||
result["threadsTruncated"] = len(thread_dirs) > 64
|
||||
descendants = []
|
||||
pending = [process.pid]
|
||||
seen = {process.pid}
|
||||
while pending and len(descendants) < 64:
|
||||
pid = pending.pop(0)
|
||||
children = read(Path(f"/proc/{pid}/task/{pid}/children"))
|
||||
for value in children.split():
|
||||
if not value.isdecimal():
|
||||
continue
|
||||
child = int(value)
|
||||
if child in seen:
|
||||
continue
|
||||
seen.add(child)
|
||||
pending.append(child)
|
||||
descendants.append(describe(child))
|
||||
if len(descendants) == 64:
|
||||
break
|
||||
result["descendants"] = descendants
|
||||
result["descendantsTruncated"] = len(descendants) == 64
|
||||
return result
|
||||
|
||||
|
||||
def _consume_task_result(task: asyncio.Task[object]) -> None:
|
||||
try:
|
||||
task.exception()
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from moli_cdp_smoke.groups import browser_close, target_semantics
|
||||
from moli_cdp_smoke.process import terminate_process_tree, wait_for_process_exit
|
||||
|
||||
|
||||
class BrowserCloseDiagnosticsTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_acknowledgment_without_exit_still_fails_and_preserves_pre_cleanup_state(self) -> None:
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(30)",
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
start_new_session=True,
|
||||
)
|
||||
await asyncio.wait_for(process.stdout.readline(), 5)
|
||||
serving = SimpleNamespace(process=process, logs=["stderr: shutdown checkpoint"])
|
||||
socket = SimpleNamespace(
|
||||
state=SimpleNamespace(name="OPEN"), close_code=None, close_reason=None,
|
||||
close=AsyncMock(),
|
||||
)
|
||||
root = SimpleNamespace(websocket=socket)
|
||||
acknowledgment = {"id": 1000, "result": {}}
|
||||
identity = {"engine": "chromium"}
|
||||
|
||||
async def bounded_exit(child, *, timeout_seconds):
|
||||
self.assertEqual(timeout_seconds, 10, "natural exit keeps the original deadline")
|
||||
return await wait_for_process_exit(child, timeout_seconds=0.02)
|
||||
|
||||
async def stop(child):
|
||||
socket.close.assert_awaited_once()
|
||||
await terminate_process_tree(child.process)
|
||||
|
||||
try:
|
||||
with (
|
||||
patch.object(browser_close, "browser_identity", AsyncMock(return_value=identity)),
|
||||
patch.object(browser_close, "start_browser", AsyncMock(return_value=serving)),
|
||||
patch.object(browser_close, "wait_for_browser_endpoint", AsyncMock(return_value="http://child")),
|
||||
patch.object(browser_close, "connect_raw_cdp", AsyncMock(return_value=root)),
|
||||
patch.object(browser_close, "_close_request", AsyncMock(return_value=acknowledgment)),
|
||||
patch.object(browser_close, "wait_for_process_exit", bounded_exit),
|
||||
patch.object(browser_close, "stop_browser", stop),
|
||||
):
|
||||
with self.assertRaisesRegex(browser_close.BrowserCloseError, "root .*natural process exit.*pid=") as raised:
|
||||
await browser_close.browser_close_instance_lifecycle("http://survivor", "http://fixture")
|
||||
|
||||
observed = raised.exception.observed
|
||||
failed = observed["failedCase"]
|
||||
self.assertEqual(failed["closeResponse"], acknowledgment)
|
||||
self.assertEqual(failed["stage"], "natural process exit")
|
||||
self.assertEqual(failed["process"]["pid"], process.pid)
|
||||
self.assertIsNone(failed["process"]["returncode"], "snapshot precedes forced cleanup")
|
||||
self.assertEqual(failed["connections"][0]["state"], "OPEN")
|
||||
self.assertEqual(failed["browserLogTail"], serving.logs)
|
||||
self.assertEqual(observed["completedCases"], [])
|
||||
self.assertIsNotNone(process.returncode, "failed scenario still cleans up its child")
|
||||
finally:
|
||||
await terminate_process_tree(process)
|
||||
|
||||
async def test_worker_result_retains_structured_failure_evidence(self) -> None:
|
||||
observed = {"completedCases": [{"case": "root", "exitCode": 0}],
|
||||
"failedCase": {"case": "nested-browser-session", "stage": "natural process exit"}}
|
||||
error = browser_close.BrowserCloseError("natural process exit timed out", observed)
|
||||
scenario = AsyncMock(side_effect=error)
|
||||
contract = target_semantics.RawSemanticContract("close", "contract", "Chromium", ["Browser.close"], scenario)
|
||||
results = []
|
||||
|
||||
with patch.object(target_semantics, "_raw_semantic_contracts", return_value=(contract,)):
|
||||
await target_semantics.run_target_semantics_group("http://child", "http://fixture", results)
|
||||
|
||||
self.assertFalse(results[0]["ok"])
|
||||
self.assertEqual(results[0]["observed"], observed)
|
||||
self.assertEqual(results[0]["error"], str(error))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -9,7 +9,11 @@ import unittest
|
||||
|
||||
from moli_cdp_smoke.assertions import SmokeError
|
||||
from moli_cdp_smoke.groups.external_process import run_external_json_process
|
||||
from moli_cdp_smoke.process import terminate_process_tree, wait_for_process_exit
|
||||
from moli_cdp_smoke.process import (
|
||||
process_exit_diagnostics,
|
||||
terminate_process_tree,
|
||||
wait_for_process_exit,
|
||||
)
|
||||
|
||||
|
||||
class ExternalProcessTests(unittest.IsolatedAsyncioTestCase):
|
||||
@@ -84,6 +88,52 @@ class ExternalProcessTests(unittest.IsolatedAsyncioTestCase):
|
||||
|
||||
|
||||
class NaturalProcessExitTests(unittest.IsolatedAsyncioTestCase):
|
||||
@unittest.skipUnless(sys.platform.startswith("linux"), "diagnostic regression uses procfs")
|
||||
async def test_failed_exit_captures_owned_process_and_descendant_before_cleanup(self) -> None:
|
||||
descendant = "import time; print('ready', flush=True); time.sleep(30)"
|
||||
parent = (
|
||||
"import subprocess, sys; "
|
||||
f"child = subprocess.Popen([sys.executable, '-c', {descendant!r}]); "
|
||||
"child.wait()"
|
||||
)
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
sys.executable, "-c", parent,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
start_new_session=True,
|
||||
)
|
||||
try:
|
||||
self.assertEqual(await asyncio.wait_for(process.stdout.readline(), 5), b"ready\n")
|
||||
with self.assertRaises(TimeoutError):
|
||||
await wait_for_process_exit(process, timeout_seconds=0.02)
|
||||
|
||||
diagnostics = process_exit_diagnostics(process)
|
||||
|
||||
self.assertEqual(diagnostics["pid"], process.pid)
|
||||
self.assertIsNone(diagnostics["returncode"])
|
||||
self.assertIn("State", diagnostics["process"])
|
||||
self.assertTrue(diagnostics["threads"])
|
||||
self.assertEqual(diagnostics["threads"][0]["tid"], process.pid)
|
||||
self.assertEqual(len(diagnostics["descendants"]), 1)
|
||||
self.assertEqual(int(diagnostics["descendants"][0]["PPid"]), process.pid)
|
||||
self.assertIsNone(process.returncode, "diagnostics must not stop the process")
|
||||
json.dumps(diagnostics)
|
||||
finally:
|
||||
await terminate_process_tree(process)
|
||||
|
||||
@unittest.skipUnless(sys.platform.startswith("linux"), "diagnostic regression uses procfs")
|
||||
async def test_diagnostics_preserve_a_reaped_process_status(self) -> None:
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
sys.executable, "-c", "raise SystemExit(7)",
|
||||
)
|
||||
await wait_for_process_exit(process, timeout_seconds=5)
|
||||
|
||||
diagnostics = process_exit_diagnostics(process)
|
||||
|
||||
self.assertEqual(diagnostics["returncode"], 7)
|
||||
self.assertIn("FileNotFoundError", diagnostics["process"]["procfsError"])
|
||||
self.assertEqual(diagnostics["threads"], [])
|
||||
self.assertEqual(diagnostics["descendants"], [])
|
||||
|
||||
@unittest.skipUnless(os.name == "posix", "inherited-pipe regression is POSIX-specific")
|
||||
async def test_parent_exit_does_not_wait_for_descendant_output_eof(self) -> None:
|
||||
descendant = "import time; time.sleep(30)"
|
||||
|
||||
Reference in New Issue
Block a user