refactor!(validation): make it stricter to reflect the new API

Check the interface type, ownership and aggregation in the user
validation. Return an hard error if one is incorrect.

Signed-off-by: Joshua Chapman <joshua.chapman@secomind.com>

Author: joshuachp

src/client.rs

@@ -26,7 +26,10 @@ use tokio::{
 };
 use tracing::{debug, error, trace};
 
-use crate::{aggregate::AstarteObject, interface::mapping::path::MappingError};
+use crate::{
+    aggregate::AstarteObject, error::InterfaceTypeError, interface::mapping::path::MappingError,
+    validate::UserValidationError,
+};
 use crate::{
     connection::ClientMessage,
     event::DeviceEvent,
@@ -43,7 +46,7 @@ use crate::{
     Error, Interface,
 };
 use crate::{
-    error::{AggregateError, DynError},
+    error::{AggregationError, DynError},
     store::PropertyMapping,
 };
 
@@ -79,7 +82,7 @@ pub enum RecvError {
 
     /// Invalid aggregation between the interface and the data.
     #[error(transparent)]
-    Aggregation(#[from] AggregateError),
+    Aggregation(#[from] AggregationError),
 
     /// Error when the Device is disconnected from Astarte or client.
     ///
@@ -365,7 +368,7 @@ impl<S> DeviceClient<S> {
             return Ok(());
         }
 
-        let validated = ValidatedIndividual::validate(mapping, path, individual, timestamp)?;
+        let validated = ValidatedIndividual::validate(mapping, individual, timestamp)?;
 
         let msg = if mapping.interface().is_property() {
             ClientMessage::Property {
@@ -394,8 +397,8 @@ impl<S> DeviceClient<S> {
             })?;
 
         let object = interface.as_object_ref().ok_or_else(|| {
-            let aggr_err = AggregateError::for_interface(
-                interface_name,
+            let aggr_err = AggregationError::new(
+                interface_name.to_string(),
                 path.to_string(),
                 InterfaceAggregation::Object,
                 interface.aggregation(),
@@ -428,12 +431,15 @@ impl<S> DeviceClient<S> {
                 mapping: path.to_string(),
             })?;
 
-        let mapping = mapping.as_prop().ok_or_else(|| Error::InterfaceType {
-            exp: InterfaceTypeDef::Properties,
-            got: interface.interface_type(),
+        let mapping = mapping.as_prop().ok_or_else(|| {
+            Error::Validation(UserValidationError::InterfaceType(InterfaceTypeError::new(
+                interface.interface_name(),
+                InterfaceTypeDef::Properties,
+                interface.interface_type(),
+            )))
         })?;
 
-        let validated = ValidatedUnset::validate(mapping, path)?;
+        let validated = ValidatedUnset::validate(mapping)?;
 
         debug!("unsetting property {interface_name}{path}");
 

src/connection.rs

@@ -32,7 +32,7 @@ use tokio::{
 };
 use tracing::{debug, error, info, trace, warn};
 
-use crate::error::AggregateError;
+use crate::error::AggregationError;
 use crate::store::PropertyMapping;
 use crate::transport::TransportError;
 use crate::{
@@ -929,8 +929,8 @@ impl<S, C> DeviceReceiver<S, C> {
         C: Receive + Sync,
     {
         let Some(object) = interface.as_object_ref() else {
-            let aggr_err = AggregateError::for_interface(
-                interface.interface_name(),
+            let aggr_err = AggregationError::new(
+                interface.interface_name().to_string(),
                 path.to_string(),
                 InterfaceAggregation::Object,
                 InterfaceAggregation::Individual,

src/error.rs

@@ -20,7 +20,7 @@
 
 use crate::interface::error::InterfaceError;
 use crate::interface::mapping::path::MappingError;
-use crate::interface::{Aggregation, InterfaceTypeDef};
+use crate::interface::{Aggregation, InterfaceTypeDef, Ownership};
 use crate::introspection::AddInterfaceError;
 use crate::properties::PropertiesError;
 use crate::retention::RetentionError;
@@ -49,14 +49,6 @@ pub enum Error {
     /// Error while operating on the device introspection.
     #[error("couldn't complete introspection operation")]
     AddInterface(#[from] AddInterfaceError),
-    /// Invalid interface type when sending or receiving.
-    #[error("invalid interface type, expected {exp} but got {got}")]
-    InterfaceType {
-        /// Expected interface type.
-        exp: InterfaceTypeDef,
-        /// Actual interface type.
-        got: InterfaceTypeDef,
-    },
     /// Couldn't find an interface with the given name.
     #[error("couldn't find interface '{name}'")]
     InterfaceNotFound {
@@ -88,7 +80,7 @@ pub enum Error {
     Validation(#[from] UserValidationError),
     /// Invalid aggregation between the interface and the data.
     #[error(transparent)]
-    Aggregation(#[from] AggregateError),
+    Aggregation(#[from] AggregationError),
     /// Infallible conversion.
     #[error(transparent)]
     Infallible(#[from] Infallible),
@@ -110,88 +102,78 @@ pub enum Error {
     Grpc(#[from] crate::transport::grpc::GrpcError),
 }
 
-/// Aggregate error variant.
+/// Aggregation error.
 ///
 /// This provides additional context in case of an aggregation error
 #[derive(Debug, thiserror::Error)]
-#[error("invalid aggregation in {ctx} for {interface}{path}, expected {exp} but got {got}")]
+#[error("invalid aggregation for {interface}{path}, expected {exp} but got {got}")]
 #[non_exhaustive]
-pub struct AggregateError {
+pub struct AggregationError {
     /// Interface name
     interface: String,
     /// Path
     path: String,
-    /// Context to differentiate interface from payload error
-    ctx: AggregationCtx,
     /// Expected aggregation of the interface.
     exp: Aggregation,
     /// The actual aggregation.
     got: Aggregation,
 }
 
-impl AggregateError {
-    pub(crate) fn new(
-        interface: String,
-        path: String,
-        ctx: AggregationCtx,
-        exp: Aggregation,
-        got: Aggregation,
-    ) -> Self {
+impl AggregationError {
+    pub(crate) fn new(interface: String, path: String, exp: Aggregation, got: Aggregation) -> Self {
         Self {
             interface,
             path,
-            ctx,
             exp,
             got,
         }
     }
+}
 
-    pub(crate) fn for_interface(
-        interface: impl Into<String>,
-        path: impl Into<String>,
-        exp: Aggregation,
-        got: Aggregation,
-    ) -> Self {
-        Self::new(
-            interface.into(),
-            path.into(),
-            AggregationCtx::Interface,
-            exp,
-            got,
-        )
-    }
+/// Invalid interface type when sending or receiving.
+#[derive(Debug, thiserror::Error)]
+#[error("invalid interface type for {name}, expected {exp} but got {got}")]
+pub struct InterfaceTypeError {
+    /// Name of the interface.
+    name: String,
+    /// Expected interface type.
+    exp: InterfaceTypeDef,
+    /// Actual interface type.
+    got: InterfaceTypeDef,
+}
 
-    #[cfg(feature = "message-hub")]
-    pub(crate) fn for_payload(
-        interface: impl Into<String>,
-        path: impl Into<String>,
-        exp: Aggregation,
-        got: Aggregation,
+impl InterfaceTypeError {
+    pub(crate) fn new(
+        name: impl Into<String>,
+        exp: InterfaceTypeDef,
+        got: InterfaceTypeDef,
     ) -> Self {
-        Self::new(
-            interface.into(),
-            path.into(),
-            AggregationCtx::Payload,
+        Self {
+            name: name.into(),
             exp,
             got,
-        )
+        }
     }
 }
 
-/// Context for the [AggregateError] error variant.
-#[derive(Debug, Clone)]
-pub enum AggregationCtx {
-    /// Error occurring when checking the interface.
-    Interface,
-    /// Error occurring when checking the payload data.
-    Payload,
+/// Sending data on an interface not owned by the device
+#[derive(Debug, thiserror::Error)]
+#[error("invalid ownership for interface {name}, expected {exp} but got {got}")]
+pub struct OwnershipError {
+    /// Name of the interface.
+    name: String,
+    /// Expected interface ownership.
+    exp: Ownership,
+    /// Actual interface ownership.
+    got: Ownership,
 }
 
-impl Display for AggregationCtx {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        match self {
-            AggregationCtx::Interface => write!(f, "interface"),
-            AggregationCtx::Payload => write!(f, "payload"),
+impl OwnershipError {
+    pub(crate) fn new(name: impl Into<String>, exp: Ownership, got: Ownership) -> Self {
+        Self {
+            name: name.into(),
+            exp,
+            got,
         }
     }
 }

src/interfaces.rs

@@ -27,11 +27,14 @@ use itertools::Itertools;
 use tracing::{debug, trace};
 
 use crate::{
+    error::InterfaceTypeError,
     interface::{
         error::InterfaceError,
         mapping::path::MappingPath,
         reference::{MappingRef, PropertyRef},
+        InterfaceTypeDef,
     },
+    validate::UserValidationError,
     Error, Interface,
 };
 
@@ -162,9 +165,12 @@ impl Interfaces {
                 name: interface_name.to_string(),
             })
             .and_then(|interface| {
-                interface.as_prop_ref().ok_or_else(|| Error::InterfaceType {
-                    exp: crate::interface::InterfaceTypeDef::Properties,
-                    got: interface.interface_type(),
+                interface.as_prop_ref().ok_or_else(|| {
+                    Error::Validation(UserValidationError::InterfaceType(InterfaceTypeError::new(
+                        interface_name,
+                        InterfaceTypeDef::Properties,
+                        interface.interface_type(),
+                    )))
                 })
             })
             .and_then(|interface| {

src/lib.rs

@@ -115,6 +115,19 @@ mod test {
         "../e2e-test/interfaces/additional/org.astarte-platform.rust.e2etest.DeviceProperty.json"
     );
 
+    pub(crate) const DEVICE_PROPERTIES_NO_UNSET: &str = r#"{
+    "interface_name": "org.astarte-platform.rust.examples.individual-properties.DevicePropertyNoUnset",
+    "version_major": 0,
+    "version_minor": 1,
+    "type": "properties",
+    "ownership": "device",
+    "mappings": [{
+        "endpoint": "/%{sensor_id}/enable",
+        "type": "boolean",
+        "allow_unset": false
+    }]
+}"#;
+
     pub(crate) async fn mock_astarte_device<I>(
         client: AsyncClient,
         eventloop: MqttEventLoop,

src/transport/grpc/mod.rs

@@ -56,7 +56,7 @@ use super::{
 use crate::aggregate::AstarteObject;
 use crate::builder::ConnectionBuildConfig;
 use crate::client::RecvError;
-use crate::error::AggregateError;
+use crate::error::AggregationError;
 use crate::interface::Aggregation;
 use crate::retention::memory::SharedVolatileStore;
 use crate::retention::{PublishInfo, RetentionId};
@@ -442,8 +442,8 @@ where
         };
 
         let individual = data.take_individual().ok_or_else(|| {
-            let aggr_err = AggregateError::for_payload(
-                mapping.interface().interface_name(),
+            let aggr_err = AggregationError::new(
+                mapping.interface().interface_name().to_string(),
                 mapping.path().to_string(),
                 Aggregation::Individual,
                 Aggregation::Object,
@@ -469,8 +469,8 @@ where
             .take_data()
             .and_then(|d| d.take_object())
             .ok_or_else(|| {
-                RecvError::Aggregation(AggregateError::for_payload(
-                    object.interface.interface_name(),
+                RecvError::Aggregation(AggregationError::new(
+                    object.interface.interface_name().to_string(),
                     path.to_string(),
                     Aggregation::Object,
                     Aggregation::Individual,
@@ -1203,7 +1203,6 @@ mod test {
 
             let validated_individual = mock_validate_individual(
                 mapping_ref,
-                &path,
                 AstarteType::String(STRING_VALUE.to_string()),
                 None,
             )

src/transport/mod.rs

@@ -216,7 +216,7 @@ pub trait Disconnect {
 #[cfg(test)]
 mod test {
     use crate::aggregate::AstarteObject;
-    use crate::error::AggregateError;
+    use crate::error::AggregationError;
     use crate::{
         interface::{mapping::path::MappingPath, reference::MappingRef},
         types::{AstarteType, TypeError},
@@ -231,8 +231,8 @@ mod test {
         timestamp: Option<chrono::DateTime<chrono::Utc>>,
     ) -> Result<ValidatedObject, crate::Error> {
         let object = interface.as_object_ref().ok_or_else(|| {
-            let aggr_err = AggregateError::for_interface(
-                interface.interface_name(),
+            let aggr_err = AggregationError::new(
+                interface.interface_name().to_string(),
                 path.to_string(),
                 crate::interface::Aggregation::Object,
                 interface.aggregation(),
@@ -245,7 +245,6 @@ mod test {
 
     pub(crate) fn mock_validate_individual<'a, D>(
         mapping_ref: MappingRef<'a, &'a Interface>,
-        path: &'a MappingPath<'a>,
         data: D,
         timestamp: Option<chrono::DateTime<chrono::Utc>>,
     ) -> Result<ValidatedIndividual, crate::Error>
@@ -254,7 +253,6 @@ mod test {
     {
         let individual = data.try_into().map_err(|_| TypeError::Conversion)?;
 
-        ValidatedIndividual::validate(mapping_ref, path, individual, timestamp)
-            .map_err(|uve| uve.into())
+        ValidatedIndividual::validate(mapping_ref, individual, timestamp).map_err(|uve| uve.into())
     }
 }