#![cfg(not(target_arch = "wasm32"))]
use anyhow::{anyhow, Context, Result};
use futures::StreamExt;
use iroh::RelayMode;
use openrtc::{
client::Client,
native_node::{AcceptEvent, IncomingStreamType},
signaling::{SessionEvent, SignalingSession},
};
use serde_json::json;
use tokio::time::{sleep, timeout, Duration};
fn install_rustls_provider() {
rustls::crypto::ring::default_provider()
.install_default()
.ok();
}
#[tokio::test]
#[ignore = "requires local/relay reachability and may be flaky on constrained CI networks"]
async fn rust_to_rust_iroh_connectivity_and_stream_open() -> Result<()> {
install_rustls_provider();
let client_a = Client::new(
"test-project".into(),
"pk_test_2222222222222222222222222222222222222222".into(),
Box::new(|| None),
);
let client_b = Client::new(
"test-project".into(),
"pk_test_2222222222222222222222222222222222222222".into(),
Box::new(|| None),
);
let endpoint_a = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
.alpns(vec![b"plutonium/p2p/1".to_vec()])
.relay_mode(RelayMode::Disabled)
.bind_addr("127.0.0.1:0")?
.bind()
.await?;
let endpoint_b = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
.alpns(vec![b"plutonium/p2p/1".to_vec()])
.relay_mode(RelayMode::Disabled)
.bind_addr("127.0.0.1:0")?
.bind()
.await?;
let mut endpoint_b_addr = endpoint_b.addr();
let addr_wait_start = std::time::Instant::now();
while endpoint_b_addr.is_empty() && addr_wait_start.elapsed() < Duration::from_secs(10) {
sleep(Duration::from_millis(100)).await;
endpoint_b_addr = endpoint_b.addr();
}
eprintln!(
"[rtc-matrix] endpoint_b addr ready={} elapsed_ms={}",
!endpoint_b_addr.is_empty(),
addr_wait_start.elapsed().as_millis()
);
if endpoint_b_addr.is_empty() {
return Err(anyhow!(
"endpoint_b did not publish a dialable address in time"
));
}
client_a.adopt_endpoint(endpoint_a).await;
client_b.adopt_endpoint(endpoint_b).await;
let endpoint_a = client_a.get_endpoint().await?;
let mut accept_events = client_b.subscribe_accept_events().await?;
sleep(Duration::from_millis(250)).await;
let connection = endpoint_a
.connect(endpoint_b_addr, b"plutonium/p2p/1")
.await
.context("rust-rust: endpoint connect failed")?;
let accept_event = timeout(Duration::from_secs(12), accept_events.next())
.await
.context("rust-rust: timed out waiting for accept event")?
.ok_or_else(|| anyhow!("missing accept event"))?;
assert!(matches!(accept_event, AcceptEvent::Accepted { .. }));
let (mut send, _recv) = connection.open_bi().await?;
send.write_all(b"ping").await?;
let incoming = client_b.incoming_streams().await?;
let stream = timeout(Duration::from_secs(12), incoming.recv())
.await
.context("rust-rust: timed out waiting for incoming stream")?
.map_err(|error| anyhow!("incoming stream recv failed: {}", error))?;
assert!(matches!(stream.stream, IncomingStreamType::Bi(_, _)));
Ok(())
}
#[tokio::test]
async fn rust_endpoint_handle_roundtrip_is_valid() -> Result<()> {
install_rustls_provider();
let client = Client::new(
"test-project".into(),
"pk_test_2222222222222222222222222222222222222222".into(),
Box::new(|| None),
);
client.init_iroh(None, vec![]).await?;
let endpoint = client.export_endpoint_handle().await?;
assert!(!endpoint.node_id.trim().is_empty());
assert!(!endpoint.node_addr.trim().is_empty());
let parsed_addr: iroh::EndpointAddr = serde_json::from_str(&endpoint.node_addr)?;
assert!(!parsed_addr.is_empty());
Ok(())
}
fn live_enabled() -> bool {
std::env::var("OPENRTC_LIVE_TESTS")
.map(|value| value == "1" || value.eq_ignore_ascii_case("true"))
.unwrap_or(false)
}
async fn get_anonymous_token() -> Result<(String, String)> {
let api_key = std::env::var("OPENRTC_FIREBASE_API_KEY")
.unwrap_or_else(|_| "AIzaSyAxyDE1xSYNAk5Ohe8VvKWi3xHOxB4cNV8".to_string());
let url = format!(
"https://identitytoolkit.googleapis.com/v1/accounts:signUp?key={}",
api_key
);
let client = reqwest::Client::new();
let response = client
.post(&url)
.json(&json!({ "returnSecureToken": true }))
.send()
.await?;
if !response.status().is_success() {
let body = response.text().await.unwrap_or_default();
return Err(anyhow!("firebase anonymous auth failed: {}", body));
}
let payload: serde_json::Value = response.json().await?;
let token = payload["idToken"]
.as_str()
.ok_or_else(|| anyhow!("missing idToken"))?;
let uid = payload["localId"]
.as_str()
.ok_or_else(|| anyhow!("missing localId"))?;
Ok((token.to_string(), uid.to_string()))
}
async fn wait_for_device(client: &Client, uid: &str, device_id: &str) -> Result<()> {
for _ in 0..20 {
let devices = client.search_devices(uid).await?;
if devices.iter().any(|device| device.device_id == device_id) {
return Ok(());
}
sleep(Duration::from_millis(500)).await;
}
Err(anyhow!(
"device '{}' not found in discovery window",
device_id
))
}
#[tokio::test]
#[ignore = "requires live Firebase credentials/network and can be flaky in CI"]
async fn live_auth_discovery_and_signaling_handshake() -> Result<()> {
if !live_enabled() {
eprintln!("skipping live test: set OPENRTC_LIVE_TESTS=1 to run");
return Ok(());
}
install_rustls_provider();
let (token, uid) = get_anonymous_token().await?;
let user_provider_a = {
let token = token.clone();
Box::new(move || Some(token.clone())) as Box<dyn Fn() -> Option<String> + Send + Sync>
};
let user_provider_b = {
let token = token.clone();
Box::new(move || Some(token.clone())) as Box<dyn Fn() -> Option<String> + Send + Sync>
};
let client_a = Client::new(
"pluto-rtc-prod".into(),
"pk_test_3333333333333333333333333333333333333333".into(),
user_provider_a,
);
let client_b = Client::new(
"pluto-rtc-prod".into(),
"pk_test_3333333333333333333333333333333333333333".into(),
user_provider_b,
);
let device_a = format!("device-a-{}", uuid::Uuid::new_v4());
let device_b = format!("device-b-{}", uuid::Uuid::new_v4());
client_a.set_node_id(device_a.clone()).await;
client_b.set_node_id(device_b.clone()).await;
client_a
.update_presence(
&uid,
"Rust Live A",
"ticket-a",
Some(&format!("{{\"deviceId\":\"{}\"}}", device_a)),
)
.await?;
client_b
.update_presence(
&uid,
"Rust Live B",
"ticket-b",
Some(&format!("{{\"deviceId\":\"{}\"}}", device_b)),
)
.await?;
wait_for_device(&client_a, &uid, &device_b).await?;
wait_for_device(&client_b, &uid, &device_a).await?;
let mut sessions = client_b.subscribe_sessions(&device_b).await?;
let session_id = format!("session-{}", uuid::Uuid::new_v4());
let session = SignalingSession {
connection_id: session_id.clone(),
initiator: uid.clone(),
target: uid.clone(),
initiator_device_id: device_a.clone(),
target_device_id: device_b.clone(),
connection_type: Some("webrtc".to_string()),
offer: Some(json!({ "type": "offer", "sdp": "fake-offer" })),
offer_e2ee: None,
answer: None,
answer_e2ee: None,
ice_candidates: vec![],
initiator_node_id: Some(device_a.clone()),
target_node_id: Some(device_b.clone()),
initiator_endpoint_addr: None,
target_endpoint_addr: None,
intent: Some("p2p".to_string()),
app_tag: Some("matrix-test".to_string()),
created_at: None,
expires_at: None,
state: "offer".to_string(),
};
client_a.create_session(session).await?;
let added_events = timeout(Duration::from_secs(20), sessions.next())
.await?
.ok_or_else(|| anyhow!("missing added session event"))??;
assert!(
added_events.iter().any(|event| {
matches!(
event,
SessionEvent::Added { session } if session.connection_id == session_id
)
}),
"expected added session event for created session"
);
client_a
.update_session(
&session_id,
json!({ "state": "answer", "answer": { "type": "answer", "sdp": "fake-answer" } }),
)
.await?;
let modified_events = timeout(Duration::from_secs(20), sessions.next())
.await?
.ok_or_else(|| anyhow!("missing modified session event"))??;
assert!(
modified_events.iter().any(|event| {
matches!(
event,
SessionEvent::Modified { session }
if session.connection_id == session_id && session.state == "answer"
)
}),
"expected modified session event after update"
);
let message_id = client_a
.send_message(&device_b, "hello-live", Some("offer"), None)
.await?;
assert!(!message_id.is_empty());
client_a.set_offline(&uid).await?;
client_b.set_offline(&uid).await?;
Ok(())
}