Add ecu_functinal_address and send_diagnostic_func for DoIPClient

doipclient/client.py

@@ -699,12 +699,23 @@ def request_entity_status(self):
     def send_diagnostic(self, diagnostic_payload, timeout=A_PROCESSING_TIME):
         """Send a raw diagnostic payload (ie: UDS) to the ECU.
 
+        :param diagnostic_payload: UDS payload to transmit to the ECU
+        :type diagnostic_payload: bytearray
+        :raises IOError: DoIP negative acknowledgement received
+        """
+        self.send_diagnostic_to_address(self._ecu_logical_address, diagnostic_payload, timeout)
+
+    def send_diagnostic_to_address(self, address, diagnostic_payload, timeout=A_PROCESSING_TIME):
+        """Send a raw diagnostic payload (ie: UDS) to the specified address.
+
+        :param address: The logical address to send the diagnostic payload to
+        :type address: int
         :param diagnostic_payload: UDS payload to transmit to the ECU
         :type diagnostic_payload: bytearray
         :raises IOError: DoIP negative acknowledgement received
         """
         message = DiagnosticMessage(
-            self._client_logical_address, self._ecu_logical_address, diagnostic_payload
+            self._client_logical_address, address, diagnostic_payload
         )
         self.send_doip_message(message)
         start_time = time.time()

tests/test_client.py

@@ -99,6 +99,9 @@
 diagnostic_request = bytearray(
     [int(x, 16) for x in "02 fd 80 01 00 00 00 07 0e 00 00 01 00 01 02".split(" ")]
 )
+diagnostic_request_to_address = bytearray(
+    [int(x, 16) for x in "02 fd 80 01 00 00 00 07 0e 00 12 34 00 01 02".split(" ")]
+)
 unknown_mercedes_message = bytearray(
     [
         int(x, 16)
@@ -500,6 +503,22 @@ def test_send_diagnostic_negative(mock_socket):
     assert mock_socket.tx_queue[-1] == diagnostic_request
 
 
+def test_send_diagnostic_to_address_positive(mock_socket):
+    sut = DoIPClient(test_ip, test_logical_address)
+    mock_socket.rx_queue.append(diagnostic_positive_response)
+    assert None == sut.send_diagnostic_to_address(0x1234, bytearray([0, 1, 2]))
+    assert mock_socket.tx_queue[-1] == diagnostic_request_to_address
+
+def test_send_diagnostic_to_address_negative(mock_socket):
+    sut = DoIPClient(test_ip, test_logical_address)
+    mock_socket.rx_queue.append(diagnostic_negative_response)
+    with pytest.raises(
+        IOError, match=r"Diagnostic request rejected with negative acknowledge code"
+    ):
+        result = sut.send_diagnostic_to_address(0x1234, bytearray([0, 1, 2]))
+    assert mock_socket.tx_queue[-1] == diagnostic_request_to_address
+
+
 def test_receive_diagnostic(mock_socket):
     sut = DoIPClient(test_ip, test_logical_address)
     mock_socket.rx_queue.append(diagnostic_result)