Update pipeline op & event tests to follow new pattern (#104)

* Update pipeline op & event tests to follow new pattern, make /request_id use consistent

* Add documentation

* flake8 and test fixes

* code-review feedback

Author: BertKleewein

azure-iot-device/azure/iot/device/common/pipeline/pipeline_events_base.py

@@ -56,7 +56,7 @@ class IotResponseEvent(PipelineEvent):
     """
 
     def __init__(self, request_id, status_code, response_body):
-        super(IotResponseEvent, self).__init()
+        super(IotResponseEvent, self).__init__()
         self.request_id = request_id
         self.status_code = status_code
         self.response_body = response_body

azure-iot-device/azure/iot/device/common/pipeline/pipeline_stages_base.py

@@ -664,7 +664,7 @@ def _handle_pipeline_event(self, event):
                 op.response_body = event.response_body
             else:
                 logger.warning(
-                    "IotResponseEvent with unknown RID received.  Nothing to do. Dropping"
+                    "IotResponseEvent with unknown request_id received.  Nothing to do. Dropping"
                 )
         else:
             super(CoordinateRequestAndResponse, self)._handle_pipeline_event(event)

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

@@ -51,8 +51,8 @@ def get_method_topic_for_publish(request_id, status):
     :return: The topic for publishing method responses. It is of the format
     "$iothub/methods/res/<status>/?$rid=<requestId>
     """
-    return "$iothub/methods/res/{status}/?$rid={rid}".format(
-        status=urllib.parse.quote_plus(status), rid=urllib.parse.quote_plus(request_id)
+    return "$iothub/methods/res/{status}/?$rid={request_id}".format(
+        status=urllib.parse.quote_plus(status), request_id=urllib.parse.quote_plus(request_id)
     )
 
 

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

@@ -146,12 +146,13 @@ class GetTwin(PipelineOperation):
     :type twin: Twin
     """
 
-    def __init__(self):
+    def __init__(self, callback=None):
         """
         Initializer for GetTwin objects.
         """
-        super(GetTwin, self).__init()
+        super(GetTwin, self).__init__(callback=callback)
         self.twin = None
+        self.needs_connection = True
 
 
 class PatchTwinReportedProperties(PipelineOperation):
@@ -160,12 +161,13 @@ class PatchTwinReportedProperties(PipelineOperation):
     IoT Hub or Azure IoT Edge Hub service.
     """
 
-    def __init__(self, patch):
+    def __init__(self, patch, callback=None):
         """
         Initializer for PatchTwinReportedProperties object
 
         :param patch: The reported properties patch to send to the service.
         :type patch: dict, str, int, float, bool, or None (JSON compatible values)
         """
-        super(PatchTwinReportedProperties, self).__init()
+        super(PatchTwinReportedProperties, self).__init__(callback=callback)
         self.patch = patch
+        self.needs_connection = True

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

@@ -154,21 +154,21 @@ def _handle_pipeline_event(self, event):
                 )
 
             elif mqtt_topic_iothub.is_method_topic(topic):
-                rid = mqtt_topic_iothub.get_method_request_id_from_topic(topic)
+                request_id = mqtt_topic_iothub.get_method_request_id_from_topic(topic)
                 method_name = mqtt_topic_iothub.get_method_name_from_topic(topic)
                 method_received = MethodRequest(
-                    request_id=rid,
+                    request_id=request_id,
                     name=method_name,
                     payload=json.loads(event.payload.decode("utf-8")),
                 )
                 self.handle_pipeline_event(pipeline_events_iothub.MethodRequest(method_received))
 
             elif mqtt_topic_iothub.is_twin_response_topic(topic):
-                rid = mqtt_topic_iothub.get_twin_request_id_from_topic(topic)
+                request_id = mqtt_topic_iothub.get_twin_request_id_from_topic(topic)
                 status_code = mqtt_topic_iothub.get_twin_status_code_from_topic(topic)
                 self.handle_pipeline_event(
                     pipeline_events_base.IotResponseEvent(
-                        request_id=rid, status_code=status_code, response_body=event.payload
+                        request_id=request_id, status_code=status_code, response_body=event.payload
                     )
                 )
 

azure-iot-device/azure/iot/device/provisioning/internal/polling_machine.py

@@ -186,11 +186,11 @@ def _send_register_request(self, event_data):
         logger.info("Sending registration request")
         self._set_query_timer()
 
-        rid = str(uuid.uuid4())
+        request_id = str(uuid.uuid4())
 
-        self._operations[rid] = constant.PUBLISH_TOPIC_REGISTRATION.format(rid)
+        self._operations[request_id] = constant.PUBLISH_TOPIC_REGISTRATION.format(request_id)
         self._request_response_provider.send_request(
-            rid=rid,
+            request_id=request_id,
             request_payload=" ",
             operation_id=None,
             callback_on_response=self._on_register_response_received,
@@ -203,22 +203,24 @@ def _query_operation_status(self, event_data):
         logger.info("Querying operation status from polling machine")
         self._set_query_timer()
 
-        rid = str(uuid.uuid4())
+        request_id = str(uuid.uuid4())
         result = event_data.args[0].args[0]
 
         operation_id = result.operation_id
-        self._operations[rid] = constant.PUBLISH_TOPIC_QUERYING.format(rid, operation_id)
+        self._operations[request_id] = constant.PUBLISH_TOPIC_QUERYING.format(
+            request_id, operation_id
+        )
         self._request_response_provider.send_request(
-            rid=rid,
+            request_id=request_id,
             request_payload=" ",
             operation_id=operation_id,
             callback_on_response=self._on_query_response_received,
         )
 
-    def _on_register_response_received(self, rid, status_code, key_values_dict, response):
+    def _on_register_response_received(self, request_id, status_code, key_values_dict, response):
         """
         The function to call in case of a response from a registration request.
-        :param rid: The id of the original register request.
+        :param request_id: The id of the original register request.
         :param status_code: The status code in the response.
         :param key_values_dict: The dictionary containing the query parameters of the returned topic.
         :param response: The complete response from the service.
@@ -228,21 +230,21 @@ def _on_register_response_received(self, rid, status_code, key_values_dict, resp
         retry_after = (
             None if "retry-after" not in key_values_dict else str(key_values_dict["retry-after"][0])
         )
-        intermediate_registration_result = RegistrationQueryStatusResult(rid, retry_after)
+        intermediate_registration_result = RegistrationQueryStatusResult(request_id, retry_after)
 
         if int(status_code, 10) >= 429:
-            del self._operations[rid]
+            del self._operations[request_id]
             self._trig_wait(intermediate_registration_result)
         elif int(status_code, 10) >= 300:  # pure failure
             self._registration_error = ValueError("Incoming message failure")
             self._trig_error()
         else:  # successful case, transition into complete or poll status
-            self._process_successful_response(rid, retry_after, response)
+            self._process_successful_response(request_id, retry_after, response)
 
-    def _on_query_response_received(self, rid, status_code, key_values_dict, response):
+    def _on_query_response_received(self, request_id, status_code, key_values_dict, response):
         """
         The function to call in case of a response from a polling/query request.
-        :param rid: The id of the original query request.
+        :param request_id: The id of the original query request.
         :param status_code: The status code in the response.
         :param key_values_dict: The dictionary containing the query parameters of the returned topic.
         :param response: The complete response from the service.
@@ -253,12 +255,12 @@ def _on_query_response_received(self, rid, status_code, key_values_dict, respons
         retry_after = (
             None if "retry-after" not in key_values_dict else str(key_values_dict["retry-after"][0])
         )
-        intermediate_registration_result = RegistrationQueryStatusResult(rid, retry_after)
+        intermediate_registration_result = RegistrationQueryStatusResult(request_id, retry_after)
 
         if int(status_code, 10) >= 429:
-            if rid in self._operations:
-                publish_query_topic = self._operations[rid]
-                del self._operations[rid]
+            if request_id in self._operations:
+                publish_query_topic = self._operations[request_id]
+                del self._operations[request_id]
                 topic_parts = publish_query_topic.split("$")
                 key_values_publish_topic = urllib.parse.parse_qs(
                     topic_parts[POS_QUERY_PARAM_PORTION]
@@ -273,17 +275,17 @@ def _on_query_response_received(self, rid, status_code, key_values_dict, respons
             self._registration_error = ValueError("Incoming message failure")
             self._trig_error()
         else:  # successful status code case, transition into complete or another poll status
-            self._process_successful_response(rid, retry_after, response)
+            self._process_successful_response(request_id, retry_after, response)
 
-    def _process_successful_response(self, rid, retry_after, response):
+    def _process_successful_response(self, request_id, retry_after, response):
         """
         Fucntion to call in case of 200 response from the service
-        :param rid: The request id
+        :param request_id: The request id
         :param retry_after: The time after which to try again.
         :param response: The complete response
         """
-        del self._operations[rid]
-        successful_result = self._decode_json_response(rid, retry_after, response)
+        del self._operations[request_id]
+        successful_result = self._decode_json_response(request_id, retry_after, response)
         if successful_result.status == "assigning":
             self._trig_wait(successful_result)
         elif successful_result.status == "assigned" or successful_result.status == "failed":
@@ -389,17 +391,17 @@ def _decode_complete_json_response(self, query_result, response):
             )
 
         registration_result = RegistrationResult(
-            rid=query_result.rid,
+            request_id=query_result.request_id,
             operation_id=query_result.operation_id,
             status=query_result.status,
             registration_state=registration_state,
         )
         return registration_result
 
-    def _decode_json_response(self, rid, retry_after, response):
+    def _decode_json_response(self, request_id, retry_after, response):
         """
         Decodes the json response for operation id and status
-        :param rid: The request id.
+        :param request_id: The request id.
         :param retry_after: The time in secs after which to retry.
         :param response: The complete response from the service.
         """
@@ -410,7 +412,7 @@ def _decode_json_response(self, rid, retry_after, response):
         )
         status = None if "status" not in decoded_result else str(decoded_result["status"])
 
-        return RegistrationQueryStatusResult(rid, retry_after, operation_id, status)
+        return RegistrationQueryStatusResult(request_id, retry_after, operation_id, status)
 
     def _on_disconnect_completed_error(self):
         logger.info("on_disconnect_completed for Device Provisioning Service")

azure-iot-device/azure/iot/device/provisioning/internal/registration_query_status_result.py

@@ -15,22 +15,22 @@ class RegistrationQueryStatusResult(object):
     from the provisioning service.
     """
 
-    def __init__(self, rid=None, retry_after=None, operation_id=None, status=None):
+    def __init__(self, request_id=None, retry_after=None, operation_id=None, status=None):
         """
-        :param rid: The request id to which the response is being obtained
+        :param request_id: The request id to which the response is being obtained
         :param retry_after : Number of secs after which to retry again.
         :param operation_id: The id of the operation as returned by the initial registration request.
         :param status: The status of the registration process.
         Values can be "unassigned", "assigning", "assigned", "failed", "disabled"
         from the provisioning service.
         """
-        self._request_id = rid
+        self._request_id = request_id
         self._operation_id = operation_id
         self._status = status
         self._retry_after = retry_after
 
     @property
-    def rid(self):
+    def request_id(self):
         return self._request_id
 
     @property

azure-iot-device/azure/iot/device/provisioning/internal/request_response_provider.py

@@ -28,17 +28,19 @@ def __init__(self, provisioning_pipeline):
 
         self._pending_requests = {}
 
-    def send_request(self, rid, request_payload, operation_id=None, callback_on_response=None):
+    def send_request(
+        self, request_id, request_payload, operation_id=None, callback_on_response=None
+    ):
         """
         Sends a request
-        :param rid: Id of the request
+        :param request_id: Id of the request
         :param request_payload: The payload of the request.
         :param operation_id: A id of the operation in case it is an ongoing process.
         :param callback_on_response: callback which is called when response comes back for this request.
         """
-        self._pending_requests[rid] = callback_on_response
+        self._pending_requests[request_id] = callback_on_response
         self._provisioning_pipeline.send_request(
-            rid=rid,
+            request_id=request_id,
             request_payload=request_payload,
             operation_id=operation_id,
             callback=self._on_publish_completed,
@@ -64,10 +66,10 @@ def disable_responses(self, callback=None):
             callback = self._on_unsubscribe_completed
         self._provisioning_pipeline.disable_responses(callback=callback)
 
-    def _receive_response(self, rid, status_code, key_value_dict, response_payload):
+    def _receive_response(self, request_id, status_code, key_value_dict, response_payload):
         """
         Handler that processes the response from the service.
-        :param rid: The id of the request which is being responded to.
+        :param request_id: The id of the request which is being responded to.
         :param status_code: The status code inside the response
         :param key_value_dict: A dictionary of keys mapped to a list of values extracted from the topic of the response.
         :param response_payload: String payload of the message received.
@@ -79,13 +81,13 @@ def _receive_response(self, rid, status_code, key_value_dict, response_payload):
         # """
         logger.info("Received response {}:".format(response_payload))
 
-        if rid in self._pending_requests:
-            callback = self._pending_requests[rid]
+        if request_id in self._pending_requests:
+            callback = self._pending_requests[request_id]
             # Only send the status code and the extracted topic
-            callback(rid, status_code, key_value_dict, response_payload)
-            del self._pending_requests[rid]
+            callback(request_id, status_code, key_value_dict, response_payload)
+            del self._pending_requests[request_id]
 
-        # TODO : What happens when rid if not there ? trigger error ?
+        # TODO : What happens when request_id if not there ? trigger error ?
 
     def _on_connection_state_change(self, new_state):
         """Handler to be called by the pipeline upon a connection state change."""

azure-iot-device/azure/iot/device/provisioning/models/registration_result.py

@@ -16,16 +16,16 @@ class RegistrationResult(object):
     from the provisioning service.
     """
 
-    def __init__(self, rid, operation_id, status, registration_state=None):
+    def __init__(self, request_id, operation_id, status, registration_state=None):
         """
-        :param rid: The request id to which the response is being obtained
+        :param request_id: The request id to which the response is being obtained
         :param operation_id: The id of the operation as returned by the initial registration request.
         :param status: The status of the registration process.
         Values can be "unassigned", "assigning", "assigned", "failed", "disabled"
         :param registration_state : Details like device id, assigned hub , date times etc returned
         from the provisioning service.
         """
-        self._request_id = rid
+        self._request_id = request_id
         self._operation_id = operation_id
         self._status = status
         self._registration_state = registration_state

azure-iot-device/azure/iot/device/provisioning/pipeline/mqtt_topic.py

@@ -24,21 +24,24 @@ def get_topic_for_subscribe():
     return _get_topic_base() + "res/#"
 
 
-def get_topic_for_register(rid):
+def get_topic_for_register(request_id):
     """
     return the topic string used to publish telemetry
     """
-    return (_get_topic_base() + "PUT/iotdps-register/?$rid={rid}").format(rid=rid)
+    return (_get_topic_base() + "PUT/iotdps-register/?$rid={request_id}").format(
+        request_id=request_id
+    )
 
 
-def get_topic_for_query(rid, operation_id):
+def get_topic_for_query(request_id, operation_id):
     """
     :return: The topic for cloud to device messages.It is of the format
     "devices/<deviceid>/messages/devicebound/#"
     """
     return (
-        _get_topic_base() + "GET/iotdps-get-operationstatus/?$rid={rid}&operationId={operation_id}"
-    ).format(rid=rid, operation_id=operation_id)
+        _get_topic_base()
+        + "GET/iotdps-get-operationstatus/?$rid={request_id}&operationId={operation_id}"
+    ).format(request_id=request_id, operation_id=operation_id)
 
 
 def get_topic_for_response():