more test cases and fix in pub rel, comp and rec

Author: GunnarMorrigan

fuzz/fuzz_targets/fuzz_target_1.rs

@@ -7,6 +7,7 @@ async fn test(mut data: &[u8]) {
     let _ = mqrstt::packets::Packet::async_read(&mut data).await;
 }
 
+#[cfg(target_os = "linux")]
 fuzz_target!(|data: &[u8]| {
     test(data);
 });

mqrstt/Cargo.toml

@@ -52,6 +52,8 @@ tokio = { version = "1", features = [
 smol = { version = "2", optional = true }
 
 [dev-dependencies]
+pretty_assertions = "1.4.1"
+
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 
 smol = { version = "2.0.0" }

mqrstt/src/packets/macros/properties_macros.rs

@@ -399,17 +399,17 @@ macro_rules! properties_read_match_branch_body {
         $properties.authentication_data = Some(prop_body);
     }};
     ($stream:ident, $properties:ident, $read_property_bytes:ident, PropertyType::RequestResponseInformation) => {{
-        if $properties.authentication_data.is_some() {
+        if $properties.request_response_information.is_some() {
             return Err($crate::packets::error::ReadError::DeserializeError(DeserializeError::DuplicateProperty(
                 PropertyType::RequestResponseInformation,
             )));
         }
         let (prop_body, read_bytes) = u8::async_read($stream).await?;
         $read_property_bytes += read_bytes;
-        $properties.request_problem_information = Some(prop_body);
+        $properties.request_response_information = Some(prop_body);
     }};
     ($stream:ident, $properties:ident, $read_property_bytes:ident, PropertyType::RequestProblemInformation) => {{
-        if $properties.authentication_data.is_some() {
+        if $properties.request_problem_information.is_some() {
             return Err($crate::packets::error::ReadError::DeserializeError(DeserializeError::DuplicateProperty(
                 PropertyType::RequestProblemInformation,
             )));

mqrstt/src/packets/mod.rs

@@ -453,6 +453,7 @@ mod tests {
     use crate::tests::test_packets::*;
 
     #[rstest::rstest]
+    #[case::connect_case(connect_case())]
     #[case::ping_req_case(ping_req_case().1)]
     #[case::ping_resp_case(ping_resp_case().1)]
     #[case::connack_case(connack_case().1)]
@@ -472,6 +473,10 @@ mod tests {
     #[case::suback(suback_case())]
     #[case::unsubscribe(unsubscribe_case())]
     #[case::unsuback(unsuback_case())]
+    #[case::pubcomp_case(pubcomp_case())]
+    #[case::pubrec_case(pubrec_case())]
+    #[case::pubrec_case(pubrel_case2())]
+    #[case::auth_case(auth_case())]
     fn test_write_read_write_read_cases(#[case] packet: Packet) {
         use crate::packets::WireLength;
 
@@ -481,9 +486,11 @@ mod tests {
 
         let wire_len = packet.wire_len();
         assert_eq!(wire_len, buffer.len());
-        dbg!(wire_len);
-        let a: Vec<_> = buffer.iter().map(|f| *f as u16).collect();
-        println!("{:?}", a);
+
+        // dbg!(wire_len);
+
+        // let a: Vec<_> = buffer.iter().map(|f| *f as u16).collect();
+        // println!("{:?}", a);
 
         let res1 = Packet::read(&mut buffer).unwrap();
 
@@ -496,6 +503,50 @@ mod tests {
         assert_eq!(res1, res2);
     }
 
+    #[rstest::rstest]
+    #[case::connect_case(connect_case())]
+    #[case::ping_req_case(ping_req_case().1)]
+    #[case::ping_resp_case(ping_resp_case().1)]
+    #[case::connack_case(connack_case().1)]
+    #[case::create_subscribe_packet(create_subscribe_packet(1))]
+    #[case::create_subscribe_packet(create_subscribe_packet(65335))]
+    #[case::create_puback_packet(create_puback_packet(1))]
+    #[case::create_puback_packet(create_puback_packet(65335))]
+    #[case::create_disconnect_packet(create_disconnect_packet())]
+    #[case::create_connack_packet(create_connack_packet(true))]
+    #[case::create_connack_packet(create_connack_packet(false))]
+    #[case::publish_packet_1(publish_packet_1())]
+    #[case::publish_packet_2(publish_packet_2())]
+    #[case::publish_packet_3(publish_packet_3())]
+    #[case::publish_packet_4(publish_packet_4())]
+    #[case::create_empty_publish_packet(create_empty_publish_packet())]
+    #[case::subscribe(subscribe_case())]
+    #[case::suback(suback_case())]
+    #[case::unsubscribe(unsubscribe_case())]
+    #[case::unsuback(unsuback_case())]
+    #[case::pubcomp_case(pubcomp_case())]
+    #[case::pubrec_case(pubrec_case())]
+    #[case::pubrec_case(pubrel_case2())]
+    #[case::auth_case(auth_case())]
+    #[tokio::test]
+    async fn test_async_write_read_write_read_cases(#[case] packet: Packet) {
+        use crate::packets::WireLength;
+
+        let mut buffer = Vec::with_capacity(1000);
+        let res = packet.async_write(&mut buffer).await.unwrap();
+
+        let wire_len = packet.wire_len();
+
+        assert_eq!(res, buffer.len());
+        assert_eq!(wire_len, buffer.len());
+
+        let mut buf = buffer.as_slice();
+
+        let res1 = Packet::async_read(&mut buf).await.unwrap();
+
+        pretty_assertions::assert_eq!(packet, res1);
+    }
+
     #[rstest::rstest]
     #[case::disconnect(disconnect_case())]
     #[case::ping_req(ping_req_case())]
@@ -548,45 +599,6 @@ mod tests {
         assert_eq!(out, input)
     }
 
-    #[rstest::rstest]
-    #[case::ping_req_case(ping_req_case().1)]
-    #[case::ping_resp_case(ping_resp_case().1)]
-    #[case::connack_case(connack_case().1)]
-    #[case::create_subscribe_packet(create_subscribe_packet(1))]
-    #[case::create_subscribe_packet(create_subscribe_packet(65335))]
-    #[case::create_puback_packet(create_puback_packet(1))]
-    #[case::create_puback_packet(create_puback_packet(65335))]
-    #[case::create_disconnect_packet(create_disconnect_packet())]
-    #[case::create_connack_packet(create_connack_packet(true))]
-    #[case::create_connack_packet(create_connack_packet(false))]
-    #[case::publish_packet_1(publish_packet_1())]
-    #[case::publish_packet_2(publish_packet_2())]
-    #[case::publish_packet_3(publish_packet_3())]
-    #[case::publish_packet_4(publish_packet_4())]
-    #[case::create_empty_publish_packet(create_empty_publish_packet())]
-    #[case::subscribe(subscribe_case())]
-    #[case::suback(suback_case())]
-    #[case::unsubscribe(unsubscribe_case())]
-    #[case::unsuback(unsuback_case())]
-    #[tokio::test]
-    async fn test_async_write_read_write_read_cases(#[case] packet: Packet) {
-        use crate::packets::WireLength;
-
-        let mut buffer = Vec::with_capacity(1000);
-        let res = packet.async_write(&mut buffer).await.unwrap();
-
-        let wire_len = packet.wire_len();
-
-        assert_eq!(res, buffer.len());
-        assert_eq!(wire_len, buffer.len());
-
-        let mut buf = buffer.as_slice();
-
-        let res1 = Packet::async_read(&mut buf).await.unwrap();
-
-        assert_eq!(packet, res1);
-    }
-
     // #[rstest::rstest]
     // #[case(&[59, 1, 0, 59])]
     // #[case(&[16, 14, 0, 4, 77, 81, 84, 84, 5, 247, 247, 252, 1, 17, 247, 247, 247])]

mqrstt/src/packets/pubcomp/mod.rs

@@ -7,6 +7,7 @@ pub use properties::PubCompProperties;
 use super::{
     error::{DeserializeError, ReadError},
     mqtt_trait::{MqttAsyncRead, MqttRead, MqttWrite, PacketAsyncRead, PacketRead, PacketWrite, WireLength},
+    VariableInteger,
 };
 use bytes::BufMut;
 use tokio::io::AsyncReadExt;
@@ -148,7 +149,8 @@ impl WireLength for PubComp {
         } else if self.properties.reason_string.is_none() && self.properties.user_properties.is_empty() {
             3
         } else {
-            2 + 1 + self.properties.wire_len()
+            let prop_wire_len = self.properties.wire_len();
+            2 + 1 + prop_wire_len.variable_integer_len() + prop_wire_len
         }
     }
 }

mqrstt/src/packets/pubrec/mod.rs

@@ -11,7 +11,7 @@ use tokio::io::AsyncReadExt;
 use super::{
     error::DeserializeError,
     mqtt_trait::{MqttAsyncRead, MqttRead, MqttWrite, PacketRead, PacketWrite, WireLength},
-    PacketAsyncRead,
+    PacketAsyncRead, VariableInteger,
 };
 
 /// The [`PubRec`] (Publish Received) packet is part of the acknowledgment flow for a [`crate::packets::Publish`] with QoS 2.
@@ -141,7 +141,8 @@ impl WireLength for PubRec {
         } else if self.properties.reason_string.is_none() && self.properties.user_properties.is_empty() {
             3
         } else {
-            2 + 1 + self.properties.wire_len()
+            let prop_wire_len = self.properties.wire_len();
+            2 + 1 + prop_wire_len.variable_integer_len() + prop_wire_len
         }
     }
 }

mqrstt/src/packets/pubrel/mod.rs

@@ -10,6 +10,7 @@ use tokio::io::AsyncReadExt;
 use super::{
     error::{DeserializeError, ReadError},
     mqtt_trait::{MqttAsyncRead, MqttRead, MqttWrite, PacketAsyncRead, PacketRead, PacketWrite, WireLength},
+    VariableInteger,
 };
 
 /// The [`PubRel`] (Publish Release) packet acknowledges the reception of a [`crate::packets::PubRec`] Packet.
@@ -139,7 +140,8 @@ impl WireLength for PubRel {
         } else if self.properties.reason_string.is_none() && self.properties.user_properties.is_empty() {
             3
         } else {
-            2 + 1 + self.properties.wire_len()
+            let prop_wire_len = self.properties.wire_len();
+            2 + 1 + prop_wire_len.variable_integer_len() + prop_wire_len
         }
     }
 }

mqrstt/src/tests/test_packets.rs

@@ -119,6 +119,31 @@ pub fn pubrel_smallest_case() -> (&'static [u8], Packet) {
     (packet, Packet::PubRel(expected))
 }
 
+pub fn connect_case() -> Packet {
+    let connect = Connect {
+        protocol_version: ProtocolVersion::V5,
+        clean_start: true,
+        last_will: Some(LastWill::new(QoS::ExactlyOnce, true, "will/topic", b"will payload".to_vec())),
+        username: Some("ThisIsTheUsername".into()),
+        password: Some("ThisIsThePassword".into()),
+        keep_alive: 60,
+        connect_properties: ConnectProperties {
+            session_expiry_interval: Some(5),
+            receive_maximum: Some(10),
+            maximum_packet_size: Some(100),
+            topic_alias_maximum: Some(10),
+            user_properties: vec![("test".into(), "test".into()), ("test2".into(), "test2".into())],
+            authentication_method: Some("AuthenticationMethod".into()),
+            authentication_data: Some(b"AuthenticationData".to_vec()),
+            request_response_information: Some(0),
+            request_problem_information: Some(1),
+        },
+        client_id: "ThisIsTheClientID".into(),
+    };
+
+    Packet::Connect(connect)
+}
+
 pub fn publish_packet_1() -> Packet {
     Packet::Publish(Publish {
         dup: false,
@@ -324,12 +349,73 @@ pub fn unsubscribe_case() -> Packet {
     let expected = Unsubscribe {
         packet_identifier: 3,
         topics: vec!["test/topic".into()],
-        properties: UnsubscribeProperties { user_properties: vec![("written += 1;".into(), "value".into())] },
+        properties: UnsubscribeProperties {
+            user_properties: vec![("written += 1;".into(), "value".into())],
+        },
     };
 
     Packet::Unsubscribe(expected)
 }
 
+pub fn pubrec_case() -> Packet {
+    let expected = PubRec {
+        packet_identifier: 3,
+        reason_code: PubRecReasonCode::Success,
+        properties: PubRecProperties {
+            reason_string: Some("test".into()),
+            user_properties: vec![("test5asdf".into(), "test3".into()), ("test4".into(), "test2".into())],
+        },
+    };
+
+    Packet::PubRec(expected)
+}
+
+pub fn pubcomp_case() -> Packet {
+    let expected = PubComp {
+        packet_identifier: 3,
+        reason_code: PubCompReasonCode::PacketIdentifierNotFound,
+        properties: PubCompProperties {
+            reason_string: Some("test".into()),
+            user_properties: vec![
+                ("test5asdf".into(), "test3".into()),
+                ("test⌚5asdf".into(), "test3".into()),
+                ("test5asdf".into(), "test3".into()),
+                ("test5asdf".into(), "test3".into()),
+                ("test4".into(), "test2".into()),
+            ],
+        },
+    };
+
+    Packet::PubComp(expected)
+}
+
+pub fn pubrel_case2() -> Packet {
+    let expected = PubRel {
+        packet_identifier: 3,
+        reason_code: PubRelReasonCode::Success,
+        properties: PubRelProperties {
+            reason_string: Some("test".into()),
+            user_properties: vec![("test5asdf".into(), "test3".repeat(10000).into()), ("test4".into(), "test2".into())],
+        },
+    };
+
+    Packet::PubRel(expected)
+}
+
+pub fn auth_case() -> Packet {
+    let expected = Auth {
+        reason_code: AuthReasonCode::ContinueAuthentication,
+        properties: AuthProperties {
+            authentication_method: Some("SomeRandomDataHere".into()),
+            authentication_data: Some(b"VeryRandomStuff".to_vec()),
+            reason_string: Some("⌚this_is_for_sure_a_test_⌚".into()),
+            user_properties: vec![("SureHopeThisWorks".into(), "😰".into())],
+        },
+    };
+
+    Packet::Auth(expected)
+}
+
 #[rstest]
 #[case(create_subscribe_packet(1))]
 #[case(create_subscribe_packet(65335))]