Added reportDTCExtDataRecordByRecordNumber subfunction support

Author: pylessard

doc/source/udsoncan/client.rst

@@ -400,6 +400,7 @@ Methods by services
 .. automethod:: udsoncan.client.Client.get_user_defined_dtc_snapshot_by_dtc_number
 .. automethod:: udsoncan.client.Client.get_dtc_snapshot_by_record_number
 .. automethod:: udsoncan.client.Client.get_dtc_extended_data_by_dtc_number
+.. automethod:: udsoncan.client.Client.get_dtc_extended_data_by_record_number
 .. automethod:: udsoncan.client.Client.get_user_defined_dtc_extended_data_by_dtc_number
 .. automethod:: udsoncan.client.Client.get_mirrormemory_dtc_extended_data_by_dtc_number
 

test/client/test_read_dtc_information.py

@@ -2584,6 +2584,319 @@ def __init__(self, *args, **kwargs):
 
 
 
+class TestreportDTCExtDataRecordByRecordNumber(ClientServerTest):   # Subfn = 0x16
+    sb = struct.pack('B', 0x16)
+    badsb = struct.pack('B', 0x16+1)
+
+    def assert_single_data_response(self, response):
+        self.assertEqual(len(response.service_data.dtcs), 1)
+        self.assertEqual(response.service_data.dtc_count, 1)
+
+        dtc = response.service_data.dtcs[0]
+
+        self.assertEqual(dtc.id, 0x123456)
+        self.assertEqual(dtc.status.get_byte_as_int(), 0x20)
+
+        self.assertEqual(len(dtc.extended_data), 1)
+        extended_data = dtc.extended_data[0]
+
+        self.assertTrue(isinstance(extended_data, Dtc.ExtendedData))
+        self.assertEqual(extended_data.record_number, 0x33) 
+
+        self.assertEqual(extended_data.raw_data, b'\x01\x02\x03\x04\x05')
+
+    def test_single_data(self):
+        for i in range(3):
+            request = self.conn.touserqueue.get(timeout=0.2)
+            self.assertEqual(request, b'\x19' + self.sb + b'\x33')
+            self.conn.fromuserqueue.put(b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05')
+
+    def _test_single_data(self):
+        response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = 5)
+        self.assert_single_data_response(response)
+
+        response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5})
+        self.assert_single_data_response(response)
+
+        self.udsclient.config['extended_data_size'] = {0x123456 : 5}
+        response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33)
+        self.assert_single_data_response(response)
+
+    def test_single_data_missing_size(self):
+        for i in range(2):
+            self.wait_request_and_respond(b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05')
+
+    def _test_single_data_missing_size(self):
+        with self.assertRaises(ConfigError):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x111111 : 5})  # 123456 is not there!
+        
+        with self.assertRaises(ConfigError):
+            self.udsclient.config['extended_data_size'] = {0x111111 : 5}    # 123456 is not there!
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33)
+
+    def test_single_data_zeropadding_ok(self):
+        data = b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05'
+        for i in range(8):
+            self.wait_request_and_respond(data + b'\x00' * (i+1))
+
+    def _test_single_data_zeropadding_ok(self):
+        self.udsclient.config['tolerate_zero_padding'] = True
+        for i in range(8):
+            response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5})
+            self.assert_single_data_response(response)
+
+    def test_single_data_zeropadding_notok_exception(self):
+        data = b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05'
+        for i in range(8):
+            self.wait_request_and_respond(data + b'\x00' * (i+1))
+
+    def _test_single_data_zeropadding_notok_exception(self):
+        self.udsclient.config['tolerate_zero_padding'] = False
+        for i in range(8):
+            with self.assertRaises(InvalidResponseException):
+                self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5})
+
+
+    def test_single_data_zeropadding_notok_no_exception(self):
+        data = b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05'
+        for i in range(8):
+            self.wait_request_and_respond(data + b'\x00' * (i+1))
+
+    def _test_single_data_zeropadding_notok_no_exception(self):
+        self.udsclient.config['tolerate_zero_padding'] = False
+        self.udsclient.config['exception_on_invalid_response'] = False
+        for i in range(8):
+            response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5})
+            self.assertFalse(response.valid)
+
+
+    def test_double_data(self):
+        self.wait_request_and_respond(b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05\x78\x9a\xbc\x30\xaa\xbb\xcc')
+
+    def _test_double_data(self):
+        response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+
+        self.assertEqual(len(response.service_data.dtcs), 2)
+        self.assertEqual(response.service_data.dtc_count, 2)
+
+        dtc = response.service_data.dtcs[0]
+
+        self.assertEqual(dtc.id, 0x123456)
+        self.assertEqual(dtc.status.get_byte_as_int(), 0x20)
+
+        self.assertEqual(len(dtc.extended_data), 1)
+
+        self.assertTrue(isinstance(dtc.extended_data[0], Dtc.ExtendedData))
+        self.assertEqual(dtc.extended_data[0].record_number, 0x33)  
+        self.assertEqual(dtc.extended_data[0].raw_data, b'\x01\x02\x03\x04\x05')
+
+
+        dtc = response.service_data.dtcs[1]
+
+        self.assertEqual(dtc.id, 0x789abc)
+        self.assertEqual(dtc.status.get_byte_as_int(), 0x30)
+
+        self.assertEqual(len(dtc.extended_data), 1)
+
+        self.assertTrue(isinstance(dtc.extended_data[0], Dtc.ExtendedData))
+        self.assertEqual(dtc.extended_data[0].record_number, 0x33)  
+        self.assertEqual(dtc.extended_data[0].raw_data, b'\xaa\xbb\xcc')
+
+
+    def test_no_data(self):
+        self.wait_request_and_respond(b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20')
+
+    def _test_no_data(self):
+        with self.assertRaises(InvalidResponseException):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5})
+
+    def test_zero_sized_data(self):
+        self.wait_request_and_respond(b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20')
+
+    def _test_zero_sized_data(self):
+        response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 0})
+
+        self.assertEqual(len(response.service_data.dtcs), 1)
+        self.assertEqual(response.service_data.dtc_count, 1)
+        dtc = response.service_data.dtcs[0]
+
+        self.assertEqual(dtc.id, 0x123456)
+        self.assertEqual(dtc.status.get_byte_as_int(), 0x20)
+        self.assertEqual(len(dtc.extended_data), 1)
+
+        self.assertEqual(len(dtc.extended_data[0].raw_data), 0)
+ 
+    def test_zero_sized_data_zeropadding_ok(self):
+        data = b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20'
+        for i in range(8):
+            self.wait_request_and_respond(data + b'\x00' * (i+1) )
+
+    def _test_zero_sized_data_zeropadding_ok(self):
+        self.udsclient.config['tolerate_zero_padding'] = True
+        for i in range(8):
+            response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 0})
+
+            self.assertEqual(len(response.service_data.dtcs), 1)
+            self.assertEqual(response.service_data.dtc_count, 1)
+            dtc = response.service_data.dtcs[0]
+
+            self.assertEqual(dtc.id, 0x123456)
+            self.assertEqual(dtc.status.get_byte_as_int(), 0x20)
+            self.assertEqual(len(dtc.extended_data), 1)
+
+        self.assertEqual(len(dtc.extended_data[0].raw_data), 0)
+
+    def test_zero_sized_data_zeropadding_not_ok_exception(self):
+        data = b'\x59'  + self.sb + b'\x33\x12\x34\x56\x20'
+        for i in range(8):
+            self.wait_request_and_respond(data + b'\x00' * (i+1) )
+
+    def _test_zero_sized_data_zeropadding_not_ok_exception(self):
+        self.udsclient.config['tolerate_zero_padding'] = False
+        for i in range(8):
+            with self.assertRaises(InvalidResponseException):
+                self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 0})
+
+    def invalid_length_no_response_server_task(self):
+        self.wait_request_and_respond(b'')
+        self.wait_request_and_respond(b'\x59')
+        self.wait_request_and_respond(b'\x59'+ self.sb)
+
+    def test_invalid_length_no_response_exception(self):
+        self.invalid_length_no_response_server_task()
+
+    def _test_invalid_length_no_response_exception(self):
+        for i in range(3):
+            with self.assertRaises(InvalidResponseException):
+                self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+
+    def test_invalid_length_no_response_no_exception(self):
+        self.invalid_length_no_response_server_task()
+
+    def _test_invalid_length_no_response_no_exception(self):
+        self.udsclient.config['exception_on_invalid_response'] = False
+        for i in range(3):
+            response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+            self.assertFalse(response.valid)
+
+
+    def invalid_length_incomplete_dtc_server_task(self):
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12')
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34')
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34\x56')
+
+    def test_invalid_length_incomplete_dtc_exception(self):
+        self.invalid_length_incomplete_dtc_server_task()
+
+    def _test_invalid_length_incomplete_dtc_exception(self):
+        for i in range(3):
+            with self.assertRaises(InvalidResponseException):
+                self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+
+    def test_invalid_length_incomplete_dtc_no_exception(self):
+        self.invalid_length_incomplete_dtc_server_task()
+
+    def _test_invalid_length_incomplete_dtc_no_exception(self):
+        self.udsclient.config['exception_on_invalid_response'] = False
+        for i in range(3):
+            response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+            self.assertFalse(response.valid)
+
+    def invalid_length_missing_data_server_task(self):
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34\x56\x20')
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34\x56\x20\x01')
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34\x56\x20\x01\x02')
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03')
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04')
+
+    def test_invalid_length_missing_data_exception(self):
+        self.invalid_length_missing_data_server_task()
+
+    def _test_invalid_length_missing_data_exception(self):
+        for i in range(5):
+            with self.assertRaises(InvalidResponseException):
+                self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+
+    def test_invalid_length_missing_data_no_exception(self):
+        self.invalid_length_missing_data_server_task()
+
+    def _test_invalid_length_missing_data_no_exception(self):
+        self.udsclient.config['exception_on_invalid_response'] = False
+        for i in range(5):
+            response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+            self.assertFalse(response.valid)
+
+    def test_wrong_subfn_response_exception(self):
+        self.wait_request_and_respond(b'\x59' + self.badsb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05')
+
+    def _test_wrong_subfn_response_exception(self):
+        with self.assertRaises(UnexpectedResponseException):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33)
+
+    def test_wrong_subfn_response_no_exception(self):
+        self.wait_request_and_respond(b'\x59' + self.badsb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05')
+
+    def _test_wrong_subfn_response_no_exception(self):
+        self.udsclient.config['exception_on_unexpected_response'] = False
+        response = self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+        self.assertTrue(response.valid)
+        self.assertTrue(response.unexpected)
+
+    def test_wrong_record_number_response_exception(self):
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x34\x12\x34\x56\x20\x01\x02\x03\x04\x05')
+
+    def _test_wrong_record_number_response_exception(self):
+        with self.assertRaises(UnexpectedResponseException):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+
+    def test_duplicate_dtc(self):
+        self.wait_request_and_respond(b'\x59' + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05\x12\x34\x56\x20\x01\x02\x03\x04\x05')
+
+    def _test_duplicate_dtc(self):
+        with self.assertRaises(InvalidResponseException):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33,  data_size = {0x123456 : 5, 0x789abc : 3})        
+
+    def test_wrong_service_response_exception(self):
+        self.wait_request_and_respond(b'\x6F' + self.sb + b'\x33\x12\x34\x56\x20\x01\x02\x03\x04\x05')
+
+    def _test_wrong_service_response_exception(self):
+        with self.assertRaises(UnexpectedResponseException):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : 5, 0x789abc : 3})
+
+    def test_oob_values(self):
+        pass
+
+    def _test_oob_values(self):
+        with self.assertRaises(ValueError):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = -1, data_size=5)
+        
+        with self.assertRaises(ValueError):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0xF0, data_size=5) # Limited to 0xEF
+
+        with self.assertRaises(ValueError):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 'asd', data_size=5)
+
+        with self.assertRaises(NotImplementedError):
+            self.udsclient.set_config('standard_version', 2013)
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size=5)
+        self.udsclient.set_config('standard_version', latest_standard)    
+
+    def test_oob_values_data_size(self): # validation is made at interpret_response
+        self.wait_request_and_respond(b'\x59'  + self.sb + b'\x12\x34\x56\x20\x99\x01\x02\x03')
+        self.wait_request_and_respond(b'\x59'  + self.sb + b'\x12\x34\x56\x20\x99\x01\x02\x03')
+        self.wait_request_and_respond(b'\x59'  + self.sb + b'\x12\x34\x56\x20\x99\x01\x02\x03')
+
+    def _test_oob_values_data_size(self):
+        with self.assertRaises(ValueError):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = -1)
+
+        with self.assertRaises(ValueError):
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33, data_size = {0x123456 : -1})
+
+        with self.assertRaises(ValueError):
+            self.udsclient.config['extended_data_size'] = {0x123456 : -1}
+            self.udsclient.get_dtc_extended_data_by_record_number(record_number = 0x33)
+
 
 class TestReportUserDefMemoryDTCByStatusMask(ClientServerTest): # Subfn = 0x17
     # Almost identical copy of duplicate of GenericTestStatusMaskRequest_DtcAndStatusMaskResponse
@@ -2860,7 +3173,6 @@ def _test_oob_value(self):
         self.udsclient.set_config('standard_version', latest_standard)
 
 
-
 class TestReportUserDefMemoryDTCSnapshotRecordByDTCNumber(ClientServerTest): # Subfn = 0x18
 
     class Codec4711(DidCodec):
@@ -3301,8 +3613,6 @@ def _test_oob_values(self):
         self.udsclient.set_config('standard_version', latest_standard)
 
 
-
-
 class TestTeportUserDefMemoryDTCExtDataRecordByDTCNumber(ClientServerTest): # Subfn = 0x19
     sb = struct.pack('B', 0x19)
     badsb = struct.pack('B', 0x19+1)