Pass all qualification tests in CI

Author: MathiasKoch

.gitignore

@@ -3,3 +3,4 @@
 .gdb_history
 Cargo.lock
 target/
+device_advisor_integration.log

mqttrust_core/Cargo.toml

@@ -40,11 +40,14 @@ defmt = { version = "^0.3", optional = true }
 [dev-dependencies]
 native-tls = { version = "^0.2" }
 dns-lookup = "1.0.3"
-env_logger = "0.9.0"
+env_logger = "0.11"
+static_cell = "2.1"
 
 [features]
 default = []
 
 std = []
 
-defmt-impl = ["defmt", "mqttrust/defmt-impl", "heapless/defmt-impl", "fugit/defmt"]
+log = ["dep:log", "mqttrust/log"]
+
+defmt-impl = ["dep:defmt", "mqttrust/defmt-impl", "heapless/defmt-impl", "fugit/defmt"]

mqttrust_core/examples/aws_device_advisor.rs

@@ -8,38 +8,49 @@ use mqttrust_core::{EventLoop, MqttOptions, Notification};
 use common::clock::SysClock;
 use common::network::Network;
 use native_tls::TlsConnector;
+use static_cell::StaticCell;
+use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::thread;
 
 use crate::common::credentials;
 
-static mut Q: BBBuffer<{ 1024 * 6 }> = BBBuffer::new();
+static mut Q: BBBuffer<{ 1024 * 60 }> = BBBuffer::new();
 
 fn main() {
     env_logger::init();
 
     let (p, c) = unsafe { Q.try_split_framed().unwrap() };
 
-    let hostname = credentials::HOSTNAME.unwrap();
+    static HOSTNAME: StaticCell<String> = StaticCell::new();
+    let hostname = HOSTNAME.init(credentials::hostname());
+
+    log::info!(
+        "Starting device advisor test on endpoint {}",
+        hostname.as_str()
+    );
 
     let connector = TlsConnector::builder()
         .identity(credentials::identity())
         .add_root_certificate(credentials::root_ca())
         .build()
         .unwrap();
 
-    let mut network = Network::new_tls(connector, String::from(hostname));
+    let mut network = Network::new_tls(connector, hostname.clone());
 
     let thing_name = "mqttrust";
 
     let mut mqtt_eventloop = EventLoop::new(
         c,
         SysClock::new(),
-        MqttOptions::new(thing_name, hostname.into(), 8883),
+        MqttOptions::new(thing_name, hostname.as_str().into(), 8883),
     );
 
     let mqtt_client = mqttrust_core::Client::new(p, thing_name);
 
+    let connected = Arc::new(AtomicBool::new(false));
+    let con = connected.clone();
+
     thread::Builder::new()
         .name("eventloop".to_string())
         .spawn(move || loop {
@@ -49,12 +60,18 @@ fn main() {
                     session_present,
                     code: ConnectReturnCode::Accepted,
                 }))) => {
-                    log::info!("Successfully connected to broker");
+                    log::info!(
+                        "Successfully connected to broker. session_present: {}",
+                        session_present
+                    );
+                    con.store(true, std::sync::atomic::Ordering::Release);
+                }
+                Ok(n) => {
+                    log::info!("Received {:?} during connect", n);
                 }
-                Ok(_) => {}
             }
 
-            match mqtt_eventloop.yield_event(&mut network) {
+            match nb::block!(mqtt_eventloop.yield_event(&mut network)) {
                 Ok(Notification::Publish(_)) => {}
                 Ok(n) => {
                     log::trace!("{:?}", n);
@@ -66,19 +83,21 @@ fn main() {
 
     loop {
         thread::sleep(std::time::Duration::from_millis(5000));
-        mqtt_client
-            .subscribe(&[SubscribeTopic {
-                topic_path: format!("plc/output/{}", thing_name).as_str(),
-                qos: QoS::AtLeastOnce,
-            }])
-            .unwrap();
-
-        mqtt_client
-            .publish(
-                format!("plc/input/{}", thing_name).as_str(),
-                format!("Hello from {}", thing_name).as_bytes(),
-                QoS::AtLeastOnce,
-            )
-            .unwrap();
+        if connected.load(std::sync::atomic::Ordering::Acquire) {
+            mqtt_client
+                .subscribe(&[SubscribeTopic {
+                    topic_path: format!("plc/output/{}", thing_name).as_str(),
+                    qos: QoS::AtLeastOnce,
+                }])
+                .unwrap();
+
+            mqtt_client
+                .publish(
+                    format!("plc/input/{}", thing_name).as_str(),
+                    format!("Hello from {}", thing_name).as_bytes(),
+                    QoS::AtLeastOnce,
+                )
+                .unwrap();
+        }
     }
 }

mqttrust_core/examples/common/clock.rs

@@ -5,14 +5,12 @@ use std::{
 };
 pub struct SysClock {
     start_time: u32,
-    countdown_end: Option<u32>,
 }
 
 impl SysClock {
     pub fn new() -> Self {
         Self {
             start_time: Self::epoch(),
-            countdown_end: None,
         }
     }
 

mqttrust_core/examples/common/credentials.rs

@@ -11,4 +11,6 @@ pub fn root_ca() -> Certificate {
     Certificate::from_pem(include_bytes!("../secrets/root-ca.pem")).unwrap()
 }
 
-pub const HOSTNAME: Option<&'static str> = option_env!("AWS_HOSTNAME");
+pub fn hostname() -> String {
+    env::var("AWS_HOSTNAME").unwrap()
+}

mqttrust_core/examples/secrets/identity.pfx

@@ -219,7 +219,7 @@ where
     }
 
     fn backoff(&self) -> TimerDurationU32<TIMER_HZ> {
-        let base_time_ms: u32 = 1000;
+        let base_time_ms: u32 = 200;
         let backoff = base_time_ms.saturating_mul(u32::pow(2, self.connect_counter as u32));
 
         core::cmp::min(50.secs(), backoff.millis())
@@ -265,14 +265,15 @@ where
             MqttConnectionStatus::Handshake => {
                 let now = self.last_outgoing_timer.now();
 
-                let backoff_time = core::cmp::max(50.secs(), self.backoff());
+                let backoff_time = self.backoff();
 
                 if self
                     .state
                     .last_ping_entry()
                     .or_insert(now)
                     .has_elapsed(&now, backoff_time)
                 {
+                    warn!("Timed out waiting for connect packet!");
                     return Err(nb::Error::Other(EventError::Timeout));
                 }
 
@@ -562,6 +563,7 @@ impl<'a> PacketDecoder<'a> {
                 Err(EventError::Encoding(e).into())
             }
             Ok(Some(packet)) => {
+                warn!("Got packet! {:?}", packet);
                 self.is_err.replace(false);
                 state
                     .handle_incoming_packet(packet)

mqttrust_core/src/eventloop.rs

@@ -44,6 +44,15 @@ function report_status {
     done
 }
 
+function check_pid {
+    if [ -n "$pid" ]; then
+        ps -p $pid > /dev/null;
+        return $?
+    else
+        return 0
+    fi
+}
+
 while test ${IN_PROGRESS} == 1; do
     # Fetch the current status and stash in /tmp so we can use it throughout the status fetch process.
 
@@ -68,13 +77,16 @@ while test ${IN_PROGRESS} == 1; do
     elif test x"${overall_status}" == x${STATUS_PASS}; then
         MONITOR_STATUS=0
         IN_PROGRESS=0
+    elif test x"${overall_status}" == x${STATUS_PASS_WITH_WARNINGS}; then
+        MONITOR_STATUS=0
+        IN_PROGRESS=0
     elif test x"${overall_status}" == x${STATUS_STOPPING}; then
         MONITOR_STATUS=1
         IN_PROGRESS=0
     elif test x"${overall_status}" == x${STATUS_STOPPED}; then
         MONITOR_STATUS=1
         IN_PROGRESS=0
-    elif { ps -p $pid > /dev/null; }; [ "$?" = 1 ]; then
+    elif check_pid; [ "$?" = 1 ]; then
         echo Binary is not running any more?
 
         MONITOR_STATUS=1

scripts/da_monitor.sh

@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+
+if [[ -z "$DEVICE_ADVISOR_PASSWORD" ]]; then
+  echo "DEVICE_ADVISOR_PASSWORD environment variable must be set!"
+  return 1;
+fi
+
+set -e
+
+DAEMONIZE=true
+
+THING_NAME="mqttrust"
+SUITE_ID="1gaev57dq6i5"
+export RUST_LOG=debug 
+
+THING_ARN="arn:aws:iot:eu-west-1:411974994697:thing/$THING_NAME"
+SCRIPT_DIR="$(dirname "$(readlink -f "$0")")"
+
+
+export AWS_HOSTNAME=$(aws iotdeviceadvisor get-endpoint --thing-arn $THING_ARN --output text --query endpoint)
+
+cargo build --features=log --example aws_device_advisor --release
+
+SUITE_RUN_ID=$(aws iotdeviceadvisor start-suite-run --suite-definition-id $SUITE_ID --suite-run-configuration "primaryDevice={thingArn=$THING_ARN},parallelRun=true" --output text --query suiteRunId)
+if $DAEMONIZE; then
+    nohup ./target/release/examples/aws_device_advisor > device_advisor_integration.log &
+    PID=$!
+else
+    echo "You can now run 'DEVICE_ADVISOR_PASSWORD=$DEVICE_ADVISOR_PASSWORD AWS_HOSTNAME=$AWS_HOSTNAME ./target/release/examples/aws_device_advisor' in a seperate terminal"
+fi
+
+always() {
+    kill $PID || true
+    cat device_advisor_integration.log
+}
+
+on_failure() {
+    if $DAEMONIZE; then
+        always || true
+    fi
+    aws iotdeviceadvisor stop-suite-run --suite-definition-id $SUITE_ID --suite-run-id $SUITE_RUN_ID
+}
+
+trap "on_failure" ERR INT
+
+$SCRIPT_DIR/da_monitor.sh $SUITE_ID $SUITE_RUN_ID $PID
+
+always
+

scripts/da_run_test.sh

@@ -13,6 +13,12 @@ CERT_PATH="$SECRETS_DIR/cert.pem"
 PRIV_KEY_PATH="$SECRETS_DIR/priv.key.pem"
 
 CERT_ARN=$(aws iot create-keys-and-certificate --set-as-active --certificate-pem-outfile $CERT_PATH --private-key-outfile $PRIV_KEY_PATH | jq -r .certificateArn);
+for OLD_CERT in $(aws iot list-thing-principals --thing-name $THING_NAME | jq -r '.principals[]' | xargs); do
+  CERT_ID=$(echo $OLD_CERT | cut -d "/" -f 2)
+  aws iot detach-thing-principal --thing-name $THING_NAME --principal $OLD_CERT
+  aws iot update-certificate --new-status INACTIVE --certificate-id $CERT_ID
+  aws iot delete-certificate --certificate-id $CERT_ID --force-delete
+done
 aws iot attach-thing-principal --thing-name $THING_NAME --principal $CERT_ARN > /dev/null 2>&1
 aws iot attach-policy --policy-name Connect --target $CERT_ARN > /dev/null 2>&1
 aws iot attach-policy --policy-name Input --target $CERT_ARN > /dev/null 2>&1