feat: add automotive diagnostic drivers (DoIP, UDS-CAN, UDS-DoIP)#295
feat: add automotive diagnostic drivers (DoIP, UDS-CAN, UDS-DoIP)#295mangelajo merged 15 commits intojumpstarter-dev:mainfrom
Conversation
Add four new driver packages for automotive diagnostic communication: jumpstarter-driver-uds: Shared UDS (ISO-14229) base package - UdsInterface base class with all 9 @export methods (change_session, ecu_reset, tester_present, read/write DID, security access, clear/read DTC), following the PowerInterface pattern - UdsClient shared client class - Pydantic models: UdsResponse, DidValue, DtcInfo, SecuritySeedResponse, UdsSessionType, UdsResetType - Handles NegativeResponseException for all operations - Unit tests for all UDS operations including NRC handling jumpstarter-driver-doip: Raw DoIP (ISO-13400) driver - DoIP class wrapping doipclient for low-level DoIP operations: routing activation, entity status, alive check, diagnostic power mode, vehicle identification, send/receive diagnostic, reconnect - Pydantic response models for all DoIP operations - DoIPDriverClient with typed client methods - Unit tests (mocked), error path tests, config validation tests - Integration tests using MockDoIPServer (TCP, port=0 dynamic allocation, daemon threads) for alive_check, routing_activation, and send/receive_diagnostic jumpstarter-driver-uds-can: UDS over CAN/ISO-TP transport - UdsCan(UdsInterface, Driver) -- transport-only, inherits all UDS operations from jumpstarter-driver-uds - Uses python-can + can-isotp + udsoncan libraries - Imports IsoTpParams from jumpstarter-driver-can for typed config - Proper close() for CAN bus + notifier + UDS client cleanup - Unit tests (mocked), error path tests, config validation tests - Integration tests using MockUdsEcu on virtual CAN bus (interface="virtual", request.node.name for channel isolation, ISO-TP framing with swapped rxid/txid, UDS response logic in daemon thread) for change_session, ecu_reset, tester_present, request_seed, send_key, clear_dtc, read_dtc jumpstarter-driver-uds-doip: UDS over DoIP transport - UdsDoip(UdsInterface, Driver) -- transport-only, inherits all UDS operations from jumpstarter-driver-uds - Uses doipclient + DoIPClientUDSConnector + udsoncan libraries - Proper close() for UDS + DoIP client cleanup - Unit tests (mocked), error path tests, config validation tests - Integration tests using MockDoIPUdsServer (independent DoIP TCP server with UDS response logic, self-contained -- no cross-package test imports) for change_session, ecu_reset, tester_present, request_seed, send_key, clear_dtc, read_dtc Workspace changes: - Register all 4 packages in python/pyproject.toml workspace sources - Add all 4 packages to jumpstarter-all/pyproject.toml meta-package - Add doc symlinks in docs/source/reference/package-apis/drivers/ - Update uv.lock Bug fixes discovered by integration tests: - Fix Dtc.Status access: .raw -> .get_byte_as_int() (udsoncan API) - Fix Dtc.Severity serialization: convert Severity object to int via .get_byte_as_int() for Pydantic DtcInfo model - Fix DiagnosticSessionControl mock response: include P2/P2* timing bytes required by udsoncan client validation All tests pass: 84 tests across 4 packages (10 + 22 + 27 + 25), linting clean.
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds four new Python driver packages (DoIP, UDS shared, UDS over CAN, UDS over DoIP) with drivers, clients, Pydantic models, test fixtures/mock servers, extensive unit/integration tests, packaging/pyproject updates, example mock ECU and end-to-end tests, and documentation README links and meta-package dependency updates. Changes
Sequence Diagram(s)sequenceDiagram
participant C as Client
participant D as DoIP Driver
participant DC as DoIPClient
participant ECU as ECU
C->>D: send_diagnostic(payload: bytes)
D->>DC: send_diagnostic(hex_payload)
DC->>ECU: TCP DoIP frame (routing + diagnostic)
ECU-->>DC: DoIP response frames
DC-->>D: ack / diagnostic response (hex)
D-->>C: return (DiagnosticPayload)
C->>D: receive_diagnostic(timeout)
D->>DC: receive_diagnostic(timeout)
DC->>ECU: await/read socket
ECU-->>DC: diagnostic frame
DC-->>D: raw bytes
D-->>C: DiagnosticPayload (hex)
sequenceDiagram
participant C as Client
participant U as UDS Driver
participant I as UdsInterface
participant T as Transport (CAN/DoIP)
participant ECU as ECU
C->>U: change_session(session)
U->>I: change_session(UdsSessionType)
I->>T: send DiagnosticSessionControl
T->>ECU: CAN frame / DoIP message
ECU-->>T: positive response
T-->>I: response bytes
I-->>U: UdsResponse(success=True,...)
U-->>C: return UdsResponse
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
📝 Coding Plan
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment Tip You can validate your CodeRabbit configuration file in your editor.If your editor has YAML language server, you can enable auto-completion and validation by adding |
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (8)
python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/conftest.py (1)
143-152: Consider catchingsocket.timeoutexplicitly in the accept loop.On Python 3.11+,
socket.timeoutis an alias forTimeoutErrorwhich is a subclass ofOSError, so the currentexcept OSErrorclause will catch it. However, the intent would be clearer with an explicit mention, and there's a subtle difference: ifsettimeoutbehavior changes or debugging is needed, distinguishing between timeout and other OS errors can be helpful.Optional: explicit timeout handling
def _accept_loop(self): while self._running: try: conn, _ = self._server.accept() conn.settimeout(1.0) self._clients.append(conn) handler = threading.Thread(target=self._handle_client, args=(conn,), daemon=True) handler.start() - except OSError: + except (OSError, socket.timeout): pass🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/conftest.py` around lines 143 - 152, The accept loop in _accept_loop currently swallows all OSError exceptions; update it to explicitly handle socket.timeout from self._server.accept() (or from conn.settimeout) by adding an except socket.timeout: pass branch, and keep a separate except OSError as e: log or re-raise unexpected OS errors; refer to _accept_loop, self._server.accept, conn.settimeout, self._clients, and _handle_client to locate where to add the explicit socket.timeout handler so timeouts are distinguished from other OS errors.python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.py (1)
151-152: Consider using a named constant for Alive Check Request.Line 151 uses magic number
0x0007while other payload types use named constants. For consistency:+ALIVE_CHECK_REQUEST = 0x0007 + # In _dispatch method: - if payload_type == 0x0007: # Alive Check Request + if payload_type == ALIVE_CHECK_REQUEST:🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.py` around lines 151 - 152, Replace the magic literal 0x0007 with a descriptive constant (e.g., PAYLOAD_TYPE_ALIVE_CHECK or PAYLOAD_ALIVE_CHECK) and use that constant in the conditional where payload_type is compared; update the comparison from "if payload_type == 0x0007" to "if payload_type == PAYLOAD_TYPE_ALIVE_CHECK" and ensure the constant is defined near other payload type constants or enums used in the module so _handle_alive_check() remains called the same way.python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py (1)
34-48: Consider adding error handling for initialization failures.If
UdsoncanClientcreation oropen()fails afterDoIPClientis created, the DoIP connection won't be cleaned up.Proposed fix
def __post_init__(self): if hasattr(super(), "__post_init__"): super().__post_init__() self._doip_client = DoIPClient( self.ecu_ip, self.ecu_logical_address, tcp_port=self.tcp_port, protocol_version=self.protocol_version, client_logical_address=self.client_logical_address, auto_reconnect_tcp=self.auto_reconnect_tcp, ) - conn = DoIPClientUDSConnector(self._doip_client) - self._uds_client = UdsoncanClient(conn, request_timeout=self.request_timeout) - self._uds_client.open() + try: + conn = DoIPClientUDSConnector(self._doip_client) + self._uds_client = UdsoncanClient(conn, request_timeout=self.request_timeout) + self._uds_client.open() + except Exception: + self._doip_client.close() + raise🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py` around lines 34 - 48, In __post_init__, wrap the UdsoncanClient creation and _uds_client.open() calls in a try/except (or try/finally) so that if UdsoncanClient(...) or _uds_client.open() raises, you explicitly clean up the previously created DoIPClient stored in self._doip_client (call its proper shutdown/close method) before re-raising the exception; ensure you reference DoIPClient, DoIPClientUDSConnector, UdsoncanClient, self._doip_client and self._uds_client so the cleanup runs only when the DoIP client exists and avoids leaving an open connection.python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py (1)
39-56: Consider adding error handling for initialization failures.If initialization fails partway through
__post_init__(e.g., stack creation fails after bus is opened), previously created resources won't be cleaned up, potentially leaking the CAN bus connection.Proposed fix with try/except cleanup
def __post_init__(self): if hasattr(super(), "__post_init__"): super().__post_init__() - self._bus = can.Bus(channel=self.channel, interface=self.interface) - self._notifier = can.Notifier(self._bus, []) - - address = isotp.Address(rxid=self.rxid, txid=self.txid) - self._stack = isotp.NotifierBasedCanStack( - self._bus, - self._notifier, - address=address, - params=self.isotp_params.model_dump() if self.isotp_params else None, - ) - - self._uds_conn = PythonIsoTpConnection(self._stack) - self._uds_client = UdsoncanClient(self._uds_conn, request_timeout=self.request_timeout) - self._uds_client.open() + self._bus = can.Bus(channel=self.channel, interface=self.interface) + try: + self._notifier = can.Notifier(self._bus, []) + address = isotp.Address(rxid=self.rxid, txid=self.txid) + self._stack = isotp.NotifierBasedCanStack( + self._bus, + self._notifier, + address=address, + params=self.isotp_params.model_dump() if self.isotp_params else None, + ) + self._uds_conn = PythonIsoTpConnection(self._stack) + self._uds_client = UdsoncanClient(self._uds_conn, request_timeout=self.request_timeout) + self._uds_client.open() + except Exception: + self._bus.shutdown() + raise🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py` around lines 39 - 56, The __post_init__ method can leak resources if an exception occurs mid-initialization; wrap the initialization sequence that creates self._bus, self._notifier, self._stack, self._uds_conn, and self._uds_client in a try/except (or try/finally) and on error ensure any already-created resources are cleanly closed or shutdown (call the CAN bus close, Notifier.stop/close, stack cleanup, and close the UdsoncanClient/PythonIsoTpConnection as appropriate); use the unique symbols self._bus, self._notifier, self._stack, self._uds_conn, self._uds_client, PythonIsoTpConnection, and UdsoncanClient to locate code and ensure exceptions are re-raised after cleanup so callers still see the failure.python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.py (2)
119-130: Consider error handling for invalid hex input insend_diagnostic.If
payload.datacontains an invalid hex string,bytes.fromhex()will raise aValueError. While Pydantic validation may catch some malformed inputs, an odd-length or non-hex string could still reach this code. You may want to either:
- Add a
patternvalidator onDiagnosticPayload.datato ensure valid hex format- Catch
ValueErrorhere and raise a more descriptive errorThis is minor since the validation layer likely handles most cases.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.py` around lines 119 - 130, send_diagnostic can raise a ValueError when bytes.fromhex(payload.data) receives invalid hex; either add a hex pattern validator to DiagnosticPayload.data so malformed strings are rejected earlier, or wrap the conversion in send_diagnostic (the bytes.fromhex call) with a try/except that catches ValueError and raises a clearer error (including the offending payload.data and context) before calling self._doip_client.send_diagnostic; update the send_diagnostic function signature and error handling accordingly so callers get a descriptive failure instead of an unhandled ValueError.
99-105: Potential issue with.hex()on non-bytes types.The
isinstancechecks forvincorrectly handle the bytes/string case with.decode(). However, foreidandgid, if they're already strings (not bytes), calling.hex()would fail since strings don't have a.hex()method. The else branch just returnsresp.eiddirectly, which is correct, but the conditional structure is inconsistent.Consider aligning the pattern with
vin:♻️ Suggested fix
- eid=resp.eid.hex() if isinstance(resp.eid, bytes) else resp.eid, - gid=resp.gid.hex() if isinstance(resp.gid, bytes) else resp.gid, + eid=resp.eid.hex() if isinstance(resp.eid, (bytes, bytearray)) else resp.eid, + gid=resp.gid.hex() if isinstance(resp.gid, (bytes, bytearray)) else resp.gid,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.py` around lines 99 - 105, The eid/gid hex handling can fail for non-bytes types; update the conditional logic in the response mapping (where vin, logical_address, eid, gid, further_action, sync_status are set from resp) to mirror vin's safe handling by only calling .hex() when the value is a bytes-like object (e.g., isinstance(resp.eid, (bytes, bytearray)) and isinstance(resp.gid, (bytes, bytearray))) and otherwise return resp.eid/resp.gid as-is; also consider accepting bytearray for vin similarly (use the same isinstance check) so .decode()/.hex() are only invoked on appropriate types.python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py (2)
85-97: Silent failure onNegativeResponseExceptionloses diagnostic context.When a negative response is received, the method returns an empty list, making it impossible for callers to distinguish between "no DIDs available" and "request was rejected by the ECU." Consider returning the NRC information similarly to other methods, or at minimum logging the exception.
If this is intentional behavior for simplified client usage, a brief comment explaining the design choice would help future maintainers.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py` around lines 85 - 97, The method read_data_by_identifier silently swallows NegativeResponseException and returns an empty list, losing NRC/diagnostic context; update read_data_by_identifier to capture the NegativeResponseException from self._uds_client.read_data_by_identifier and either (a) include the NRC/negative response info in the returned DidValue(s) (mirroring other methods) or (b) log the exception with details before returning, so callers can distinguish a rejected request from legitimately empty results; reference NegativeResponseException, read_data_by_identifier, _uds_client.read_data_by_identifier and DidValue when making the change.
145-161: Same silent failure pattern forread_dtc_by_status_mask.Similar to
read_data_by_identifier, this method returns an empty list onNegativeResponseException, losing diagnostic context. Consider consistency with methods likerequest_seedthat preserve NRC information.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py` around lines 145 - 161, The method read_dtc_by_status_mask currently swallows NegativeResponseException and returns an empty list; change the except block to capture the exception (except NegativeResponseException as e) and return a DtcInfo-preserving representation that includes the NRC/negative response information instead of dropping it. In practice, catch the exception from _uds_client.get_dtc_by_status_mask, extract the NRC/response code from the NegativeResponseException (e.g., e.nrc or other exception attribute available on NegativeResponseException), and return a list containing a DtcInfo instance populated to convey that negative response (e.g., dtc_id or status/severity set to None and a field or encoding that preserves the NRC) so callers receive the diagnostic code; mirror the pattern used in request_seed for how NRC is propagated.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@python/packages/jumpstarter-driver-doip/examples/exporter.yaml`:
- Around line 3-4: The YAML is invalid because the `export` key is not nested
under `endpoint` for the ExporterConfig; update the example so that `export` is
a child of `endpoint` (i.e., move `export` under `endpoint`) so the `endpoint`
mapping contains the `export` mapping as expected by ExporterConfig.
In
`@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver_test.py`:
- Around line 229-236: The test function name is misleading:
test_doip_invalid_request_timeout_type actually validates the type of the
auto_reconnect_tcp argument. Rename the test to reflect the checked parameter
(e.g., test_doip_invalid_auto_reconnect_tcp_type) or change the instantiation to
test request_timeout if you intended to validate that parameter; update the test
function name and/or the DoIP(...) call accordingly and keep references to DoIP
and auto_reconnect_tcp in the updated test.
In `@python/packages/jumpstarter-driver-uds-can/examples/exporter.yaml`:
- Around line 3-4: The YAML has an incorrect structure: the export key must be
nested under endpoint to form a valid ExporterConfig; move the export mapping
under endpoint (indent export and its children beneath endpoint) so endpoint
contains the export configuration (ensure the nested keys required by
ExporterConfig are placed under export).
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver_test.py`:
- Around line 177-183: The test is passing bytes to send_key but the method
signature is send_key(self, level: int, key_hex: str); update the test to pass a
hex string instead of raw bytes (e.g., "AABBCCDD" or "aabbccdd") when calling
client.send_key(1, ...), so the argument matches send_key's key_hex parameter
and the assertions remain valid for SecurityAccess.
- Around line 445-451: test_uds_can_virtual_send_key is passing a bytes object
to client.send_key but send_key expects a hex string; update the test to pass
the key as a hex string (e.g. use b"\xAA\xBB\xCC\xDD".hex() or a literal like
"AABBCCDD") when calling client.send_key so the argument type matches the
send_key signature (refer to client.send_key / send_key in UdsCan).
- Around line 131-137: The test is passing raw bytes to write_data_by_identifier
but the method signature write_data_by_identifier(self, did: int, value_hex:
str) and `@validate_call` expect a hex string; update the test to pass a proper
hex string (e.g., use b"ABC123456789".hex() or the explicit hex literal) when
calling client.write_data_by_identifier(0xF190, ...), so the input matches the
method's value_hex parameter and real validation behavior in
UdsCan.write_data_by_identifier and the served client.
- Around line 294-301: The test is passing raw bytes to write_data_by_identifier
but the API expects a hex string; update the call in driver_test.py to pass a
hex string (e.g., use b"BADDATA".hex() or the literal "42414444415441") to
client.write_data_by_identifier so it matches the expected input type for
UdsCan.write_data_by_identifier and still triggers
uds_mock.write_data_by_identifier side_effect (NegativeResponseException) for
the asserted NRC checks.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py`:
- Around line 58-62: The close() cleanup must stop the CAN Notifier and
explicitly tear down the ISO‑TP stack before shutting down the bus; update
close() to first call self._notifier.stop() (or the actual notifier attribute)
before self._bus.shutdown(), then perform the ISO‑TP cleanup on the UDS/ISO‑TP
objects (e.g., a dedicated shutdown/close on self._uds_client or
self._isotp_stack/_isotp_socket) inside a safe try/except to avoid exceptions
during teardown, then proceed with self._uds_client.close() and super().close().
In `@python/packages/jumpstarter-driver-uds-doip/examples/exporter.yaml`:
- Around line 3-4: The YAML has an invalid structure for ExporterConfig: the
top-level key "export" must be nested inside "endpoint" (mirroring the UDS-CAN
exporter fix). Modify the YAML so that the "export" mapping is a child of the
"endpoint" mapping (i.e., under the same "endpoint" key) to conform to the
ExporterConfig schema, ensuring keys expected by ExporterConfig are placed under
endpoint.export.
---
Nitpick comments:
In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.py`:
- Around line 151-152: Replace the magic literal 0x0007 with a descriptive
constant (e.g., PAYLOAD_TYPE_ALIVE_CHECK or PAYLOAD_ALIVE_CHECK) and use that
constant in the conditional where payload_type is compared; update the
comparison from "if payload_type == 0x0007" to "if payload_type ==
PAYLOAD_TYPE_ALIVE_CHECK" and ensure the constant is defined near other payload
type constants or enums used in the module so _handle_alive_check() remains
called the same way.
In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.py`:
- Around line 119-130: send_diagnostic can raise a ValueError when
bytes.fromhex(payload.data) receives invalid hex; either add a hex pattern
validator to DiagnosticPayload.data so malformed strings are rejected earlier,
or wrap the conversion in send_diagnostic (the bytes.fromhex call) with a
try/except that catches ValueError and raises a clearer error (including the
offending payload.data and context) before calling
self._doip_client.send_diagnostic; update the send_diagnostic function signature
and error handling accordingly so callers get a descriptive failure instead of
an unhandled ValueError.
- Around line 99-105: The eid/gid hex handling can fail for non-bytes types;
update the conditional logic in the response mapping (where vin,
logical_address, eid, gid, further_action, sync_status are set from resp) to
mirror vin's safe handling by only calling .hex() when the value is a bytes-like
object (e.g., isinstance(resp.eid, (bytes, bytearray)) and isinstance(resp.gid,
(bytes, bytearray))) and otherwise return resp.eid/resp.gid as-is; also consider
accepting bytearray for vin similarly (use the same isinstance check) so
.decode()/.hex() are only invoked on appropriate types.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py`:
- Around line 39-56: The __post_init__ method can leak resources if an exception
occurs mid-initialization; wrap the initialization sequence that creates
self._bus, self._notifier, self._stack, self._uds_conn, and self._uds_client in
a try/except (or try/finally) and on error ensure any already-created resources
are cleanly closed or shutdown (call the CAN bus close, Notifier.stop/close,
stack cleanup, and close the UdsoncanClient/PythonIsoTpConnection as
appropriate); use the unique symbols self._bus, self._notifier, self._stack,
self._uds_conn, self._uds_client, PythonIsoTpConnection, and UdsoncanClient to
locate code and ensure exceptions are re-raised after cleanup so callers still
see the failure.
In
`@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/conftest.py`:
- Around line 143-152: The accept loop in _accept_loop currently swallows all
OSError exceptions; update it to explicitly handle socket.timeout from
self._server.accept() (or from conn.settimeout) by adding an except
socket.timeout: pass branch, and keep a separate except OSError as e: log or
re-raise unexpected OS errors; refer to _accept_loop, self._server.accept,
conn.settimeout, self._clients, and _handle_client to locate where to add the
explicit socket.timeout handler so timeouts are distinguished from other OS
errors.
In
`@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py`:
- Around line 34-48: In __post_init__, wrap the UdsoncanClient creation and
_uds_client.open() calls in a try/except (or try/finally) so that if
UdsoncanClient(...) or _uds_client.open() raises, you explicitly clean up the
previously created DoIPClient stored in self._doip_client (call its proper
shutdown/close method) before re-raising the exception; ensure you reference
DoIPClient, DoIPClientUDSConnector, UdsoncanClient, self._doip_client and
self._uds_client so the cleanup runs only when the DoIP client exists and avoids
leaving an open connection.
In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py`:
- Around line 85-97: The method read_data_by_identifier silently swallows
NegativeResponseException and returns an empty list, losing NRC/diagnostic
context; update read_data_by_identifier to capture the NegativeResponseException
from self._uds_client.read_data_by_identifier and either (a) include the
NRC/negative response info in the returned DidValue(s) (mirroring other methods)
or (b) log the exception with details before returning, so callers can
distinguish a rejected request from legitimately empty results; reference
NegativeResponseException, read_data_by_identifier,
_uds_client.read_data_by_identifier and DidValue when making the change.
- Around line 145-161: The method read_dtc_by_status_mask currently swallows
NegativeResponseException and returns an empty list; change the except block to
capture the exception (except NegativeResponseException as e) and return a
DtcInfo-preserving representation that includes the NRC/negative response
information instead of dropping it. In practice, catch the exception from
_uds_client.get_dtc_by_status_mask, extract the NRC/response code from the
NegativeResponseException (e.g., e.nrc or other exception attribute available on
NegativeResponseException), and return a list containing a DtcInfo instance
populated to convey that negative response (e.g., dtc_id or status/severity set
to None and a field or encoding that preserves the NRC) so callers receive the
diagnostic code; mirror the pattern used in request_seed for how NRC is
propagated.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 25715e6b-6bf2-4e16-9281-f0b002bd5968
⛔ Files ignored due to path filters (1)
python/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (40)
python/docs/source/reference/package-apis/drivers/doip.mdpython/docs/source/reference/package-apis/drivers/uds-can.mdpython/docs/source/reference/package-apis/drivers/uds-doip.mdpython/docs/source/reference/package-apis/drivers/uds.mdpython/packages/jumpstarter-all/pyproject.tomlpython/packages/jumpstarter-driver-doip/.gitignorepython/packages/jumpstarter-driver-doip/README.mdpython/packages/jumpstarter-driver-doip/examples/exporter.yamlpython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/__init__.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/client.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/common.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver_test.pypython/packages/jumpstarter-driver-doip/pyproject.tomlpython/packages/jumpstarter-driver-uds-can/.gitignorepython/packages/jumpstarter-driver-uds-can/README.mdpython/packages/jumpstarter-driver-uds-can/examples/exporter.yamlpython/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/__init__.pypython/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/conftest.pypython/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.pypython/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver_test.pypython/packages/jumpstarter-driver-uds-can/pyproject.tomlpython/packages/jumpstarter-driver-uds-doip/.gitignorepython/packages/jumpstarter-driver-uds-doip/README.mdpython/packages/jumpstarter-driver-uds-doip/examples/exporter.yamlpython/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/__init__.pypython/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/conftest.pypython/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.pypython/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver_test.pypython/packages/jumpstarter-driver-uds-doip/pyproject.tomlpython/packages/jumpstarter-driver-uds/.gitignorepython/packages/jumpstarter-driver-uds/README.mdpython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/__init__.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/client.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/common.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver_test.pypython/packages/jumpstarter-driver-uds/pyproject.tomlpython/pyproject.toml
| endpoint: | ||
| export: |
There was a problem hiding this comment.
Invalid YAML structure for ExporterConfig.
Same issue as other exporter examples: export should be nested under endpoint.
Proposed fix
apiVersion: jumpstarter.dev/v1alpha1
kind: ExporterConfig
-endpoint:
-export:
+endpoint:
+ export:
doip:
type: jumpstarter_driver_doip.driver.DoIP📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| endpoint: | |
| export: | |
| apiVersion: jumpstarter.dev/v1alpha1 | |
| kind: ExporterConfig | |
| endpoint: | |
| export: | |
| doip: | |
| type: jumpstarter_driver_doip.driver.DoIP |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@python/packages/jumpstarter-driver-doip/examples/exporter.yaml` around lines
3 - 4, The YAML is invalid because the `export` key is not nested under
`endpoint` for the ExporterConfig; update the example so that `export` is a
child of `endpoint` (i.e., move `export` under `endpoint`) so the `endpoint`
mapping contains the `export` mapping as expected by ExporterConfig.
| endpoint: | ||
| export: |
There was a problem hiding this comment.
Invalid YAML structure for ExporterConfig.
The export key should be nested under endpoint. Currently, both endpoint: and export: appear at the same indentation level with endpoint having no value.
Proposed fix
apiVersion: jumpstarter.dev/v1alpha1
kind: ExporterConfig
-endpoint:
-export:
+endpoint:
+ export:
uds:
type: jumpstarter_driver_uds_can.driver.UdsCan📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| endpoint: | |
| export: | |
| apiVersion: jumpstarter.dev/v1alpha1 | |
| kind: ExporterConfig | |
| endpoint: | |
| export: | |
| uds: | |
| type: jumpstarter_driver_uds_can.driver.UdsCan | |
| channel: can0 | |
| interface: socketcan | |
| rxid: 0x641 | |
| txid: 0x642 | |
| request_timeout: 10 | |
| isotp_params: | |
| stmin: 0 | |
| blocksize: 0 | |
| tx_data_length: 8 |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@python/packages/jumpstarter-driver-uds-can/examples/exporter.yaml` around
lines 3 - 4, The YAML has an incorrect structure: the export key must be nested
under endpoint to form a valid ExporterConfig; move the export mapping under
endpoint (indent export and its children beneath endpoint) so endpoint contains
the export configuration (ensure the nested keys required by ExporterConfig are
placed under export).
| uds_mock.write_data_by_identifier.return_value = _make_mock_uds_response() | ||
|
|
||
| driver = UdsCan(channel="vcan0", rxid=0x641, txid=0x642) | ||
| with serve(driver) as client: | ||
| resp = client.write_data_by_identifier(0xF190, b"ABC123456789") | ||
| assert resp.success is True | ||
| assert resp.service == "WriteDataByIdentifier" |
There was a problem hiding this comment.
Test passes bytes but method signature expects str.
The write_data_by_identifier method signature is write_data_by_identifier(self, did: int, value_hex: str), expecting a hex string. However, this test passes b"ABC123456789" (bytes). With @validate_call, Pydantic will attempt to coerce bytes to str, but b"ABC123456789" would become the string representation "b'ABC123456789'", not a valid hex string.
This test may pass due to the mock bypassing actual validation, but it doesn't test realistic usage.
🧪 Suggested fix
- resp = client.write_data_by_identifier(0xF190, b"ABC123456789")
+ resp = client.write_data_by_identifier(0xF190, "414243313233343536373839") # hex for "ABC123456789"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| uds_mock.write_data_by_identifier.return_value = _make_mock_uds_response() | |
| driver = UdsCan(channel="vcan0", rxid=0x641, txid=0x642) | |
| with serve(driver) as client: | |
| resp = client.write_data_by_identifier(0xF190, b"ABC123456789") | |
| assert resp.success is True | |
| assert resp.service == "WriteDataByIdentifier" | |
| uds_mock.write_data_by_identifier.return_value = _make_mock_uds_response() | |
| driver = UdsCan(channel="vcan0", rxid=0x641, txid=0x642) | |
| with serve(driver) as client: | |
| resp = client.write_data_by_identifier(0xF190, "414243313233343536373839") # hex for "ABC123456789" | |
| assert resp.success is True | |
| assert resp.service == "WriteDataByIdentifier" |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver_test.py`
around lines 131 - 137, The test is passing raw bytes to
write_data_by_identifier but the method signature write_data_by_identifier(self,
did: int, value_hex: str) and `@validate_call` expect a hex string; update the
test to pass a proper hex string (e.g., use b"ABC123456789".hex() or the
explicit hex literal) when calling client.write_data_by_identifier(0xF190, ...),
so the input matches the method's value_hex parameter and real validation
behavior in UdsCan.write_data_by_identifier and the served client.
| uds_mock.send_key.return_value = _make_mock_uds_response() | ||
|
|
||
| driver = UdsCan(channel="vcan0", rxid=0x641, txid=0x642) | ||
| with serve(driver) as client: | ||
| resp = client.send_key(1, b"\xAA\xBB\xCC\xDD") | ||
| assert resp.success is True | ||
| assert resp.service == "SecurityAccess" |
There was a problem hiding this comment.
Test passes bytes but method signature expects str.
Similar to write_data_by_identifier, the send_key method signature is send_key(self, level: int, key_hex: str), but this test passes b"\xAA\xBB\xCC\xDD" (bytes).
🧪 Suggested fix
- resp = client.send_key(1, b"\xAA\xBB\xCC\xDD")
+ resp = client.send_key(1, "aabbccdd")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| uds_mock.send_key.return_value = _make_mock_uds_response() | |
| driver = UdsCan(channel="vcan0", rxid=0x641, txid=0x642) | |
| with serve(driver) as client: | |
| resp = client.send_key(1, b"\xAA\xBB\xCC\xDD") | |
| assert resp.success is True | |
| assert resp.service == "SecurityAccess" | |
| uds_mock.send_key.return_value = _make_mock_uds_response() | |
| driver = UdsCan(channel="vcan0", rxid=0x641, txid=0x642) | |
| with serve(driver) as client: | |
| resp = client.send_key(1, "aabbccdd") | |
| assert resp.success is True | |
| assert resp.service == "SecurityAccess" |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver_test.py`
around lines 177 - 183, The test is passing bytes to send_key but the method
signature is send_key(self, level: int, key_hex: str); update the test to pass a
hex string instead of raw bytes (e.g., "AABBCCDD" or "aabbccdd") when calling
client.send_key(1, ...), so the argument matches send_key's key_hex parameter
and the assertions remain valid for SecurityAccess.
| uds_mock.write_data_by_identifier.side_effect = NegativeResponseException(nrc_response) | ||
|
|
||
| driver = UdsCan(channel="vcan0", rxid=0x641, txid=0x642) | ||
| with serve(driver) as client: | ||
| resp = client.write_data_by_identifier(0xF190, b"BADDATA") | ||
| assert resp.success is False | ||
| assert resp.nrc == 0x72 | ||
| assert resp.nrc_name == "generalProgrammingFailure" |
There was a problem hiding this comment.
Test passes bytes instead of hex string.
Same pattern as above - write_data_by_identifier expects a hex string.
🧪 Suggested fix
- resp = client.write_data_by_identifier(0xF190, b"BADDATA")
+ resp = client.write_data_by_identifier(0xF190, "424144444154") # hex for "BADDAT" (truncated to valid hex length)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver_test.py`
around lines 294 - 301, The test is passing raw bytes to
write_data_by_identifier but the API expects a hex string; update the call in
driver_test.py to pass a hex string (e.g., use b"BADDATA".hex() or the literal
"42414444415441") to client.write_data_by_identifier so it matches the expected
input type for UdsCan.write_data_by_identifier and still triggers
uds_mock.write_data_by_identifier side_effect (NegativeResponseException) for
the asserted NRC checks.
| def test_uds_can_virtual_send_key(mock_uds_ecu): | ||
| channel, rxid, txid = mock_uds_ecu | ||
| driver = UdsCan(channel=channel, interface="virtual", rxid=rxid, txid=txid, request_timeout=5.0) | ||
| with serve(driver) as client: | ||
| resp = client.send_key(1, b"\xAA\xBB\xCC\xDD") | ||
| assert resp.success is True | ||
| assert resp.service == "SecurityAccess" |
There was a problem hiding this comment.
Integration test passes bytes instead of hex string.
The integration test at line 449 passes b"\xAA\xBB\xCC\xDD" but send_key expects a hex string.
🧪 Suggested fix
- resp = client.send_key(1, b"\xAA\xBB\xCC\xDD")
+ resp = client.send_key(1, "aabbccdd")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def test_uds_can_virtual_send_key(mock_uds_ecu): | |
| channel, rxid, txid = mock_uds_ecu | |
| driver = UdsCan(channel=channel, interface="virtual", rxid=rxid, txid=txid, request_timeout=5.0) | |
| with serve(driver) as client: | |
| resp = client.send_key(1, b"\xAA\xBB\xCC\xDD") | |
| assert resp.success is True | |
| assert resp.service == "SecurityAccess" | |
| def test_uds_can_virtual_send_key(mock_uds_ecu): | |
| channel, rxid, txid = mock_uds_ecu | |
| driver = UdsCan(channel=channel, interface="virtual", rxid=rxid, txid=txid, request_timeout=5.0) | |
| with serve(driver) as client: | |
| resp = client.send_key(1, "aabbccdd") | |
| assert resp.success is True | |
| assert resp.service == "SecurityAccess" |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver_test.py`
around lines 445 - 451, test_uds_can_virtual_send_key is passing a bytes object
to client.send_key but send_key expects a hex string; update the test to pass
the key as a hex string (e.g. use b"\xAA\xBB\xCC\xDD".hex() or a literal like
"AABBCCDD") when calling client.send_key so the argument type matches the
send_key signature (refer to client.send_key / send_key in UdsCan).
| endpoint: | ||
| export: |
There was a problem hiding this comment.
Invalid YAML structure for ExporterConfig.
Same issue as the UDS-CAN exporter: export should be nested under endpoint.
Proposed fix
apiVersion: jumpstarter.dev/v1alpha1
kind: ExporterConfig
-endpoint:
-export:
+endpoint:
+ export:
uds:
type: jumpstarter_driver_uds_doip.driver.UdsDoip📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| endpoint: | |
| export: | |
| apiVersion: jumpstarter.dev/v1alpha1 | |
| kind: ExporterConfig | |
| endpoint: | |
| export: | |
| uds: | |
| type: jumpstarter_driver_uds_doip.driver.UdsDoip |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@python/packages/jumpstarter-driver-uds-doip/examples/exporter.yaml` around
lines 3 - 4, The YAML has an invalid structure for ExporterConfig: the top-level
key "export" must be nested inside "endpoint" (mirroring the UDS-CAN exporter
fix). Modify the YAML so that the "export" mapping is a child of the "endpoint"
mapping (i.e., under the same "endpoint" key) to conform to the ExporterConfig
schema, ensuring keys expected by ExporterConfig are placed under
endpoint.export.
- Align exporter.yaml examples with project convention (endpoint/token fields) - Rename misleading test (test_doip_invalid_request_timeout_type -> test_doip_invalid_auto_reconnect_tcp_type) - Fix UDS-CAN close() to stop ISO-TP stack and CAN notifier before bus shutdown, preventing thread leak warnings - Add ALIVE_CHECK_REQUEST constant to replace 0x0007 magic literal in DoIP conftest - Add hex string validation on DiagnosticPayload.data via Pydantic field_validator - Extend eid/gid/vin isinstance checks to include bytearray in DoIP vehicle identification response - Log NRC details on NegativeResponseException in read_data_by_identifier and read_dtc_by_status_mask instead of silently swallowing - Add try/except guards in UDS-DoIP close() for teardown robustness
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.py (1)
77-79: Use the defined constant instead of magic number.Line 79 uses the literal
0x0008while the constantALIVE_CHECK_RESPONSEis defined on line 15 with the same value.Proposed fix
def _handle_alive_check() -> bytes: resp = struct.pack("!H", ECU_LOGICAL_ADDRESS) - return _pack_doip(0x0008, resp) + return _pack_doip(ALIVE_CHECK_RESPONSE, resp)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.py` around lines 77 - 79, Replace the magic literal 0x0008 in _handle_alive_check with the existing constant ALIVE_CHECK_RESPONSE: in function _handle_alive_check() use ALIVE_CHECK_RESPONSE when calling _pack_doip instead of the hard-coded 0x0008 so the code references the defined constant and avoids the magic number.python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/common.py (1)
58-63: Consider whether empty hex strings should be allowed.The regex
r"[0-9a-fA-F]*"allows empty strings due to the*quantifier. If empty diagnostic payloads are invalid, consider using+instead:- if not re.fullmatch(r"[0-9a-fA-F]*", v): + if not re.fullmatch(r"[0-9a-fA-F]+", v):If empty payloads are intentionally valid (e.g., for certain protocol operations), then the current implementation is correct.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/common.py` around lines 58 - 63, The current validator _validate_hex (field_validator for "data") uses the regex r"[0-9a-fA-F]*" which permits empty strings; if empty diagnostic payloads should be invalid, change the regex to r"[0-9a-fA-F]+" so at least one hex digit is required and retain the same ValueError message; if empty payloads are intentionally allowed, leave the implementation as-is but add a comment above _validate_hex explaining that empty strings are permitted by design.python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.py (1)
44-56: Consider overridingclose()to clean up the DoIPClient.The
DoIPClientis initialized in__post_init__but there's noclose()override to ensure it's properly cleaned up when the driver is destroyed. This could lead to leaked connections.Proposed fix
def __post_init__(self): if hasattr(super(), "__post_init__"): super().__post_init__() self._doip_client = DoIPClient( self.ecu_ip, self.ecu_logical_address, tcp_port=self.tcp_port, protocol_version=self.protocol_version, client_logical_address=self.client_logical_address, auto_reconnect_tcp=self.auto_reconnect_tcp, activation_type=self.activation_type, ) + def close(self): + """Close the driver and underlying DoIP connection.""" + try: + self._doip_client.close() + except Exception: + pass + super().close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.py` around lines 44 - 56, Add an override of close(self) in the same class that created _doip_client (initialized in __post_init__) to properly clean up the DoIPClient: check if self._doip_client exists, call its shutdown/close/disconnect method (use the DoIPClient's public close API), swallow/log exceptions as needed, set self._doip_client to None, and then call super().close() (or the parent close if present) to ensure proper teardown.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py`:
- Around line 91-102: The current read_data_by_identifier method swallows
NegativeResponseException and returns an empty list, hiding NRCs; change it to
preserve the failure by removing the except block that returns [] and instead
re-raise the NegativeResponseException so callers can distinguish ECU NRCs from
“no data” (apply the same change to the other method that currently collapses
NRCs to [] at lines 152-167). Keep the existing logger warning if desired, but
do not convert NRCs into a successful empty list—raise the exception
(NegativeResponseException) from read_data_by_identifier (and the analogous
method) so callers receive the original error and can handle retries/conditional
logic.
- Around line 106-112: Both write_data_by_identifier and send_key call
bytes.fromhex() without validation so malformed hex raises an uncaught
ValueError; wrap the bytes.fromhex(...) call in each method (in
write_data_by_identifier for value_hex and in send_key for key_hex) with a
try/except ValueError that re-raises a new ValueError with a descriptive message
like "Invalid hex for value_hex: '<value>'" or "Invalid hex for key_hex:
'<key>'" (keep original exception chained). Keep the existing
NegativeResponseException handling and return paths
(_build_uds_response/_build_nrc_response) unchanged, and add unit tests that
pass malformed hex strings to write_data_by_identifier and send_key to assert
the new ValueError is raised with the descriptive message.
---
Nitpick comments:
In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/common.py`:
- Around line 58-63: The current validator _validate_hex (field_validator for
"data") uses the regex r"[0-9a-fA-F]*" which permits empty strings; if empty
diagnostic payloads should be invalid, change the regex to r"[0-9a-fA-F]+" so at
least one hex digit is required and retain the same ValueError message; if empty
payloads are intentionally allowed, leave the implementation as-is but add a
comment above _validate_hex explaining that empty strings are permitted by
design.
In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.py`:
- Around line 77-79: Replace the magic literal 0x0008 in _handle_alive_check
with the existing constant ALIVE_CHECK_RESPONSE: in function
_handle_alive_check() use ALIVE_CHECK_RESPONSE when calling _pack_doip instead
of the hard-coded 0x0008 so the code references the defined constant and avoids
the magic number.
In `@python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.py`:
- Around line 44-56: Add an override of close(self) in the same class that
created _doip_client (initialized in __post_init__) to properly clean up the
DoIPClient: check if self._doip_client exists, call its
shutdown/close/disconnect method (use the DoIPClient's public close API),
swallow/log exceptions as needed, set self._doip_client to None, and then call
super().close() (or the parent close if present) to ensure proper teardown.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 011adb37-4b77-42d6-b218-af2c01e46c0c
📒 Files selected for processing (10)
python/packages/jumpstarter-driver-doip/examples/exporter.yamlpython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/common.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/conftest.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver.pypython/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver_test.pypython/packages/jumpstarter-driver-uds-can/examples/exporter.yamlpython/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.pypython/packages/jumpstarter-driver-uds-doip/examples/exporter.yamlpython/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py
🚧 Files skipped from review as they are similar to previous changes (4)
- python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py
- python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py
- python/packages/jumpstarter-driver-uds-can/examples/exporter.yaml
- python/packages/jumpstarter-driver-doip/examples/exporter.yaml
| try: | ||
| self._uds_client.close() | ||
| except Exception: | ||
| pass |
There was a problem hiding this comment.
I would at least log as a warning, unless we know it happens by routine/always.
There was a problem hiding this comment.
This is still showing the
except Exception:
pass
block, I suspect it was not commited in git.
| try: | ||
| self._stack.stop() | ||
| except Exception: | ||
| pass |
There was a problem hiding this comment.
I would at least log as a warning, unless we know it happens by routine/always.
There was a problem hiding this comment.
Ok, I added logging in all such similar cases.
| try: | ||
| self._notifier.stop() | ||
| except Exception: | ||
| pass |
There was a problem hiding this comment.
I would at least log as a warning, unless we know it happens by routine/always.
(same for any general Exception catch, which in general should be avoided, but for a closing safeguard I think it's fine but I'd still log it)
There was a problem hiding this comment.
same.
What I meant is adding a warning log, printing exception details instead of just pass.
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (3)
python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/common.py (1)
62-80: Consider documenting the string encoding behavior.The
encode()method encodes strings as UTF-8, which is a valid design choice. However, for automotive diagnostics where DID payloads are often binary, users might expect hex strings (e.g.,"AABBCC") to be decoded to bytes (b'\xaa\xbb\xcc').Consider either:
- Adding a note in the docstring clarifying that binary data should be passed as
bytes, not hex strings, or- Accepting hex strings via
bytes.fromhex()for convenience.The current implementation is correct—just a documentation clarity suggestion.
📝 Documentation enhancement option
class RawDidCodec(DidCodec): """Pass-through codec that treats DID values as raw bytes. Used as the default codec so that any DID can be read/written without requiring per-DID configuration in the udsoncan client. + + For encoding: + - bytes/bytearray: passed through unchanged + - str: UTF-8 encoded (for binary data, pass bytes directly) """🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/common.py` around lines 62 - 80, The RawDidCodec.encode behavior currently treats str inputs as UTF-8 bytes; update the RawDidCodec class docstring to explicitly state that strings are encoded as UTF-8 and that binary DID payloads should be provided as bytes/bytearray (or, if you prefer the convenience option, add support in RawDidCodec.encode to detect hex strings and convert them via bytes.fromhex()); reference the RawDidCodec.encode method when clarifying the expected input types and behavior so callers know to pass raw bytes for binary DIDs or opt-in to hex-string handling if you implement the bytes.fromhex() branch.python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py (1)
52-62: Consider logging suppressed exceptions during close.The bare
except Exception: passblocks silently swallow errors during cleanup. While this is acceptable for cleanup code (ensuring both clients get a chance to close), logging at debug level would help diagnose connection issues in production.♻️ Optional: Add debug logging for suppressed exceptions
+import logging + +logger = logging.getLogger(__name__) + ... def close(self): """Close the UDS and DoIP connections.""" try: self._uds_client.close() - except Exception: - pass + except Exception as e: + logger.debug("Error closing UDS client: %s", e) try: self._doip_client.close() - except Exception: - pass + except Exception as e: + logger.debug("Error closing DoIP client: %s", e) super().close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py` around lines 52 - 62, The close method currently suppresses exceptions from self._uds_client.close() and self._doip_client.close() with bare except blocks; update close to catch Exception as e for each client and log the caught exception at debug level (e.g., using a module/class logger) before continuing so both clients still get closed, then call super().close(); reference the close method and the attributes self._uds_client, self._doip_client and the final super().close() when making the change.python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py (1)
294-302: Client handler threads are not joined on stop.The
stop()method joins only the accept thread but not the individual client handler threads (spawned at line 146). Since they are daemon threads, this won't block shutdown, but active handlers may continue briefly afterstop()returns. For a test mock this is acceptable, but consider tracking and joining handler threads if deterministic cleanup becomes important.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py` around lines 294 - 302, The stop() method only joins the accept thread (_thread) and closes sockets but does not join the per-client handler threads created when connections are accepted (the client handler threads spawned alongside clients added to _clients); update the implementation to track those handler Thread objects (e.g., store them in a list or map when creating them in the accept/handler spawn site) and in stop() iterate and join each handler thread with a short timeout after closing the sockets, handling exceptions as needed, before returning to ensure deterministic cleanup of client handlers.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In
`@python/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.py`:
- Around line 52-54: The test passes a bytes object to ecu_client.send_key but
UdsInterface.send_key expects a hex string; convert the derived key to hex
before calling send_key. In the test code replace the bytes argument returned by
derive_key(bytes.fromhex(seed_resp.seed)) with its hex representation (e.g.,
call .hex() on the bytes) and pass that string into ecu_client.send_key(level=1,
key_hex) so send_key receives a hex string; keep references to derive_key,
seed_resp.seed and ecu_client.send_key to locate and update the call.
- Around line 57-60: The test currently passes a bytes object new_vin =
b"NEWVIN12345678901" to write_data_by_identifier, but the API expects a hex
string; change new_vin to a hex string (e.g.,
"4e455656494e3132333435363738393031" or use new_vin.hex()) and call
ecu_client.write_data_by_identifier(0xF190, new_vin_hex) so
write_data_by_identifier receives the hex string; ensure you update the variable
name (e.g., new_vin_hex) and keep the assert resp.success is True.
- Around line 94-98: The test calls write_data_by_identifier(0xF190,
b"SHOULDFAIL0000000") but the API expects a hex string payload, not raw bytes;
update the call in test_diagnostic_flow.py to pass a hex-encoded string (e.g.
payload.hex() or a literal hex string) for the second argument to
write_data_by_identifier so the function receives the expected hex string
format.
- Around line 86-89: The test is calling ecu_client.send_key with a bytes object
(wrong_key = b"\x00\x00\x00\x00") but the API expects a hex string; change
wrong_key to a hex string (e.g., "00000000" or "0x00000000" per the send_key
contract) and keep the same call to send_key(1, wrong_key) and assertions
(resp.success is False, resp.nrc == 0x35) so the test exercises the same failure
path; update any nearby tests using send_key to the same string format if
present.
- Around line 23-26: The test is passing bytes to write_data_by_identifier, but
the API expects a hex-encoded string; update the three call sites to pass hex
strings instead of bytes by calling .hex() on the byte literals/variables:
replace b"NEWVIN12345678901" and b"SHOULDFAIL0000000" with their hex string
equivalents and change the variable new_vin (used with write_data_by_identifier)
to new_vin.hex() when calling write_data_by_identifier so the function receives
a value_hex: str as required.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py`:
- Around line 74-75: Wrap the call to self._bus.shutdown() in a try/except
inside the close() implementation so any exception from self._bus.shutdown() is
caught and logged (or suppressed) and does not prevent calling super().close();
specifically modify the close method where self._bus.shutdown() is invoked to
catch exceptions from _bus.shutdown(), handle/log the error (using the module's
logger or an existing error path), and then always call super().close() to
ensure child driver resources are cleaned up.
In
`@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py`:
- Around line 35-50: In __post_init__, if UdsoncanClient.open() raises an
exception the already-created _doip_client (instance of DoIPClient) may leak
resources; wrap the UdsoncanClient creation/open in a try/except (or
try/finally) and on failure explicitly clean up the DoIPClient (call its
shutdown/close/disconnect method, e.g., _doip_client.close() or the appropriate
DoIPClient cleanup API) before re-raising the exception; ensure this change
touches the block where _doip_client is instantiated, the
DoIPClientUDSConnector(...) usage, and the UdsoncanClient(...).open() call so
the client is always cleaned up on init failure.
---
Nitpick comments:
In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py`:
- Around line 294-302: The stop() method only joins the accept thread (_thread)
and closes sockets but does not join the per-client handler threads created when
connections are accepted (the client handler threads spawned alongside clients
added to _clients); update the implementation to track those handler Thread
objects (e.g., store them in a list or map when creating them in the
accept/handler spawn site) and in stop() iterate and join each handler thread
with a short timeout after closing the sockets, handling exceptions as needed,
before returning to ensure deterministic cleanup of client handlers.
In
`@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py`:
- Around line 52-62: The close method currently suppresses exceptions from
self._uds_client.close() and self._doip_client.close() with bare except blocks;
update close to catch Exception as e for each client and log the caught
exception at debug level (e.g., using a module/class logger) before continuing
so both clients still get closed, then call super().close(); reference the close
method and the attributes self._uds_client, self._doip_client and the final
super().close() when making the change.
In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/common.py`:
- Around line 62-80: The RawDidCodec.encode behavior currently treats str inputs
as UTF-8 bytes; update the RawDidCodec class docstring to explicitly state that
strings are encoded as UTF-8 and that binary DID payloads should be provided as
bytes/bytearray (or, if you prefer the convenience option, add support in
RawDidCodec.encode to detect hex strings and convert them via bytes.fromhex());
reference the RawDidCodec.encode method when clarifying the expected input types
and behavior so callers know to pass raw bytes for binary DIDs or opt-in to
hex-string handling if you implement the bytes.fromhex() branch.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: c740226a-50dd-4b11-9534-4802fc9939b3
📒 Files selected for processing (10)
python/examples/automotive/README.mdpython/examples/automotive/jumpstarter_example_automotive/conftest.pypython/examples/automotive/jumpstarter_example_automotive/hello.pypython/examples/automotive/jumpstarter_example_automotive/hello_test.pypython/examples/automotive/jumpstarter_example_automotive/mock_ecu.pypython/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.pypython/examples/automotive/pyproject.tomlpython/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.pypython/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/common.py
💤 Files with no reviewable changes (2)
- python/examples/automotive/jumpstarter_example_automotive/hello_test.py
- python/examples/automotive/jumpstarter_example_automotive/hello.py
Add a stateful MockDiagnosticEcu (DoIP+UDS) and representative diagnostic tests under examples/automotive/ demonstrating jumpstarter for automotive ECU diagnostics. The mock ECU implements session management, security access (seed/key), DID read/write with precondition checks, clearable DTC memory, and proper negative response codes. Also fix a bug in the UDS drivers where read/write DID operations failed because the udsoncan client was not configured with a default DID codec.
dc38c97 to
bd64522
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (2)
python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py (1)
60-75:⚠️ Potential issue | 🟡 MinorWrap
_bus.shutdown()in try/except and add logging for caught exceptions.Two concerns remain in the cleanup logic:
_bus.shutdown()at line 74 is not wrapped in try/except, unlike the other cleanup calls. If it raises,super().close()won't execute.The silent
passin each except block swallows errors without logging. Per previous feedback, these should at least log a warning so issues during teardown are visible.,
Proposed fix
+import logging + +logger = logging.getLogger(__name__) + + `@dataclass`(kw_only=True, config=ConfigDict(arbitrary_types_allowed=True)) class UdsCan(UdsInterface, Driver): ... def close(self): """Close the UDS connection and CAN bus.""" try: self._uds_client.close() except Exception: - pass + logger.warning("Error closing UDS client", exc_info=True) try: self._stack.stop() except Exception: - pass + logger.warning("Error stopping ISO-TP stack", exc_info=True) try: self._notifier.stop() except Exception: - pass - self._bus.shutdown() + logger.warning("Error stopping CAN notifier", exc_info=True) + try: + self._bus.shutdown() + except Exception: + logger.warning("Error shutting down CAN bus", exc_info=True) super().close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py` around lines 60 - 75, In close(), wrap the call to self._bus.shutdown() in a try/except like the other teardown calls and, for every except block (for self._uds_client.close, self._stack.stop, self._notifier.stop and the new bus shutdown try), log the caught exception instead of silently passing; use an existing logger (e.g. self._logger.warning) or the logging module to emit a warning that includes context and the exception details, then proceed to call super().close() regardless of teardown errors.python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py (1)
35-50:⚠️ Potential issue | 🟡 MinorPotential resource leak if
open()fails.If
_uds_client.open()raises an exception at line 50, the already-instantiated_doip_client(which may have opened a socket) won't be cleaned up.,
Proposed fix
def __post_init__(self): if hasattr(super(), "__post_init__"): super().__post_init__() self._doip_client = DoIPClient( self.ecu_ip, self.ecu_logical_address, tcp_port=self.tcp_port, protocol_version=self.protocol_version, client_logical_address=self.client_logical_address, auto_reconnect_tcp=self.auto_reconnect_tcp, ) - conn = DoIPClientUDSConnector(self._doip_client) - config = make_uds_client_config(request_timeout=self.request_timeout) - self._uds_client = UdsoncanClient(conn, config=config) - self._uds_client.open() + try: + conn = DoIPClientUDSConnector(self._doip_client) + config = make_uds_client_config(request_timeout=self.request_timeout) + self._uds_client = UdsoncanClient(conn, config=config) + self._uds_client.open() + except Exception: + self._doip_client.close() + raise🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py` around lines 35 - 50, Wrap the UdsoncanClient creation/open sequence in a try/except so that if UdsoncanClient.open() raises, you explicitly clean up the already-created DoIPClient; e.g., in __post_init__ after constructing self._doip_client and before/around self._uds_client.open(), catch exceptions, call the DoIPClient teardown method (e.g. self._doip_client.close() or the appropriate disconnect/shutdown method on DoIPClient) to release sockets/resources, and then re-raise the exception so initialization fails cleanly.
🧹 Nitpick comments (2)
python/examples/automotive/pyproject.toml (1)
13-17: Keeppytestout of runtime dependencies.Line 14 puts the test runner in the package's install metadata, so anyone installing this example also gets
pytesteven if they only need the runtime pieces. Unless the example importspytestoutside tests, this should move to a test extra or the repo's test/dev dependency group.♻️ Proposed change
dependencies = [ - "pytest>=8.3.2", "jumpstarter", "jumpstarter-driver-uds-doip", ] + +[project.optional-dependencies] +test = [ + "pytest>=8.3.2", +]#!/bin/bash set -euo pipefail echo "Checking for pytest imports outside tests in python/examples/automotive..." rg -n --type=py '\b(import|from)\s+pytest\b' python/examples/automotive \ -g '!**/*test*.py' \ -g '!**/conftest.py' || true echo echo "All pytest imports in python/examples/automotive..." rg -n --type=py '\b(import|from)\s+pytest\b' python/examples/automotive || true echo echo "Current dependency block:" sed -n '1,80p' python/examples/automotive/pyproject.tomlExpected result: no
pytestimports outside test files. If that holds,pytestshould live in a test-only dependency bucket instead of[project.dependencies].🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/examples/automotive/pyproject.toml` around lines 13 - 17, The pyproject.toml currently lists "pytest>=8.3.2" in the runtime [project.dependencies]; remove that entry from the dependencies array and instead add pytest to a test-only dependency group (e.g., [project.optional-dependencies."test"] or your tool's dev/test section such as [tool.poetry.dev-dependencies]) so it is not installed for runtime consumers; ensure you run the provided grep checks to confirm there are no pytest imports outside test files before moving it.python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py (1)
52-62: Add logging for caught exceptions during cleanup.For consistency with the feedback on the UdsCan driver, consider logging warnings when exceptions occur during cleanup rather than silently passing. This helps diagnose teardown issues.
Proposed fix
+import logging + +logger = logging.getLogger(__name__) + + `@dataclass`(kw_only=True, config=ConfigDict(arbitrary_types_allowed=True)) class UdsDoip(UdsInterface, Driver): ... def close(self): """Close the UDS and DoIP connections.""" try: self._uds_client.close() except Exception: - pass + logger.warning("Error closing UDS client", exc_info=True) try: self._doip_client.close() except Exception: - pass + logger.warning("Error closing DoIP client", exc_info=True) super().close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py` around lines 52 - 62, The close method currently swallows exceptions from self._uds_client.close() and self._doip_client.close() which hides teardown failures; update close to catch exceptions but log a warning including the exception details and context (e.g., "failed closing UDS client" / "failed closing DoIP client") instead of simply passing, and ensure super().close() is still called; use the module/class logger (e.g., self._logger or logging.getLogger(__name__)) to emit the warnings so diagnostic information is available.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py`:
- Around line 218-225: In _uds_session_ctrl validate the requested sub-function
byte before mutating state or returning a positive response: check that sub is
one of the allowed session IDs (e.g., SESSION_DEFAULT and other supported
constants used in this module) and only if valid set self.state.session, reset
security_unlocked/seed, and return the positive response (sid + 0x40 with
session payload); if invalid return a UDS negative response (format 0x7F,
original SID, NRC 0x12 for "sub-function not supported") and do not change
state. Ensure you reference _uds_session_ctrl, self.state.session, and
SESSION_DEFAULT (and any other session constants) when implementing this check.
- Around line 259-276: In _uds_security_access detect and reject key submissions
when there is no outstanding seed: check self.state.security_seed is non-empty
before computing derive_key(self.state.security_seed) and comparing to the
provided_key, and return an NRC (e.g., NRC_INVALID_KEY or
NRC_CONDITIONS_NOT_CORRECT) if the seed is empty; this prevents an empty
send_key from unlocking the ECU—ensure you reference and update
self.state.security_unlocked only after a valid non-empty seed/key pair is
verified and clear the seed afterward as currently done.
- Around line 71-85: The code currently treats socket.timeout as a fatal OSError
and tears down healthy DoIP sessions; update the error handling so
socket.timeout is handled separately and non-fatally: in places that call
_read_doip_message (and where you set per-client timeouts) catch socket.timeout
explicitly (except socket.timeout:) and treat it as an idle read (e.g., return
None or continue the loop) while preserving a separate except OSError: branch
for real socket errors; adjust callers of _read_doip_message to expect None for
timeouts vs fatal errors so sessions are not closed on transient idle gaps.
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py`:
- Around line 40-58: The initialization can leak resources if
UdsoncanClient.open() raises: wrap the creation/opening of self._uds_conn,
self._uds_client and the call to self._uds_client.open() in a try/except (or
try/finally) and on exception explicitly clean up any partially-initialized
objects (call the appropriate close/shutdown/stop on self._uds_client (e.g.,
close()), self._stack (e.g., shutdown/stop), self._notifier (e.g., stop()), and
self._bus (e.g., shutdown/close) if they were created), then re-raise the
exception; update __post_init__ to perform this guarded initialization around
the symbols _bus, _notifier, _stack, _uds_conn, and _uds_client.
---
Duplicate comments:
In
`@python/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.py`:
- Around line 60-75: In close(), wrap the call to self._bus.shutdown() in a
try/except like the other teardown calls and, for every except block (for
self._uds_client.close, self._stack.stop, self._notifier.stop and the new bus
shutdown try), log the caught exception instead of silently passing; use an
existing logger (e.g. self._logger.warning) or the logging module to emit a
warning that includes context and the exception details, then proceed to call
super().close() regardless of teardown errors.
In
`@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py`:
- Around line 35-50: Wrap the UdsoncanClient creation/open sequence in a
try/except so that if UdsoncanClient.open() raises, you explicitly clean up the
already-created DoIPClient; e.g., in __post_init__ after constructing
self._doip_client and before/around self._uds_client.open(), catch exceptions,
call the DoIPClient teardown method (e.g. self._doip_client.close() or the
appropriate disconnect/shutdown method on DoIPClient) to release
sockets/resources, and then re-raise the exception so initialization fails
cleanly.
---
Nitpick comments:
In `@python/examples/automotive/pyproject.toml`:
- Around line 13-17: The pyproject.toml currently lists "pytest>=8.3.2" in the
runtime [project.dependencies]; remove that entry from the dependencies array
and instead add pytest to a test-only dependency group (e.g.,
[project.optional-dependencies."test"] or your tool's dev/test section such as
[tool.poetry.dev-dependencies]) so it is not installed for runtime consumers;
ensure you run the provided grep checks to confirm there are no pytest imports
outside test files before moving it.
In
`@python/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.py`:
- Around line 52-62: The close method currently swallows exceptions from
self._uds_client.close() and self._doip_client.close() which hides teardown
failures; update close to catch exceptions but log a warning including the
exception details and context (e.g., "failed closing UDS client" / "failed
closing DoIP client") instead of simply passing, and ensure super().close() is
still called; use the module/class logger (e.g., self._logger or
logging.getLogger(__name__)) to emit the warnings so diagnostic information is
available.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b945b7c7-52ab-4da9-9509-1ad7fd6a5412
⛔ Files ignored due to path filters (1)
python/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (10)
python/examples/automotive/README.mdpython/examples/automotive/jumpstarter_example_automotive/conftest.pypython/examples/automotive/jumpstarter_example_automotive/hello.pypython/examples/automotive/jumpstarter_example_automotive/hello_test.pypython/examples/automotive/jumpstarter_example_automotive/mock_ecu.pypython/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.pypython/examples/automotive/pyproject.tomlpython/packages/jumpstarter-driver-uds-can/jumpstarter_driver_uds_can/driver.pypython/packages/jumpstarter-driver-uds-doip/jumpstarter_driver_uds_doip/driver.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/common.py
💤 Files with no reviewable changes (2)
- python/examples/automotive/jumpstarter_example_automotive/hello_test.py
- python/examples/automotive/jumpstarter_example_automotive/hello.py
🚧 Files skipped from review as they are similar to previous changes (2)
- python/examples/automotive/jumpstarter_example_automotive/conftest.py
- python/examples/automotive/README.md
|
@vtz This looks awesome! I was hoping someone would implement UDS for Jumpstarter! |
…r services Expose three new UDS services end-to-end through the driver/client pipeline, with matching stateful mock ECU handlers and integration tests: - RoutineControl (0x31): start, stop, and query routine results - Authentication (0x29): challenge-based PKI auth per ISO-14229-1:2020 - RequestFileTransfer (0x38): add, read, delete files and read directories
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (4)
python/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.py (1)
30-31:⚠️ Potential issue | 🟠 MajorPass hex strings to
write_data_by_identifierandsend_key.
python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.pyconverts these payloads withbytes.fromhex(...), so the rawbytesarguments here will fail before the requests ever reach the mock ECU. That means these assertions are not exercising the intended NRC/success paths.🛠️ Suggested fix
- resp = ecu_client.write_data_by_identifier(0xF190, b"NEWVIN12345678901") + resp = ecu_client.write_data_by_identifier(0xF190, b"NEWVIN12345678901".hex()) @@ - resp = ecu_client.send_key(1, key) + resp = ecu_client.send_key(1, key.hex()) @@ - resp = ecu_client.write_data_by_identifier(0xF190, new_vin) + resp = ecu_client.write_data_by_identifier(0xF190, new_vin.hex()) @@ - resp = ecu_client.send_key(1, wrong_key) + resp = ecu_client.send_key(1, wrong_key.hex()) @@ - resp = ecu_client.write_data_by_identifier(0xF190, b"SHOULDFAIL0000000") + resp = ecu_client.write_data_by_identifier(0xF190, b"SHOULDFAIL0000000".hex())Also applies to: 59-60, 65-67, 93-95, 103-104
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.py` around lines 30 - 31, The tests are passing raw bytes to ecu_client.write_data_by_identifier and ecu_client.send_key but the driver expects hex string payloads (it calls bytes.fromhex(...)); update all test calls (e.g., ecu_client.write_data_by_identifier and ecu_client.send_key invocations at the noted locations) to supply hex-encoded strings for the payload argument instead of raw b"...", ensuring the strings represent the same byte sequence previously used so the driver's bytes.fromhex conversion succeeds and the assertions exercise the intended NRC/success flows.python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py (3)
301-318:⚠️ Potential issue | 🟠 MajorRequire an outstanding seed before accepting
send_key.With
self.state.security_seed == b"",derive_key()also returnsb"", so an emptysend_keycan unlock the ECU in a non-default session. That weakens the mock and can false-pass the security flow.🛠️ Suggested fix
def _uds_security_access(self, data: bytes) -> bytes: sub = data[1] if len(data) > 1 else 0x01 with self.state._lock: if self.state.session == SESSION_DEFAULT: return _nrc(data[0], NRC_CONDITIONS_NOT_CORRECT) if sub % 2 == 1: seed = os.urandom(4) self.state.security_seed = seed return bytes([data[0] + 0x40, sub]) + seed else: + if not self.state.security_seed: + return _nrc(data[0], NRC_CONDITIONS_NOT_CORRECT) expected_key = derive_key(self.state.security_seed) provided_key = data[2:] if provided_key != expected_key: return _nrc(data[0], NRC_INVALID_KEY)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py` around lines 301 - 318, In _uds_security_access, require that a seed is present before accepting the "send_key" (even-sub) branch: inside the else branch where provided_key is checked, first verify that self.state.security_seed is non-empty (e.g., if not self.state.security_seed: return _nrc(data[0], NRC_CONDITIONS_NOT_CORRECT) or an appropriate NRC) so derive_key is not called on an empty seed and an empty provided_key cannot unlock the ECU; keep the rest of the logic (derive_key(self.state.security_seed), compare provided_key, set security_unlocked and clear security_seed) unchanged.
97-111:⚠️ Potential issue | 🟠 MajorDon't tear down healthy DoIP sessions on
socket.timeout.Each client socket is put on a 1s timeout, and
socket.timeoutsubclassesOSError, so the currentexcept OSError:path drops otherwise healthy sessions after short idle gaps. That will make this example flaky under slower CI or while debugging.🛠️ Suggested fix
def _read_doip_message(conn: socket.socket) -> tuple[int, bytes] | None: header = b"" while len(header) < 8: - chunk = conn.recv(8 - len(header)) + try: + chunk = conn.recv(8 - len(header)) + except socket.timeout: + continue if not chunk: return None header += chunk _ver, _inv, payload_type, payload_len = struct.unpack("!BBHL", header) payload = b"" while len(payload) < payload_len: - chunk = conn.recv(payload_len - len(payload)) + try: + chunk = conn.recv(payload_len - len(payload)) + except socket.timeout: + continue if not chunk: return None payload += chunk return payload_type, payloadAlso applies to: 183-204
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py` around lines 97 - 111, The code currently treats socket.timeout as an OSError and tears down otherwise healthy DoIP sessions; in _read_doip_message (and the similar reader at lines 183-204) change socket.recv error handling to catch socket.timeout separately and treat it as a non-fatal, retryable condition (e.g., continue the read loop) while still handling other OSErrors by closing the session; ensure you only close/return on empty recv (peer closed) or non-timeout OSError, and keep the payload assembly logic unchanged.
260-267:⚠️ Potential issue | 🟡 MinorReject unsupported
DiagnosticSessionControlsub-functions before mutating state.This currently stores any byte in
self.state.sessionand always replies positively. Invalid session IDs should return an NRC instead of changing state, otherwise the example flow can mask bad session mappings.🛠️ Suggested fix
def _uds_session_ctrl(self, data: bytes) -> bytes: - sub = data[1] if len(data) > 1 else SESSION_DEFAULT + if len(data) < 2: + return _nrc(data[0], 0x13) + sub = data[1] + if sub not in {SESSION_DEFAULT, SESSION_PROGRAMMING, SESSION_EXTENDED}: + return _nrc(data[0], 0x12) with self.state._lock: self.state.session = sub self.state.security_unlocked = False self.state.security_seed = b""🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py` around lines 260 - 267, The _uds_session_ctrl handler currently writes any sub-function byte into self.state.session and always returns a positive response; change it to first validate that the sub value is one of the supported session IDs (e.g., SESSION_DEFAULT and any other defined session constants such as SESSION_PROGRAMMING/SESSION_EXTENDED) and only mutate self.state.session and related security fields after validation; if the sub-function is unsupported, return a proper UDS negative response (0x7F, original SID, NRC 0x12 for "sub-function not supported") without changing state; ensure you reference and use the existing symbols _uds_session_ctrl, self.state.session, positive_sid and SESSION_DEFAULT when implementing the check.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py`:
- Around line 402-419: The proof-of-ownership verification allows bypass when
there is no outstanding challenge because derive_key(self.state.auth_challenge)
yields b""; update the authentication branch handling task == 0x06 in
mock_ecu.py to first verify that self.state.auth_challenge is present/non-empty
(e.g., raise/return an NRC like NRC_CONDITIONS_NOT_CORRECT or another
appropriate error) before calling derive_key and comparing proof, and only then
proceed to mark self.state.authenticated, clear self.state.auth_challenge, and
return the positive response including session_key.
In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py`:
- Around line 179-195: The bytes.fromhex(data_hex) conversion can raise
ValueError for malformed non-empty hex strings; wrap the conversion in a small
try/except ValueError inside start_routine (and mirror the same change in
stop_routine and get_routine_result) so a malformed data_hex returns a
RoutineControlResponse with routine_id=routine_id, control_type set to the
correct string ("startRoutine"/"stopRoutine"/"getRoutineResult"), success=False,
status_record=None (and include/log the invalid hex error message), instead of
letting the exception propagate; reference the bytes.fromhex call and the
RoutineControlResponse construction to locate where to add the try/except and
the error response.
- Around line 239-283: The authentication method currently calls
bytes.fromhex(...) directly on certificate_client_hex, challenge_client_hex,
algorithm_indicator_hex, and proof_of_ownership_client_hex which raises
unhandled ValueError for malformed hex; fix by validating/converting these
inputs before calling self._uds_client.authentication (e.g., implement a small
helper or wrap each conversion in a try/except ValueError), and on ValueError
return an AuthenticationResponse with authentication_task=authentication_task,
return_value=0, success=False and include the error text in nrc_name (or another
available error field) instead of letting the exception propagate; update the
conversion sites for certificate_client_hex, challenge_client_hex,
algorithm_indicator_hex, and proof_of_ownership_client_hex and ensure the
_uds_client.authentication call only receives bytes or None.
---
Duplicate comments:
In `@python/examples/automotive/jumpstarter_example_automotive/mock_ecu.py`:
- Around line 301-318: In _uds_security_access, require that a seed is present
before accepting the "send_key" (even-sub) branch: inside the else branch where
provided_key is checked, first verify that self.state.security_seed is non-empty
(e.g., if not self.state.security_seed: return _nrc(data[0],
NRC_CONDITIONS_NOT_CORRECT) or an appropriate NRC) so derive_key is not called
on an empty seed and an empty provided_key cannot unlock the ECU; keep the rest
of the logic (derive_key(self.state.security_seed), compare provided_key, set
security_unlocked and clear security_seed) unchanged.
- Around line 97-111: The code currently treats socket.timeout as an OSError and
tears down otherwise healthy DoIP sessions; in _read_doip_message (and the
similar reader at lines 183-204) change socket.recv error handling to catch
socket.timeout separately and treat it as a non-fatal, retryable condition
(e.g., continue the read loop) while still handling other OSErrors by closing
the session; ensure you only close/return on empty recv (peer closed) or
non-timeout OSError, and keep the payload assembly logic unchanged.
- Around line 260-267: The _uds_session_ctrl handler currently writes any
sub-function byte into self.state.session and always returns a positive
response; change it to first validate that the sub value is one of the supported
session IDs (e.g., SESSION_DEFAULT and any other defined session constants such
as SESSION_PROGRAMMING/SESSION_EXTENDED) and only mutate self.state.session and
related security fields after validation; if the sub-function is unsupported,
return a proper UDS negative response (0x7F, original SID, NRC 0x12 for
"sub-function not supported") without changing state; ensure you reference and
use the existing symbols _uds_session_ctrl, self.state.session, positive_sid and
SESSION_DEFAULT when implementing the check.
In
`@python/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.py`:
- Around line 30-31: The tests are passing raw bytes to
ecu_client.write_data_by_identifier and ecu_client.send_key but the driver
expects hex string payloads (it calls bytes.fromhex(...)); update all test calls
(e.g., ecu_client.write_data_by_identifier and ecu_client.send_key invocations
at the noted locations) to supply hex-encoded strings for the payload argument
instead of raw b"...", ensuring the strings represent the same byte sequence
previously used so the driver's bytes.fromhex conversion succeeds and the
assertions exercise the intended NRC/success flows.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 87bbaa0b-fcb2-4f99-b7b1-e37da2169fa5
📒 Files selected for processing (5)
python/examples/automotive/jumpstarter_example_automotive/mock_ecu.pypython/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/client.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/common.pypython/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py
| def start_routine(self, routine_id: int, data_hex: str = "") -> RoutineControlResponse: | ||
| """Start a routine on the ECU.""" | ||
| data = bytes.fromhex(data_hex) if data_hex else None | ||
| try: | ||
| resp = self._uds_client.start_routine(routine_id, data) | ||
| return RoutineControlResponse( | ||
| routine_id=resp.service_data.routine_id_echo, | ||
| control_type="startRoutine", | ||
| success=resp.positive, | ||
| status_record=resp.service_data.routine_status_record.hex() | ||
| if resp.service_data.routine_status_record else None, | ||
| ) | ||
| except NegativeResponseException as e: | ||
| return RoutineControlResponse( | ||
| routine_id=routine_id, control_type="startRoutine", success=False, | ||
| nrc=e.response.code, nrc_name=e.response.code_name, | ||
| ) |
There was a problem hiding this comment.
Add input validation for data_hex parameter in RoutineControl methods.
The pattern bytes.fromhex(data_hex) if data_hex else None will raise an unhandled ValueError if data_hex is a non-empty but malformed hex string. This breaks the structured error handling pattern used elsewhere in this interface.
Suggested fix for start_routine (apply similarly to stop_routine and get_routine_result)
def start_routine(self, routine_id: int, data_hex: str = "") -> RoutineControlResponse:
"""Start a routine on the ECU."""
- data = bytes.fromhex(data_hex) if data_hex else None
+ try:
+ data = bytes.fromhex(data_hex) if data_hex else None
+ except ValueError as exc:
+ raise ValueError("data_hex must be a valid even-length hex string") from exc
try:
resp = self._uds_client.start_routine(routine_id, data)Also applies to: 199-215, 219-235
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py`
around lines 179 - 195, The bytes.fromhex(data_hex) conversion can raise
ValueError for malformed non-empty hex strings; wrap the conversion in a small
try/except ValueError inside start_routine (and mirror the same change in
stop_routine and get_routine_result) so a malformed data_hex returns a
RoutineControlResponse with routine_id=routine_id, control_type set to the
correct string ("startRoutine"/"stopRoutine"/"getRoutineResult"), success=False,
status_record=None (and include/log the invalid hex error message), instead of
letting the exception propagate; reference the bytes.fromhex call and the
RoutineControlResponse construction to locate where to add the try/except and
the error response.
There was a problem hiding this comment.
Same reasoning applies. The data_hex parameter comes from UdsClient.start_routine(data: bytes | None) which calls data.hex() if data else "" before sending over gRPC. The driver never receives user-supplied raw strings -- it always gets valid hex produced by the client layer. No fix needed.
There was a problem hiding this comment.
@vtz, understood — this is the same client/driver boundary pattern used throughout the UDS interface: the client serializes bytes → hex via .hex() before the gRPC call, and the driver always receives well-formed hex strings produced by that layer. The ValueError concern doesn't apply here. Thanks for the clarification!
✏️ Learnings added
Learnt from: vtz
Repo: jumpstarter-dev/jumpstarter PR: 295
File: python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py:179-195
Timestamp: 2026-03-09T20:12:25.677Z
Learning: In jumpstarter-dev/jumpstarter, `UdsClient.start_routine`, `stop_routine`, and `get_routine_result` (python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/client.py) accept `data: bytes | None` and convert to hex (`data.hex() if data else ""`) before sending over gRPC. The driver-side `UdsInterface` methods receive `data_hex: str` and convert back with `bytes.fromhex(data_hex)`. The driver never receives user-supplied raw strings — the client/driver boundary is where the hex conversion happens, so `data_hex` is always valid hex at the driver layer.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
🧠 Learnings used
Learnt from: CR
Repo: jumpstarter-dev/jumpstarter PR: 0
File: .cursor/rules/creating-new-drivers.mdc:0-0
Timestamp: 2026-02-04T18:12:55.555Z
Learning: Applies to python/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py : Drivers with child drivers should be composite drivers, with the client interface inheriting from `CompositeClient` in jumpstarter_driver_composite.client
Learnt from: vtz
Repo: jumpstarter-dev/jumpstarter PR: 295
File: python/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.py:30-33
Timestamp: 2026-03-09T19:51:26.688Z
Learning: In jumpstarter-dev/jumpstarter, `UdsClient.write_data_by_identifier` (python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/client.py) accepts `bytes` as the value parameter and converts it to a hex string (`value.hex()`) before sending over gRPC. The driver-side `UdsInterface.write_data_by_identifier` (python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py) receives `value_hex: str` and converts back with `bytes.fromhex(value_hex)`. Test call sites correctly pass bytes to the client — the client/driver boundary is where the hex conversion happens.
Learnt from: vtz
Repo: jumpstarter-dev/jumpstarter PR: 295
File: python/examples/automotive/jumpstarter_example_automotive/test_diagnostic_flow.py:59-61
Timestamp: 2026-03-09T19:52:26.318Z
Learning: In jumpstarter-dev/jumpstarter, `UdsClient.send_key` (python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/client.py) accepts `key: bytes` as the key parameter and converts it to a hex string (`key.hex()`) before sending over gRPC. The driver-side `UdsInterface.send_key` (python/packages/jumpstarter-driver-uds/jumpstarter_driver_uds/driver.py) receives `key_hex: str` and converts back with `bytes.fromhex(key_hex)`. Test call sites correctly pass bytes to the client — the client/driver boundary is where the hex conversion happens. This mirrors the same pattern used by `UdsClient.write_data_by_identifier`.
Wrap __post_init__ resource creation in try/except so partially initialized resources are cleaned up if open() fails. Replace bare except/pass in close() with logger.warning to surface teardown errors.
Same fix as uds-can: wrap __post_init__ in try/except to clean up partially initialized resources, and log warnings on close() failures instead of silently swallowing exceptions.
Prevents empty seed/key pair from silently unlocking the ECU.
…ding Same pattern as the security access fix — prevents empty challenge/proof from silently authenticating.
| @@ -0,0 +1,488 @@ | |||
| """Stateful mock ECU speaking DoIP + UDS for representative diagnostic testing. | |||
|
It looks great, Just a broken test, and you need to add your .md files for the drivers to this index: |
The driver now passes a full config dict via make_uds_client_config() instead of request_timeout as a keyword arg. Update the test assertion to verify the config dict contents rather than exact call signature.
Add DoIP, UDS, UDS-over-DoIP, and UDS-over-CAN driver pages to the driver documentation index under a new Automotive Diagnostics section.
Add required metadata block to DoIP, UDS-DoIP, and UDS-CAN exporter YAML examples. Add automotive diagnostics drivers to docs toctree.
mangelajo
left a comment
There was a problem hiding this comment.
just the wide exception capture without a warning print.
✅ Deploy Preview for jumpstarter-docs ready!
To edit notification comments on pull requests, go to your Netlify project configuration. |
|
@mangelajo I've just fixed the test. The other point regarding silent failure is logged here: 8358549 Or do you mean something else? |
Add four new driver packages for automotive diagnostic communication:
jumpstarter-driver-uds: Shared UDS (ISO-14229) base package
jumpstarter-driver-doip: Raw DoIP (ISO-13400) driver
jumpstarter-driver-uds-can: UDS over CAN/ISO-TP transport
jumpstarter-driver-uds-doip: UDS over DoIP transport
Workspace changes:
Bug fixes discovered by integration tests:
All tests pass: 84 tests across 4 packages (10 + 22 + 27 + 25), linting clean.
Summary by CodeRabbit