diff --git a/moli-cdp-smoke/README.md b/moli-cdp-smoke/README.md index f7516cd9c7..1e22f96390 100644 --- a/moli-cdp-smoke/README.md +++ b/moli-cdp-smoke/README.md @@ -81,6 +81,22 @@ The `network` and `multi-context` groups require the encoded Fetch body and flag then independently verify the original page-visible text after continuation. Their Network body and `IO.read` stream assertions retain their own contracts. +The Fetch byte-counter and EventSource-close contracts were probed on 2026-10-20 +with Playwright 1.58.0's pinned Chrome for Testing 145.0.7632.6 and local HTTP +fixtures. Of 150 ordinary 45-byte JSON fetches, one reported `dataReceived` +counters `(45, 0)` followed by `(0, 45)`; 146 of 150 gzip responses split their +decoded and encoded counters similarly. The `network` group therefore checks +the totals over all data events, their nonnegative lengths, and their order +between the response and successful terminal. A finite SSE response completed +with `loadingFinished`, while a response with a withheld trailing chunk ended +with `loadingFailed`, `net::ERR_ABORTED`, and `canceled: true` when the message +handler called `close()`. The EventSource scenario accepts those two outcomes +and still rejects other errors, wrong order, and duplicate terminals. The same +probe against the Moli release produced both SSE terminals, with its Fetch byte +counters delivered together. Run +`--browser chromium --group network --group browser-semantics --repeat 20 --jobs 4` +to exercise both groups with fresh browser processes. + The default `cdp-ordering` group locks down Renderer native and Inspector wire publication order. It pipelines Page trees/layout, DOMSnapshot, DOM queries and objects, CSS, AX, and a native backend error before `Runtime.enable`; stylesheet diff --git a/moli-cdp-smoke/moli_cdp_smoke/groups/browser_semantics.py b/moli-cdp-smoke/moli_cdp_smoke/groups/browser_semantics.py index bd5d61345d..43bbb12f19 100644 --- a/moli-cdp-smoke/moli_cdp_smoke/groups/browser_semantics.py +++ b/moli-cdp-smoke/moli_cdp_smoke/groups/browser_semantics.py @@ -1358,7 +1358,7 @@ async def _event_source_runtime_and_network(state: SmokeState) -> dict[str, Any] async with _isolated_page(state) as (_context, page, cdp): events = attach_cdp_event_collector( cdp, - ["Network.requestWillBeSent", "Network.responseReceived", "Network.loadingFinished"], + ["Network.requestWillBeSent", "Network.responseReceived", "Network.loadingFinished", "Network.loadingFailed"], ) await cdp.send("Network.enable") await page.goto(f"{state.fixture}/plain", wait_until="load", timeout=10_000) @@ -1402,34 +1402,40 @@ async def _event_source_runtime_and_network(state: SmokeState) -> dict[str, Any] request_id = response_event["params"]["requestId"] await wait_until( lambda: any( - event["method"] == "Network.loadingFinished" + event["method"] in ("Network.loadingFinished", "Network.loadingFailed") and event["params"].get("requestId") == request_id for event in events ), - "EventSource Network.loadingFinished", - ) - event_methods = [ - event["method"] - for event in events - if event["params"].get("requestId") == request_id - ] - assert_equal( - event_methods, - [ - "Network.requestWillBeSent", - "Network.responseReceived", - "Network.loadingFinished", - ], - "EventSource Network event order", + "EventSource Network terminal after close()", ) return { "runtime": runtime_value, "requestId": request_id, "resourceType": response_event["params"]["type"], - "eventMethods": event_methods, + **_event_source_network_summary(events, request_id), } +def _event_source_network_summary(events: list[dict[str, Any]], request_id: str) -> dict[str, Any]: + matching = [event for event in events if event["params"].get("requestId") == request_id] + methods = [event["method"] for event in matching] + terminal = matching[-1] if matching else {} + terminal_method = terminal.get("method") + if terminal_method not in ("Network.loadingFinished", "Network.loadingFailed"): + raise SmokeError(f"EventSource close() has no Network terminal: {matching}") + assert_equal( + methods, + ["Network.requestWillBeSent", "Network.responseReceived", terminal_method], + "EventSource Network event order", + ) + # close() races the finite fixture's EOF. If EOF has not committed yet, + # Chromium reports the intentional cancellation instead of successful EOF. + if terminal_method == "Network.loadingFailed": + assert_equal(terminal["params"].get("errorText"), "net::ERR_ABORTED", "EventSource close() error") + assert_equal(terminal["params"].get("canceled"), True, "EventSource close() cancellation") + return {"eventMethods": methods, "terminal": terminal} + + async def _http_cache_event_and_reuse(state: SmokeState) -> dict[str, Any]: cache_route = "/semantic-cache.js" state.fixture_server.reset_request_count(cache_route) diff --git a/moli-cdp-smoke/moli_cdp_smoke/groups/network.py b/moli-cdp-smoke/moli_cdp_smoke/groups/network.py index c3709b003b..e32e4ca8b5 100644 --- a/moli-cdp-smoke/moli_cdp_smoke/groups/network.py +++ b/moli-cdp-smoke/moli_cdp_smoke/groups/network.py @@ -321,37 +321,42 @@ async def _verify_fetch_network_events(state: SmokeState) -> None: observed_fetch = await page.evaluate("async () => await fetch('/api-continue').then(response => response.json())") assert_equal(observed_fetch.get("routeHeader"), None, "unrouted fetch should reach fixture server") - def saw_fetch_events() -> bool: - events = subresource_events[fetch_network_start:] - request = _find_request(events, "Fetch", f"{fixture}/api-continue") - request_id = request and request["params"].get("requestId") - if not request_id: - return False - matching_indices = { - method: next( - ( - index - for index, event in enumerate(events) - if event["method"] == method and event["params"].get("requestId") == request_id - ), - None, - ) - for method in ("Network.responseReceived", "Network.dataReceived", "Network.loadingFinished") - } - response_index = matching_indices["Network.responseReceived"] - data_index = matching_indices["Network.dataReceived"] - finished_index = matching_indices["Network.loadingFinished"] - if response_index is None or data_index is None or finished_index is None: - return False - data_params = events[data_index]["params"] - return bool( - response_index < data_index < finished_index - and data_params.get("dataLength", 0) > 0 - and data_params.get("encodedDataLength", 0) > 0 - ) + await wait_until( + lambda: _completed_request_id( + subresource_events[fetch_network_start:], "Fetch", f"{fixture}/api-continue" + ) is not None, + "Fetch Network events", + ) + events = subresource_events[fetch_network_start:] + request_id = _completed_request_id(events, "Fetch", f"{fixture}/api-continue") + if request_id is None: + raise SmokeError("completed Fetch request disappeared from the event trace") + state.record("fetch_network_events", { + "requestId": request_id, + **_fetch_network_event_summary(events, request_id), + }) - await wait_until(saw_fetch_events, "Fetch Network events") - state.record("fetch_network_events") + +def _fetch_network_event_summary(events: list[dict[str, Any]], request_id: str) -> dict[str, Any]: + matching = [event for event in events if event["params"].get("requestId") == request_id] + methods = [event["method"] for event in matching] + assert_equal(methods[:2], ["Network.requestWillBeSent", "Network.responseReceived"], "Fetch Network start order") + assert_equal(methods[-1:], ["Network.loadingFinished"], "Fetch Network terminal") + data_events = matching[2:-1] + assert_equal({event["method"] for event in data_events}, {"Network.dataReceived"}, "Fetch Network data order") + totals = {"dataLength": 0, "encodedDataLength": 0} + for event in data_events: + for key in totals: + value = event["params"].get(key) + if type(value) is not int or value < 0: + raise SmokeError(f"invalid Fetch Network {key}: {event}") + totals[key] += value + # Chromium can deliver decoded bytes first, then flush encoded byte counts + # in a later dataReceived event whose dataLength is zero. Neither counter + # has to be positive in the first chunk; the completed nonempty body does. + if any(value <= 0 for value in totals.values()): + raise SmokeError(f"nonempty Fetch Network body must report positive byte totals: {matching}") + return {"eventMethods": methods, **totals} async def _verify_xhr_network_events(state: SmokeState) -> None: diff --git a/moli-cdp-smoke/moli_cdp_smoke/supervisor.py b/moli-cdp-smoke/moli_cdp_smoke/supervisor.py index c3e2761467..0a0d38d859 100644 --- a/moli-cdp-smoke/moli_cdp_smoke/supervisor.py +++ b/moli-cdp-smoke/moli_cdp_smoke/supervisor.py @@ -307,7 +307,9 @@ async def _run_worker_job( else: status = "failed" worker_error = payload.get("error") - error = str(worker_error) if worker_error else f"worker exited with {exit_code}" + error = str(worker_error) if worker_error else ( + _coverage_error(job, payload) or f"worker exited with {exit_code}" + ) outcome = WorkerOutcome( job=job, diff --git a/moli-cdp-smoke/tests/test_network_events.py b/moli-cdp-smoke/tests/test_network_events.py new file mode 100644 index 0000000000..1640120918 --- /dev/null +++ b/moli-cdp-smoke/tests/test_network_events.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import unittest +from typing import Any + +from moli_cdp_smoke.assertions import SmokeError +from moli_cdp_smoke.groups.browser_semantics import _event_source_network_summary +from moli_cdp_smoke.groups.network import _fetch_network_event_summary + + +REQUEST = "request" + + +def event(method: str, **params: Any) -> dict[str, Any]: + return {"method": f"Network.{method}", "params": {"requestId": REQUEST, **params}} + + +def fetch_trace(chunks: list[tuple[int, int]]) -> list[dict[str, Any]]: + return [ + event("requestWillBeSent"), + event("responseReceived"), + *(event("dataReceived", dataLength=decoded, encodedDataLength=encoded) + for decoded, encoded in chunks), + event("loadingFinished"), + ] + + +class FetchNetworkEventsTests(unittest.TestCase): + def test_counters_can_arrive_together_or_in_separate_chunks(self) -> None: + # Observed on pinned Chromium 145.0.7632.6: a normal 45-byte JSON + # response can split into (45, 0), (0, 45). Gzip does this frequently. + for chunks, decoded, encoded in [ + ([(45, 45)], 45, 45), + ([(45, 0), (0, 45)], 45, 45), + ([(45, 0), (0, 35)], 45, 35), + ([(1044, 0), (31737, 82)], 32781, 82), + ]: + with self.subTest(chunks=chunks): + summary = _fetch_network_event_summary(fetch_trace(chunks), REQUEST) + self.assertEqual(summary["dataLength"], decoded) + self.assertEqual(summary["encodedDataLength"], encoded) + self.assertEqual(summary["eventMethods"][-1], "Network.loadingFinished") + + def test_other_requests_cannot_supply_missing_byte_counts(self) -> None: + events = fetch_trace([(45, 0)]) + events.insert(-1, event("dataReceived", requestId="other", dataLength=0, encodedDataLength=45)) + with self.assertRaisesRegex(SmokeError, "positive byte totals"): + _fetch_network_event_summary(events, REQUEST) + + def test_nonempty_body_requires_both_positive_totals(self) -> None: + for chunks in [[(45, 0)], [(0, 45)], [(0, 0)], []]: + with self.subTest(chunks=chunks), self.assertRaises(SmokeError): + _fetch_network_event_summary(fetch_trace(chunks), REQUEST) + + def test_invalid_chunk_lengths_are_rejected_before_summing(self) -> None: + for key in ["dataLength", "encodedDataLength"]: + for value in [-1, None, "45", True]: + events = fetch_trace([(45, 45), (1, 1)]) + events[2]["params"][key] = value + with self.subTest(key=key, value=value), self.assertRaisesRegex(SmokeError, "invalid Fetch"): + _fetch_network_event_summary(events, REQUEST) + + def test_data_must_follow_response_and_precede_one_successful_terminal(self) -> None: + valid = fetch_trace([(45, 45)]) + for events in [ + [valid[0], valid[2], valid[1], valid[3]], + [valid[0], valid[1], valid[3], valid[2]], + [*valid, valid[-1]], + [*valid[:-1], event("loadingFailed", errorText="net::ERR_ABORTED", canceled=True)], + ]: + with self.subTest(events=events), self.assertRaises(SmokeError): + _fetch_network_event_summary(events, REQUEST) + + +class EventSourceNetworkEventsTests(unittest.TestCase): + def test_close_accepts_committed_eof_or_an_explicit_abort(self) -> None: + for terminal in [ + event("loadingFinished"), + event("loadingFailed", errorText="net::ERR_ABORTED", canceled=True), + ]: + events = [event("requestWillBeSent"), event("responseReceived"), terminal] + with self.subTest(terminal=terminal): + summary = _event_source_network_summary(events, REQUEST) + self.assertEqual(summary["terminal"], terminal) + self.assertEqual(summary["eventMethods"], [entry["method"] for entry in events]) + + def test_close_does_not_accept_unrelated_network_errors(self) -> None: + for params in [ + {"errorText": "net::ERR_CONNECTION_RESET", "canceled": True}, + {"errorText": "net::ERR_ABORTED", "canceled": False}, + {"errorText": "net::ERR_ABORTED"}, + ]: + events = [event("requestWillBeSent"), event("responseReceived"), event("loadingFailed", **params)] + with self.subTest(params=params), self.assertRaises(SmokeError): + _event_source_network_summary(events, REQUEST) + + def test_close_requires_ordered_response_and_exactly_one_terminal(self) -> None: + request, response, finished = event("requestWillBeSent"), event("responseReceived"), event("loadingFinished") + aborted = event("loadingFailed", errorText="net::ERR_ABORTED", canceled=True) + for events in [ + [request, response], + [request, finished, response], + [request, response, finished, aborted], + [request, response, finished, finished], + ]: + with self.subTest(events=events), self.assertRaises(SmokeError): + _event_source_network_summary(events, REQUEST) + + def test_other_request_terminals_are_ignored(self) -> None: + terminal = event("loadingFinished") + summary = _event_source_network_summary([ + event("requestWillBeSent"), event("responseReceived"), + event("loadingFailed", requestId="other", errorText="net::ERR_FAILED"), terminal, + ], REQUEST) + self.assertEqual(summary["terminal"], terminal) + + +if __name__ == "__main__": + unittest.main() diff --git a/moli-cdp-smoke/tests/test_supervisor.py b/moli-cdp-smoke/tests/test_supervisor.py index 51af6e9c00..989e0fecf4 100644 --- a/moli-cdp-smoke/tests/test_supervisor.py +++ b/moli-cdp-smoke/tests/test_supervisor.py @@ -22,6 +22,40 @@ from moli_cdp_smoke.supervisor import ( class SupervisorWorkerTests(unittest.IsolatedAsyncioTestCase): + async def test_failed_scenario_is_reported_when_worker_has_no_top_level_error(self) -> None: + job = WorkerJob(group=GROUPS_BY_NAME["browser-semantics"], attempt=1, repeat=1) + for worker_error in (None, "worker traceback takes precedence"): + with self.subTest(worker_error=worker_error), tempfile.TemporaryDirectory() as temporary: + payload = { + "ok": False, + "group": "browser-semantics", + "results": [{ + "name": "cdp_contract_event_source_runtime_and_network", + "ok": False, + "errorType": "SmokeError", + "error": "timed out waiting for EventSource Network terminal", + }], + } + if worker_error is not None: + payload["error"] = worker_error + + def fake_argv(_: WorkerJob, result_path: Path) -> list[str]: + return [sys.executable, "-c", ( + "from pathlib import Path; import sys; " + f"Path(sys.argv[1]).write_text({json.dumps(payload)!r}); " + "print('later scenario passed'); sys.exit(1)" + ), str(result_path)] + + with patch("moli_cdp_smoke.supervisor._worker_argv", fake_argv): + outcome = await _run_worker_job(job, Path(temporary), 5) + self.assertEqual(outcome.status, "failed") + if worker_error is not None: + self.assertEqual(outcome.error, worker_error) + else: + self.assertIn("cdp_contract_event_source_runtime_and_network", outcome.error) + self.assertIn("SmokeError", outcome.error) + self.assertIn("timed out waiting for EventSource", outcome.error) + async def test_worker_result_is_read_from_its_isolated_file(self) -> None: group = GROUPS_BY_NAME["protocol"] job = WorkerJob(group=group, attempt=1, repeat=1)