diff --git a/crates/core/c8y_api/src/smartrest/inventory.rs b/crates/core/c8y_api/src/smartrest/inventory.rs index 0bd878f6fc0..773548ef135 100644 --- a/crates/core/c8y_api/src/smartrest/inventory.rs +++ b/crates/core/c8y_api/src/smartrest/inventory.rs @@ -141,6 +141,13 @@ impl From for MqttMessage { } } +/// Create a SmartREST payload for setting/updating the current state of the target profile +/// in its own managed object. When all individual operations are finished (i.e. firmware update, software update +/// and configuration update), the `profile_executed` field should be set to `true`, otherwise it should be `false`. +pub fn set_c8y_profile_target_payload(profile_executed: bool) -> String { + fields_to_csv_string(&["121", &profile_executed.to_string()]) +} + #[derive(thiserror::Error, Debug)] #[error("Field `{field_name}` contains invalid value: {value:?}")] pub struct InvalidValueError { diff --git a/crates/extensions/c8y_mapper_ext/src/operations/handlers/device_profile.rs b/crates/extensions/c8y_mapper_ext/src/operations/handlers/device_profile.rs index 13e9d9ea407..85f5d07790d 100644 --- a/crates/extensions/c8y_mapper_ext/src/operations/handlers/device_profile.rs +++ b/crates/extensions/c8y_mapper_ext/src/operations/handlers/device_profile.rs @@ -1,3 +1,74 @@ +use anyhow::Context; +use c8y_api::smartrest; +use c8y_api::smartrest::inventory::set_c8y_profile_target_payload; +use c8y_api::smartrest::smartrest_serializer::CumulocitySupportedOperations; +use tedge_api::device_profile::DeviceProfileCmd; +use tedge_api::CommandStatus; +use tedge_mqtt_ext::MqttMessage; +use tracing::warn; + +use super::EntityTarget; +use super::OperationContext; +use super::OperationError; +use super::OperationOutcome; + +impl OperationContext { + pub async fn handle_device_profile_state_change( + &self, + target: &EntityTarget, + cmd_id: &str, + message: &MqttMessage, + ) -> Result { + if !self.capabilities.device_profile { + warn!("Received a device_profile command, however, device_profile feature is disabled"); + return Ok(OperationOutcome::Ignored); + } + + let command = match DeviceProfileCmd::try_from_bytes( + target.topic_id.to_owned(), + cmd_id.into(), + message.payload_bytes(), + ) + .context("Could not parse command as a device profile command")? + { + Some(command) => command, + None => { + // The command has been fully processed + return Ok(OperationOutcome::Ignored); + } + }; + + let sm_topic = &target.smartrest_publish_topic; + + match command.status() { + CommandStatus::Executing => { + let c8y_target_profile = + MqttMessage::new(sm_topic, set_c8y_profile_target_payload(false)); // Set target profile + + Ok(OperationOutcome::Executing { + extra_messages: vec![c8y_target_profile], + }) + } + CommandStatus::Successful => { + let c8y_target_profile = + MqttMessage::new(sm_topic, set_c8y_profile_target_payload(true)); // Set the target profile as executed + + let smartrest_set_operation = + smartrest::smartrest_serializer::succeed_operation_no_payload( + CumulocitySupportedOperations::C8yDeviceProfile, + ); + let c8y_notification = MqttMessage::new(sm_topic, smartrest_set_operation); + + Ok(OperationOutcome::Finished { + messages: vec![c8y_target_profile, c8y_notification], + }) + } + CommandStatus::Failed { reason } => Err(anyhow::anyhow!(reason).into()), + _ => Ok(OperationOutcome::Ignored), + } + } +} + #[cfg(test)] mod tests { use crate::tests::skip_init_messages; @@ -9,6 +80,7 @@ mod tests { use std::time::Duration; use tedge_actors::test_helpers::MessageReceiverExt; use tedge_actors::Sender; + use tedge_mqtt_ext::test_helpers::assert_received_contains_str; use tedge_mqtt_ext::test_helpers::assert_received_includes_json; use tedge_mqtt_ext::MqttMessage; use tedge_mqtt_ext::Topic; @@ -723,4 +795,433 @@ mod tests { ) .await; } + + #[tokio::test] + async fn handle_config_update_executing_and_failed_cmd_for_main_device() { + let ttd = TempTedgeDir::new(); + let test_handle = spawn_c8y_mapper_actor(&ttd, true).await; + let TestHandle { mqtt, .. } = test_handle; + let mut mqtt = mqtt.with_timeout(TEST_TIMEOUT_MS); + + skip_init_messages(&mut mqtt).await; + + // Simulate config_snapshot command with "executing" state + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/main///cmd/device_profile/c8y-mapper-123456"), + json!({ + "status": "executing", + "name": "test-profile", + "operations": [ + { + "operation": "firmware_update", + "skip": false, + "payload": { + "name": "test-firmware", + "version": "1.0", + "url": "http://www.my.url" + } + }, + { + "operation": "software_update", + "skip": false, + "payload": { + "updateList": [ + { + "type": "apt", + "modules": [ + { + "name": "test-software-1", + "version": "latest", + "action": "install" + }, + { + "name": "test-software-2", + "version": "latest", + "action": "install" + } + ] + } + ] + } + }, + { + "operation": "config_update", + "skip": false, + "payload": { + "type": "path/config/test-software-1", + "remoteUrl":"http://www.my.url" + } + } + ] + }) + .to_string(), + )) + .await + .expect("Send failed"); + + // Expect `501` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us", "501,c8y_DeviceProfile")]).await; + + // Expect `121` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us", "121,false")]).await; + + // Simulate config_snapshot command with "failed" state + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/main///cmd/device_profile/c8y-mapper-123456"), + json!({ + "status": "failed", + "reason": "Something went wrong", + "name": "test-profile", + "operations": [ + { + "operation": "firmware_update", + "skip": false, + "payload": { + "name": "test-firmware", + "version": "1.0", + "url": "http://www.my.url" + } + }, + { + "operation": "software_update", + "skip": false, + "payload": { + "updateList": [ + { + "type": "apt", + "modules": [ + { + "name": "test-software-1", + "version": "latest", + "action": "install" + }, + { + "name": "test-software-2", + "version": "latest", + "action": "install" + } + ] + } + ] + } + }, + { + "operation": "config_update", + "skip": false, + "payload": { + "type": "path/config/test-software-1", + "remoteUrl":"http://www.my.url" + } + } + ] + }) + .to_string(), + )) + .await + .expect("Send failed"); + + // Expect `502` smartrest message on `c8y/s/us`. + assert_received_contains_str( + &mut mqtt, + [("c8y/s/us", "502,c8y_DeviceProfile,Something went wrong")], + ) + .await; + } + + #[tokio::test] + async fn handle_config_update_executing_and_failed_cmd_for_child_device() { + let ttd = TempTedgeDir::new(); + let test_handle = spawn_c8y_mapper_actor(&ttd, true).await; + let TestHandle { mqtt, .. } = test_handle; + let mut mqtt = mqtt.with_timeout(TEST_TIMEOUT_MS); + + skip_init_messages(&mut mqtt).await; + + // The child device must be registered first + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/child1//"), + r#"{ "@type":"child-device", "@id":"child1" }"#, + )) + .await + .expect("fail to register the child-device"); + + mqtt.skip(1).await; // Skip child device registration messages + + // Simulate config_snapshot command with "executing" state + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/child1///cmd/device_profile/c8y-mapper-123456"), + json!({ + "status": "executing", + "name": "test-profile", + "operations": [ + { + "operation": "firmware_update", + "skip": false, + "payload": { + "name": "test-firmware", + "version": "1.0", + "url": "http://www.my.url" + } + }, + { + "operation": "software_update", + "skip": false, + "payload": { + "updateList": [ + { + "type": "apt", + "modules": [ + { + "name": "test-software-1", + "version": "latest", + "action": "install" + }, + { + "name": "test-software-2", + "version": "latest", + "action": "install" + } + ] + } + ] + } + }, + { + "operation": "config_update", + "skip": false, + "payload": { + "type": "path/config/test-software-1", + "remoteUrl":"http://www.my.url" + } + } + ] + }) + .to_string(), + )) + .await + .expect("Send failed"); + + // Expect `501` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us/child1", "501,c8y_DeviceProfile")]) + .await; + + // Expect `121` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us/child1", "121,false")]).await; + + // Simulate config_snapshot command with "failed" state + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/child1///cmd/device_profile/c8y-mapper-123456"), + json!({ + "status": "failed", + "reason": "Something went wrong", + "name": "test-profile", + "operations": [ + { + "operation": "firmware_update", + "skip": false, + "payload": { + "name": "test-firmware", + "version": "1.0", + "url": "http://www.my.url" + } + }, + { + "operation": "software_update", + "skip": false, + "payload": { + "updateList": [ + { + "type": "apt", + "modules": [ + { + "name": "test-software-1", + "version": "latest", + "action": "install" + }, + { + "name": "test-software-2", + "version": "latest", + "action": "install" + } + ] + } + ] + } + }, + { + "operation": "config_update", + "skip": false, + "payload": { + "type": "path/config/test-software-1", + "remoteUrl":"http://www.my.url" + } + } + ] + }) + .to_string(), + )) + .await + .expect("Send failed"); + + // Expect `502` smartrest message on `c8y/s/us`. + assert_received_contains_str( + &mut mqtt, + [( + "c8y/s/us/child1", + "502,c8y_DeviceProfile,Something went wrong", + )], + ) + .await; + } + + #[tokio::test] + async fn handle_device_profile_successful_cmd_for_main_device() { + let ttd = TempTedgeDir::new(); + let test_handle = spawn_c8y_mapper_actor(&ttd, true).await; + let TestHandle { mqtt, .. } = test_handle; + let mut mqtt = mqtt.with_timeout(TEST_TIMEOUT_MS); + + skip_init_messages(&mut mqtt).await; + + // Simulate config_update command with "successful" state + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/main///cmd/device_profile/c8y-mapper-123456"), + json!({ + "status": "successful", + "name": "test-profile", + "operations": [ + { + "operation": "firmware_update", + "skip": false, + "payload": { + "name": "test-firmware", + "version": "1.0", + "url": "http://www.my.url" + } + }, + { + "operation": "software_update", + "skip": false, + "payload": { + "updateList": [ + { + "type": "apt", + "modules": [ + { + "name": "test-software-1", + "version": "latest", + "action": "install" + }, + { + "name": "test-software-2", + "version": "latest", + "action": "install" + } + ] + } + ] + } + }, + { + "operation": "config_update", + "skip": false, + "payload": { + "type": "path/config/test-software-1", + "remoteUrl":"http://www.my.url" + } + } + ] + }) + .to_string(), + )) + .await + .expect("Send failed"); + + // Expect `121` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us", "121,true")]).await; + + // Expect `503` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us", "503,c8y_DeviceProfile")]).await; + } + + #[tokio::test] + async fn handle_device_profile_successful_cmd_for_child_device() { + let ttd = TempTedgeDir::new(); + let test_handle = spawn_c8y_mapper_actor(&ttd, true).await; + let TestHandle { mqtt, .. } = test_handle; + let mut mqtt = mqtt.with_timeout(TEST_TIMEOUT_MS); + + skip_init_messages(&mut mqtt).await; + + // The child device must be registered first + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/child1//"), + r#"{ "@type":"child-device", "@id":"child1" }"#, + )) + .await + .expect("fail to register the child-device"); + + mqtt.skip(1).await; // Skip child device registration messages + + // Simulate config_update command with "successful" state + mqtt.send(MqttMessage::new( + &Topic::new_unchecked("te/device/child1///cmd/device_profile/c8y-mapper-123456"), + json!({ + "status": "successful", + "name": "test-profile", + "operations": [ + { + "operation": "firmware_update", + "skip": false, + "payload": { + "name": "test-firmware", + "version": "1.0", + "url": "http://www.my.url" + } + }, + { + "operation": "software_update", + "skip": false, + "payload": { + "updateList": [ + { + "type": "apt", + "modules": [ + { + "name": "test-software-1", + "version": "latest", + "action": "install" + }, + { + "name": "test-software-2", + "version": "latest", + "action": "install" + } + ] + } + ] + } + }, + { + "operation": "config_update", + "skip": false, + "payload": { + "type": "path/config/test-software-1", + "remoteUrl":"http://www.my.url" + } + } + ] + }) + .to_string(), + )) + .await + .expect("Send failed"); + + // Expect `121` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us/child1", "121,true")]).await; + + // Expect `503` smartrest message on `c8y/s/us`. + assert_received_contains_str(&mut mqtt, [("c8y/s/us/child1", "503,c8y_DeviceProfile")]) + .await; + } } diff --git a/crates/extensions/c8y_mapper_ext/src/operations/handlers/mod.rs b/crates/extensions/c8y_mapper_ext/src/operations/handlers/mod.rs index 66b7aaa294c..c16795f2151 100644 --- a/crates/extensions/c8y_mapper_ext/src/operations/handlers/mod.rs +++ b/crates/extensions/c8y_mapper_ext/src/operations/handlers/mod.rs @@ -136,7 +136,8 @@ impl OperationContext { .await } OperationType::DeviceProfile => { - Ok(OperationOutcome::Ignored) // to do handle state change for device profile + self.handle_device_profile_state_change(&entity, &cmd_id, &message) + .await } };