Add some unit tests for the RPC client.

src/network/client.rs

@@ -204,7 +204,7 @@ pub struct RPCInterface<PacketType, ErrorType> {
 }
 
 impl<PacketType, ErrorType> RPCInterface<PacketType, ErrorType> {
-    pub async fn call(&mut self, call: PacketType) -> RPC<PacketType, ErrorType> {
+    pub fn call(&mut self, call: PacketType) -> RPC<PacketType, ErrorType> {
         let (sender, receiver) = oneshot::channel();
         let call_data = CallData {
             data: call,
@@ -292,13 +292,136 @@ impl<PacketType> Stream for ServerEventStream<PacketType> {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq)]
 pub enum RPCError<StreamError> {
     Closed,
     Timeout,
     Stream(StreamError),
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::network::test_utils::{panic_after, Pipe};
+    use crate::network::Packet;
+
+    type TestPayload = u32;
+
+    fn serialize_packet<T>(id: u32, response: bool, payload: T) -> Vec<u8>
+    where
+        T: Serialize,
+    {
+        let packet = Packet {
+            id,
+            response,
+            payload,
+        };
+        let mut serialized = Vec::new();
+        packet
+            .serialize(
+                &mut Serializer::new(&mut serialized)
+                    .with_struct_map()
+                    .with_string_variants(),
+            )
+            .unwrap();
+        serialized
+    }
+
+    /// Test the basic RPC interface.
+    #[tokio::test]
+    async fn test_rpc_client() {
+        panic_after(Duration::from_secs(1), async move {
+            let (pipe1, mut pipe2) = Pipe::new();
+            let (mut client, events) = create::<_, _, TestPayload>(pipe1);
+            println!("init");
+
+            // Client task.
+            let tasks = (
+                tokio::spawn(async move {
+                    assert_eq!(client.call(42).await, Ok(1337));
+                    let delayed = client.call(43);
+                    assert_eq!(client.call(44).await, Ok(1339));
+                    assert_eq!(delayed.await, Ok(1338));
+                }),
+                tokio::spawn(async move {
+                    // Invalid ID, should be ignored.
+                    pipe2.send(serialize_packet(42, true, 42)).await.unwrap();
+
+                    assert_eq!(pipe2.next().await, Some(Ok(serialize_packet(0, false, 42))));
+                    pipe2.send(serialize_packet(0, true, 1337)).await.unwrap();
+
+                    // Network errors shall be ignored if the stream is not closed.
+                    pipe2.inject_error();
+
+                    // The ID has to be reused, and a second concurrent call has to have a
+                    // different ID.
+                    assert_eq!(pipe2.next().await, Some(Ok(serialize_packet(0, false, 43))));
+                    assert_eq!(pipe2.next().await, Some(Ok(serialize_packet(1, false, 44))));
+                    pipe2.send(serialize_packet(1, true, 1339)).await.unwrap();
+                    pipe2.send(serialize_packet(0, true, 1338)).await.unwrap();
+                }),
+            );
+
+            tasks.0.await.unwrap();
+            tasks.1.await.unwrap();
+        })
+        .await;
+    }
+
+    /// Test correct behaviour if the connection is closed during a call.
+    #[tokio::test]
+    async fn test_connection_closed() {
+        panic_after(Duration::from_secs(1), async move {
+            let (pipe1, pipe2) = Pipe::new();
+            let (mut client, _events) = create::<_, _, TestPayload>(pipe1);
+
+            // Client task.
+            let tasks = (
+                tokio::spawn(async move {
+                    assert_eq!(client.call(42).await, Err(RPCError::Closed));
+                }),
+                tokio::spawn(async move {
+                    drop(pipe2);
+                }),
+            );
+
+            tasks.0.await.unwrap();
+            tasks.1.await.unwrap();
+        })
+        .await;
+    }
+
+    /// Test server events.
+    #[tokio::test]
+    async fn test_events() {
+        panic_after(Duration::from_secs(1), async move {
+            let (pipe1, mut pipe2) = Pipe::new();
+            let (client, mut events) = create::<_, _, TestPayload>(pipe1);
+
+            // Client task.
+            let tasks = (
+                tokio::spawn(async move {
+                    assert_eq!(events.next().await, Some(42));
+                    assert_eq!(events.next().await, None);
+                }),
+                tokio::spawn(async move {
+                    pipe2.send(serialize_packet(SERVER_EVENT_ID, false, 42)).await.unwrap();
+                    drop(pipe2);
+                }),
+            );
+
+            tasks.0.await.unwrap();
+            tasks.1.await.unwrap();
+        }).await;
+    }
+
+    // TODO:
+    // - Test for unparseable response.
+    // - Test for server events.
+    // - Test wether the client correctly closes the connection and whether futures return an error
+    // in this case.
+}
+
 pub async fn low_level_integration_test_client(_address: String) {
     /*use super::websocket::WebSocketConnection;