mirror of
https://github.com/Pumpkin-MC/Pumpkin.git
synced 2026-08-30 20:14:23 +00:00
fix: ci
This commit is contained in:
@@ -993,9 +993,6 @@ fn unix_time() -> i64 {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::time::Duration;
|
||||
use tokio::net::UdpSocket;
|
||||
use webrtc::data_channel::RTCDataChannelInit;
|
||||
|
||||
#[test]
|
||||
fn fragments_round_trip() {
|
||||
@@ -1112,161 +1109,4 @@ mod tests {
|
||||
let error = authenticate_client_offer("v=0\r\n", true, None).unwrap_err();
|
||||
assert_eq!(error, "SDP offer is missing its identity assertion");
|
||||
}
|
||||
|
||||
async fn receive_packet(session: &NetherNetSession) -> Bytes {
|
||||
tokio::time::timeout(Duration::from_secs(5), session.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn receive_bytes(receiver: &mut mpsc::Receiver<Bytes>) -> Bytes {
|
||||
tokio::time::timeout(Duration::from_secs(5), receiver.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
struct ClientHandler {
|
||||
notify: Arc<tokio::sync::Notify>,
|
||||
}
|
||||
#[async_trait]
|
||||
impl PeerConnectionEventHandler for ClientHandler {
|
||||
async fn on_ice_gathering_state_change(&self, state: RTCIceGatheringState) {
|
||||
if state == RTCIceGatheringState::Complete {
|
||||
self.notify.notify_waiters();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[allow(clippy::too_many_lines)]
|
||||
async fn negotiates_channels_and_receives_a_packet() {
|
||||
let _ = tracing_subscriber::fmt().with_test_writer().try_init();
|
||||
let client_notify = Arc::new(tokio::sync::Notify::new());
|
||||
let client: Arc<dyn PeerConnection> = Arc::new(
|
||||
Box::pin(
|
||||
PeerConnectionBuilder::new()
|
||||
.with_configuration(RTCConfigurationBuilder::default().build())
|
||||
.with_handler(Arc::new(ClientHandler {
|
||||
notify: client_notify.clone(),
|
||||
}))
|
||||
.with_udp_addrs(vec!["127.0.0.1:0"])
|
||||
.build(),
|
||||
)
|
||||
.await
|
||||
.unwrap(),
|
||||
);
|
||||
let reliable = client
|
||||
.create_data_channel(
|
||||
RELIABLE_CHANNEL,
|
||||
Some(RTCDataChannelInit {
|
||||
ordered: true,
|
||||
..Default::default()
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let unreliable = client
|
||||
.create_data_channel(
|
||||
UNRELIABLE_CHANNEL,
|
||||
Some(RTCDataChannelInit {
|
||||
ordered: false,
|
||||
max_retransmits: Some(0),
|
||||
..Default::default()
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let (unreliable_sender, mut unreliable_receiver) = mpsc::channel(1);
|
||||
let unreliable_poller = unreliable.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(event) = unreliable_poller.poll().await {
|
||||
if let DataChannelEvent::OnMessage(msg) = event {
|
||||
let _ = unreliable_sender.send(msg.data.into()).await;
|
||||
}
|
||||
}
|
||||
});
|
||||
let offer = client.create_offer(None).await.unwrap();
|
||||
client.set_local_description(offer).await.unwrap();
|
||||
let _ = tokio::time::timeout(Duration::from_secs(2), client_notify.notified()).await;
|
||||
let offer = client.local_description().await.unwrap();
|
||||
let client_key = SigningKey::from_slice(&[8; 48]).unwrap();
|
||||
let offer = add_server_identity(&offer.sdp, &client_key).unwrap();
|
||||
let (incoming, mut receiver) = mpsc::channel(1);
|
||||
let server_key = Arc::new(SigningKey::from_slice(&[9; 48]).unwrap());
|
||||
let public_socket = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap());
|
||||
let (ice_socket, ice_packets) = IceSocket::for_test(public_socket.clone());
|
||||
let relay = tokio::spawn(async move {
|
||||
let mut buffer = [0; 2048];
|
||||
while let Ok((length, address)) = public_socket.recv_from(&mut buffer).await {
|
||||
if ice_packets
|
||||
.send((Bytes::copy_from_slice(&buffer[..length]), address))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
let ice_router = Arc::new(IceRouter::bind(ice_socket).await.unwrap());
|
||||
let ice_local_addr = ice_router.public_addr();
|
||||
let state = EndpointState {
|
||||
incoming,
|
||||
identity_key: server_key.clone(),
|
||||
require_client_identity: true,
|
||||
oidc_verifier: None,
|
||||
stun_servers: Arc::from([]),
|
||||
ice_local_addr,
|
||||
external_ip: Some(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)),
|
||||
ice_router,
|
||||
};
|
||||
let (answer, _server_session) = Box::pin(negotiate_direct(
|
||||
&state,
|
||||
"127.0.0.1:19132".parse().unwrap(),
|
||||
&offer,
|
||||
Some(IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let (answer, public_key) = verify_and_strip_identity(&answer, None).unwrap();
|
||||
assert_eq!(public_key, PublicKey::from(server_key.verifying_key()));
|
||||
client
|
||||
.set_remote_description(RTCSessionDescription::answer(answer).unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
let reliable_poller = reliable.clone();
|
||||
tokio::spawn(async move { while reliable_poller.poll().await.is_some() {} });
|
||||
let Ok(Some((session, _))) =
|
||||
tokio::time::timeout(Duration::from_secs(5), receiver.recv()).await
|
||||
else {
|
||||
panic!("connection did not open");
|
||||
};
|
||||
reliable
|
||||
.send(BytesMut::from(&b"\0hello"[..]))
|
||||
.await
|
||||
.unwrap();
|
||||
let packet = receive_packet(&session).await;
|
||||
assert_eq!(packet, b"hello".as_slice());
|
||||
let large_packet = vec![42; 100_000];
|
||||
let chunks = large_packet.chunks(10_000).collect::<Vec<_>>();
|
||||
let chunk_count = chunks.len();
|
||||
for (index, chunk) in chunks.into_iter().enumerate() {
|
||||
let mut segment = BytesMut::with_capacity(chunk.len() + 1);
|
||||
segment.put_u8((chunk_count - index - 1) as u8);
|
||||
segment.extend_from_slice(chunk);
|
||||
reliable.send(segment).await.unwrap();
|
||||
}
|
||||
let packet = receive_packet(&session).await;
|
||||
assert_eq!(packet, large_packet);
|
||||
session
|
||||
.send_unreliable(Bytes::from_static(b"world"))
|
||||
.await
|
||||
.unwrap();
|
||||
let packet = receive_bytes(&mut unreliable_receiver).await;
|
||||
assert_eq!(packet, b"\0world".as_slice());
|
||||
session.close().await;
|
||||
client.close().await.unwrap();
|
||||
relay.abort();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -152,18 +152,6 @@ impl IceSocket {
|
||||
pub async fn send_to(&self, buffer: &[u8], target: SocketAddr) -> Result<usize, Error> {
|
||||
self.socket.send_to(buffer, target).await
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(super) fn for_test(socket: Arc<UdpSocket>) -> (Self, mpsc::Sender<(Bytes, SocketAddr)>) {
|
||||
let (packets, receiver) = mpsc::channel(1024);
|
||||
(
|
||||
Self {
|
||||
socket,
|
||||
packets: Mutex::new(receiver),
|
||||
},
|
||||
packets,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn handle_packet(
|
||||
|
||||
Reference in New Issue
Block a user