Add EnsureDesiredPropertiesStage kwarg (#1013)

- Added ensure_desired_properties to abstract_clients and shared_client_tests

- Added unit tests

Author: lev-i

azure-iot-device/azure/iot/device/iothub/abstract_clients.py

@@ -39,6 +39,7 @@ def _validate_kwargs(exclude=[], **kwargs):
         "auto_connect",
         "connection_retry",
         "connection_retry_interval",
+        "ensure_desired_properties",
     ]
 
     for kwarg in kwargs:
@@ -59,6 +60,7 @@ def _get_config_kwargs(**kwargs):
         "auto_connect",
         "connection_retry",
         "connection_retry_interval",
+        "ensure_desired_properties",
     ]
 
     config_kwargs = {}
@@ -264,6 +266,8 @@ def create_from_connection_string(cls, connection_string, **kwargs):
         :param bool connection_retry: Attempt to re-establish a dropped connection (Default: True)
         :param int connection_retry_interval: Interval, in seconds, between attempts to
             re-establish a dropped connection (Default: 10)
+        :param bool ensure_desired_properties: Ensure the most recent desired properties patch has
+            been received upon re-connections (Default:True)
 
         :raises: ValueError if given an invalid connection_string.
         :raises: TypeError if given an unsupported parameter.
@@ -345,6 +349,8 @@ def create_from_sastoken(cls, sastoken, **kwargs):
         :param bool connection_retry: Attempt to re-establish a dropped connection (Default: True)
         :param int connection_retry_interval: Interval, in seconds, between attempts to
             re-establish a dropped connection (Default: 10)
+        :param bool ensure_desired_properties: Ensure the most recent desired properties patch has
+            been received upon re-connections (Default:True)
 
         :raises: TypeError if given an unsupported parameter.
         :raises: ValueError if the sastoken parameter is invalid.
@@ -554,6 +560,8 @@ def create_from_x509_certificate(cls, x509, hostname, device_id, **kwargs):
         :param bool connection_retry: Attempt to re-establish a dropped connection (Default: True)
         :param int connection_retry_interval: Interval, in seconds, between attempts to
             re-establish a dropped connection (Default: 10)
+        :param bool ensure_desired_properties: Ensure the most recent desired properties patch has
+            been received upon re-connections (Default:True)
 
         :raises: TypeError if given an unsupported parameter.
 
@@ -569,6 +577,7 @@ def create_from_x509_certificate(cls, x509, hostname, device_id, **kwargs):
             device_id=device_id, hostname=hostname, x509=x509, **config_kwargs
         )
         pipeline_configuration.blob_upload = True  # Blob Upload is a feature on Device Clients
+        pipeline_configuration.ensure_desired_properties = True
 
         # Pipeline setup
         http_pipeline = pipeline.HTTPPipeline(pipeline_configuration)
@@ -611,6 +620,8 @@ def create_from_symmetric_key(cls, symmetric_key, hostname, device_id, **kwargs)
         :param bool connection_retry: Attempt to re-establish a dropped connection (Default: True)
         :param int connection_retry_interval: Interval, in seconds, between attempts to
             re-establish a dropped connection (Default: 10)
+        :param bool ensure_desired_properties: Ensure the most recent desired properties patch has
+            been received upon re-connections (Default:True)
 
         :raises: TypeError if given an unsupported parameter.
         :raises: ValueError if the provided parameters are invalid.
@@ -637,6 +648,7 @@ def create_from_symmetric_key(cls, symmetric_key, hostname, device_id, **kwargs)
             device_id=device_id, hostname=hostname, sastoken=sastoken, **config_kwargs
         )
         pipeline_configuration.blob_upload = True  # Blob Upload is a feature on Device Clients
+        pipeline_configuration.ensure_desired_properties = True
 
         # Pipeline setup
         http_pipeline = pipeline.HTTPPipeline(pipeline_configuration)
@@ -799,6 +811,8 @@ def create_from_edge_environment(cls, **kwargs):
             server_verification_cert=server_verification_cert,
             **config_kwargs
         )
+        pipeline_configuration.ensure_desired_properties = True
+
         pipeline_configuration.method_invoke = (
             True  # Method Invoke is allowed on modules created from edge environment
         )
@@ -847,6 +861,8 @@ def create_from_x509_certificate(cls, x509, hostname, device_id, module_id, **kw
         :param bool connection_retry: Attempt to re-establish a dropped connection (Default: True)
         :param int connection_retry_interval: Interval, in seconds, between attempts to
             re-establish a dropped connection (Default: 10)
+        :param bool ensure_desired_properties: Ensure the most recent desired properties patch has
+            been received upon re-connections (Default:True)
 
         :raises: TypeError if given an unsupported parameter.
 
@@ -861,6 +877,7 @@ def create_from_x509_certificate(cls, x509, hostname, device_id, module_id, **kw
         pipeline_configuration = pipeline.IoTHubPipelineConfig(
             device_id=device_id, module_id=module_id, hostname=hostname, x509=x509, **config_kwargs
         )
+        pipeline_configuration.ensure_desired_properties = True
 
         # Pipeline setup
         http_pipeline = pipeline.HTTPPipeline(pipeline_configuration)

azure-iot-device/azure/iot/device/iothub/pipeline/config.py

@@ -13,7 +13,15 @@
 class IoTHubPipelineConfig(BasePipelineConfig):
     """A class for storing all configurations/options for IoTHub clients in the Azure IoT Python Device Client Library."""
 
-    def __init__(self, hostname, device_id, module_id=None, product_info="", **kwargs):
+    def __init__(
+        self,
+        hostname,
+        device_id,
+        module_id=None,
+        product_info="",
+        ensure_desired_properties=True,
+        **kwargs
+    ):
         """Initializer for IoTHubPipelineConfig which passes all unrecognized keyword-args down to BasePipelineConfig
         to be evaluated. This stacked options setting is to allow for unique configuration options to exist between the
         multiple clients, while maintaining a base configuration class with shared config options.
@@ -22,6 +30,8 @@ def __init__(self, hostname, device_id, module_id=None, product_info="", **kwarg
         :param str device_id: The device identity being used with the IoTHub
         :param str module_id: The module identity being used with the IoTHub
         :param str product_info: A custom identification string for the type of device connecting to Azure IoT Hub.
+        :param bool ensure_desired_properties: Indicates if twin_patches should ensure the most
+            recent desired properties patch has been received upon re-connections
         """
         super().__init__(hostname=hostname, **kwargs)
 
@@ -32,6 +42,9 @@ def __init__(self, hostname, device_id, module_id=None, product_info="", **kwarg
         # Product Info
         self.product_info = product_info
 
+        # Stage Behavior
+        self.ensure_desired_properties = ensure_desired_properties
+
         # Now, the parameters below are not exposed to the user via kwargs. They need to be set by manipulating the IoTHubPipelineConfig object.
         # They are not in the BasePipelineConfig because these do not apply to the provisioning client.
         self.blob_upload = False

azure-iot-device/azure/iot/device/iothub/pipeline/pipeline_stages_iothub.py

@@ -34,16 +34,19 @@ def __init__(self):
 
     @pipeline_thread.runs_on_pipeline_thread
     def _run_op(self, op):
-        if isinstance(op, pipeline_ops_base.EnableFeatureOperation):
-            # If we're enabling twin patches, we set last_version_seen to -1
-            # as a way of enabling this functionality.  If the ConnectedEvent handler
-            # sees this -1, it will send a GetTwinOperation to refresh desired properties.
-
-            if op.feature_name == constant.TWIN_PATCHES:
-                logger.debug(
-                    "{}: enabling twin patches.  setting last_version_seen".format(self.name)
-                )
-                self.last_version_seen = -1
+        if self.nucleus.pipeline_configuration.ensure_desired_properties:
+            if isinstance(op, pipeline_ops_base.EnableFeatureOperation):
+                # Ensure_desired_properties enables twin patches, when true, by setting last version
+                # to -1. The ConnectedEvent handler sees this and sends a GetTwinOperation to refresh
+                # desired properties. Setting ensure_desired_properties to false causes the GetTwinOp
+                # to not be sent. The rest of the functions in this stage stem from the GetTwinOperation,
+                # so disabling ensure_desired_properties effectively disables this stage.
+
+                if op.feature_name == constant.TWIN_PATCHES:
+                    logger.debug(
+                        "{}: enabling twin patches.  setting last_version_seen".format(self.name)
+                    )
+                    self.last_version_seen = -1
         self.send_op_down(op)
 
     @pipeline_thread.runs_on_pipeline_thread
@@ -102,20 +105,21 @@ def _on_get_twin_complete(self, op, error):
 
     @pipeline_thread.runs_on_pipeline_thread
     def _handle_pipeline_event(self, event):
-        if isinstance(event, pipeline_events_iothub.TwinDesiredPropertiesPatchEvent):
-            # remember the $version when we get a patch.
-            version = event.patch["$version"]
-            logger.debug(
-                "{}: Desired patch received.  Saving $version={}".format(self.name, version)
-            )
-            self.last_version_seen = version
-        elif isinstance(event, pipeline_events_base.ConnectedEvent):
-            # If last_version_seen is truthy, that means we've seen desired property patches
-            # before (or we've enabled them at least).  If this is the case, get the twin to
-            # see if the desired props have been updated.
-            if self.last_version_seen:
-                logger.info("{}: Reconnected.  Getting twin".format(self.name))
-                self._ensure_get_op()
+        if self.nucleus.pipeline_configuration.ensure_desired_properties:
+            if isinstance(event, pipeline_events_iothub.TwinDesiredPropertiesPatchEvent):
+                # remember the $version when we get a patch.
+                version = event.patch["$version"]
+                logger.debug(
+                    "{}: Desired patch received.  Saving $version={}".format(self.name, version)
+                )
+                self.last_version_seen = version
+            elif isinstance(event, pipeline_events_base.ConnectedEvent):
+                # If last_version_seen is truthy, that means we've seen desired property patches
+                # before (or we've enabled them at least).  If this is the case, get the twin to
+                # see if the desired props have been updated.
+                if self.last_version_seen:
+                    logger.info("{}: Reconnected.  Getting twin".format(self.name))
+                    self._ensure_get_op()
         self.send_event_up(event)