starred/moli
1
0
Fork 0
mirror of https://github.com/lexmount/moli.git synced 2026-11-04 16:06:44 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
4c7ad5fc9bdbfd123710ec0d2333204cd708a46a
moli/moli-protocol/src/testing.rs
T
ldm0 30e36e64a3 fix(cdp): preserve auxiliary Page ownership through document navigation 
Create script auxiliary Pages synchronously and let their CDP targets adopt
those exact Pages. Preserve Page and WindowProxy identity through document
navigation, sharing browsing-context names, opener references, message endpoints
and close state with related windows.

Commit browser responses into the existing Page with exact document admission,
cancellation and failure disposition. Capture accepted navigation origin, base,
policy, storage and referrer, preserve auxiliary history and restore related
browsing-context groups without issuing duplicate popup requests.

Reuse the initial same-origin Window while replacing its Document, and centralize
Document owner transitions, descendant retirement and Inspector publication.
Give related Pages independent Inspector routes and shared per-Page startup
pause state, route isolate foreground work through surviving Pages, and cache
cross-origin Window functions in each accessing realm with receiver-aware access.

Cover actual popup state, navigation and parser rebinding, accepted environments,
history, cross-origin access, multiple frontends, cancellation and task retirement.

Validation: cargo fmt --all; cargo clippy --workspace --all-targets --all-features -- -D warnings; cargo nextest run --no-fail-fast.
2026-10-06 01:01:03 +08:00

2267 lines
93 KiB
Rust
Raw Blame History

//! Test helpers for exercising the CDP connection and its scheduler.
//!
//! `TestContext` owns a `CdpConnection`, accepts JSON commands, and exposes
//! focused assertions over emitted results, events, and errors.
use serde_json::{Value, json};
use std::collections::VecDeque;
use std::time::Duration;
use super::conn::{
BackgroundProtocolEvent, CdpCommandTaskStep, CdpConnection, CdpInitialStoragePartition,
CdpSchedulerEvent, CommandDispatchContext, CommandResponseFlushPermit,
LoadedNavigationRendererAttachmentCommit, ParsedCdpCommand, PendingCdpCommandDispatch,
RuntimeInspectorResponseReady,
};
use crate::automation::{AutomationCommand, AutomationResult, DevToolsError};
use crate::domains::activity::{
ProtocolSchedulerWork, ProtocolSchedulerWorkKind, RuntimeCommandOutputBarrierCompletion,
RuntimeCommandOutputBarrierPermit, RuntimeCommandOutputBarriers,
};
use moli_core::{
LayoutPolicy, OptionalResourceFetchMask, RendererOutputFence, RendererOutputStreamControl,
RendererOutputStreamIdentity, RendererOutputTransportMessage, runtime::NavigationRuntimeConfig,
};
use moli_fetch::FetchConfig;
#[cfg(test)]
use std::net::SocketAddr;
#[cfg(test)]
use tokio::io::AsyncWriteExt;
#[cfg(test)]
use tokio::net::TcpListener;
#[cfg(test)]
use tokio::task::JoinHandle;
#[cfg(test)]
// Full-workspace runs execute CPU-heavy crypto vectors beside protocol tests.
// Keep this as a bounded diagnostic guard, but leave enough headroom for a
// real renderer-owner wake to be scheduled under that contention.
const TEST_SCHEDULER_INPUT_TIMEOUT: Duration = Duration::from_secs(30);
pub struct TestContext {
pub conn: CdpConnection,
pub sent: Vec<Value>,
pending_runtime_deferred_replies: VecDeque<PendingTestRuntimeDeferredReply>,
pending_protocol_scheduler_work: VecDeque<ProtocolSchedulerWork>,
runtime_command_output_barriers: RuntimeCommandOutputBarriers,
runtime_inspector_response_ready_rx:
tokio::sync::mpsc::UnboundedReceiver<RuntimeInspectorResponseReady>,
renderer_publication_rx: moli_core::RendererOutputTransportReceiver,
background_event_tx: tokio::sync::mpsc::UnboundedSender<BackgroundProtocolEvent>,
background_event_rx: tokio::sync::mpsc::UnboundedReceiver<BackgroundProtocolEvent>,
background_navigation_completion_tx:
tokio::sync::mpsc::UnboundedSender<crate::domains::page::BackgroundNavigationCompletion>,
background_navigation_completion_rx:
tokio::sync::mpsc::UnboundedReceiver<crate::domains::page::BackgroundNavigationCompletion>,
background_navigation_scheduler_enabled: bool,
}
struct PendingTestRuntimeDeferredReply {
pending: PendingCdpCommandDispatch,
command_context: CommandDispatchContext,
runtime_output_barrier: Option<RuntimeCommandOutputBarrierPermit>,
}
impl PendingTestRuntimeDeferredReply {
fn new(
pending: PendingCdpCommandDispatch,
command_context: CommandDispatchContext,
runtime_output_barrier: Option<RuntimeCommandOutputBarrierPermit>,
) -> Self {
Self {
pending,
command_context,
runtime_output_barrier,
}
}
fn command_id(&self) -> Option<u64> {
self.pending.command_id()
}
}
enum TestSchedulerWork {
ProtocolEvents(Vec<BackgroundProtocolEvent>),
SchedulerEvents(Vec<CdpSchedulerEvent>),
BackgroundEvent(BackgroundProtocolEvent),
BackgroundNavigationCompletion(crate::domains::page::BackgroundNavigationCompletion),
RuntimeDeferredReplyReady(RuntimeInspectorResponseReady),
RendererPublication(RendererOutputTransportMessage),
ReleaseRuntimeOutputBarrier(RuntimeCommandOutputBarrierPermit),
CancelRuntimeOutputBarrier(RuntimeCommandOutputBarrierPermit),
}
#[must_use = "the held test command response must be released exactly once"]
pub(crate) struct TestCommandResponseFlushPermit {
response_flush: CommandResponseFlushPermit,
runtime_output_barrier: Option<RuntimeCommandOutputBarrierPermit>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TestSchedulerTurnOutcome {
Idle,
Processed(TestSchedulerInputKind),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TestSchedulerInputKind {
BackgroundEvent,
BackgroundNavigationCompletion,
RuntimeDeferredReply,
RendererPublication,
}
impl Default for TestContext {
fn default() -> Self {
Self::new()
}
}
fn set_test_target_discovery(conn: &mut CdpConnection, enabled: bool) {
conn.set_root_target_discovery_enabled(enabled);
}
fn real_layout_test_runtime_config(
optional_resource_fetch_mask: OptionalResourceFetchMask,
) -> NavigationRuntimeConfig {
NavigationRuntimeConfig::new(
FetchConfig::default(),
optional_resource_fetch_mask,
true,
LayoutPolicy::OnDemand,
)
}
pub(crate) fn real_layout_test_connection() -> CdpConnection {
CdpConnection::new_with_initial_storage_partition_and_runtime_config(
CdpInitialStoragePartition::memory(),
real_layout_test_runtime_config(OptionalResourceFetchMask::NONE),
)
}
impl TestContext {
/// Explicitly publish fixture geometry before geometry-dependent commands.
pub(crate) async fn capture_fixture_layout(&mut self, session_id: Option<&str>) {
let mut request = json!({"id":9_000_000,"method":"Page.captureScreenshot"});
if let Some(session_id) = session_id {
request["sessionId"] = json!(session_id);
}
self.process_async(request).await;
let response = self.take_response_by_id(9_000_000);
assert!(
response["result"]["data"].is_string(),
"fixture screenshot: {response}"
);
}
/// Build the default CDP test harness used by older internal unit tests.
///
/// Historically those tests treated `Target.targetCreated` as a baseline
/// event after `Target.createTarget`. Real CDP clients only receive target
/// discovery events after enabling discovery, so keep that convenience in
/// `TestContext` instead of changing `CdpConnection::new()`.
pub fn new() -> Self {
Self::new_with_target_discovery(true)
}
/// Build a CDP test harness with explicit Target discovery state.
///
/// Use `true` for internal tests that assert Target-domain event payloads
/// without spelling out the setup command. Use `false` for Chromium-parity
/// tests that need to verify the default protocol behavior before
/// `Target.setDiscoverTargets(true)` is called.
pub fn new_with_target_discovery(target_discovery_enabled: bool) -> Self {
let mut conn = real_layout_test_connection();
set_test_target_discovery(&mut conn, target_discovery_enabled);
Self::from_conn(conn)
}
pub fn new_with_layout_policy(layout_policy: LayoutPolicy) -> Self {
let conn = CdpConnection::new_with_initial_storage_partition_and_runtime_config(
CdpInitialStoragePartition::memory(),
NavigationRuntimeConfig::new(
FetchConfig::default(),
OptionalResourceFetchMask::NONE,
true,
layout_policy,
),
);
Self::from_conn(conn)
}
pub fn new_with_target_discovery_and_image_fetch(
target_discovery_enabled: bool,
image_fetch_enabled: bool,
) -> Self {
Self::new_with_target_discovery_and_optional_resource_fetch_mask(
target_discovery_enabled,
if image_fetch_enabled {
OptionalResourceFetchMask::IMAGE
} else {
OptionalResourceFetchMask::NONE
},
)
}
pub fn new_with_target_discovery_and_optional_resource_fetch_mask(
target_discovery_enabled: bool,
optional_resource_fetch_mask: OptionalResourceFetchMask,
) -> Self {
let mut conn = CdpConnection::new_with_initial_storage_partition_and_runtime_config(
CdpInitialStoragePartition::memory(),
real_layout_test_runtime_config(optional_resource_fetch_mask),
);
set_test_target_discovery(&mut conn, target_discovery_enabled);
Self::from_conn(conn)
}
pub fn from_conn(mut conn: CdpConnection) -> Self {
let (renderer_publication_tx, renderer_publication_rx) =
moli_core::renderer_output_transport_channel();
let (runtime_inspector_response_ready_tx, runtime_inspector_response_ready_rx) =
tokio::sync::mpsc::unbounded_channel();
let (background_event_tx, background_event_rx) = tokio::sync::mpsc::unbounded_channel();
let (background_navigation_completion_tx, background_navigation_completion_rx) =
tokio::sync::mpsc::unbounded_channel();
conn.set_renderer_publication_sender(renderer_publication_tx);
conn.set_runtime_inspector_response_ready_sender(runtime_inspector_response_ready_tx);
Self {
conn,
sent: Vec::new(),
pending_runtime_deferred_replies: VecDeque::new(),
pending_protocol_scheduler_work: VecDeque::new(),
runtime_command_output_barriers: RuntimeCommandOutputBarriers::default(),
runtime_inspector_response_ready_rx,
renderer_publication_rx,
background_event_tx,
background_event_rx,
background_navigation_completion_tx,
background_navigation_completion_rx,
background_navigation_scheduler_enabled: false,
}
}
/// Enables the same asynchronous navigation channels used by the socket
/// scheduler.
///
/// Most protocol unit tests intentionally dispatch a domain command to
/// completion without owning an actor. Chromium-ordering and lifecycle
/// tests must opt into this production boundary: Page.navigate emits its
/// start/early response first, while the later renderer Page commit arrives
/// independently and is joined by its exact concrete-output cursor.
pub(crate) fn enable_background_navigation_scheduler_for_test(&mut self) {
if self.background_navigation_scheduler_enabled {
return;
}
self.conn
.set_background_event_sender(self.background_event_tx.clone());
self.conn.set_background_navigation_completion_sender(
self.background_navigation_completion_tx.clone(),
);
self.background_navigation_scheduler_enabled = true;
}
/// Loads and installs one production-shaped navigation fixture for the
/// exact Page target addressed by `session_id`.
///
/// Tests that expect owner-produced lifecycle or child-frame output must
/// not insert a bare `Page` into the runtime slot: production installs the
/// renderer Page and its exact root-Document lifecycle binding together.
/// Keeping that invariant in one helper prevents protocol tests from
/// accidentally depending on a state that the real navigation path never
/// exposes.
pub(crate) async fn install_navigation_fixture_for_session_owner(
&mut self,
raw_url: &str,
session_id: Option<&str>,
) {
self.conn.commit_declared_session_fixtures_for_test();
let navigation = self
.conn
.load_navigation_via_runtime_for_session_owner_async(session_id, raw_url)
.await
.expect("navigation fixture should load");
self.install_loaded_navigation_fixture_for_session_owner(navigation, session_id)
.await;
}
/// Installs an in-memory response through the same target/Page ownership
/// transaction as a real navigation.
///
/// Tests that need a non-fetchable origin (for example an IndexedDB
/// fixture at `https://example.test`) use this entry point. Building the
/// renderer Page first and assigning it directly to `TargetRuntimeSlot`
/// skips the concrete-output owner binding and is not a state production
/// can expose.
pub(crate) async fn install_buffered_navigation_fixture_for_session_owner(
&mut self,
requested_url: url::Url,
response_body: String,
session_id: Option<&str>,
) {
self.conn.commit_declared_session_fixtures_for_test();
let navigation = self
.conn
.build_loaded_navigation_from_buffered_response_for_session_owner_async(
session_id,
requested_url,
"GET".into(),
Vec::new().into(),
200,
Vec::new(),
response_body,
)
.await
.expect("buffered navigation fixture should build");
self.install_loaded_navigation_fixture_for_session_owner(navigation, session_id)
.await;
}
async fn install_loaded_navigation_fixture_for_session_owner(
&mut self,
navigation: crate::conn::LoadedNavigation,
session_id: Option<&str>,
) {
let owner = crate::conn::CommandOwnerScope::capture(&self.conn, session_id);
let (_, target_id) = self
.conn
.target_owner_identity_for_owner(&owner)
.expect("navigation fixture requires an installed browser context");
let target_id = target_id.expect("navigation fixture requires an exact target");
let renderer_output_predecessor = navigation.renderer_output_predecessor;
let page_creation_artifacts = navigation.page_creation_artifacts;
let final_url = navigation.final_url;
let main_document_commit = navigation
.main_document_commit
.expect("navigation fixture must retain its frozen Document commit identity");
let page_commit = self
.conn
.commit_loaded_navigation_page_for_owner_async(
&owner,
navigation.page,
LoadedNavigationRendererAttachmentCommit::Prepare(None),
&final_url,
)
.await
.expect("navigation fixture target must remain installed")
.expect("navigation fixture Page commit must succeed");
assert!(
page_commit
.committed_document_post_response_continuation
.is_none(),
"lifecycle-target fixture must not retain a DocumentCommit response gate"
);
let _ = self
.conn
.commit_loaded_navigation_target_identity_for_owner(
&owner,
&main_document_commit,
&final_url,
);
let (binding, _) = self.conn.bind_renderer_document_lifecycle_for_owner(
&owner,
page_creation_artifacts,
None,
target_id,
crate::domains::page::LOADER_ID.to_owned(),
);
let binding =
binding.expect("navigation fixture must install its exact renderer Document binding");
assert_eq!(
self.conn
.target_root_document_lifecycle_identity_for_owner(&owner,),
Some(binding.renderer_document_identity()),
"navigation fixture must retain its exact renderer Document binding"
);
if let Some(predecessor) = renderer_output_predecessor {
// Production does not expose a completed navigation response until
// the Page-creation cursor has crossed ordered protocol ingress.
// Mirror that boundary here so a later enable command observes
// the already-loaded target tail instead of racing publications
// that merely happen to remain queued in the test transport.
self.route_renderer_output_predecessor_before_command_response(predecessor)
.await;
}
}
/// Feed a JSON-serialisable message through the async CDP entrypoint and
/// route the scheduler work directly requested by that command. Tests
/// waiting for later external lifecycle input must use the event-wait
/// helpers, which consume one renderer wake or runtime reply at a time.
pub async fn process_async(&mut self, msg: impl serde::Serialize) {
let command = ParsedCdpCommand::from_serializable(msg)
.expect("test message must be a valid serialisable CDP command");
let command_id = command.request().id();
let session_id = command.request().session_id().map(str::to_owned);
let command_awaits_promise = match command.request().method() {
"Runtime.awaitPromise" => true,
"Runtime.evaluate" | "Runtime.callFunctionOn" | "Runtime.runScript" => command
.request()
.params()
.and_then(|params| params.get("awaitPromise"))
.and_then(Value::as_bool)
.unwrap_or(false),
_ => false,
};
let response_start = self.sent.len();
Box::pin(self.process_parsed_command_like_scheduler(&command, true)).await;
Box::pin(self.route_ready_test_command_response(command_id, response_start)).await;
if self
.conn
.renderer_runtime_command_cause_for_frontend(session_id.as_deref(), command_id)
.is_some()
&& !(command_awaits_promise
&& self
.conn
.has_pending_inspector_awaits_for_session_owner(session_id.as_deref()))
&& !self
.pending_runtime_deferred_replies
.iter()
.any(|pending| pending.command_id() == Some(command_id))
{
Box::pin(self.wait_for_test_command_response(command_id, response_start)).await;
}
}
/// Dispatch one command and keep running the real scheduler inputs until
/// that command's response is routed. This mirrors Chromium's synchronous
/// DevTools test client without draining unrelated page work to idle.
pub async fn process_and_wait_for_response_async(&mut self, msg: impl serde::Serialize) {
let command = ParsedCdpCommand::from_serializable(msg)
.expect("test message must be a valid serialisable CDP command");
let command_id = command.request().id();
let response_start = self.sent.len();
Box::pin(self.process_parsed_command_like_scheduler(&command, true)).await;
Box::pin(self.wait_for_test_command_response(command_id, response_start)).await;
}
#[cfg(test)]
pub(crate) fn enable_page_events_for_test(&mut self, session_id: Option<&str>) {
assert!(
self.conn
.set_page_domain_enabled_for_session_owner(session_id, true),
"test Page subscription requires a loaded target session owner"
);
}
#[cfg(test)]
pub(crate) fn enable_dom_events_for_test(&mut self, session_id: Option<&str>) {
assert!(
self.conn
.with_target_devtools_session_state_for_session_mut(session_id, |state| {
state.dom_session_state.enabled = true;
})
.is_some(),
"test DOM subscription requires a loaded target session owner"
);
}
/// Wait for and remove one protocol message produced by a real scheduler
/// input. It does not synthesize a capture or ask a command path to advance
/// renderer work: it only routes renderer publications and deferred
/// inspector replies already published by the production owner scheduler.
#[cfg(test)]
pub(crate) async fn wait_for_scheduler_message(
&mut self,
description: &str,
mut matches: impl FnMut(&Value) -> bool,
) -> Value {
let message = tokio::time::timeout(TEST_SCHEDULER_INPUT_TIMEOUT, async {
loop {
if let Some(position) = self.sent.iter().position(&mut matches) {
return self.sent.remove(position);
}
assert!(
matches!(
Box::pin(self.wait_for_one_test_scheduler_turn()).await,
TestSchedulerTurnOutcome::Processed(_)
),
"test scheduler lost all external input while waiting for {description}"
);
}
})
.await;
match message {
Ok(message) => message,
Err(_) => panic!(
"timed out waiting for {description} from a real scheduler input; sent={:?}",
self.sent
),
}
}
/// Wait until routing real scheduler input makes connection state satisfy
/// `predicate`. This observes owner-published wakes only; it does not
/// synthesize a capture or ask protocol code to advance renderer lifecycle.
#[cfg(test)]
pub(crate) async fn wait_until_scheduler_state(
&mut self,
description: &str,
predicate: impl Fn(&CdpConnection) -> bool,
) {
let waited = tokio::time::timeout(TEST_SCHEDULER_INPUT_TIMEOUT, async {
loop {
if predicate(&self.conn) {
return;
}
assert!(
matches!(
Box::pin(self.wait_for_one_test_scheduler_turn()).await,
TestSchedulerTurnOutcome::Processed(_)
),
"test scheduler lost all external input while waiting for {description}"
);
}
})
.await;
if waited.is_err() {
panic!(
"timed out waiting for {description} from a real scheduler input; sent={:?}; diagnostics={}",
self.sent,
self.conn.moli_memory_diagnostics()
);
}
}
/// Feed one command through the async CDP entrypoint without completing
/// deferred protocol residences afterwards, returning the scheduler events
/// that production would handle after the command response turn.
///
/// Use this for protocol sequences that intentionally send another command
/// before idle/runtime work gets a chance to run. The production socket
/// actor has a client-turn boundary between command output and deferred
/// CDP activity; eager test draining would otherwise make the test observe
/// a stronger, more synchronous ordering than real clients get.
#[cfg(test)]
pub(crate) async fn process_command_only_async(
&mut self,
msg: impl serde::Serialize,
) -> Vec<CdpSchedulerEvent> {
let command = ParsedCdpCommand::from_serializable(msg)
.expect("test message must be a valid serialisable CDP command");
Box::pin(self.process_parsed_command_like_scheduler(&command, false)).await
}
/// Dispatch one command through the production response-flush boundary,
/// but leave that boundary held so tests can inspect causally later page
/// work before the wire response is considered flushed.
#[cfg(test)]
pub(crate) async fn process_command_holding_response_flush_for_test(
&mut self,
msg: impl serde::Serialize,
) -> TestCommandResponseFlushPermit {
let command = ParsedCdpCommand::from_serializable(msg)
.expect("test message must be a valid serialisable CDP command");
let command_id = command.request().id();
let response_start = self.sent.len();
let (response_flush_permit, response_flush_context) =
self.conn.begin_command_response_flush_permit();
let mut command_context = CommandDispatchContext::new(response_flush_context);
let step = self
.conn
.start_parsed_command_dispatch_with_context(&command, &mut command_context);
let mut runtime_output_barrier = self.admit_runtime_output_barrier(&command);
let mut protocol_events = Vec::new();
let mut scheduler_events = Vec::new();
let completed = Box::pin(self.complete_command_step_like_scheduler(
step,
&mut command_context,
&mut protocol_events,
&mut scheduler_events,
&mut runtime_output_barrier,
))
.await;
assert!(
completed,
"held-flush test helper requires an immediate command response boundary"
);
protocol_events.extend(command_context.take_protocol_events());
protocol_events.extend(command_context.take_post_response_events());
scheduler_events.extend(self.conn.take_scheduler_events());
Box::pin(self.route_test_scheduler_causal_batch(protocol_events, scheduler_events)).await;
Box::pin(self.route_ready_test_command_response(command_id, response_start)).await;
TestCommandResponseFlushPermit {
response_flush: response_flush_permit,
runtime_output_barrier,
}
}
/// Releases a response boundary held by
/// [`Self::process_command_holding_response_flush_for_test`].
///
/// The response-flush permit becomes visible before command-owned
/// after-response output, matching the production scheduler's
/// `finish_command_dispatch_output_flush` ordering.
pub(crate) async fn finish_held_command_response_flush_for_test(
&mut self,
permit: TestCommandResponseFlushPermit,
) {
permit.response_flush.finish();
if let Some(runtime_output_barrier) = permit.runtime_output_barrier {
Box::pin(
self.release_runtime_output_barrier_like_scheduler(runtime_output_barrier, true),
)
.await;
}
}
/// Collects the renderer events routed during completion together with
/// the command messages. Domain tests need that prefix to inspect native
/// notifications rather than observing only the terminal response.
#[cfg(test)]
pub(crate) async fn complete_command_task_step_with_events_for_test(
&mut self,
step: CdpCommandTaskStep,
) -> (Vec<Value>, Vec<CdpSchedulerEvent>) {
let sent_start = self.sent.len();
let (messages, scheduler_events) = self.complete_command_task_step_for_test(step).await;
let mut published: Vec<_> = self.sent.drain(sent_start..).collect();
published.extend(messages);
(published, scheduler_events)
}
#[cfg(test)]
pub(crate) async fn complete_command_task_step_for_test(
&mut self,
step: CdpCommandTaskStep,
) -> (Vec<Value>, Vec<CdpSchedulerEvent>) {
let sent_start = self.sent.len();
let command_id = match &step {
CdpCommandTaskStep::Pending(pending) => pending.command_id(),
CdpCommandTaskStep::Complete(_) => None,
};
let mut protocol_events = Vec::new();
let mut scheduler_events = Vec::new();
let mut runtime_output_barrier = None;
let completed = Box::pin(self.complete_command_step_like_scheduler(
step,
&mut crate::conn::CommandDispatchContext::default(),
&mut protocol_events,
&mut scheduler_events,
&mut runtime_output_barrier,
))
.await;
if completed {
let mut messages = protocol_events_into_messages(protocol_events);
if let Some(command_id) = command_id
&& !messages
.iter()
.any(|message| message.get("id").and_then(Value::as_u64) == Some(command_id))
{
// Synchronous renderer-owned Runtime commands publish their
// terminal response through the DevTools-session output
// transport. The command completion only proves that the
// publication was committed; admit that independently
// transported response just as `process_async` does, while
// leaving sibling renderer events in the scheduler queue.
Box::pin(self.route_ready_test_command_response(command_id, sent_start)).await;
if let Some(position) = self.sent.get(sent_start..).and_then(|sent| {
sent.iter()
.position(|message| {
message.get("id").and_then(Value::as_u64) == Some(command_id)
})
.map(|position| sent_start + position)
}) {
messages.push(self.sent.remove(position));
}
}
return (messages, scheduler_events);
}
Box::pin(self.route_test_scheduler_causal_batch(protocol_events, scheduler_events)).await;
if !self.pending_runtime_deferred_replies.is_empty() {
let _ = Box::pin(self.run_one_ready_test_scheduler_turn()).await;
}
let messages = self.sent.drain(sent_start..).collect();
(messages, Vec::new())
}
/// Routes the output of one direct protocol-neutral command through the
/// stateful test scheduler.
///
/// Direct `AutomationCommand` fixtures bypass parsed CDP dispatch but can
/// still publish protocol work whose exact owner action is not ready yet.
/// Keeping that work resident here mirrors the production scheduler;
/// callers must use the scheduler wait helpers for later renderer input.
#[cfg(test)]
pub(crate) async fn route_direct_command_output_for_test(
&mut self,
protocol_events: Vec<BackgroundProtocolEvent>,
scheduler_events: Vec<CdpSchedulerEvent>,
) {
Box::pin(self.route_test_scheduler_causal_batch(protocol_events, scheduler_events)).await;
}
/// Waits until every concrete scheduler work item published by a direct
/// command has completed.
///
/// This is the protocol-neutral counterpart of waiting for a CDP event:
/// WebDriver commands can require the same owner action without enabling a
/// CDP domain, so their tests must synchronize with the work residence
/// itself rather than manufacture a frontend subscription.
#[cfg(test)]
pub(crate) async fn wait_for_direct_command_work_completion_for_test(
&mut self,
description: &str,
) {
let waited = tokio::time::timeout(TEST_SCHEDULER_INPUT_TIMEOUT, async {
while !self.pending_protocol_scheduler_work.is_empty() {
assert!(
matches!(
Box::pin(self.wait_for_one_test_scheduler_turn()).await,
TestSchedulerTurnOutcome::Processed(_)
),
"test scheduler lost all external input while waiting for {description}"
);
}
})
.await;
if waited.is_err() {
panic!(
"timed out waiting for {description}; pending protocol work={:?}",
self.pending_protocol_scheduler_work
);
}
}
async fn process_parsed_command_like_scheduler(
&mut self,
command: &ParsedCdpCommand,
drain_after_command: bool,
) -> Vec<CdpSchedulerEvent> {
// Some low-level fixtures mutate target-side session declarations
// after installing their BrowserContext. Materialize those declarations
// as committed attachments at the test scheduler boundary; production
// route lookup remains control-plane-only.
self.conn.commit_declared_session_fixtures_for_test();
let output_session_id = command.command_output_session_id().map(str::to_owned);
let output_owner =
crate::conn::CommandOwnerScope::capture(&self.conn, output_session_id.as_deref());
let mut protocol_events = Vec::new();
let mut scheduler_events = Vec::new();
let mut command_context = crate::conn::CommandDispatchContext::default();
let step = self
.conn
.start_parsed_command_dispatch_with_context(command, &mut command_context);
let mut runtime_output_barrier = self.admit_runtime_output_barrier(command);
let completed = Box::pin(self.complete_command_step_like_scheduler(
step,
&mut command_context,
&mut protocol_events,
&mut scheduler_events,
&mut runtime_output_barrier,
))
.await;
if drain_after_command && completed {
crate::domains::activity::project_protocol_local_command_outputs(
&mut self.conn,
&output_owner,
&mut command_context,
)
.await;
protocol_events.extend(command_context.take_protocol_events());
protocol_events.extend(command_context.take_post_response_events());
scheduler_events.extend(self.conn.take_scheduler_events());
}
if drain_after_command {
Box::pin(self.route_test_scheduler_causal_batch(protocol_events, scheduler_events))
.await;
if completed {
if let Some(runtime_output_barrier) = runtime_output_barrier {
Box::pin(self.release_runtime_output_barrier_like_scheduler(
runtime_output_barrier,
true,
))
.await;
}
} else {
assert!(
runtime_output_barrier.is_none(),
"a pending Runtime command must transfer its output barrier to the pending reply"
);
}
if !completed && !self.pending_runtime_deferred_replies.is_empty() {
let _ = Box::pin(self.run_one_ready_test_scheduler_turn()).await;
}
Vec::new()
} else {
Box::pin(self.route_test_scheduler_causal_batch(protocol_events, Vec::new())).await;
if completed {
if let Some(runtime_output_barrier) = runtime_output_barrier {
let mut release_scheduler_events =
Box::pin(self.release_runtime_output_barrier_like_scheduler(
runtime_output_barrier,
false,
))
.await;
scheduler_events.append(&mut release_scheduler_events);
}
} else {
assert!(
runtime_output_barrier.is_none(),
"a pending Runtime command must transfer its output barrier to the pending reply"
);
}
if !completed && !self.pending_runtime_deferred_replies.is_empty() {
let _ = Box::pin(self.run_one_ready_test_scheduler_turn()).await;
}
scheduler_events
}
}
fn admit_runtime_output_barrier(
&mut self,
command: &ParsedCdpCommand,
) -> Option<RuntimeCommandOutputBarrierPermit> {
command
.runtime_command_executes_page_javascript()
.then(|| {
self.runtime_command_output_barriers.admit(
&self.conn,
command.request().id(),
command.command_output_session_id(),
)
})
.flatten()
}
async fn complete_command_step_like_scheduler(
&mut self,
mut step: CdpCommandTaskStep,
command_context: &mut CommandDispatchContext,
protocol_events: &mut Vec<BackgroundProtocolEvent>,
scheduler_events: &mut Vec<CdpSchedulerEvent>,
runtime_output_barrier: &mut Option<RuntimeCommandOutputBarrierPermit>,
) -> bool {
loop {
match step {
CdpCommandTaskStep::Complete(outcome) => {
let (
mut events,
mut post_renderer_output_events,
renderer_output_boundary,
mut post_response_events,
mut new_scheduler_events,
mut renderer_output_predecessor,
) = outcome.into_renderer_owner_turn_parts();
if let Some(command_predecessor) =
command_context.take_renderer_output_predecessor()
{
command_predecessor
.merge_into_same_stream_tail(&mut renderer_output_predecessor);
}
if let Some(predecessor) = renderer_output_predecessor {
Box::pin(
self.route_renderer_output_predecessor_before_command_response(
predecessor,
),
)
.await;
}
protocol_events.append(&mut events);
if let Some(renderer_output_boundary) = renderer_output_boundary {
// The production actor flushes the already-materialized
// prefix, admits the exact renderer publication, then
// continues with the suffix. Do the same here instead
// of letting the test harness's final batch flatten the
// independently transported commit in front of its
// Page.navigate/Fetch responses.
Box::pin(self.route_test_scheduler_causal_batch(
std::mem::take(protocol_events),
Vec::new(),
))
.await;
Box::pin(
self.route_renderer_output_predecessor_before_command_response(
renderer_output_boundary,
),
)
.await;
}
protocol_events.append(&mut post_renderer_output_events);
protocol_events.append(&mut post_response_events);
scheduler_events.append(&mut new_scheduler_events);
return true;
}
CdpCommandTaskStep::Pending(mut pending)
if pending.waits_for_scheduler_deferred_inspector_reply() =>
{
let owner = pending
.runtime_command_owner_scope()
.expect("deferred Runtime commands have an exact owner")
.clone();
protocol_events
.extend(pending.take_scheduler_deferred_inspector_reply_events());
crate::domains::activity::project_protocol_local_command_outputs(
&mut self.conn,
&owner,
command_context,
)
.await;
protocol_events.extend(command_context.take_protocol_events());
protocol_events.extend(command_context.take_post_response_events());
scheduler_events.extend(self.conn.take_scheduler_events());
self.enqueue_pending_runtime_deferred_reply(
*pending,
std::mem::take(command_context),
runtime_output_barrier.take(),
);
return false;
}
CdpCommandTaskStep::Pending(pending) => {
// Keep the test scheduler's pending-command boundary shaped
// like production. Some domain completions carry a complete
// renderer Page build, so composing both futures inline can
// exceed Rust's default test-thread stack before the owner
// turn gets a chance to yield.
let completed = Box::pin(pending.wait()).await;
step = Box::pin(self.conn.complete_pending_command_dispatch_with_context(
completed,
command_context,
))
.await;
}
}
}
}
/// Project the exact concrete renderer output owned by a command before
/// exposing that command's response.
///
/// Production performs the same fence in
/// `flush_renderer_publication_predecessor`: the command carries a cursor,
/// while the publication itself arrives over the renderer transport. The
/// test scheduler must consume that real transport input rather than
/// rescanning renderer state or draining unrelated work. Merely removing
/// the publication from the ordered transport is insufficient: async
/// owner actions in that batch must have returned as well.
async fn route_renderer_output_predecessor_before_command_response(
&mut self,
predecessor: RendererOutputFence,
) {
let mut observed_transport = Vec::new();
let cursor = predecessor.cursor();
let projected = tokio::time::timeout(TEST_SCHEDULER_INPUT_TIMEOUT, async {
while !self.conn.renderer_output_cursor_is_projected(cursor) {
let publication = self
.renderer_publication_rx
.recv()
.await
.expect("renderer output transport closed before command predecessor");
observed_transport.push(format!("{publication:?}"));
let mut work =
VecDeque::from([TestSchedulerWork::RendererPublication(publication)]);
Box::pin(self.route_test_scheduler_work_queue(&mut work)).await;
}
})
.await;
assert!(
projected.is_ok(),
"timed out waiting for renderer output predecessor {predecessor:?}; \
observed transport={observed_transport:#?}"
);
}
/// Admits one direct command's exact renderer fence through the production-
/// shaped transport and ordered ingress.
///
/// Protocol-neutral command tests do not have a parsed CDP response for
/// `complete_command_step_like_scheduler()` to hold. They still must route
/// the cursor rather than inspecting renderer state or manufacturing the
/// owner action that the concrete publication contains.
#[cfg(test)]
pub(crate) async fn route_direct_command_renderer_predecessor_for_test(
&mut self,
predecessor: RendererOutputFence,
) {
Box::pin(self.route_renderer_output_predecessor_before_command_response(predecessor)).await;
}
/// Completes one protocol-neutral command across the same concrete
/// renderer-output boundary as the production actor.
pub(crate) async fn execute_automation_command_through_renderer_fence_for_test(
&mut self,
command: AutomationCommand,
) -> Result<AutomationResult, DevToolsError> {
let (result, scheduler_events, protocol_events, renderer_output_predecessor) = self
.conn
.execute_automation_command(command)
.await
.into_complete_parts();
if let Some(predecessor) = renderer_output_predecessor {
self.route_direct_command_renderer_predecessor_for_test(predecessor)
.await;
}
self.route_direct_command_output_for_test(protocol_events, scheduler_events)
.await;
result
}
/// Routes an explicitly completed command turn through the same ordered
/// renderer boundary used by the production actor.
///
/// A few ownership tests drive `PendingCdpCommandDispatch` directly so
/// they can inspect its scheduler sidecars. They must still admit concrete
/// renderer output at its exact position rather than flattening the
/// outcome into a message-only vector.
pub(crate) async fn route_completed_command_outcome_for_test(
&mut self,
outcome: crate::CdpRendererOwnerTurnOutcome,
) -> (Vec<Value>, Vec<CdpSchedulerEvent>) {
let sent_start = self.sent.len();
let (
before_renderer_output,
post_renderer_output,
renderer_output_boundary,
post_response_events,
scheduler_events,
renderer_output_predecessor,
) = outcome.into_renderer_owner_turn_parts();
if let Some(predecessor) = renderer_output_predecessor {
Box::pin(self.route_renderer_output_predecessor_before_command_response(predecessor))
.await;
}
Box::pin(self.route_test_scheduler_causal_batch(before_renderer_output, Vec::new())).await;
if let Some(boundary) = renderer_output_boundary {
Box::pin(self.route_renderer_output_predecessor_before_command_response(boundary))
.await;
} else {
assert!(
post_renderer_output.is_empty(),
"post-renderer output requires an exact boundary"
);
}
let mut suffix = post_renderer_output;
suffix.extend(post_response_events);
Box::pin(self.route_test_scheduler_causal_batch(suffix, Vec::new())).await;
(self.sent.drain(sent_start..).collect(), scheduler_events)
}
async fn release_runtime_output_barrier_like_scheduler(
&mut self,
permit: RuntimeCommandOutputBarrierPermit,
route_scheduler_events: bool,
) -> Vec<CdpSchedulerEvent> {
let completion = self
.conn
.release_runtime_command_output_barrier_turn_async(
&mut self.runtime_command_output_barriers,
permit,
)
.await;
let (protocol_events, scheduler_events) =
completion.into_outcome().into_protocol_event_parts();
if route_scheduler_events {
Box::pin(self.route_test_scheduler_causal_batch(protocol_events, scheduler_events))
.await;
Vec::new()
} else {
Box::pin(self.route_test_scheduler_causal_batch(protocol_events, Vec::new())).await;
scheduler_events
}
}
fn enqueue_runtime_output_barrier_completion(
&mut self,
completion: RuntimeCommandOutputBarrierCompletion,
work: &mut VecDeque<TestSchedulerWork>,
) {
let (protocol_events, scheduler_events) =
completion.into_outcome().into_protocol_event_parts();
if !scheduler_events.is_empty() {
work.push_front(TestSchedulerWork::SchedulerEvents(scheduler_events));
}
if !protocol_events.is_empty() {
work.push_front(TestSchedulerWork::ProtocolEvents(protocol_events));
}
}
async fn route_test_scheduler_causal_batch(
&mut self,
initial_events: Vec<BackgroundProtocolEvent>,
initial_scheduler_events: Vec<CdpSchedulerEvent>,
) {
// Route only output caused by the current scheduler input. Deferred
// external inputs remain separate turns, matching the actor.
let mut work = VecDeque::new();
if !initial_events.is_empty() {
work.push_back(TestSchedulerWork::ProtocolEvents(initial_events));
}
if !initial_scheduler_events.is_empty() {
work.push_back(TestSchedulerWork::SchedulerEvents(initial_scheduler_events));
}
Box::pin(self.route_test_scheduler_work_queue(&mut work)).await;
}
/// Routes one explicit renderer publication through the same capture,
/// barrier, and concrete-residence path as the production adapter.
///
/// This helper is for boundary tests that manufacture a typed publication
/// instead of receiving it from a live renderer. It does not provide a
/// direct output drain or a broad source scan.
pub(crate) async fn route_renderer_publication_for_test(
&mut self,
publication: RendererOutputTransportMessage,
) -> Vec<Value> {
let sent_start = self.sent.len();
let mut work = VecDeque::from([TestSchedulerWork::RendererPublication(publication)]);
Box::pin(self.route_test_scheduler_work_queue(&mut work)).await;
self.sent.drain(sent_start..).collect()
}
fn enqueue_pending_runtime_deferred_reply(
&mut self,
mut pending: PendingCdpCommandDispatch,
command_context: CommandDispatchContext,
runtime_output_barrier: Option<RuntimeCommandOutputBarrierPermit>,
) {
if let Some(command_id) = pending.command_id()
&& let Some(response_rx) = pending.take_scheduler_deferred_inspector_reply_receiver()
{
let session_id = pending.session_id().map(str::to_owned);
let response_tx = self
.conn
.runtime_inspector_response_ready_sender()
.expect("test scheduler should install its typed runtime-response channel");
tokio::spawn(async move {
let response = response_rx
.await
.map_err(|_| "RuntimeDeferredInspectorResponseCanceled".to_owned());
let _ = response_tx.send(RuntimeInspectorResponseReady::new(
command_id,
session_id.as_deref(),
response,
));
});
}
self.pending_runtime_deferred_replies
.push_back(PendingTestRuntimeDeferredReply::new(
pending,
command_context,
runtime_output_barrier,
));
}
async fn route_test_scheduler_work_queue(&mut self, work: &mut VecDeque<TestSchedulerWork>) {
self.route_ready_protocol_scheduler_work_for_test_context(work)
.await;
while let Some(item) = work.pop_front() {
match item {
TestSchedulerWork::ProtocolEvents(events) => {
Box::pin(self.route_protocol_events_like_scheduler(events, work)).await;
}
TestSchedulerWork::SchedulerEvents(scheduler_events) => {
Box::pin(self.route_scheduler_events_for_test_context(scheduler_events, work))
.await;
}
TestSchedulerWork::BackgroundEvent(event) => {
Box::pin(self.route_protocol_events_like_scheduler(vec![event], work)).await;
}
TestSchedulerWork::BackgroundNavigationCompletion(completion) => {
Box::pin(
self.route_background_navigation_completion_like_scheduler(
completion, work,
),
)
.await;
}
TestSchedulerWork::RuntimeDeferredReplyReady(response) => {
Box::pin(self.complete_runtime_response_ready_like_scheduler(
response,
Vec::new(),
work,
))
.await;
}
TestSchedulerWork::RendererPublication(publication) => {
Box::pin(self.ingest_renderer_publication_like_scheduler(publication, work))
.await;
}
TestSchedulerWork::ReleaseRuntimeOutputBarrier(permit) => {
let completion = self
.conn
.release_runtime_command_output_barrier_turn_async(
&mut self.runtime_command_output_barriers,
permit,
)
.await;
self.enqueue_runtime_output_barrier_completion(completion, work);
}
TestSchedulerWork::CancelRuntimeOutputBarrier(permit) => {
let completion = self
.conn
.cancel_runtime_command_output_barrier_turn_async(
&mut self.runtime_command_output_barriers,
permit,
)
.await;
self.enqueue_runtime_output_barrier_completion(completion, work);
}
}
self.route_ready_protocol_scheduler_work_for_test_context(work)
.await;
}
}
async fn route_background_navigation_completion_like_scheduler(
&mut self,
completion: crate::domains::page::BackgroundNavigationCompletion,
work: &mut VecDeque<TestSchedulerWork>,
) {
// Match the production actor's three-part boundary:
//
// already-produced navigation output
// -> exact renderer Page cursor
// -> navigation commit output
//
// The event and renderer transports are independent, so flattening
// them after the fact would allow the commit cursor to move the new
// realm in front of frameStartedNavigating/Page.navigate's response.
let mut prefix = Vec::new();
while let Ok(event) = self.background_event_rx.try_recv() {
prefix.push(event);
}
if !prefix.is_empty() {
Box::pin(self.route_protocol_events_like_scheduler(prefix, work)).await;
}
let outcome = self
.conn
.drain_background_navigation_completion_turn_async(completion)
.await;
let (
mut completion_prefix,
mut completion_suffix,
renderer_output_boundary,
mut post_response_events,
scheduler_events,
renderer_output_predecessor,
) = outcome.into_renderer_owner_turn_parts();
assert!(
renderer_output_predecessor.is_none(),
"background navigation completion must use an insertion boundary"
);
if !completion_prefix.is_empty() {
Box::pin(self.route_protocol_events_like_scheduler(
std::mem::take(&mut completion_prefix),
work,
))
.await;
}
if let Some(boundary) = renderer_output_boundary {
Box::pin(self.route_renderer_output_predecessor_before_command_response(boundary))
.await;
}
completion_suffix.append(&mut post_response_events);
while let Ok(event) = self.background_event_rx.try_recv() {
completion_suffix.push(event);
}
if !completion_suffix.is_empty() {
work.push_back(TestSchedulerWork::ProtocolEvents(completion_suffix));
}
if !scheduler_events.is_empty() {
work.push_back(TestSchedulerWork::SchedulerEvents(scheduler_events));
}
}
async fn run_one_ready_test_scheduler_turn(&mut self) -> TestSchedulerTurnOutcome {
let mut work = VecDeque::new();
let input_kind = if self.background_navigation_scheduler_enabled
&& let Ok(completion) = self.background_navigation_completion_rx.try_recv()
{
work.push_back(TestSchedulerWork::BackgroundNavigationCompletion(
completion,
));
TestSchedulerInputKind::BackgroundNavigationCompletion
} else if self.background_navigation_scheduler_enabled
&& let Ok(event) = self.background_event_rx.try_recv()
{
work.push_back(TestSchedulerWork::BackgroundEvent(event));
TestSchedulerInputKind::BackgroundEvent
} else if !self.pending_runtime_deferred_replies.is_empty() {
match self.runtime_inspector_response_ready_rx.try_recv() {
Ok(response) => {
work.push_back(TestSchedulerWork::RuntimeDeferredReplyReady(response));
TestSchedulerInputKind::RuntimeDeferredReply
}
Err(tokio::sync::mpsc::error::TryRecvError::Empty) => {
if let Ok(publication) = self.renderer_publication_rx.try_recv() {
work.push_back(TestSchedulerWork::RendererPublication(publication));
TestSchedulerInputKind::RendererPublication
} else {
return TestSchedulerTurnOutcome::Idle;
}
}
Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
return TestSchedulerTurnOutcome::Idle;
}
}
} else if let Ok(publication) = self.renderer_publication_rx.try_recv() {
work.push_back(TestSchedulerWork::RendererPublication(publication));
TestSchedulerInputKind::RendererPublication
} else if let Ok(response) = self.runtime_inspector_response_ready_rx.try_recv() {
work.push_back(TestSchedulerWork::RuntimeDeferredReplyReady(response));
TestSchedulerInputKind::RuntimeDeferredReply
} else {
return TestSchedulerTurnOutcome::Idle;
};
Box::pin(self.route_test_scheduler_work_queue(&mut work)).await;
TestSchedulerTurnOutcome::Processed(input_kind)
}
pub(crate) async fn wait_for_test_command_response(
&mut self,
command_id: u64,
response_start: usize,
) {
let response = tokio::time::timeout(TEST_SCHEDULER_INPUT_TIMEOUT, async {
loop {
if self
.sent
.get(response_start..)
.unwrap_or_default()
.iter()
.any(|message| message.get("id").and_then(Value::as_u64) == Some(command_id))
{
return;
}
assert!(
matches!(
Box::pin(self.wait_for_one_test_scheduler_turn()).await,
TestSchedulerTurnOutcome::Processed(_)
),
"CDP command `{command_id}` lost all scheduler input before its response"
);
}
})
.await;
if response.is_err() {
panic!(
"timed out waiting for CDP command `{command_id}` response; sent={:?}; diagnostics={}",
self.sent,
self.conn.moli_memory_diagnostics()
);
}
}
async fn route_ready_test_command_response(&mut self, command_id: u64, response_start: usize) {
while !self
.sent
.get(response_start..)
.unwrap_or_default()
.iter()
.any(|message| message.get("id").and_then(Value::as_u64) == Some(command_id))
&& matches!(
Box::pin(self.run_one_ready_test_scheduler_turn()).await,
TestSchedulerTurnOutcome::Processed(_)
)
{}
}
async fn wait_for_one_test_scheduler_turn(&mut self) -> TestSchedulerTurnOutcome {
let ready = Box::pin(self.run_one_ready_test_scheduler_turn()).await;
if matches!(ready, TestSchedulerTurnOutcome::Processed(_)) {
return ready;
}
let mut work = VecDeque::new();
let background_navigation_scheduler_enabled = self.background_navigation_scheduler_enabled;
let input_kind = if !self.pending_runtime_deferred_replies.is_empty() {
tokio::select! {
biased;
maybe_response = self.runtime_inspector_response_ready_rx.recv() => {
let Some(response) = maybe_response else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::RuntimeDeferredReplyReady(response));
TestSchedulerInputKind::RuntimeDeferredReply
}
maybe_completion = self.background_navigation_completion_rx.recv(), if background_navigation_scheduler_enabled => {
let Some(completion) = maybe_completion else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::BackgroundNavigationCompletion(completion));
TestSchedulerInputKind::BackgroundNavigationCompletion
}
maybe_event = self.background_event_rx.recv(), if background_navigation_scheduler_enabled => {
let Some(event) = maybe_event else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::BackgroundEvent(event));
TestSchedulerInputKind::BackgroundEvent
}
maybe_publication = self.renderer_publication_rx.recv() => {
let Some(publication) = maybe_publication else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::RendererPublication(publication));
TestSchedulerInputKind::RendererPublication
}
}
} else {
tokio::select! {
biased;
maybe_completion = self.background_navigation_completion_rx.recv(), if background_navigation_scheduler_enabled => {
let Some(completion) = maybe_completion else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::BackgroundNavigationCompletion(completion));
TestSchedulerInputKind::BackgroundNavigationCompletion
}
maybe_event = self.background_event_rx.recv(), if background_navigation_scheduler_enabled => {
let Some(event) = maybe_event else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::BackgroundEvent(event));
TestSchedulerInputKind::BackgroundEvent
}
maybe_publication = self.renderer_publication_rx.recv() => {
let Some(publication) = maybe_publication else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::RendererPublication(publication));
TestSchedulerInputKind::RendererPublication
}
maybe_response = self.runtime_inspector_response_ready_rx.recv() => {
let Some(response) = maybe_response else {
return TestSchedulerTurnOutcome::Idle;
};
work.push_back(TestSchedulerWork::RuntimeDeferredReplyReady(response));
TestSchedulerInputKind::RuntimeDeferredReply
}
}
};
Box::pin(self.route_test_scheduler_work_queue(&mut work)).await;
TestSchedulerTurnOutcome::Processed(input_kind)
}
async fn route_protocol_events_like_scheduler(
&mut self,
mut events: Vec<BackgroundProtocolEvent>,
work: &mut VecDeque<TestSchedulerWork>,
) {
loop {
if events.is_empty() {
return;
}
if let Some(position) = events
.iter()
.position(|event| event.as_runtime_inspector_response_ready().is_some())
{
self.sent.extend(
events
.drain(..position)
.map(BackgroundProtocolEvent::into_protocol_message),
);
let response = events
.remove(0)
.take_runtime_inspector_response_ready()
.expect("runtime response event position should contain typed response");
Box::pin(
self.complete_runtime_response_ready_like_scheduler(response, events, work),
)
.await;
return;
}
let pending_ids = self.pending_runtime_deferred_reply_command_ids();
if pending_ids.is_empty() {
self.sent.extend(
events
.into_iter()
.map(BackgroundProtocolEvent::into_protocol_message),
);
return;
}
let Some(position) = events.iter().position(|event| {
event
.protocol_message()
.and_then(|message| message.get("id"))
.and_then(Value::as_u64)
.is_some_and(|id| pending_ids.contains(&id))
}) else {
self.sent.extend(
events
.into_iter()
.map(BackgroundProtocolEvent::into_protocol_message),
);
return;
};
self.sent.extend(
events
.drain(..position)
.map(BackgroundProtocolEvent::into_protocol_message),
);
let message = events.remove(0).into_protocol_message();
let Some(command_id) = message.get("id").and_then(Value::as_u64) else {
self.sent.push(message);
continue;
};
let Some(index) = self
.pending_runtime_deferred_replies
.iter()
.position(|pending| pending.command_id() == Some(command_id))
else {
self.sent.push(message);
continue;
};
let mut pending = self
.pending_runtime_deferred_replies
.remove(index)
.expect("pending runtime deferred reply index should exist");
pending
.pending
.forget_scheduler_deferred_inspector_reply(&mut self.conn);
self.sent.push(json!({
"id": command_id,
"error": {
"code": -32000,
"message": "RuntimeDeferredReplyLooseProtocolResponse",
},
}));
if let Some(runtime_output_barrier) = pending.runtime_output_barrier.take() {
work.push_front(TestSchedulerWork::CancelRuntimeOutputBarrier(
runtime_output_barrier,
));
}
if !events.is_empty() {
work.push_front(TestSchedulerWork::ProtocolEvents(events));
}
return;
}
}
async fn route_scheduler_events_for_test_context(
&mut self,
scheduler_events: Vec<CdpSchedulerEvent>,
work: &mut VecDeque<TestSchedulerWork>,
) {
let mut queue = VecDeque::new();
enqueue_scheduler_events_like_scheduler(&mut queue, scheduler_events);
while let Some(TestDeferredSchedulerWork(protocol_work)) = queue.pop_front() {
self.pending_protocol_scheduler_work
.push_back(protocol_work);
let scheduler_events = self.conn.take_scheduler_events();
if !scheduler_events.is_empty() {
work.push_back(TestSchedulerWork::SchedulerEvents(scheduler_events));
}
}
}
async fn route_ready_protocol_scheduler_work_for_test_context(
&mut self,
work: &mut VecDeque<TestSchedulerWork>,
) {
loop {
let selected_index = match self.pending_protocol_scheduler_work.front() {
Some(front) if front.is_ready() => 0,
Some(front)
if front.kind() == ProtocolSchedulerWorkKind::MainDocumentLoadOwnerAction =>
{
let Some(index) =
self.pending_protocol_scheduler_work
.iter()
.position(|candidate| {
candidate.is_ready()
&& candidate.navigation_dependency()
== crate::ProtocolNavigationDependency::ReplacesPendingLoad
})
else {
return;
};
// Production checks the pending load observer out of the
// FIFO while it waits for the renderer. An unconstrained
// location owner action may then run and replace that
// exact source Document, which completes the observer as
// Superseded. Keeping the pending observer at the front in
// this protocol-only harness would deadlock the action
// needed to make it terminal.
index
}
Some(_) | None => return,
};
if !self.background_navigation_scheduler_enabled
&& self
.pending_protocol_scheduler_work
.get(selected_index)
.is_some_and(ProtocolSchedulerWork::requires_background_navigation_scheduler)
{
// The default protocol fixture has no owner task lane. Keep
// independent popup navigation resident rather than invoking
// the production function's synchronous fallback while the
// exact renderer cursor is still being projected. Tests that
// assert navigation progress opt into the production-shaped
// background scheduler and drive its typed completions.
return;
}
let protocol_work = self
.pending_protocol_scheduler_work
.remove(selected_index)
.expect("ready protocol work must remain resident");
let (events, scheduler_events) = self
.conn
.complete_ready_protocol_scheduler_work_turn(protocol_work)
.await
.into_protocol_event_parts();
if !events.is_empty() {
work.push_back(TestSchedulerWork::ProtocolEvents(events));
}
if !scheduler_events.is_empty() {
work.push_back(TestSchedulerWork::SchedulerEvents(scheduler_events));
}
}
}
async fn ingest_renderer_publication_like_scheduler(
&mut self,
publication: RendererOutputTransportMessage,
work: &mut VecDeque<TestSchedulerWork>,
) {
let scheduler_events = self.conn.take_scheduler_events();
if !scheduler_events.is_empty() {
work.push_front(TestSchedulerWork::RendererPublication(publication));
work.push_front(TestSchedulerWork::SchedulerEvents(scheduler_events));
return;
}
// Load-relative scheduling belongs to the adapter-level
// `CdpScheduler`, which captures this publication first and delays
// only its already-projected protocol batch. This protocol-only
// harness must not park the raw publication: doing so prevents the
// ordered ingress from admitting later records in the same renderer
// stream and can deadlock sequence N behind output produced at N + 1.
let outcome = self
.conn
.ingest_renderer_output_turn_async(
publication,
&mut self.runtime_command_output_barriers,
)
.await;
let (protocol_events, scheduler_events) = outcome.into_protocol_event_parts();
if !protocol_events.is_empty() {
work.push_back(TestSchedulerWork::ProtocolEvents(protocol_events));
}
if !scheduler_events.is_empty() {
work.push_back(TestSchedulerWork::SchedulerEvents(scheduler_events));
}
}
fn pending_runtime_deferred_reply_command_ids(&self) -> Vec<u64> {
self.pending_runtime_deferred_replies
.iter()
.filter_map(PendingTestRuntimeDeferredReply::command_id)
.collect()
}
async fn complete_runtime_response_ready_like_scheduler(
&mut self,
response: RuntimeInspectorResponseReady,
suffix_events: Vec<BackgroundProtocolEvent>,
work: &mut VecDeque<TestSchedulerWork>,
) {
// Production admits an exact renderer cursor before resolving the
// response correlation. Keep the test scheduler on the same boundary:
// a SharedWorker `Destroyed` publication must terminate the pending
// call with `Target closed` before a later V8 context-destruction
// response can become visible.
if let Some(predecessor) = response.renderer_output_predecessor() {
Box::pin(self.route_renderer_output_predecessor_before_command_response(predecessor))
.await;
}
let command_id = response.command_id();
let Some(index) = self
.pending_runtime_deferred_replies
.iter()
.position(|pending| pending.command_id() == Some(command_id))
else {
let mut response_events = Vec::new();
let mut background_events = Vec::new();
self.conn.route_registered_runtime_inspector_response_into(
response,
&mut response_events,
&mut background_events,
);
if !background_events.is_empty() {
work.push_front(TestSchedulerWork::ProtocolEvents(background_events));
}
self.sent.extend(
response_events
.into_iter()
.map(BackgroundProtocolEvent::into_protocol_message),
);
if !suffix_events.is_empty() {
work.push_front(TestSchedulerWork::ProtocolEvents(suffix_events));
}
return;
};
let pending = self
.pending_runtime_deferred_replies
.remove(index)
.expect("pending runtime deferred reply index should exist");
Box::pin(
self.complete_renderer_runtime_deferred_response_like_scheduler(
pending,
response,
suffix_events,
work,
),
)
.await;
}
async fn complete_renderer_runtime_deferred_response_like_scheduler(
&mut self,
mut pending: PendingTestRuntimeDeferredReply,
response: RuntimeInspectorResponseReady,
suffix_events: Vec<BackgroundProtocolEvent>,
work: &mut VecDeque<TestSchedulerWork>,
) {
if pending.pending.command_id().is_none() {
self.pending_runtime_deferred_replies.push_back(pending);
return;
}
pending
.pending
.route_scheduler_deferred_inspector_response(&mut self.conn, response)
.await;
Box::pin(self.complete_runtime_deferred_reply_like_scheduler(pending, suffix_events, work))
.await;
}
async fn complete_runtime_deferred_reply_like_scheduler(
&mut self,
mut pending: PendingTestRuntimeDeferredReply,
suffix_events: Vec<BackgroundProtocolEvent>,
work: &mut VecDeque<TestSchedulerWork>,
) {
let completed = pending
.pending
.complete_scheduler_deferred_inspector_reply(&mut self.conn);
let step = self
.conn
.complete_pending_command_dispatch_with_context(completed, &mut pending.command_context)
.await;
let mut protocol_events = Vec::new();
let mut scheduler_events = Vec::new();
let completed = Box::pin(self.complete_command_step_like_scheduler(
step,
&mut pending.command_context,
&mut protocol_events,
&mut scheduler_events,
&mut pending.runtime_output_barrier,
))
.await;
if !suffix_events.is_empty() {
work.push_front(TestSchedulerWork::ProtocolEvents(suffix_events));
}
if completed {
if let Some(runtime_output_barrier) = pending.runtime_output_barrier {
work.push_front(TestSchedulerWork::ReleaseRuntimeOutputBarrier(
runtime_output_barrier,
));
}
} else {
assert!(
pending.runtime_output_barrier.is_none(),
"a still-pending Runtime command must transfer its output barrier"
);
}
if !scheduler_events.is_empty() {
work.push_front(TestSchedulerWork::SchedulerEvents(scheduler_events));
}
if !protocol_events.is_empty() {
work.push_front(TestSchedulerWork::ProtocolEvents(protocol_events));
}
}
// ── Assertion helpers ─────────────────────────────────────────────────────
/// Assert that a `{id, result, sessionId?}` message exists in the sent
/// queue and remove it. `session_id = None` means "no sessionId field".
pub fn expect_result(&mut self, id: u64, result: Value, session_id: Option<&str>) {
let expected = build_result(id, &result, session_id);
self.find_and_remove(&expected, "result");
}
/// Assert that a `{id, error: {code, message}}` message exists.
pub fn expect_error(&mut self, id: u64, code: i32, message: &str) {
let expected = json!({ "id": id, "error": { "code": code, "message": message } });
self.find_and_remove(&expected, "error");
}
/// Assert that an event `{method, params, sessionId?}` message exists.
/// When `params` is `None`, only the method name is checked.
pub fn expect_event(&mut self, method: &str, params: Option<&Value>) {
let pos = self.sent.iter().position(|v| {
if v["method"].as_str() != Some(method) {
return false;
}
if let Some(expected_params) = params {
values_subset(expected_params, &v["params"])
} else {
true // any params
}
});
match pos {
Some(i) => {
self.sent.remove(i);
}
None => {
let queue: String = self.sent.iter().map(|v| format!(" {}\n", v)).collect();
panic!("expected event '{method}' not found in sent queue:\n{queue}");
}
}
}
/// Take and return the next message; panics if the queue is empty.
pub fn take_one(&mut self) -> Value {
if self.sent.is_empty() {
panic!("expected a message in the sent queue but it is empty");
}
self.sent.remove(0)
}
/// Take and return the first sent message matching the predicate.
pub fn take_first_matching(
&mut self,
description: &str,
matches: impl FnMut(&Value) -> bool,
) -> Value {
let pos = self
.sent
.iter()
.position(matches)
.unwrap_or_else(|| panic!("expected {description} in sent queue: {:?}", self.sent));
self.sent.remove(pos)
}
/// Take and return the response with the requested id.
pub fn take_response_by_id(&mut self, id: u64) -> Value {
let pos = self
.sent
.iter()
.position(|message| message["id"] == json!(id))
.unwrap_or_else(|| {
panic!(
"expected a response with id {id} in sent queue: {:?}",
self.sent
)
});
self.sent.remove(pos)
}
/// Drain all pending sent messages (useful to discard setup noise).
pub fn take_all(&mut self) -> Vec<Value> {
std::mem::take(&mut self.sent)
}
/// Completes at most one scheduler input that is already ready.
///
/// Unlike the removed broad test capture, this cannot snapshot an
/// arbitrary session. It only consumes a concrete renderer publication, Runtime
/// response, scheduler event, or ready `ProtocolSchedulerWork` already
/// resident in the production-shaped harness.
#[cfg(test)]
pub(crate) async fn complete_one_ready_scheduler_input_for_test(&mut self) {
let scheduler_events = self.conn.take_scheduler_events();
Box::pin(self.route_test_scheduler_causal_batch(Vec::new(), scheduler_events)).await;
let _ = Box::pin(self.run_one_ready_test_scheduler_turn()).await;
}
// ── Internal ──────────────────────────────────────────────────────────────
fn find_and_remove(&mut self, expected: &Value, kind: &str) {
let pos = self.sent.iter().position(|v| values_subset(expected, v));
match pos {
Some(i) => {
self.sent.remove(i);
}
None => {
let queue: String = self.sent.iter().map(|v| format!(" {}\n", v)).collect();
panic!("expected {kind} not found.\nExpected:\n {expected}\nSent queue:\n{queue}");
}
}
}
}
fn protocol_events_into_messages(events: Vec<BackgroundProtocolEvent>) -> Vec<Value> {
events
.into_iter()
.map(BackgroundProtocolEvent::into_protocol_message)
.collect()
}
#[cfg(test)]
pub(crate) fn protocol_events_into_internal_messages(
events: Vec<BackgroundProtocolEvent>,
) -> Vec<Value> {
events
.into_iter()
.map(|event| event.into_parts().0)
.collect()
}
#[cfg(test)]
pub(crate) async fn drain_scheduler_events_like_scheduler(
conn: &mut CdpConnection,
out: &mut Vec<Value>,
scheduler_events: Vec<CdpSchedulerEvent>,
) {
drain_scheduler_events_like_scheduler_with_materializer(
conn,
out,
scheduler_events,
BackgroundProtocolEvent::into_protocol_message,
)
.await;
}
#[cfg(test)]
pub(crate) async fn drain_scheduler_events_like_scheduler_preserving_internal_fields(
conn: &mut CdpConnection,
out: &mut Vec<Value>,
scheduler_events: Vec<CdpSchedulerEvent>,
) {
drain_scheduler_events_like_scheduler_with_materializer(
conn,
out,
scheduler_events,
protocol_event_into_internal_message,
)
.await;
}
#[cfg(test)]
async fn drain_scheduler_events_like_scheduler_with_materializer(
conn: &mut CdpConnection,
out: &mut Vec<Value>,
scheduler_events: Vec<CdpSchedulerEvent>,
materialize_event: fn(BackgroundProtocolEvent) -> Value,
) {
let mut queue = VecDeque::new();
enqueue_scheduler_events_like_scheduler(&mut queue, scheduler_events);
while let Some(TestDeferredSchedulerWork(protocol_work)) = queue.pop_front() {
assert!(
protocol_work.is_ready(),
"the stateless compatibility materializer cannot own pending protocol work; use TestContext or the production CdpScheduler"
);
let (events, nested_scheduler_events) = conn
.complete_ready_protocol_scheduler_work_turn(protocol_work)
.await
.into_protocol_event_parts();
out.extend(events.into_iter().map(materialize_event));
enqueue_scheduler_events_like_scheduler(&mut queue, nested_scheduler_events);
enqueue_scheduler_events_like_scheduler(&mut queue, conn.take_scheduler_events());
}
}
#[cfg(test)]
fn protocol_event_into_internal_message(event: BackgroundProtocolEvent) -> Value {
event.into_parts().0
}
#[cfg(test)]
/// Compatibility materializer for protocol-domain fixtures.
///
/// It preserves scheduler-event FIFO and concrete work residence, but does not
/// model adapter client-turn predecessors. Tests that claim scheduling or
/// ordering behavior must use the production `CdpScheduler` instead.
struct TestDeferredSchedulerWork(ProtocolSchedulerWork);
#[cfg(test)]
fn enqueue_scheduler_events_like_scheduler(
queue: &mut VecDeque<TestDeferredSchedulerWork>,
events: Vec<CdpSchedulerEvent>,
) {
for event in events {
match event {
CdpSchedulerEvent::ProtocolWorkPublished { work } => {
queue.push_back(TestDeferredSchedulerWork(work));
}
CdpSchedulerEvent::PageScreencastStarted { .. } => {}
}
}
}
#[cfg(test)]
pub(crate) trait TestSessionId<'a> {}
#[cfg(test)]
impl<'a> TestSessionId<'a> for Option<&'a str> {}
#[cfg(test)]
impl<'a> TestSessionId<'a> for &'a str {}
#[cfg(test)]
pub async fn spawn_connection_drop_server() -> (SocketAddr, JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let Ok((mut stream, _)) = listener.accept().await else {
return;
};
let _ = stream.shutdown().await;
});
(addr, server)
}
#[cfg(test)]
pub(crate) async fn wait_until_message(
ctx: &mut TestContext,
session_id: impl TestSessionId<'_> + Copy,
description: &str,
predicate: impl Fn(&Value) -> bool,
) {
wait_until_messages(ctx, session_id, description, |messages| {
messages.iter().any(&predicate)
})
.await;
}
#[cfg(test)]
pub(crate) async fn wait_until_messages(
ctx: &mut TestContext,
_session_id: impl TestSessionId<'_> + Copy,
description: &str,
predicate: impl Fn(&[Value]) -> bool,
) {
for _ in 0..256 {
if predicate(&ctx.sent) {
return;
}
ctx.complete_one_ready_scheduler_input_for_test().await;
if predicate(&ctx.sent) {
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
ctx.complete_one_ready_scheduler_input_for_test().await;
if predicate(&ctx.sent) {
return;
}
panic!("timed out waiting for {description}; sent={:?}", ctx.sent);
}
/// Wait for a protocol message produced by a real scheduler input without
/// synthesizing broad capture turns or moving messages already collected in
/// `ctx.sent`.
///
/// Resource and runtime completions publish typed wakes into `TestContext`.
/// Tests asserting those completion boundaries should block on that channel
/// instead of tying correctness to a polling iteration budget.
#[cfg(test)]
pub(crate) async fn wait_until_scheduler_message(
ctx: &mut TestContext,
description: &str,
predicate: impl Fn(&Value) -> bool,
) {
let waited = tokio::time::timeout(TEST_SCHEDULER_INPUT_TIMEOUT, async {
loop {
if ctx.sent.iter().any(&predicate) {
return;
}
assert!(
matches!(
Box::pin(ctx.wait_for_one_test_scheduler_turn()).await,
TestSchedulerTurnOutcome::Processed(_)
),
"test scheduler lost all external input while waiting for {description}"
);
}
})
.await;
if waited.is_err() {
panic!(
"timed out waiting for {description} from a real scheduler input; sent={:?}",
ctx.sent
);
}
}
/// Wait for the terminal loading event of one concrete frame while preserving
/// the complete event sequence for assertions that follow.
///
/// `Page.navigate` only acknowledges that navigation was accepted; Chromium
/// likewise allows document replacement to continue after that response. DOM
/// tests must synchronize with the frame lifecycle before retaining frontend
/// node ids from the new document.
#[cfg(test)]
pub(crate) async fn wait_until_frame_stopped_loading(ctx: &mut TestContext, frame_id: &str) {
let description = format!("Page.frameStoppedLoading for {frame_id}");
wait_until_scheduler_message(ctx, &description, |message| {
message["method"] == json!("Page.frameStoppedLoading")
&& message["params"]["frameId"] == json!(frame_id)
})
.await;
}
/// Wait for the renderer-owned load fact of one exact document generation.
///
/// Unlike `Page.frameStoppedLoading`, the authoritative binding carries the
/// loader id, so a test cannot accidentally accept a terminal event left by an
/// older document in the same frame.
#[cfg(test)]
pub(crate) async fn wait_until_renderer_document_load(
ctx: &mut TestContext,
session_id: Option<&str>,
frame_id: &str,
loader_id: &str,
) {
let description = format!("renderer load for {frame_id}/{loader_id}");
ctx.wait_until_scheduler_state(&description, |conn| {
conn.renderer_document_lifecycle_authoritative_state_for_session_owner(session_id)
.is_some_and(|(binding, snapshot)| {
binding.frame_id == frame_id
&& binding.loader_id == loader_id
&& snapshot.load.is_some()
})
})
.await;
}
/// Build a result message, omitting sessionId when it is None.
fn build_result(id: u64, result: &Value, session_id: Option<&str>) -> Value {
let mut v = json!({ "id": id, "result": result });
if let Some(sid) = session_id {
v["sessionId"] = json!(sid);
}
v
}
/// Return true when every field of `expected` appears in `actual` with the
/// same value. Arrays and nested objects are compared recursively.
fn values_subset(expected: &Value, actual: &Value) -> bool {
match expected {
Value::Object(exp_map) => {
let Value::Object(act_map) = actual else {
return false;
};
exp_map.iter().all(|(k, ev)| {
act_map
.get(k)
.map(|av| values_subset(ev, av))
.unwrap_or(false)
})
}
Value::Array(exp_arr) => {
let Value::Array(act_arr) = actual else {
return false;
};
if exp_arr.len() != act_arr.len() {
return false;
}
exp_arr
.iter()
.zip(act_arr.iter())
.all(|(e, a)| values_subset(e, a))
}
_ => expected == actual,
}
}
#[cfg(test)]
mod tests {
use moli_core::{PageId, RendererRuntimeInspectorAsyncCompletion};
use super::*;
#[tokio::test]
async fn test_context_drops_unmatched_typed_runtime_response_like_scheduler() {
let mut ctx = TestContext::new();
let response = RuntimeInspectorResponseReady::new(
42,
None,
Ok(
RendererRuntimeInspectorAsyncCompletion::from_protocol_message(
42,
json!({
"id": 42,
"result": {}
}),
),
),
);
ctx.complete_runtime_response_ready_like_scheduler(
response,
Vec::new(),
&mut VecDeque::new(),
)
.await;
assert!(
ctx.sent.is_empty(),
"unmatched typed runtime completion must stay internal in the test harness too"
);
}
#[tokio::test]
async fn test_context_consumes_selected_task_output_publications_one_per_turn() {
let mut conn = CdpConnection::new();
let (publication_tx, publication_rx) = moli_core::renderer_output_transport_channel();
let (runtime_response_tx, runtime_response_rx) = tokio::sync::mpsc::unbounded_channel();
let (background_event_tx, background_event_rx) = tokio::sync::mpsc::unbounded_channel();
let (background_navigation_completion_tx, background_navigation_completion_rx) =
tokio::sync::mpsc::unbounded_channel();
conn.set_renderer_publication_sender(publication_tx.clone());
conn.set_runtime_inspector_response_ready_sender(runtime_response_tx);
let mut ctx = TestContext {
conn,
sent: Vec::new(),
pending_runtime_deferred_replies: VecDeque::new(),
pending_protocol_scheduler_work: VecDeque::new(),
runtime_command_output_barriers: RuntimeCommandOutputBarriers::default(),
runtime_inspector_response_ready_rx: runtime_response_rx,
renderer_publication_rx: publication_rx,
background_event_tx,
background_event_rx,
background_navigation_completion_tx,
background_navigation_completion_rx,
background_navigation_scheduler_enabled: false,
};
let opened = |page_id| {
RendererOutputTransportMessage::from(RendererOutputStreamControl::Opened {
stream: RendererOutputStreamIdentity::new_page_for_protocol_test(
PageId::new_for_testing(page_id),
),
})
};
let first = opened(1);
let second = opened(2);
let second_stream = match &second {
RendererOutputTransportMessage::StreamControl(
RendererOutputStreamControl::Opened { stream },
) => *stream,
RendererOutputTransportMessage::StreamControl(
RendererOutputStreamControl::Closed { .. },
)
| RendererOutputTransportMessage::PageReservationReleased { .. }
| RendererOutputTransportMessage::CursorLeaseDeclared { .. }
| RendererOutputTransportMessage::CursorLeaseReleased { .. }
| RendererOutputTransportMessage::Publication(_) => {
unreachable!("test input is an opened stream control")
}
};
publication_tx.send(first).expect("first scheduler input");
publication_tx.send(second).expect("second scheduler input");
assert_eq!(
ctx.run_one_ready_test_scheduler_turn().await,
TestSchedulerTurnOutcome::Processed(TestSchedulerInputKind::RendererPublication)
);
let queued = ctx
.renderer_publication_rx
.try_recv()
.expect("second input must remain queued for the next turn");
assert!(matches!(
queued,
RendererOutputTransportMessage::StreamControl(
RendererOutputStreamControl::Opened { stream }
) if stream == second_stream
));
}
}
Reference in New Issue View Git Blame Copy Permalink