A
""".encode() ) return if self.path == "/document-ran?a": server.a_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/redirect-to-b": self.send_response(302) self.send_header("Location", f"http://127.0.0.1:{server_port}/b") self.end_headers() return if self.path.startswith("/nested"): self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""Nested
""".encode() ) return if self.path.startswith("/state"): self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() document_ready_script = "" if self.path == "/state?replace-load": document_ready_script = "" self.wfile.write( f"""State
""".encode() ) return if self.path.startswith("/scroll"): self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Scroll
""".encode() ) return if self.path == "/reload-blocked": should_block = False with server.reload_blocked_request_lock: server.reload_blocked_request_count += 1 if server.reload_blocked_request_count > 2: server.reload_recovery_requested.set() should_block = server.reload_blocked_request_count > 1 if should_block: server.blocked_reload_requested.set() if not server.release_blocked_reload.wait(timeout=BLOCKED_RESPONSE_TIMEOUT_SECONDS): self.send_response(500) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"timed out waiting to unblock reload") return self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Reload Blocked
""".encode() ) return if self.path == "/process-swap-back-blocked": should_block = False with server.process_swap_back_blocked_request_lock: server.process_swap_back_blocked_request_count += 1 if server.process_swap_back_blocked_request_count > 2: server.process_swap_back_recovery_requested.set() should_block = server.process_swap_back_blocked_request_count > 1 if should_block: server.blocked_process_swap_back_requested.set() if not server.release_blocked_process_swap_back.wait(timeout=BLOCKED_RESPONSE_TIMEOUT_SECONDS): self.send_response(500) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"timed out waiting to unblock process-swap back") return self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""Process Swap Back Blocked
""".encode() ) return if self.path == "/forward-blocked": if server.block_forward_load: server.blocked_forward_requested.set() if not server.release_blocked_forward.wait(timeout=BLOCKED_RESPONSE_TIMEOUT_SECONDS): self.send_response(500) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"timed out waiting to unblock forward") return self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Forward Blocked
""".encode() ) return if self.path == "/frame-a": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Frame A
""".encode() ) return if self.path == "/frame-b": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Frame B
""".encode() ) return if self.path == "/frame-b-blocked": server.blocked_frame_b_requested.set() if not server.release_blocked_frame_b.wait(timeout=BLOCKED_RESPONSE_TIMEOUT_SECONDS): self.send_response(500) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"timed out waiting to unblock frame") return self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Frame B Blocked
""".encode() ) return if self.path == "/b": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""B
""".encode() ) return if self.path == "/document-ran?b": server.b_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/c": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""C
""".encode() ) return if self.path == "/document-ran?state-replace-load": server.state_replace_load_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?c": server.c_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?process-swap-back-blocked": server.process_swap_back_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?frame-b-blocked": server.frame_b_blocked_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?post-result": server.post_result_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?cross-site-post-result": server.cross_site_post_result_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?same-site-post-result": server.same_site_post_result_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?same-url-post-result": server.same_url_post_result_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/document-ran?reload-blocked": server.reload_blocked_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/d": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """D
""".encode() ) return if self.path == "/document-ran?d": server.d_document_ran.set() self.send_response(204) self.end_headers() return if self.path == "/post-form": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Post Form
""".encode() ) return if self.path == "/post-blocked-form": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Post Blocked Form
""".encode() ) return if self.path == "/post-same-url-blocked": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """Post Same URL Blocked Form
""".encode() ) return if self.path == "/cross-site-post-form": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""Cross-site Post Form
""".encode() ) return if self.path == "/cross-site-post-blocked-form": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""Cross-site Post Blocked Form
""".encode() ) return if self.path == "/post-result": self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( """GET
""".encode() ) return self.send_response(404) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"not found") def do_POST(self): server = cast(TestPageServer, self.server) if self.path == "/post-result-blocked": content_length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(content_length).decode() server.blocked_cross_site_post_requested.set() if not server.release_blocked_cross_site_post.wait(timeout=BLOCKED_RESPONSE_TIMEOUT_SECONDS): self.send_response(500) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"timed out waiting to unblock cross-site POST") return self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""POST:{body}
""".encode() ) return if self.path == "/post-result-blocked-same-site": content_length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(content_length).decode() server.blocked_same_site_post_requested.set() if not server.release_blocked_same_site_post.wait(timeout=BLOCKED_RESPONSE_TIMEOUT_SECONDS): self.send_response(500) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"timed out waiting to unblock same-site POST") return self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""POST:{body}
""".encode() ) return if self.path == "/post-same-url-blocked": content_length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(content_length).decode() server.blocked_same_url_post_requested.set() if not server.release_blocked_same_url_post.wait(timeout=BLOCKED_RESPONSE_TIMEOUT_SECONDS): self.send_response(500) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"timed out waiting to unblock same-URL POST") return self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""POST:{body}
""".encode() ) return if self.path == "/post-result": content_length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(content_length).decode() self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write( f"""POST:{body}
""".encode() ) return self.send_response(404) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"not found") def log_message(self, format, *args): pass def unused_port(): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind(("127.0.0.1", 0)) return sock.getsockname()[1] def wait_for_port(port, timeout=EVENT_TIMEOUT_SECONDS): deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: with socket.create_connection(("127.0.0.1", port), timeout=0.1): return except OSError: time.sleep(0.05) raise RuntimeError(f"Timed out waiting for port {port}") def wait_for_event(event, label, timeout=EVENT_TIMEOUT_SECONDS): if event.wait(timeout=timeout): return raise RuntimeError(f"Timed out waiting for {label}") def request_raw(webdriver_port, method, path, body=None): encoded_body = None headers = {} if body is not None: encoded_body = json.dumps(body).encode() headers["Content-Type"] = "application/json" connection = http.client.HTTPConnection("127.0.0.1", webdriver_port, timeout=WEBDRIVER_REQUEST_TIMEOUT_SECONDS) try: connection.request(method, path, encoded_body, headers) response = connection.getresponse() response_body = response.read().decode() finally: connection.close() try: payload = json.loads(response_body) if response_body else {} except json.JSONDecodeError as error: raise RuntimeError(f"{method} {path} returned invalid JSON: {response_body}") from error return response.status, payload, response_body def request(webdriver_port, method, path, body=None): status, payload, response_body = request_raw(webdriver_port, method, path, body) if status >= 400 or isinstance(payload.get("value"), dict) and payload["value"].get("error"): raise RuntimeError(f"{method} {path} failed with HTTP {status}: {response_body}") return payload def current_url(webdriver_port, session_id): return request(webdriver_port, "GET", f"/session/{session_id}/url")["value"] def execute_script(webdriver_port, session_id, script): return request( webdriver_port, "POST", f"/session/{session_id}/execute/sync", { "script": script, "args": [], }, )["value"] def wait_for_script_result(webdriver_port, session_id, label, script, predicate, log, timeout=EVENT_TIMEOUT_SECONDS): deadline = time.monotonic() + timeout result = None while time.monotonic() < deadline: result = execute_script(webdriver_port, session_id, script) if predicate(result): log.append(f"{label}: {result}") return result time.sleep(0.05) raise AssertionError(f"Timed out waiting for {label}, got {result}\n" + "\n".join(log)) def execute_async_script(webdriver_port, session_id, script): return request( webdriver_port, "POST", f"/session/{session_id}/execute/async", { "script": script, "args": [], }, )["value"] def navigate_from_renderer_using_link( webdriver_port, session_id, link_selector, expected_url, log, document_ran_event, document_ran_label, expected_scheduled_url=None, ): actual_url = execute_script( webdriver_port, session_id, f""" const link = document.querySelector({json.dumps(link_selector)}); if (!link) return null; return link.href; """, ) log.append(f"clicked renderer link to {actual_url}") expected_scheduled_url = expected_scheduled_url or expected_url if actual_url != expected_scheduled_url: raise AssertionError( f"Expected renderer navigation target to be {expected_scheduled_url}, got {actual_url}\n" + "\n".join(log) ) document_ran_event.clear() execute_script( webdriver_port, session_id, f""" const link = document.querySelector({json.dumps(link_selector)}); link.click(); return null; """, ) wait_for_event(document_ran_event, document_ran_label) def crash_current_page(webdriver_port, session_id): request(webdriver_port, "POST", f"/session/{session_id}/ladybird/crash-current-page", {}) def crash_current_page_allowing_navigation_timeout(webdriver_port, session_id): try: status, payload, response_body = request_raw( webdriver_port, "POST", f"/session/{session_id}/ladybird/crash-current-page", {} ) except TimeoutError: return if status < 400: return value = payload.get("value") if isinstance(value, dict) and value.get("error") == "timeout": return raise RuntimeError( f"POST /session/{session_id}/ladybird/crash-current-page failed with HTTP {status}: {response_body}" ) def load_url_from_ui(webdriver_port, session_id, url): request(webdriver_port, "POST", f"/session/{session_id}/ladybird/load-url-from-ui", {"url": url}) def traverse_history_from_ui(webdriver_port, session_id, delta, wait_for_navigation_completion=True): request( webdriver_port, "POST", f"/session/{session_id}/ladybird/traverse-history-from-ui", { "delta": delta, "waitForNavigationCompletion": wait_for_navigation_completion, }, ) def mark_web_content_session_history_stale(webdriver_port, session_id): request(webdriver_port, "POST", f"/session/{session_id}/ladybird/mark-web-content-session-history-stale", {}) def refresh(webdriver_port, session_id): request(webdriver_port, "POST", f"/session/{session_id}/refresh", {}) def create_session(webdriver_port, enable_test_hooks=True): always_match = { "ladybird:headless": True, "pageLoadStrategy": "normal", "unhandledPromptBehavior": {"beforeUnload": "dismiss"}, } if enable_test_hooks: always_match["ladybird:enableTestHooks"] = True created = request( webdriver_port, "POST", "/session", {"capabilities": {"alwaysMatch": always_match}}, ) session_id = created.get("value", {}).get("sessionId") or created.get("sessionId") if not session_id: raise RuntimeError(f"Could not find session id in response: {created}") request(webdriver_port, "POST", f"/session/{session_id}/timeouts", {"pageLoad": 10000}) return session_id def expect_ladybird_test_hooks_require_capability(webdriver_port): session_id = create_session(webdriver_port, enable_test_hooks=False) try: status, payload, response_body = request_raw( webdriver_port, "GET", f"/session/{session_id}/ladybird/session-history" ) value = payload.get("value") if status != 404 or not isinstance(value, dict) or value.get("error") != "unknown command": raise AssertionError( f"Expected Ladybird test hook to fail with unknown command, got HTTP {status}: {response_body}" ) finally: request(webdriver_port, "DELETE", f"/session/{session_id}") def browser_history_shortcut_modifier(): if sys.platform == "darwin": return "\ue03d", "Meta" return "\ue00a", "Alt" def perform_browser_history_shortcut(webdriver_port, session_id, direction, log): if direction == "left": arrow = "\ue012" elif direction == "right": arrow = "\ue014" else: raise ValueError(f"Unknown arrow direction {direction}") modifier, modifier_name = browser_history_shortcut_modifier() request( webdriver_port, "POST", f"/session/{session_id}/actions", { "actions": [ { "type": "key", "id": "browser-shortcut-keyboard", "actions": [ {"type": "keyDown", "value": modifier}, {"type": "keyDown", "value": arrow}, {"type": "keyUp", "value": arrow}, {"type": "keyUp", "value": modifier}, ], } ] }, ) log.append(f"performed {modifier_name}+{direction}") def perform_pointer_click(webdriver_port, session_id, x, y, log): request( webdriver_port, "POST", f"/session/{session_id}/actions", { "actions": [ { "type": "pointer", "id": "activation-pointer", "parameters": {"pointerType": "mouse"}, "actions": [ {"type": "pointerMove", "origin": "viewport", "x": x, "y": y}, {"type": "pointerDown", "button": 0}, {"type": "pointerUp", "button": 0}, ], } ] }, ) log.append(f"clicked viewport at {x},{y}") def session_history(webdriver_port, session_id): return request(webdriver_port, "GET", f"/session/{session_id}/ladybird/session-history")["value"] def wait_for_session_history(webdriver_port, session_id, label, predicate, log, timeout=EVENT_TIMEOUT_SECONDS): deadline = time.monotonic() + timeout snapshot = None while time.monotonic() < deadline: snapshot = session_history(webdriver_port, session_id) if predicate(snapshot): log.append(f"{label} history: {summarize_history_snapshot(snapshot)}") return snapshot time.sleep(0.05) raise AssertionError( f"Timed out waiting for {label} history, got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def history_entry_urls(history): return [entry["url"] for entry in history["entries"]] def history_used_steps(history): return history_step_values(history["usedSteps"]) def history_step_values(steps): return [used_step["step"] for used_step in steps] def history_current_step(steps): current_steps = [used_step["step"] for used_step in steps if used_step.get("current", False)] if len(current_steps) > 1: raise AssertionError(f"Expected at most one current step, got {current_steps}") if not current_steps: return None return current_steps[0] def history_current_entry(history): current_entries = [entry for entry in history["entries"] if entry["current"]] if len(current_entries) != 1: raise AssertionError(f"Expected exactly one current entry, got {current_entries}") return current_entries[0] def comparable_history(history): def comparable_entry(entry): return { "step": entry["step"], "url": entry["url"], "current": entry.get("current", False), "nestedHistories": [ { "id": nested_history["id"], "entries": [comparable_entry(nested_entry) for nested_entry in nested_history["entries"]], } for nested_history in entry["nestedHistories"] ], } return { "currentUsedStepIndex": history["currentUsedStepIndex"], "entries": [comparable_entry(entry) for entry in history["entries"]], "usedSteps": history["usedSteps"], } def summarize_history_snapshot(snapshot): ui = snapshot["ui"] web_content = snapshot["webContent"] return { "ui": { "entries": history_entry_urls(ui), "usedSteps": history_used_steps(ui), "currentUsedStepIndex": ui["currentUsedStepIndex"], "back": ui["backButtonEnabled"], "forward": ui["forwardButtonEnabled"], "matches": ui["webContentHistoryMatchesUI"], "waitingToSeedWebContent": ui["waitingToSeedWebContent"], "waitingForWebContentSeedAck": ui["waitingForWebContentSeedAck"], "ignoringWebContentUpdatesUntilSeed": ui["ignoringWebContentUpdatesUntilSeed"], "reseedAfterCurrentHistoryLoad": ui["reseedAfterCurrentHistoryLoad"], "webContentUsesUIStepCoordinates": ui["webContentUsesUIStepCoordinates"], "webContentKnownUsedSteps": history_step_values(ui["webContentKnownUsedSteps"]), "webContentCurrentStep": ui["webContentCurrentStep"], "pendingWebContentHistoryStepAfterFallbackLoad": ui["pendingWebContentHistoryStepAfterFallbackLoad"], "pendingSessionHistoryNavigation": ui["pendingSessionHistoryNavigation"], "pendingSessionHistoryTraversal": ui["pendingSessionHistoryTraversal"], "currentResource": history_current_entry(ui).get("resource"), }, "webContent": { "entries": history_entry_urls(web_content), "usedSteps": history_used_steps(web_content), "currentUsedStepIndex": web_content["currentUsedStepIndex"], "currentResource": history_current_entry(web_content).get("resource"), }, } def expect_ui_session_history( webdriver_port, session_id, label, expected_entry_urls, expected_used_steps, expected_current_used_step_index, expected_back_enabled, expected_forward_enabled, log, expect_web_content_matches_ui=None, expected_web_content_known_used_steps=None, expected_web_content_current_step=None, expected_waiting_to_seed_web_content=None, expected_waiting_for_web_content_seed_ack=None, expected_ignoring_web_content_updates_until_seed=None, expected_reseed_after_current_history_load=None, ): def ui_history_matches(ui): return ( history_entry_urls(ui) == expected_entry_urls and history_used_steps(ui) == expected_used_steps and ui["currentUsedStepIndex"] == expected_current_used_step_index and ui["backButtonEnabled"] is expected_back_enabled and ui["forwardButtonEnabled"] is expected_forward_enabled and ( expect_web_content_matches_ui is None or ui["webContentHistoryMatchesUI"] is expect_web_content_matches_ui ) and ( expected_web_content_known_used_steps is None or history_step_values(ui["webContentKnownUsedSteps"]) == expected_web_content_known_used_steps ) and ( expected_web_content_current_step is None or ui["webContentCurrentStep"] == expected_web_content_current_step ) and ( expected_web_content_current_step is None or history_current_step(ui["webContentKnownUsedSteps"]) == expected_web_content_current_step ) and ( expected_waiting_to_seed_web_content is None or ui["waitingToSeedWebContent"] is expected_waiting_to_seed_web_content ) and ( expected_waiting_for_web_content_seed_ack is None or ui["waitingForWebContentSeedAck"] is expected_waiting_for_web_content_seed_ack ) and ( expected_ignoring_web_content_updates_until_seed is None or ui["ignoringWebContentUpdatesUntilSeed"] is expected_ignoring_web_content_updates_until_seed ) and ( expected_reseed_after_current_history_load is None or ui["reseedAfterCurrentHistoryLoad"] is expected_reseed_after_current_history_load ) ) def web_content_matches_ui(snapshot): ui = snapshot["ui"] web_content = snapshot["webContent"] return not ( ui["waitingToSeedWebContent"] or ui["waitingForWebContentSeedAck"] or ui["ignoringWebContentUpdatesUntilSeed"] or ui["reseedAfterCurrentHistoryLoad"] or ui["pendingWebContentHistoryStepAfterFallbackLoad"] is not None or ui["pendingSessionHistoryNavigation"] is not None or ui["pendingSessionHistoryTraversal"] is not None or comparable_history(ui) != comparable_history(web_content) ) def raise_ui_mismatch(snapshot): raise AssertionError( f"Expected {label} UI history to be entries={expected_entry_urls}, usedSteps={expected_used_steps}, " f"currentUsedStepIndex={expected_current_used_step_index}, back={expected_back_enabled}, " f"forward={expected_forward_enabled}, webContentMatchesUI={expect_web_content_matches_ui}, " f"webContentKnownUsedSteps={expected_web_content_known_used_steps}, " f"webContentCurrentStep={expected_web_content_current_step}, " f"waitingToSeedWebContent={expected_waiting_to_seed_web_content}, " f"waitingForWebContentSeedAck={expected_waiting_for_web_content_seed_ack}, " f"ignoringWebContentUpdatesUntilSeed={expected_ignoring_web_content_updates_until_seed}; " f"reseedAfterCurrentHistoryLoad={expected_reseed_after_current_history_load}; " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) if expect_web_content_matches_ui: deadline = time.monotonic() + EVENT_TIMEOUT_SECONDS while True: snapshot = session_history(webdriver_port, session_id) if ui_history_matches(snapshot["ui"]) and web_content_matches_ui(snapshot): break if time.monotonic() >= deadline: if not ui_history_matches(snapshot["ui"]): raise_ui_mismatch(snapshot) raise AssertionError( f"Expected {label} WebContent history to match UI, got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) time.sleep(0.05) log.append(f"{label} history: {summarize_history_snapshot(snapshot)}") return snapshot snapshot = session_history(webdriver_port, session_id) if not ui_history_matches(snapshot["ui"]): raise_ui_mismatch(snapshot) log.append(f"{label} history: {summarize_history_snapshot(snapshot)}") return snapshot def wait_for_ui_session_history( webdriver_port, session_id, label, expected_entry_urls, expected_used_steps, expected_current_used_step_index, expected_back_enabled, expected_forward_enabled, log, expected_web_content_known_used_steps, expected_web_content_current_step, ): def matches_expected_history(snapshot): ui = snapshot["ui"] web_content = snapshot["webContent"] return ( history_entry_urls(ui) == expected_entry_urls and history_used_steps(ui) == expected_used_steps and ui["currentUsedStepIndex"] == expected_current_used_step_index and ui["backButtonEnabled"] is expected_back_enabled and ui["forwardButtonEnabled"] is expected_forward_enabled and ui["webContentHistoryMatchesUI"] and history_step_values(ui["webContentKnownUsedSteps"]) == expected_web_content_known_used_steps and ui["webContentCurrentStep"] == expected_web_content_current_step and history_current_step(ui["webContentKnownUsedSteps"]) == expected_web_content_current_step and not ui["waitingToSeedWebContent"] and not ui["waitingForWebContentSeedAck"] and not ui["ignoringWebContentUpdatesUntilSeed"] and not ui["reseedAfterCurrentHistoryLoad"] and ui["pendingWebContentHistoryStepAfterFallbackLoad"] is None and ui["pendingSessionHistoryNavigation"] is None and ui["pendingSessionHistoryTraversal"] is None and comparable_history(ui) == comparable_history(web_content) ) return wait_for_session_history(webdriver_port, session_id, label, matches_expected_history, log) def expect_beforeunload_cancels_webdriver_navigation( webdriver_port, session_id, url, label, expected_url, previous_beforeunload_count, expected_history_snapshot, log, ): request(webdriver_port, "POST", f"/session/{session_id}/timeouts", {"pageLoad": 1000}) request(webdriver_port, "POST", f"/session/{session_id}/url", {"url": url}) request(webdriver_port, "POST", f"/session/{session_id}/timeouts", {"pageLoad": 10000}) state = execute_script( webdriver_port, session_id, "return [location.href, window.beforeUnloadCount];", ) if state[0] != expected_url or state[1] <= previous_beforeunload_count: raise AssertionError(f"Expected beforeunload to cancel {label}, got {state}\n" + "\n".join(log)) ui_history = expected_history_snapshot["ui"] expect_ui_session_history( webdriver_port, session_id, f"after blocked {label}", history_entry_urls(ui_history), history_used_steps(ui_history), ui_history["currentUsedStepIndex"], ui_history["backButtonEnabled"], ui_history["forwardButtonEnabled"], log, expect_web_content_matches_ui=True, expected_web_content_known_used_steps=history_step_values(ui_history["webContentKnownUsedSteps"]), expected_web_content_current_step=ui_history["webContentCurrentStep"], ) return state def expect_javascript_noop_ui_load_does_not_change_history( webdriver_port, session_id, expected_url, expected_history_snapshot, log, ): load_url_from_ui(webdriver_port, session_id, "javascript:window.uiJavascriptNoopRan=true;void(0)") state = execute_script( webdriver_port, session_id, "return [location.href, window.uiJavascriptNoopRan === true];", ) if state != [expected_url, True]: raise AssertionError( f"Expected javascript:void(0) UI load to stay on {expected_url}, got {state}\n" + "\n".join(log) ) ui_history = expected_history_snapshot["ui"] expect_ui_session_history( webdriver_port, session_id, "after javascript:void(0) UI load", history_entry_urls(ui_history), history_used_steps(ui_history), ui_history["currentUsedStepIndex"], ui_history["backButtonEnabled"], ui_history["forwardButtonEnabled"], log, expect_web_content_matches_ui=True, expected_web_content_known_used_steps=history_step_values(ui_history["webContentKnownUsedSteps"]), expected_web_content_current_step=ui_history["webContentCurrentStep"], ) def expect_javascript_noop_webdriver_navigation_does_not_change_history( webdriver_port, session_id, expected_url, expected_history_snapshot, log, ): request(webdriver_port, "POST", f"/session/{session_id}/timeouts", {"pageLoad": 1000}) request(webdriver_port, "POST", f"/session/{session_id}/url", {"url": "javascript:void(0)"}) request(webdriver_port, "POST", f"/session/{session_id}/timeouts", {"pageLoad": 10000}) state = execute_script(webdriver_port, session_id, "return location.href;") if state != expected_url: raise AssertionError( f"Expected javascript:void(0) WebDriver navigation to stay on {expected_url}, got {state}\n" + "\n".join(log) ) ui_history = expected_history_snapshot["ui"] expect_ui_session_history( webdriver_port, session_id, "after javascript:void(0) WebDriver navigation", history_entry_urls(ui_history), history_used_steps(ui_history), ui_history["currentUsedStepIndex"], ui_history["backButtonEnabled"], ui_history["forwardButtonEnabled"], log, expect_web_content_matches_ui=True, expected_web_content_known_used_steps=history_step_values(ui_history["webContentKnownUsedSteps"]), expected_web_content_current_step=ui_history["webContentCurrentStep"], ) def expect_beforeunload_cancels_refresh( webdriver_port, session_id, expected_url, previous_beforeunload_count, log, ): expect_current_ui_entry_reload_pending(webdriver_port, session_id, "before blocked refresh from /b", False, log) request(webdriver_port, "POST", f"/session/{session_id}/timeouts", {"pageLoad": 1000}) try: refresh(webdriver_port, session_id) finally: request(webdriver_port, "POST", f"/session/{session_id}/timeouts", {"pageLoad": 10000}) def refresh_was_canceled_by_beforeunload(result): return ( isinstance(result, list) and len(result) == 2 and result[0] == expected_url and isinstance(result[1], int) and result[1] > previous_beforeunload_count ) state = wait_for_script_result( webdriver_port, session_id, "beforeunload-canceled refresh from /b", "return [location.href, window.beforeUnloadCount];", refresh_was_canceled_by_beforeunload, log, ) expect_current_ui_entry_reload_pending(webdriver_port, session_id, "after blocked refresh from /b", False, log) return state def expect_current_top_level_history_url(webdriver_port, session_id, label, expected_url, log): snapshot = session_history(webdriver_port, session_id) current_entry = history_current_entry(snapshot["ui"]) if current_entry["url"] == expected_url: log.append(f"{label} current history URL: {expected_url}") return snapshot raise AssertionError( f"Expected {label} current history URL to be {expected_url}, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_navigation_buttons(webdriver_port, session_id, label, expected_back_enabled, expected_forward_enabled, log): snapshot = session_history(webdriver_port, session_id) ui = snapshot["ui"] if ui["backButtonEnabled"] is expected_back_enabled and ui["forwardButtonEnabled"] is expected_forward_enabled: log.append( f"{label} buttons: back={expected_back_enabled}, forward={expected_forward_enabled}, " f"history={summarize_history_snapshot(snapshot)}" ) return snapshot raise AssertionError( f"Expected {label} navigation buttons to be back={expected_back_enabled}, " f"forward={expected_forward_enabled}, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_web_content_session_history_matches_ui(webdriver_port, session_id, label, log): snapshot = session_history(webdriver_port, session_id) ui = snapshot["ui"] if ( ui["webContentHistoryMatchesUI"] and not ui["waitingToSeedWebContent"] and not ui["waitingForWebContentSeedAck"] and not ui["ignoringWebContentUpdatesUntilSeed"] and not ui["reseedAfterCurrentHistoryLoad"] and ui["pendingWebContentHistoryStepAfterFallbackLoad"] is None and ui["pendingSessionHistoryNavigation"] is None and ui["pendingSessionHistoryTraversal"] is None and comparable_history(ui) == comparable_history(snapshot["webContent"]) ): log.append(f"{label} matched history: {summarize_history_snapshot(snapshot)}") return snapshot raise AssertionError( f"Expected {label} WebContent history to match UI, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_pending_session_history_traversal( webdriver_port, session_id, label, expected_entry_urls, expected_used_steps, expected_current_used_step_index, expected_back_enabled, expected_forward_enabled, expected_stage, expected_will_replace_web_content_process, log, ): snapshot = session_history(webdriver_port, session_id) ui = snapshot["ui"] pending_navigation = ui["pendingSessionHistoryNavigation"] pending_traversal = ui["pendingSessionHistoryTraversal"] log.append(f"{label} pending traversal: {summarize_history_snapshot(snapshot)}") if history_entry_urls(ui) != expected_entry_urls: raise AssertionError(f"Expected {label} entries to be {expected_entry_urls}\n" + "\n".join(log)) if history_used_steps(ui) != expected_used_steps: raise AssertionError(f"Expected {label} used steps to be {expected_used_steps}\n" + "\n".join(log)) if ui["currentUsedStepIndex"] != expected_current_used_step_index: raise AssertionError( f"Expected {label} current used step index to be {expected_current_used_step_index}\n" + "\n".join(log) ) if ( ui["backButtonEnabled"] is not expected_back_enabled or ui["forwardButtonEnabled"] is not expected_forward_enabled ): raise AssertionError( f"Expected {label} buttons to be back={expected_back_enabled}, forward={expected_forward_enabled}\n" + "\n".join(log) ) if pending_traversal is None: raise AssertionError(f"Expected {label} to have a pending session history traversal\n" + "\n".join(log)) if pending_navigation is None: raise AssertionError(f"Expected {label} to have a pending session history navigation\n" + "\n".join(log)) if pending_navigation["url"] != expected_entry_urls[expected_current_used_step_index]: raise AssertionError( f"Expected {label} pending navigation URL to be " f"{expected_entry_urls[expected_current_used_step_index]}, got {pending_navigation['url']}\n" + "\n".join(log) ) if pending_navigation["previousCurrentURL"] == pending_navigation["url"]: raise AssertionError( f"Expected {label} pending navigation to preserve the previous current URL\n" + "\n".join(log) ) expected_step = expected_used_steps[expected_current_used_step_index] if pending_traversal["targetStep"] != expected_step: raise AssertionError( f"Expected {label} pending traversal target step to be {expected_step}, " f"got {pending_traversal['targetStep']}\n" + "\n".join(log) ) if pending_traversal["targetStepIndex"] != expected_current_used_step_index: raise AssertionError( f"Expected {label} pending traversal target step index to be {expected_current_used_step_index}, " f"got {pending_traversal['targetStepIndex']}\n" + "\n".join(log) ) if pending_traversal["stage"] != expected_stage: raise AssertionError( f"Expected {label} pending traversal stage to be {expected_stage}, " f"got {pending_traversal['stage']}\n" + "\n".join(log) ) if pending_traversal["willReplaceWebContentProcess"] is not expected_will_replace_web_content_process: raise AssertionError( f"Expected {label} pending traversal willReplaceWebContentProcess to be " f"{expected_will_replace_web_content_process}, got " f"{pending_traversal['willReplaceWebContentProcess']}\n" + "\n".join(log) ) expected_restore_mode = ( "restore-from-ui-process" if expected_will_replace_web_content_process else "preserve-current-process-state" ) if pending_navigation["webContentRestoreMode"] != expected_restore_mode: raise AssertionError( f"Expected {label} pending navigation restore mode to be {expected_restore_mode}, " f"got {pending_navigation['webContentRestoreMode']}\n" + "\n".join(log) ) if ui["webContentHistoryMatchesUI"]: raise AssertionError(f"Expected {label} WebContent history to still be catching up\n" + "\n".join(log)) return snapshot def expect_current_entry_nested_history(webdriver_port, session_id, label, expected_url, expected_nested_urls, log): snapshot = session_history(webdriver_port, session_id) ui_current_entry = history_current_entry(snapshot["ui"]) web_content_current_entry = history_current_entry(snapshot["webContent"]) ui_nested_urls = [] if ui_current_entry["nestedHistories"]: ui_nested_urls = [entry["url"] for entry in ui_current_entry["nestedHistories"][0]["entries"]] web_content_nested_urls = [] if web_content_current_entry["nestedHistories"]: web_content_nested_urls = [entry["url"] for entry in web_content_current_entry["nestedHistories"][0]["entries"]] if ( ui_current_entry["url"] == expected_url and web_content_current_entry["url"] == expected_url and ui_nested_urls == expected_nested_urls and web_content_nested_urls == expected_nested_urls ): log.append(f"{label} nested history: {expected_nested_urls}") return snapshot raise AssertionError( f"Expected {label} nested history to be {expected_nested_urls} at {expected_url}, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_current_ui_entry_reload_pending(webdriver_port, session_id, label, expected_reload_pending, log): snapshot = session_history(webdriver_port, session_id) ui_current_entry = history_current_entry(snapshot["ui"]) if ui_current_entry["reloadPending"] is expected_reload_pending: log.append(f"{label} UI reload pending: {expected_reload_pending}") return snapshot raise AssertionError( f"Expected {label} current UI entry reloadPending to be {expected_reload_pending}, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def nested_history_urls_for_entry(history, url): matching_entries = [entry for entry in history["entries"] if entry["url"] == url] if len(matching_entries) > 1: raise AssertionError(f"Expected at most one entry for {url}, got {matching_entries}") if not matching_entries: return None entry = matching_entries[0] if not entry["nestedHistories"]: return [] return [nested_entry["url"] for nested_entry in entry["nestedHistories"][0]["entries"]] def expect_entry_nested_history(webdriver_port, session_id, label, entry_url, expected_nested_urls, log): snapshot = session_history(webdriver_port, session_id) ui_nested_urls = nested_history_urls_for_entry(snapshot["ui"], entry_url) web_content_nested_urls = nested_history_urls_for_entry(snapshot["webContent"], entry_url) if ui_nested_urls == expected_nested_urls and web_content_nested_urls == expected_nested_urls: log.append(f"{label} nested history: {expected_nested_urls}") return snapshot raise AssertionError( f"Expected {label} nested history to be {expected_nested_urls} at {entry_url}, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_pending_web_content_history_step_after_fallback_load(webdriver_port, session_id, label, log): snapshot = session_history(webdriver_port, session_id) ui = snapshot["ui"] web_content = snapshot["webContent"] pending_step = ui["pendingWebContentHistoryStepAfterFallbackLoad"] pending_traversal = ui["pendingSessionHistoryTraversal"] ui_current_step = history_used_steps(ui)[ui["currentUsedStepIndex"]] web_content_current_step = history_used_steps(web_content)[web_content["currentUsedStepIndex"]] log.append( f"{label} pending fallback step: {pending_step}, " f"pendingTraversal={pending_traversal}, " f"webContentCurrentStep={ui['webContentCurrentStep']}, " f"matches={ui['webContentHistoryMatchesUI']}" ) if pending_step is None: raise AssertionError(f"Expected {label} to have a pending fallback step\n" + "\n".join(log)) if ui["webContentHistoryMatchesUI"]: raise AssertionError( f"Expected {label} WebContent history to be stale while step is pending\n" + "\n".join(log) ) if ui["waitingToSeedWebContent"] or ui["waitingForWebContentSeedAck"] or ui["reseedAfterCurrentHistoryLoad"]: raise AssertionError(f"Expected {label} WebContent seed to be applied\n" + "\n".join(log)) if pending_traversal is None: raise AssertionError(f"Expected {label} to have a pending session history traversal\n" + "\n".join(log)) if pending_traversal["targetStep"] != pending_step: raise AssertionError( f"Expected {label} pending traversal target {pending_traversal['targetStep']} " f"to match pending WebContent step {pending_step}\n" + "\n".join(log) ) if pending_traversal["stage"] != "restoring-nested-step-after-seed": raise AssertionError( f"Expected {label} pending traversal to be restoring nested step after seed, " f"got {pending_traversal['stage']}\n" + "\n".join(log) ) if pending_step != ui_current_step: raise AssertionError( f"Expected {label} pending step {pending_step} to match UI current step {ui_current_step}\n" + "\n".join(log) ) if ui["webContentCurrentStep"] != web_content_current_step: raise AssertionError( f"Expected {label} UI to report WebContent current step {web_content_current_step}, " f"got {ui['webContentCurrentStep']}\n" + "\n".join(log) ) if history_current_step(ui["webContentKnownUsedSteps"]) != web_content_current_step: raise AssertionError( f"Expected {label} known WebContent steps to mark current step {web_content_current_step}\n" + "\n".join(log) ) if ui["webContentCurrentStep"] == pending_step: raise AssertionError( f"Expected {label} WebContent to still be before pending step {pending_step}\n" + "\n".join(log) ) def expect_no_pending_web_content_history_step_after_fallback_load(webdriver_port, session_id, label, log): snapshot = session_history(webdriver_port, session_id) ui = snapshot["ui"] web_content = snapshot["webContent"] if ( ui["pendingWebContentHistoryStepAfterFallbackLoad"] is None and ui["webContentHistoryMatchesUI"] and not ui["waitingToSeedWebContent"] and not ui["waitingForWebContentSeedAck"] and not ui["ignoringWebContentUpdatesUntilSeed"] and not ui["reseedAfterCurrentHistoryLoad"] and ui["pendingSessionHistoryNavigation"] is None and ui["pendingSessionHistoryTraversal"] is None and ui["webContentCurrentStep"] == history_used_steps(ui)[ui["currentUsedStepIndex"]] and history_current_step(ui["webContentKnownUsedSteps"]) == ui["webContentCurrentStep"] and comparable_history(ui) == comparable_history(web_content) ): log.append(f"{label} restored fallback step: {summarize_history_snapshot(snapshot)}") return raise AssertionError( f"Expected {label} to clear pending fallback step and match WebContent, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_url(webdriver_port, session_id, label, expected_url, log): actual_url = current_url(webdriver_port, session_id) if actual_url == expected_url: log.append(f"{label}: {actual_url}") return raise AssertionError(f"Expected {label} to be {expected_url}, got {actual_url}\n" + "\n".join(log)) def wait_for_url(webdriver_port, session_id, label, expected_url, log, timeout=EVENT_TIMEOUT_SECONDS): deadline = time.monotonic() + timeout actual_url = None while time.monotonic() < deadline: actual_url = current_url(webdriver_port, session_id) if actual_url == expected_url: log.append(f"{label}: {actual_url}") return time.sleep(0.05) raise AssertionError(f"Timed out waiting for {label} to be {expected_url}, got {actual_url}\n" + "\n".join(log)) def expect_body_text(webdriver_port, session_id, label, expected_text, log): actual = execute_script(webdriver_port, session_id, "return [document.body.innerText.trim(), document.readyState];") if actual == [expected_text, "complete"]: log.append(f"{label}: {actual[0]}") return raise AssertionError(f"Expected {label} body text to be {expected_text}, got {actual}\n" + "\n".join(log)) def expect_current_entry_resource(webdriver_port, session_id, label, expected_resource, log): snapshot = session_history(webdriver_port, session_id) ui_current_entry = history_current_entry(snapshot["ui"]) web_content_current_entry = history_current_entry(snapshot["webContent"]) if ui_current_entry["resource"] == expected_resource and web_content_current_entry["resource"] == expected_resource: log.append(f"{label} current resource: {expected_resource}") return snapshot raise AssertionError( f"Expected {label} current resource to be {expected_resource}, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_current_ui_entry_resource(webdriver_port, session_id, label, expected_resource, log): snapshot = session_history(webdriver_port, session_id) ui_current_entry = history_current_entry(snapshot["ui"]) if ui_current_entry["resource"] == expected_resource: log.append(f"{label} current UI resource: {expected_resource}") return snapshot raise AssertionError( f"Expected {label} current UI resource to be {expected_resource}, " f"got {summarize_history_snapshot(snapshot)}\n" + "\n".join(log) ) def expect_frame_url(webdriver_port, session_id, label, expected_url, log, expected_title=None): result = execute_script( webdriver_port, session_id, """ const frame = document.getElementById('frame'); return [frame.contentWindow.location.href, frame.contentDocument.readyState, frame.contentDocument.title]; """, ) actual_url, ready_state, actual_title = result if ( actual_url == expected_url and ready_state == "complete" and (expected_title is None or actual_title == expected_title) ): log.append(f"{label}: {actual_url}") return actual = f"{actual_url} ({actual_title})" raise AssertionError(f"Expected {label} to be {expected_url}, got {actual}\n" + "\n".join(log)) def expect_history_entry_state( webdriver_port, session_id, label, expected_url, expected_classic_state, expected_navigation_state, expected_scroll_restoration, expected_navigation_key, expected_navigation_id, log, ): result = execute_script( webdriver_port, session_id, """ return [ location.href, JSON.stringify(history.state), JSON.stringify(navigation.currentEntry.getState()), history.scrollRestoration, navigation.currentEntry.key, navigation.currentEntry.id, ]; """, ) log.append(f"{label}: {result}") expected = [ expected_url, expected_classic_state, expected_navigation_state, expected_scroll_restoration, expected_navigation_key, expected_navigation_id, ] if result != expected: raise AssertionError(f"Expected {label} to be {expected}, got {result}\n" + "\n".join(log)) def expect_window_name(webdriver_port, session_id, label, expected_name, log): actual_name = execute_script(webdriver_port, session_id, "return window.name;") log.append(f"{label}: {actual_name}") if actual_name != expected_name: raise AssertionError(f"Expected {label} to be {expected_name}, got {actual_name}\n" + "\n".join(log)) def expect_scroll_position(webdriver_port, session_id, label, expected_x, expected_y, log): actual = execute_script( webdriver_port, session_id, "return [Math.round(scrollX), Math.round(scrollY), document.readyState];", ) if actual == [expected_x, expected_y, "complete"]: log.append(f"{label}: {actual}") return raise AssertionError( f"Expected {label} to be {[expected_x, expected_y, 'complete']}, got {actual}\n" + "\n".join(log) ) def run_blocked_process_swap_ui_forward_crash_recovery_test( webdriver_port, page_server, url_b, url_process_swap_back_blocked, ): session_id = create_session(webdriver_port) log = [f"blocked process-swap UI forward crash recovery initial: {current_url(webdriver_port, session_id)}"] with page_server.process_swap_back_blocked_request_lock: page_server.process_swap_back_blocked_request_count = 0 page_server.blocked_process_swap_back_requested.clear() page_server.process_swap_back_recovery_requested.clear() page_server.release_blocked_process_swap_back.clear() load_url_from_ui(webdriver_port, session_id, url_b) expect_url(webdriver_port, session_id, "after process-swap forward crash recovery setup /b", url_b, log) load_url_from_ui(webdriver_port, session_id, url_process_swap_back_blocked) expect_url( webdriver_port, session_id, "after process-swap forward crash recovery setup /process-swap-back-blocked", url_process_swap_back_blocked, log, ) expect_ui_session_history( webdriver_port, session_id, "after process-swap forward crash recovery setup", [url_b, url_process_swap_back_blocked], [0, 1], 1, True, False, log, expect_web_content_matches_ui=True, expected_web_content_known_used_steps=[0, 1], expected_web_content_current_step=1, ) traverse_history_from_ui(webdriver_port, session_id, -1) expect_url(webdriver_port, session_id, "after process-swap forward crash recovery setup back to /b", url_b, log) expect_ui_session_history( webdriver_port, session_id, "after process-swap forward crash recovery setup back to /b", [url_b, url_process_swap_back_blocked], [0, 1], 0, False, True, log, expect_web_content_matches_ui=True, expected_web_content_known_used_steps=[0, 1], expected_web_content_current_step=0, ) page_server.blocked_process_swap_back_requested.clear() page_server.process_swap_back_recovery_requested.clear() page_server.release_blocked_process_swap_back.clear() page_server.process_swap_back_document_ran.clear() traverse_history_from_ui(webdriver_port, session_id, 1, wait_for_navigation_completion=False) wait_for_event( page_server.blocked_process_swap_back_requested, "blocked process-swap UI forward before crash", ) expect_pending_session_history_traversal( webdriver_port, session_id, "while process-swap UI forward loads target before crash", [url_b, url_process_swap_back_blocked], [0, 1], 1, True, False, "replacing-webcontent-process", True, log, ) crash_current_page_allowing_navigation_timeout(webdriver_port, session_id) wait_for_event(page_server.process_swap_back_recovery_requested, "process-swap forward recovery request") expect_ui_session_history( webdriver_port, session_id, "while process-swap UI forward recovers target after crash", [url_b, url_process_swap_back_blocked], [0, 1], 1, True, False, log, expect_web_content_matches_ui=False, expected_waiting_to_seed_web_content=True, expected_waiting_for_web_content_seed_ack=False, expected_ignoring_web_content_updates_until_seed=True, expected_reseed_after_current_history_load=True, ) page_server.release_blocked_process_swap_back.set() wait_for_event(page_server.process_swap_back_document_ran, "process-swap UI forward recovery document") expect_url( webdriver_port, session_id, "after process-swap UI forward crash recovery completes", url_process_swap_back_blocked, log, ) expect_ui_session_history( webdriver_port, session_id, "after process-swap UI forward crash recovery completes", [url_b, url_process_swap_back_blocked], [0, 1], 1, True, False, log, expect_web_content_matches_ui=True, expected_web_content_known_used_steps=[0, 1], expected_web_content_current_step=1, ) request(webdriver_port, "DELETE", f"/session/{session_id}") def expect_second_ui_forward_during_pending_forward_does_not_hang( webdriver_port, session_id, page_server, url_a, url_forward_blocked, url_c, ): log = [f"second UI forward during pending forward initial: {current_url(webdriver_port, session_id)}"] page_server.block_forward_load = False page_server.blocked_forward_requested.clear() page_server.release_blocked_forward.clear() load_url_from_ui(webdriver_port, session_id, url_a) expect_url(webdriver_port, session_id, "after pending-forward setup /a", url_a, log) load_url_from_ui(webdriver_port, session_id, url_forward_blocked) expect_url(webdriver_port, session_id, "after pending-forward setup /forward-blocked", url_forward_blocked, log) load_url_from_ui(webdriver_port, session_id, url_c) expect_url(webdriver_port, session_id, "after pending-forward setup /c", url_c, log) expect_web_content_session_history_matches_ui(webdriver_port, session_id, "after pending-forward setup", log) traverse_history_from_ui(webdriver_port, session_id, -1) expect_url( webdriver_port, session_id, "after pending-forward setup back to /forward-blocked", url_forward_blocked, log ) traverse_history_from_ui(webdriver_port, session_id, -1) expect_url(webdriver_port, session_id, "after pending-forward setup back to /a", url_a, log) page_server.block_forward_load = True page_server.blocked_forward_requested.clear() page_server.release_blocked_forward.clear() traverse_history_from_ui(webdriver_port, session_id, 1, wait_for_navigation_completion=False) wait_for_event(page_server.blocked_forward_requested, "blocked first UI forward") second_forward_error = [] def request_second_forward(): try: traverse_history_from_ui(webdriver_port, session_id, 1) except Exception as error: second_forward_error.append(error) second_forward_thread = threading.Thread(target=request_second_forward) second_forward_thread.start() page_server.release_blocked_forward.set() second_forward_thread.join(timeout=EVENT_TIMEOUT_SECONDS) if second_forward_thread.is_alive(): raise AssertionError("Timed out waiting for second UI forward during pending forward\n" + "\n".join(log)) if second_forward_error: raise AssertionError( f"Second UI forward during pending forward failed: {second_forward_error[0]}\n" + "\n".join(log) ) expect_url(webdriver_port, session_id, "after second UI forward during pending forward", url_c, log) expect_ui_session_history( webdriver_port, session_id, "after second UI forward during pending forward", [url_a, url_forward_blocked, url_c], [0, 1, 2], 2, True, False, log, expect_web_content_matches_ui=True, expected_web_content_known_used_steps=[0, 1, 2], expected_web_content_current_step=2, ) page_server.block_forward_load = False def expect_sandboxed_history_back_to_be_blocked(webdriver_port, session_id, url_a, url_b, log): sandboxed_history_back_state = execute_async_script( webdriver_port, session_id, """ const done = arguments[0]; window.sandboxedHistoryBackResult = null; window.addEventListener("message", event => { if (event.data.startsWith("sandboxed-history-back:")) { window.sandboxedHistoryBackResult = event.data; done([location.href, window.sandboxedHistoryBackResult]); } }, { once: true }); const iframe = document.createElement("iframe"); iframe.sandbox = "allow-same-origin allow-scripts"; iframe.srcdoc = `