Conversation
|
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 a new NanoKVM driver package with video, HID, and composite drivers, client implementations with CLIs, tests, docs, example configs, package metadata (pyproject), and workspace registration. All changes are additive; no existing public APIs outside the new package were modified. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant Driver as NanoKVM (Composite)
participant Video as NanoKVMVideo
participant HID as NanoKVMHID
participant Client as NanoKVM API Client
participant Device as NanoKVM Device API
User->>Driver: snapshot() / paste_text() / mouse_click()
alt snapshot
Driver->>Video: delegate snapshot()
Video->>Client: _get_client() (create/reuse)
Client->>Device: authenticate / request MJPEG frame
Device-->>Client: MJPEG frame
Client-->>Video: JPEG bytes
Video-->>Driver: base64-encoded snapshot
Driver-->>User: image
else HID action
Driver->>HID: delegate HID method
HID->>Client: _get_client() (create/reuse)
Client->>Device: HID API call (paste/press/move/click/scroll)
Device-->>Client: ack
Client-->>HID: ack
HID-->>Driver: ack
Driver-->>User: success
end
alt 401 Unauthorized during call
Client-->>Driver: raises 401
Driver->>Driver: _reset_client()
Driver->>Client: new client / retry call
Device-->>Client: data
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 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 docstrings
🧪 Generate unit tests (beta)
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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`:
- Around line 460-466: The unmount behavior in _mount_impl is inconsistent: the
comment says the API expects an empty string for unmount, but the code uses
mount_file or None which sends None; change the logic so mount_file is set to an
empty string when file is falsy (e.g., mount_file = file if file else "") and
pass mount_file directly to client.mount_image (not mount_file or None); also
ensure the cdrom argument is computed from whether file is provided (e.g., cdrom
if file else False) so cdrom is False when unmounting.
- Around line 133-155: The retry path uses send_stream after its async context
has already exited, which is invalid; update stream_video to ensure send_stream
remains active for retries by moving the async with send_stream: to wrap the
entire streaming/retry logic (or re-enter a new async with send_stream: before
sending retry frames). Specifically, in stream_video, avoid using send_stream
after the except block exits its context — on _is_unauthorized_error(e) call
await self._reset_client() and await self._get_client() while still inside the
async with send_stream: context (or re-open it), then iterate
new_client.mjpeg_stream() to send frames; alternatively factor the per-client
frame-loop into a helper used inside the async with send_stream: so send_stream
is never used outside its context. Ensure references to client.mjpeg_stream(),
new_client.mjpeg_stream(), _is_unauthorized_error, _reset_client, and
_get_client are preserved.
In `@python/packages/jumpstarter-driver-nanokvm/pyproject.toml`:
- Around line 52-56: The dev dependency group in pyproject.toml is missing
pytest-asyncio, but tests rely on pytest's asyncio_mode = "auto"; add
pytest-asyncio (e.g., "pytest-asyncio>=0.20.3") to the [dependency-groups] dev
list so async tests run correctly and match the existing pytest asyncio_mode
configuration.
🧹 Nitpick comments (8)
python/packages/jumpstarter-driver-nanokvm/pyproject.toml (1)
16-16: Git dependency ondevbranch may cause instability.Using a dev branch (
@dev) for thenanokvmdependency can lead to unpredictable behavior as the branch may change unexpectedly. Consider pinning to a specific commit hash or creating a versioned release.- "nanokvm @ git+https://github.com/mangelajo/python-nanokvm.git@dev", + "nanokvm @ git+https://github.com/mangelajo/python-nanokvm.git@<commit-hash>",python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py (2)
193-196: Invalid escape sequences will raise unhandledValueError.The
unicode_escapecodec raisesValueErrorfor invalid escape sequences (e.g.,\xgg). Consider wrapping with error handling for better CLI UX.♻️ Optional: Add error handling
`@base.command`() `@click.argument`("text") def paste(text): """Paste text via keyboard HID (supports \\n for newline, \\t for tab)""" - # Decode escape sequences like \n, \t, etc. - decoded_text = text.encode().decode("unicode_escape") + try: + decoded_text = text.encode().decode("unicode_escape") + except ValueError as e: + raise click.BadParameter(f"Invalid escape sequence: {e}") self.paste_text(decoded_text) click.echo(f"Pasted: {repr(decoded_text)}")
526-535: Function namelistshadows Python built-in.The command function name
listshadows the built-inlisttype. While this works in Click's context, consider renaming for clarity.♻️ Suggested rename
`@image.command`() - def list(): + def list_images(): """List available image files""" images = self.get_images()python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py (2)
170-217: Code duplication: client management logic repeated fromNanoKVMVideo.
NanoKVMHIDduplicates the entire client lifecycle management (_client,_client_ctx,_reset_client,_get_client,close) fromNanoKVMVideo. Consider extracting this into a shared mixin class.♻️ Suggested refactor sketch
class NanoKVMClientMixin: """Mixin for NanoKVM API client lifecycle management""" host: str username: str = "admin" password: str = "admin" _client: NanoKVMAPIClient | None = field(init=False, repr=False, default=None) _client_ctx: object = field(init=False, repr=False, default=None) async def _reset_client(self): ... async def _get_client(self) -> NanoKVMAPIClient: ... def close(self): ... # Then: class NanoKVMVideo(NanoKVMClientMixin, Driver): ... class NanoKVMHID(NanoKVMClientMixin, Driver): ...
327-349:NanoKVMSerialis a non-functional placeholder.The class hardcodes
rfc2217://localhost:2217and requires manual SSH port forwarding setup. Combined with the warning at line 396, this feature is incomplete. Consider either removing it or marking it more explicitly.♻️ Suggested documentation improvement
`@dataclass`(kw_only=True) class NanoKVMSerial(PySerial): - """NanoKVM Serial console access via SSH tunnel""" + """NanoKVM Serial console access via SSH tunnel + + WARNING: This is a placeholder implementation. Users must manually + set up SSH port forwarding: ssh -L 2217:localhost:2217 root@<nanokvm> + """python/packages/jumpstarter-driver-nanokvm/README.md (1)
54-58: Consider adding a note about SSH password security.The documentation shows default SSH credentials (
root/root) in the example. While these are defaults for many embedded devices, consider adding a recommendation to change them or use key-based authentication in production environments.python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver_test.py (2)
67-74: Consider documenting why this fixture is needed.The
mock_aiohttp_sessionfixture is used by all tests but the session mock appears minimal. A brief comment explaining its purpose (e.g., preventing actual HTTP connections during driver initialization) would improve clarity.
208-227: Consider adding tests for missing functionality.The test suite covers core functionality well but is missing tests for:
mouse_move_relandmouse_scroll- Image management:
mount_image,download_image,get_mounted_image,get_cdrom_status,is_image_download_enabled,get_image_download_statusBased on learnings, driver implementations should add comprehensive tests.
Do you want me to help generate additional test cases for the missing functionality?
python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py
Outdated
Show resolved
Hide resolved
| @with_reauth | ||
| async def _mount_impl(driver): | ||
| client = await driver._get_client() | ||
| # Pass empty string or None for unmount - API expects empty string | ||
| mount_file = file if file else "" | ||
| # When unmounting, we need to pass the file as empty string or None | ||
| await client.mount_image(file=mount_file or None, cdrom=cdrom if mount_file else False) |
There was a problem hiding this comment.
Confusing unmount logic: comment contradicts code.
The comment says "API expects empty string" for unmount, but the code passes mount_file or None which evaluates to None when unmounting (since empty string is falsy). This discrepancy may cause unmount failures.
`@with_reauth`
async def _mount_impl(driver):
client = await driver._get_client()
- # Pass empty string or None for unmount - API expects empty string
- mount_file = file if file else ""
- # When unmounting, we need to pass the file as empty string or None
- await client.mount_image(file=mount_file or None, cdrom=cdrom if mount_file else False)
+ # Pass empty string for unmount, or the file path for mount
+ await client.mount_image(file=file if file else "", cdrom=cdrom if file else False)🤖 Prompt for AI Agents
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`
around lines 460 - 466, The unmount behavior in _mount_impl is inconsistent: the
comment says the API expects an empty string for unmount, but the code uses
mount_file or None which sends None; change the logic so mount_file is set to an
empty string when file is falsy (e.g., mount_file = file if file else "") and
pass mount_file directly to client.mount_image (not mount_file or None); also
ensure the cdrom argument is computed from whether file is provided (e.g., cdrom
if file else False) so cdrom is False when unmounting.
There was a problem hiding this comment.
Missing test coverage in driver_test.py
The following exported methods have no tests at all:
mouse_move_relmouse_scrollmouse_clickwith x/y coordinatesmouse_clickwithMouseButtonenum (only string"left"is tested)mount_image,download_image,get_mounted_image,get_cdrom_status,is_image_download_enabled,get_image_download_status(all exported viaNanoKVMcomposite but untested)with_reautherror recovery path (the decorator is core logic but never exercised in tests -- a test simulating a 401 followed by successful retry would verify this critical path)snapshotwhen the stream yields zero frames (should hitRuntimeError("No frames available"))press_keywith multi-char input (driver logs a warning forlen(key) > 2but this is untested)
Also: test_nanokvm_hid_press_key uses assert_called_with (checks last call) instead of assert_called_once_with, so multiple calls would silently pass.
No client_test.py - project convention gap
Other drivers in this project (power, can, flashers, ridesx) have dedicated client_test.py files. The NanoKVM client.py has non-trivial logic that warrants its own tests:
NanoKVMHIDClient.mouse_click()has branching logic forx/ybeingNonevs provided- CLI
pastecommand uses.encode().decode("unicode_escape")which can raise on invalid input - CLI
mouse clickcommand maps string button names toMouseButtonenums
| raise | ||
|
|
||
| # Start the video streaming task | ||
| task = asyncio.create_task(stream_video()) |
There was a problem hiding this comment.
This uses asyncio.create_task() directly, but the project convention is to use anyio for async concurrency (used throughout the codebase and listed as a dependency). This should use anyio.create_task_group() for portability and consistency with other drivers.
| def client(cls) -> str: | ||
| return "jumpstarter_driver_nanokvm.client.NanoKVMHIDClient" | ||
|
|
||
| async def _reset_client(self): |
There was a problem hiding this comment.
_reset_client(), _get_client(), and close() are duplicated verbatim between NanoKVMVideo (lines 65-93) and NanoKVMHID (lines 189-217). This is ~30 lines of copy-pasted code that could drift out of sync. Consider extracting to a shared base class.
| return "jumpstarter_driver_nanokvm.client.NanoKVMClient" | ||
|
|
||
| @export | ||
| async def get_info(self): |
There was a problem hiding this comment.
get_info() manually reimplements the 401 retry logic instead of using the with_reauth decorator. The same fragile pattern (defining a local function, passing video_driver as self) repeats across reboot(), mount_image(), download_image(), get_mounted_image(), get_cdrom_status(), is_image_download_enabled(), get_image_download_status(), and get_images(). This is inconsistent and fragile -- it works because with_reauth accesses self._get_client() and self._reset_client(), but a refactor could easily break it silently. If the duplicated client code (comment above) were extracted to a base class, these methods could use with_reauth directly.
| "jumpstarter", | ||
| "jumpstarter-driver-composite", | ||
| "jumpstarter-driver-pyserial", | ||
| "nanokvm @ git+https://github.com/mangelajo/python-nanokvm.git@dev", |
There was a problem hiding this comment.
This dependency points to a mutable git branch (@dev). The dev branch could change at any time, making builds non-reproducible and potentially breaking them without any change in this repo. Consider pinning to a specific commit hash (e.g., @abc1234) or using a published release.
There was a problem hiding this comment.
It's my own fork, but let me check I think the merged my PRs already
|
|
||
|
|
||
| @dataclass(kw_only=True) | ||
| class NanoKVMSerial(PySerial): |
There was a problem hiding this comment.
NanoKVMSerial is a placeholder -- it hardcodes url = "rfc2217://localhost:2217" and the NanoKVM composite just logs a warning when enable_serial=True. This is effectively dead code. Consider removing it from this PR and adding it when the implementation is ready, to avoid users trying to enable a non-functional feature.
|
|
||
| 1. **video**: Video streaming and snapshot capture | ||
| 2. **hid**: Keyboard and mouse HID control | ||
| 3. **serial**: Serial console access (optional, future feature) |
There was a problem hiding this comment.
The docs list serial console access as one of the three main interfaces and document config parameters for it (enable_serial, ssh_username, ssh_password, ssh_port), but the implementation is a non-functional placeholder. This will mislead users into thinking serial works. Either remove serial references entirely or add a prominent "not yet implemented" disclaimer.
There was a problem hiding this comment.
True , I removed that incomplete part of the implementation.
|
|
||
| ```{eval-rst} | ||
| .. autoclass:: jumpstarter_driver_nanokvm.client.NanoKVMClient() | ||
| :members: get_info, reboot, mount_image, download_image, get_mounted_image, get_cdrom_status, is_image_download_enabled, get_image_download_status |
There was a problem hiding this comment.
get_images is missing from the :members: list. It is exported, tested, and functional but will not appear in the generated API docs.
Suggested fix:
:members: get_info, reboot, mount_image, download_image, get_mounted_image, get_cdrom_status, is_image_download_enabled, get_image_download_status, get_images
ambient-code
bot
force-pushed
the
nanokvm-api-driver
branch
from
182c38c to
725b5f1
Compare
❌ Deploy Preview for jumpstarter-docs failed. Why did it fail? →
|
ambient-code
bot
force-pushed
the
nanokvm-api-driver
branch
from
725b5f1 to
ab7da1d
Compare
ambient-code
bot
force-pushed
the
nanokvm-api-driver
branch
from
ab7da1d to
adfaa2f
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py (1)
260-271: Unknown button strings silently default to LEFT.When an unrecognized button string is provided (e.g.,
"invalid"), it silently defaults toMouseButton.LEFT. Consider logging a warning or raising an error to help users identify typos.♻️ Proposed fix with warning
# Convert string to MouseButton enum for backward compatibility if isinstance(button, str): button_map = { "left": MouseButton.LEFT, "right": MouseButton.RIGHT, "middle": MouseButton.MIDDLE, } - button = button_map.get(button.lower(), MouseButton.LEFT) + button_lower = button.lower() + if button_lower not in button_map: + self.logger.warning(f"Unknown button '{button}', defaulting to LEFT") + button = button_map.get(button_lower, MouseButton.LEFT)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py` around lines 260 - 271, The string-to-enum conversion for MouseButton silently falls back to LEFT for unknown strings; update the conversion block in the driver (the section using button_map and MouseButton) to detect when button.lower() is not in button_map and log a warning via self.logger.warn/self.logger.warning (including the unknown value and that LEFT will be used) or alternatively raise a ValueError; ensure this check occurs before calling await client.mouse_click(...) and that the logged message references the original string and MouseButton.LEFT so users can spot typos.python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py (2)
525-534: Function namelistshadows Python built-in.The function name
listshadows the Python built-in. While Click commands use the function name as the command name, this can cause issues if the built-in is needed inside the function body. Consider renaming tolist_imageswith an explicitname="list".♻️ Proposed rename
`@image.command`() - def list(): + `@click.pass_context` + def list_images(ctx): """List available image files""" images = self.get_images()Or simply use the
nameparameter:- `@image.command`() - def list(): + `@image.command`(name="list") + def list_images(): """List available image files"""🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py` around lines 525 - 534, The command callback currently defines a function named list which shadows the Python built-in; rename the function to list_images (or another non-built-in name) and update the decorator to keep the CLI command name by using `@image.command`(name="list") so the CLI remains "list" while the function avoids shadowing; update any internal references to call list_images and ensure the docstring and help remain unchanged.
193-196: Consider handling invalid escape sequences gracefully.The
unicode_escapecodec will raise aValueErrorif the input contains invalid escape sequences (e.g.,\xGGor incomplete sequences like\u00). Consider wrapping this in a try-except or documenting the limitation.♻️ Proposed defensive handling
`@base.command`() `@click.argument`("text") def paste(text): """Paste text via keyboard HID (supports \\n for newline, \\t for tab)""" - # Decode escape sequences like \n, \t, etc. - decoded_text = text.encode().decode("unicode_escape") + # Decode escape sequences like \n, \t, etc. + try: + decoded_text = text.encode().decode("unicode_escape") + except ValueError: + # If escape decoding fails, use the text as-is + decoded_text = text self.paste_text(decoded_text) click.echo(f"Pasted: {repr(decoded_text)}")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py` around lines 193 - 196, The decode using "unicode_escape" can raise ValueError on bad escape sequences; wrap the decode call that produces decoded_text in a try/except (catch ValueError and UnicodeDecodeError), and on exception fall back to using the original text (or a replacement strategy) before calling self.paste_text(decoded_text); also emit a brief warning via click.echo so the user knows the escapes were not processed. Ensure you update the same local variable decoded_text and keep the subsequent self.paste_text(...) and click.echo(f"Pasted: {repr(decoded_text)}") calls.
🤖 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-nanokvm/jumpstarter_driver_nanokvm/driver.py`:
- Around line 91-98: The close method in NanoKVMDriverBase (close) performs its
own async client context teardown but does not call super().close(), so child
drivers won’t be closed; update the close method to call super().close() (either
before or after the anyio.from_thread.run(self._client_ctx.__aexit__(...))
cleanup) to ensure the base Driver.close() runs and iterates over self.children,
guaranteeing nested children are cleaned up.
---
Nitpick comments:
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py`:
- Around line 525-534: The command callback currently defines a function named
list which shadows the Python built-in; rename the function to list_images (or
another non-built-in name) and update the decorator to keep the CLI command name
by using `@image.command`(name="list") so the CLI remains "list" while the
function avoids shadowing; update any internal references to call list_images
and ensure the docstring and help remain unchanged.
- Around line 193-196: The decode using "unicode_escape" can raise ValueError on
bad escape sequences; wrap the decode call that produces decoded_text in a
try/except (catch ValueError and UnicodeDecodeError), and on exception fall back
to using the original text (or a replacement strategy) before calling
self.paste_text(decoded_text); also emit a brief warning via click.echo so the
user knows the escapes were not processed. Ensure you update the same local
variable decoded_text and keep the subsequent self.paste_text(...) and
click.echo(f"Pasted: {repr(decoded_text)}") calls.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`:
- Around line 260-271: The string-to-enum conversion for MouseButton silently
falls back to LEFT for unknown strings; update the conversion block in the
driver (the section using button_map and MouseButton) to detect when
button.lower() is not in button_map and log a warning via
self.logger.warn/self.logger.warning (including the unknown value and that LEFT
will be used) or alternatively raise a ValueError; ensure this check occurs
before calling await client.mouse_click(...) and that the logged message
references the original string and MouseButton.LEFT so users can spot typos.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: f625f95c-74ba-45f0-ad46-3efa8f8a4419
⛔ Files ignored due to path filters (1)
python/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (11)
python/docs/source/reference/package-apis/drivers/index.mdpython/docs/source/reference/package-apis/drivers/nanokvm.mdpython/packages/jumpstarter-driver-nanokvm/.gitignorepython/packages/jumpstarter-driver-nanokvm/README.mdpython/packages/jumpstarter-driver-nanokvm/examples/exporter.yamlpython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/__init__.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver_test.pypython/packages/jumpstarter-driver-nanokvm/pyproject.tomlpython/pyproject.toml
✅ Files skipped from review due to trivial changes (8)
- python/packages/jumpstarter-driver-nanokvm/.gitignore
- python/pyproject.toml
- python/docs/source/reference/package-apis/drivers/nanokvm.md
- python/packages/jumpstarter-driver-nanokvm/examples/exporter.yaml
- python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/init.py
- python/packages/jumpstarter-driver-nanokvm/pyproject.toml
- python/packages/jumpstarter-driver-nanokvm/README.md
- python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver_test.py
🚧 Files skipped from review as they are similar to previous changes (1)
- python/docs/source/reference/package-apis/drivers/index.md
| def close(self): | ||
| """Clean up resources""" | ||
| if self._client_ctx is not None: | ||
| try: | ||
| anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None)) | ||
| except Exception as e: | ||
| self.logger.debug(f"Error closing client: {e}") | ||
|
|
There was a problem hiding this comment.
Missing super().close() call may leave child drivers unclosed.
The base Driver.close() method iterates over self.children and calls close() on each child. By not calling super().close(), any nested children won't be properly cleaned up. While NanoKVMDriverBase itself doesn't have children, subclasses or future modifications might.
🛡️ Proposed fix
def close(self):
"""Clean up resources"""
if self._client_ctx is not None:
try:
anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None))
except Exception as e:
self.logger.debug(f"Error closing client: {e}")
+ super().close()📝 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 close(self): | |
| """Clean up resources""" | |
| if self._client_ctx is not None: | |
| try: | |
| anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None)) | |
| except Exception as e: | |
| self.logger.debug(f"Error closing client: {e}") | |
| def close(self): | |
| """Clean up resources""" | |
| if self._client_ctx is not None: | |
| try: | |
| anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None)) | |
| except Exception as e: | |
| self.logger.debug(f"Error closing client: {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-nanokvm/jumpstarter_driver_nanokvm/driver.py`
around lines 91 - 98, The close method in NanoKVMDriverBase (close) performs its
own async client context teardown but does not call super().close(), so child
drivers won’t be closed; update the close method to call super().close() (either
before or after the anyio.from_thread.run(self._client_ctx.__aexit__(...))
cleanup) to ensure the base Driver.close() runs and iterates over self.children,
guaranteeing nested children are cleaned up.
There was a problem hiding this comment.
♻️ Duplicate comments (1)
python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py (1)
91-98:⚠️ Potential issue | 🟡 MinorMissing
super().close()call may leave child drivers unclosed.The base
Driver.close()method iterates overself.childrenand closes each child. Without callingsuper().close(), any nested children won't be properly cleaned up.🛡️ Proposed fix
def close(self): """Clean up resources""" if self._client_ctx is not None: try: anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None)) except Exception as e: self.logger.debug(f"Error closing client: {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-nanokvm/jumpstarter_driver_nanokvm/driver.py` around lines 91 - 98, The close method in jumpstarter_driver_nanokvm.driver (method close) is missing a call to super().close(), so child drivers in self.children won't be closed; after performing the client context teardown (the anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None)) block and its exception handling), invoke super().close() to ensure base Driver.close() runs and closes nested children (or, if driver semantics require closing children first, call super().close() before the client teardown) — update the close method in the class to call super().close() in the appropriate order so child drivers are properly cleaned up.
🧹 Nitpick comments (3)
python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py (1)
144-162: Retry loop in stream doesn't handle second failure gracefully.If the re-authenticated stream also encounters an error (401 or otherwise), the exception will propagate with less context. The first stream logs "Error streaming video" but the retry path has no error logging.
♻️ Proposed improvement with retry limit and consistent logging
async def stream_video(): async with send_stream: + current_client = client + retried = False try: - async for frame in client.mjpeg_stream(): + async for frame in current_client.mjpeg_stream(): buffer = BytesIO() frame.save(buffer, format="JPEG") await send_stream.send(buffer.getvalue()) except Exception as e: - if _is_unauthorized_error(e): + if _is_unauthorized_error(e) and not retried: self.logger.warning("Received 401 Unauthorized during stream, re-authenticating...") await self._reset_client() - new_client = await self._get_client() - async for frame in new_client.mjpeg_stream(): - buffer = BytesIO() - frame.save(buffer, format="JPEG") - await send_stream.send(buffer.getvalue()) + current_client = await self._get_client() + retried = True + # Retry streaming with new client + try: + async for frame in current_client.mjpeg_stream(): + buffer = BytesIO() + frame.save(buffer, format="JPEG") + await send_stream.send(buffer.getvalue()) + except Exception as retry_error: + self.logger.error(f"Error streaming video after re-auth: {retry_error}") + raise else: self.logger.error(f"Error streaming video: {e}") raise🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py` around lines 144 - 162, stream_video's retry path after calling self._reset_client() and new_client.mjpeg_stream() lacks error handling and logging; wrap the retry stream consumption in its own try/except, detect unauthorized via _is_unauthorized_error, and enforce a retry limit (e.g., max 2 attempts) to avoid infinite retries. Specifically, modify the stream_video coroutine to loop up to a retry_count (start with client, on 401 call self._reset_client() and _get_client() and attempt once more), and for the second attempt catch exceptions from new_client.mjpeg_stream(), logging via self.logger.error with the same contextual message used for the first stream (include exception details) before re-raising; reference functions/methods stream_video, _is_unauthorized_error, _reset_client, _get_client, mjpeg_stream, and logger when making these changes.python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py (2)
160-163: Asymmetric coordinate handling may surprise callers.When only
xor onlyyis provided (but not both), the current logic sendsNone, Nonefor both coordinates, silently ignoring the provided value. Consider either:
- Raising an error if only one coordinate is provided, or
- Documenting this behavior explicitly
🛡️ Optional: Add validation for partial coordinates
def mouse_click(self, button: MouseButton | str = "left", x: float | None = None, y: float | None = None): + if (x is None) != (y is None): + raise ValueError("Both x and y must be provided together, or neither") if x is not None and y is not None: self.call("mouse_click", button, x, y) else: self.call("mouse_click", button, None, None)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py` around lines 160 - 163, The mouse-click wrapper currently sends None for both coords when only one of x/y is provided; instead, add input validation in the method (the function that calls self.call("mouse_click", ...)) so that if exactly one of x or y is provided you raise a ValueError with a clear message (e.g., "both x and y must be provided together or neither"), keep the existing behavior when both are provided (call("mouse_click", button, x, y)) and when neither is provided (call("mouse_click", button, None, None)); reference the method that invokes self.call("mouse_click", ...) in client.py to locate where to add this validation.
191-205:unicode_escapedecoding can raise on invalid sequences.If users provide malformed escape sequences (e.g.,
\xwithout hex digits),decode("unicode_escape")will raiseUnicodeDecodeError. Consider wrapping in a try/except with a helpful error message.🛡️ Proposed defensive handling
`@base.command`() `@click.argument`("text") def paste(text): """Paste text via keyboard HID (supports \\n for newline, \\t for tab)""" - # Decode escape sequences like \n, \t, etc. - decoded_text = text.encode().decode("unicode_escape") + try: + decoded_text = text.encode().decode("unicode_escape") + except UnicodeDecodeError as e: + raise click.ClickException(f"Invalid escape sequence in text: {e}") self.paste_text(decoded_text) click.echo(f"Pasted: {repr(decoded_text)}") `@base.command`() `@click.argument`("key") def press(key): """Press a single key (supports \\n for Enter, \\t for Tab)""" - # Decode escape sequences - decoded_key = key.encode().decode("unicode_escape") + try: + decoded_key = key.encode().decode("unicode_escape") + except UnicodeDecodeError as e: + raise click.ClickException(f"Invalid escape sequence in key: {e}") self.press_key(decoded_key) click.echo(f"Pressed: {repr(decoded_key)}")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py` around lines 191 - 205, The decode("unicode_escape") call in both paste and press can raise UnicodeDecodeError on malformed escapes; wrap the decoding in a try/except in paste(text) and press(key), catch UnicodeDecodeError (and optionally ValueError), and convert it into a user-friendly Click error (e.g., raise click.BadParameter or click.ClickException) that includes the offending input and the decode error message so the CLI reports a helpful message instead of crashing. Ensure both paste_text(decoded_text) and press_key(decoded_key) are only called after successful decode.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`:
- Around line 91-98: The close method in jumpstarter_driver_nanokvm.driver
(method close) is missing a call to super().close(), so child drivers in
self.children won't be closed; after performing the client context teardown (the
anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None)) block and
its exception handling), invoke super().close() to ensure base Driver.close()
runs and closes nested children (or, if driver semantics require closing
children first, call super().close() before the client teardown) — update the
close method in the class to call super().close() in the appropriate order so
child drivers are properly cleaned up.
---
Nitpick comments:
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py`:
- Around line 160-163: The mouse-click wrapper currently sends None for both
coords when only one of x/y is provided; instead, add input validation in the
method (the function that calls self.call("mouse_click", ...)) so that if
exactly one of x or y is provided you raise a ValueError with a clear message
(e.g., "both x and y must be provided together or neither"), keep the existing
behavior when both are provided (call("mouse_click", button, x, y)) and when
neither is provided (call("mouse_click", button, None, None)); reference the
method that invokes self.call("mouse_click", ...) in client.py to locate where
to add this validation.
- Around line 191-205: The decode("unicode_escape") call in both paste and press
can raise UnicodeDecodeError on malformed escapes; wrap the decoding in a
try/except in paste(text) and press(key), catch UnicodeDecodeError (and
optionally ValueError), and convert it into a user-friendly Click error (e.g.,
raise click.BadParameter or click.ClickException) that includes the offending
input and the decode error message so the CLI reports a helpful message instead
of crashing. Ensure both paste_text(decoded_text) and press_key(decoded_key) are
only called after successful decode.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`:
- Around line 144-162: stream_video's retry path after calling
self._reset_client() and new_client.mjpeg_stream() lacks error handling and
logging; wrap the retry stream consumption in its own try/except, detect
unauthorized via _is_unauthorized_error, and enforce a retry limit (e.g., max 2
attempts) to avoid infinite retries. Specifically, modify the stream_video
coroutine to loop up to a retry_count (start with client, on 401 call
self._reset_client() and _get_client() and attempt once more), and for the
second attempt catch exceptions from new_client.mjpeg_stream(), logging via
self.logger.error with the same contextual message used for the first stream
(include exception details) before re-raising; reference functions/methods
stream_video, _is_unauthorized_error, _reset_client, _get_client, mjpeg_stream,
and logger when making these changes.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3c6409a2-58b3-4f64-b4be-6ab3c1b83d6f
📒 Files selected for processing (6)
python/packages/jumpstarter-driver-nanokvm/README.mdpython/packages/jumpstarter-driver-nanokvm/examples/exporter.yamlpython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/__init__.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.pypython/packages/jumpstarter-driver-nanokvm/pyproject.toml
✅ Files skipped from review due to trivial changes (4)
- python/packages/jumpstarter-driver-nanokvm/examples/exporter.yaml
- python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/init.py
- python/packages/jumpstarter-driver-nanokvm/pyproject.toml
- python/packages/jumpstarter-driver-nanokvm/README.md
|
@ambient-code please rebase this |
- Extract NanoKVMDriverBase to deduplicate client management across NanoKVMVideo and NanoKVMHID (~30 lines of copy-paste removed) - Remove NanoKVMSerial placeholder class and all SSH/serial config (enable_serial, ssh_username, ssh_password, ssh_port) - Fix stream re-auth bug: keep send_stream context open during retry so the retry can still write to the stream - Replace asyncio.create_task with anyio.create_task_group for consistency with project conventions - Apply @with_reauth decorator directly on NanoKVM composite methods instead of fragile inner-function pattern; delegate _get_client and _reset_client to video child driver - Fix mount_image unmount logic: pass None explicitly for unmount instead of confusing `mount_file or None` expression - Add pytest-asyncio to dev dependencies (needed for asyncio_mode=auto) - Remove jumpstarter-driver-pyserial dependency (only needed for serial) - Add get_images to API docs :members: list - Remove serial/SSH references from README and documentation Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
ambient-code
bot
force-pushed
the
nanokvm-api-driver
branch
from
adfaa2f to
48575f9
Compare
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 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-nanokvm/jumpstarter_driver_nanokvm/client.py`:
- Around line 480-487: The CLI hides a valid 0% because code uses truthiness;
change checks to explicit None checks (e.g., use status.get('percentage') is not
None and status.get('file') is not None) in the download(url) handler and the
download_status handler so zero is shown, and when printing percentage include
the "%" suffix (e.g., f"{status['percentage']}%") to preserve the percent sign.
- Around line 141-163: The mouse_click method currently treats a single provided
coordinate as absent and silently clicks at the current pointer; change it to
validate that x and y are provided together: inside mouse_click check if (x is
None) != (y is None) and if so raise a ValueError (or TypeError) with a clear
message like "both x and y must be provided together or both omitted"; otherwise
call self.call("mouse_click", button, x, y) with the actual x/y (no None
substitution). Apply the same validation pattern to the other similar mouse
coordinate helper referenced around lines 234-251 to ensure symmetric behavior.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`:
- Around line 182-211: The paste_text and press_key methods currently log raw
HID input (self.logger.info f"Pasted text: {text}" and self.logger.debug
f"Pressed key: {repr(key)}"), which can leak secrets; remove or replace those
logs so they never include the actual text/key content—log only non-sensitive
metadata instead (e.g., mask content as "<REDACTED>", log length, or log an
event like "paste_text invoked" or "press_key invoked with length=X" and keep
any warnings about misuse but without including the key string). Update both
paste_text and press_key to stop emitting raw input in logs and ensure any
existing tests or callers expecting those exact messages are adjusted
accordingly.
- Around line 73-89: The code caches a partially-initialized NanoKVM client
because _client is set before authenticate() completes and _reset_client()
discards _client_ctx without exiting the async context; update _get_client to
only assign self._client after authenticate() succeeds (create the
NanoKVMAPIClient context, await __aenter__ into a temp variable, call
authenticate, then assign self._client and self._client_ctx), and update
_reset_client to properly __aexit__ or close the _client_ctx when present
(calling await self._client_ctx.__aexit__(None, None, None) or using its close
method) before clearing self._client and self._client_ctx so no entered context
is abandoned.
- Around line 117-123: The loop that skips stale frames is off-by-one because
frame_count is incremented before comparing to skip_frames; change the logic so
the comparison happens before incrementing (or use <= instead of <) to ensure
exactly skip_frames frames are skipped—e.g., in the async for over
client.mjpeg_stream() referencing frame_count and skip_frames, either move
"frame_count += 1" below the if-check or change the condition to "if frame_count
<= skip_frames: continue" (adjusting initial frame_count as needed) so
skip_frames=N truly skips N frames.
- Around line 91-97: The close method is calling
self._client_ctx.__aexit__(None, None, None) and passing the coroutine to
anyio.from_thread.run(), but anyio.from_thread.run expects an async callable and
its args separately; update Driver.close to call
anyio.from_thread.run(self._client_ctx.__aexit__, None, None, None) (i.e. pass
the __aexit__ bound method as the callable and the three None args separately)
so the async context exit runs correctly, keeping the existing try/except and
logger.debug behavior for errors.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: d0c5d7e5-36cb-41c8-a3eb-9aa841cecbc3
⛔ Files ignored due to path filters (1)
python/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (11)
python/docs/source/reference/package-apis/drivers/index.mdpython/docs/source/reference/package-apis/drivers/nanokvm.mdpython/packages/jumpstarter-driver-nanokvm/.gitignorepython/packages/jumpstarter-driver-nanokvm/README.mdpython/packages/jumpstarter-driver-nanokvm/examples/exporter.yamlpython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/__init__.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.pypython/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver_test.pypython/packages/jumpstarter-driver-nanokvm/pyproject.tomlpython/pyproject.toml
✅ Files skipped from review due to trivial changes (9)
- python/packages/jumpstarter-driver-nanokvm/.gitignore
- python/docs/source/reference/package-apis/drivers/nanokvm.md
- python/docs/source/reference/package-apis/drivers/index.md
- python/pyproject.toml
- python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/init.py
- python/packages/jumpstarter-driver-nanokvm/examples/exporter.yaml
- python/packages/jumpstarter-driver-nanokvm/pyproject.toml
- python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver_test.py
- python/packages/jumpstarter-driver-nanokvm/README.md
| def mouse_click(self, button: MouseButton | str = "left", x: float | None = None, y: float | None = None): | ||
| """ | ||
| Click a mouse button | ||
|
|
||
| Args: | ||
| button: Mouse button to click (MouseButton enum or "left", "right", "middle" string) | ||
| x: Optional X coordinate (0.0 to 1.0) for absolute positioning before click | ||
| y: Optional Y coordinate (0.0 to 1.0) for absolute positioning before click | ||
|
|
||
| Example:: | ||
|
|
||
| # Using string (backward compatible) | ||
| hid.mouse_click("left") | ||
| hid.mouse_click("left", 0.5, 0.5) | ||
|
|
||
| # Using MouseButton enum (recommended) | ||
| hid.mouse_click(MouseButton.LEFT) | ||
| hid.mouse_click(MouseButton.RIGHT, 0.75, 0.25) | ||
| """ | ||
| if x is not None and y is not None: | ||
| self.call("mouse_click", button, x, y) | ||
| else: | ||
| self.call("mouse_click", button, None, None) |
There was a problem hiding this comment.
Reject half-specified click coordinates.
If only one of x/y is provided, this falls through to self.call("mouse_click", button, None, None) and clicks at the current pointer location. The CLI exposes --x and --y independently, so this is easy to trigger from normal usage.
🛠️ Suggested fix
def mouse_click(self, button: MouseButton | str = "left", x: float | None = None, y: float | None = None):
+ if (x is None) != (y is None):
+ raise ValueError("x and y must be provided together")
if x is not None and y is not None:
self.call("mouse_click", button, x, y)
else:
self.call("mouse_click", button, None, None)
@@
`@click.option`("--x", type=float, default=None, help="Optional X coordinate (0.0-1.0)")
`@click.option`("--y", type=float, default=None, help="Optional Y coordinate (0.0-1.0)")
def mouse_click_cmd(button, x, y):
"""Click a mouse button"""
+ if (x is None) != (y is None):
+ raise click.UsageError("--x and --y must be provided together")
# Convert string to MouseButton enum
button_map = {Also applies to: 234-251
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py`
around lines 141 - 163, The mouse_click method currently treats a single
provided coordinate as absent and silently clicks at the current pointer; change
it to validate that x and y are provided together: inside mouse_click check if
(x is None) != (y is None) and if so raise a ValueError (or TypeError) with a
clear message like "both x and y must be provided together or both omitted";
otherwise call self.call("mouse_click", button, x, y) with the actual x/y (no
None substitution). Apply the same validation pattern to the other similar mouse
coordinate helper referenced around lines 234-251 to ensure symmetric behavior.
| def download(url): | ||
| """Download an image from URL""" | ||
| status = self.download_image(url) | ||
| click.echo(f"Download started: {status['status']}") | ||
| if status['file']: | ||
| click.echo(f"File: {status['file']}") | ||
| if status['percentage']: | ||
| click.echo(f"Progress: {status['percentage']}%") |
There was a problem hiding this comment.
Show 0% download progress.
Both commands use truthiness checks, so a valid 0 percentage disappears from the CLI. download_status() also drops the % suffix.
🛠️ Suggested fix
def download(url):
"""Download an image from URL"""
status = self.download_image(url)
click.echo(f"Download started: {status['status']}")
if status['file']:
click.echo(f"File: {status['file']}")
- if status['percentage']:
+ if status['percentage'] is not None:
click.echo(f"Progress: {status['percentage']}%")
@@
def download_status():
"""Get current image download status"""
status = self.get_image_download_status()
click.echo(f"Status: {status['status']}")
if status['file']:
click.echo(f"File: {status['file']}")
- if status['percentage']:
- click.echo(f"Progress: {status['percentage']}")
+ if status['percentage'] is not None:
+ click.echo(f"Progress: {status['percentage']}%")Also applies to: 516-523
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/client.py`
around lines 480 - 487, The CLI hides a valid 0% because code uses truthiness;
change checks to explicit None checks (e.g., use status.get('percentage') is not
None and status.get('file') is not None) in the download(url) handler and the
download_status handler so zero is shown, and when printing percentage include
the "%" suffix (e.g., f"{status['percentage']}%") to preserve the percent sign.
| async def _reset_client(self): | ||
| """Reset the client, forcing re-authentication""" | ||
| if self._client is not None: | ||
| try: | ||
| await self._client.close() | ||
| except Exception as e: | ||
| self.logger.debug(f"Error closing client during reset: {e}") | ||
| self._client = None | ||
| self._client_ctx = None | ||
|
|
||
| async def _get_client(self) -> NanoKVMAPIClient: | ||
| """Get or create the NanoKVM API client using context manager""" | ||
| if self._client is None: | ||
| self._client_ctx = NanoKVMAPIClient(f"http://{self.host}/api/") | ||
| self._client = await self._client_ctx.__aenter__() | ||
| await self._client.authenticate(self.username, self.password) | ||
| return self._client |
There was a problem hiding this comment.
Don't cache a half-initialized NanoKVM client.
_client is assigned before authenticate() succeeds. Any non-401 failure in client setup leaves a stale client cached, so later calls skip re-auth entirely. _reset_client() also drops _client_ctx without exiting it, which abandons the entered async context on retries.
🛠️ Suggested fix
async def _reset_client(self):
"""Reset the client, forcing re-authentication"""
- if self._client is not None:
+ ctx = self._client_ctx
+ self._client = None
+ self._client_ctx = None
+ if ctx is not None:
try:
- await self._client.close()
+ await ctx.__aexit__(None, None, None)
except Exception as e:
self.logger.debug(f"Error closing client during reset: {e}")
- self._client = None
- self._client_ctx = None
async def _get_client(self) -> NanoKVMAPIClient:
"""Get or create the NanoKVM API client using context manager"""
if self._client is None:
- self._client_ctx = NanoKVMAPIClient(f"http://{self.host}/api/")
- self._client = await self._client_ctx.__aenter__()
- await self._client.authenticate(self.username, self.password)
+ ctx = NanoKVMAPIClient(f"http://{self.host}/api/")
+ client = await ctx.__aenter__()
+ try:
+ await client.authenticate(self.username, self.password)
+ except Exception:
+ await ctx.__aexit__(None, None, None)
+ raise
+ self._client_ctx = ctx
+ self._client = client
return self._client🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`
around lines 73 - 89, The code caches a partially-initialized NanoKVM client
because _client is set before authenticate() completes and _reset_client()
discards _client_ctx without exiting the async context; update _get_client to
only assign self._client after authenticate() succeeds (create the
NanoKVMAPIClient context, await __aenter__ into a temp variable, call
authenticate, then assign self._client and self._client_ctx), and update
_reset_client to properly __aexit__ or close the _client_ctx when present
(calling await self._client_ctx.__aexit__(None, None, None) or using its close
method) before clearing self._client and self._client_ctx so no entered context
is abandoned.
| def close(self): | ||
| """Clean up resources""" | ||
| if self._client_ctx is not None: | ||
| try: | ||
| anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None)) | ||
| except Exception as e: | ||
| self.logger.debug(f"Error closing client: {e}") |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
What is the AnyIO 4 signature of anyio.from_thread.run? Does it accept a coroutine object, or must callers pass an async callable such as ctx.aexit plus its arguments?
💡 Result:
In AnyIO 4, the API reference documents:
- Signature:
anyio.from_thread.run(func, *args, token=None)(anyio.readthedocs.io) - Type of
func: a coroutine function / async callable (Callable[..., Awaitable[T]]), with*argspassed positionally (anyio.readthedocs.io)
So you’re expected to pass an async callable plus its arguments (e.g. anyio.from_thread.run(ctx.__aexit__, exc_type, exc, tb, token=...)), not a pre-created coroutine object. (anyio.readthedocs.io)
If you specifically want an interface that will accept any callable and await it if it returns a coroutine object, that behavior is documented for BlockingPortal.call(func, *args) (“If the callable returns a coroutine object, it is awaited on.”). (anyio.readthedocs.io)
Citations:
- 1: https://anyio.readthedocs.io/en/stable/api.html
- 2: https://anyio.readthedocs.io/en/stable/api.html
- 3: https://anyio.readthedocs.io/en/stable/api.html
- 4: https://anyio.readthedocs.io/en/stable/api.html
Pass __aexit__ itself to anyio.from_thread.run() with its arguments.
The current code passes a coroutine object (the result of calling __aexit__(None, None, None) immediately) to from_thread.run(). However, anyio.from_thread.run() expects an async callable plus its arguments as separate parameters. This will cause the cleanup to fail before the context is properly exited.
🛠️ Suggested fix
- anyio.from_thread.run(self._client_ctx.__aexit__(None, None, None))
+ anyio.from_thread.run(self._client_ctx.__aexit__, None, None, None)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`
around lines 91 - 97, The close method is calling
self._client_ctx.__aexit__(None, None, None) and passing the coroutine to
anyio.from_thread.run(), but anyio.from_thread.run expects an async callable and
its args separately; update Driver.close to call
anyio.from_thread.run(self._client_ctx.__aexit__, None, None, None) (i.e. pass
the __aexit__ bound method as the callable and the three None args separately)
so the async context exit runs correctly, keeping the existing try/except and
logger.debug behavior for errors.
| client = await self._get_client() | ||
| frame_count = 0 | ||
| async for frame in client.mjpeg_stream(): | ||
| frame_count += 1 | ||
| # Skip the first frames as they're normally stale | ||
| if frame_count < skip_frames: | ||
| continue |
There was a problem hiding this comment.
skip_frames is off by one.
With skip_frames=1, this returns the first frame because frame_count is incremented before the < check. The default path therefore skips only two stale frames instead of three.
🛠️ Suggested fix
- if frame_count < skip_frames:
+ if frame_count <= skip_frames:
continue🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`
around lines 117 - 123, The loop that skips stale frames is off-by-one because
frame_count is incremented before comparing to skip_frames; change the logic so
the comparison happens before incrementing (or use <= instead of <) to ensure
exactly skip_frames frames are skipped—e.g., in the async for over
client.mjpeg_stream() referencing frame_count and skip_frames, either move
"frame_count += 1" below the if-check or change the condition to "if frame_count
<= skip_frames: continue" (adjusting initial frame_count as needed) so
skip_frames=N truly skips N frames.
| async def paste_text(self, text: str): | ||
| """ | ||
| Paste text via keyboard HID simulation | ||
|
|
||
| Args: | ||
| text: Text to paste (limited character set supported) | ||
| """ | ||
| client = await self._get_client() | ||
| await client.paste_text(text) | ||
| self.logger.info(f"Pasted text: {text}") | ||
|
|
||
| @export | ||
| @with_reauth | ||
| async def press_key(self, key: str): | ||
| """ | ||
| Press a key by pasting a single character | ||
|
|
||
| Args: | ||
| key: Single character or escape sequence to press (e.g., 'a', 'A', '\\n', '\\t') | ||
|
|
||
| Note: | ||
| This uses paste_text under the hood, so it supports the same character set. | ||
| For special keys like Enter, use '\\n'. For Tab, use '\\t'. | ||
| """ | ||
| if len(key) > 2: # Allow for escape sequences like \n, \t | ||
| self.logger.warning(f"press_key should be used with single characters, got: {key}") | ||
|
|
||
| client = await self._get_client() | ||
| await client.paste_text(key) | ||
| self.logger.debug(f"Pressed key: {repr(key)}") |
There was a problem hiding this comment.
Don't log raw HID input.
These messages capture the exact text/key sent through the KVM, which will eventually include passwords, tokens, and other sensitive input.
🛠️ Suggested fix
- self.logger.info(f"Pasted text: {text}")
+ self.logger.debug(f"Pasted text via HID ({len(text)} chars)")
@@
- self.logger.warning(f"press_key should be used with single characters, got: {key}")
+ self.logger.warning("press_key should be used with single characters")
@@
- self.logger.debug(f"Pressed key: {repr(key)}")
+ self.logger.debug("Pressed key via HID")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@python/packages/jumpstarter-driver-nanokvm/jumpstarter_driver_nanokvm/driver.py`
around lines 182 - 211, The paste_text and press_key methods currently log raw
HID input (self.logger.info f"Pasted text: {text}" and self.logger.debug
f"Pressed key: {repr(key)}"), which can leak secrets; remove or replace those
logs so they never include the actual text/key content—log only non-sensitive
metadata instead (e.g., mask content as "<REDACTED>", log length, or log an
event like "paste_text invoked" or "press_key invoked with length=X" and keep
any warnings about misuse but without including the key string). Update both
paste_text and press_key to stop emitting raw input in logs and ensure any
existing tests or callers expecting those exact messages are adjusted
accordingly.
raballew
left a comment
raballew
left a comment
There was a problem hiding this comment.
Also take code rabbit into consideration. Lots of the findings seems correct.
| async def mouse_click(self, button: MouseButton | str = "left", x: float | None = None, y: float | None = None): | ||
| """ | ||
| Click a mouse button at current position or specified coordinates | ||
|
|
||
| Args: | ||
| button: Mouse button to click (MouseButton enum or "left", "right", "middle" string) | ||
| x: Optional X coordinate (0.0 to 1.0) for absolute positioning before click | ||
| y: Optional Y coordinate (0.0 to 1.0) for absolute positioning before click | ||
| """ | ||
| # Convert string to MouseButton enum for backward compatibility | ||
| if isinstance(button, str): | ||
| button_map = { | ||
| "left": MouseButton.LEFT, | ||
| "right": MouseButton.RIGHT, | ||
| "middle": MouseButton.MIDDLE, | ||
| } | ||
| button = button_map.get(button.lower(), MouseButton.LEFT) | ||
|
|
||
| client = await self._get_client() | ||
| await client.mouse_click(button, x, y) | ||
| self.logger.info(f"Mouse {button.name} clicked") |
There was a problem hiding this comment.
[HIGH] mouse_click will crash with AttributeError when the button argument arrives as a float. The gRPC framework serializes IntEnum values as protobuf number_value, which deserializes as Python float (e.g., MouseButton.LEFT becomes 1.0). The current code only handles str conversion, so (1.0).name blows up.
Suggested fix: add numeric handling before the string check:
if isinstance(button, (int, float)):
button = MouseButton(int(button))AI-generated, human reviewed
| async def _reset_client(self): | ||
| """Reset the client, forcing re-authentication""" | ||
| if self._client is not None: | ||
| try: | ||
| await self._client.close() | ||
| except Exception as e: | ||
| self.logger.debug(f"Error closing client during reset: {e}") | ||
| self._client = None | ||
| self._client_ctx = None |
There was a problem hiding this comment.
[HIGH] _reset_client calls await self._client.close(), but NanoKVMAPIClient doesn't implement close(). The AttributeError gets silently swallowed, and the aiohttp session leaks on every re-authentication.
Suggested fix: replace with await self._client_ctx.__aexit__(None, None, None) to properly clean up the context manager.
AI-generated, human reviewed
| [project.entry-points."jumpstarter.drivers"] | ||
| NanoKVM = "jumpstarter_driver_nanokvm.driver:NanoKVM" |
There was a problem hiding this comment.
[HIGH] Only NanoKVM is registered as an entry point. The child drivers NanoKVMVideo and NanoKVMHID are missing entry points, which means they can't be discovered independently. Other composite drivers in this codebase (dutlink, qemu) register all their children.
Suggested fix: add entry points for NanoKVMVideo and NanoKVMHID.
AI-generated, human reviewed
| async def press_key(self, key: str): | ||
| """ | ||
| Press a key by pasting a single character | ||
|
|
||
| Args: | ||
| key: Single character or escape sequence to press (e.g., 'a', 'A', '\\n', '\\t') | ||
|
|
||
| Note: | ||
| This uses paste_text under the hood, so it supports the same character set. | ||
| For special keys like Enter, use '\\n'. For Tab, use '\\t'. | ||
| """ | ||
| if len(key) > 2: # Allow for escape sequences like \n, \t | ||
| self.logger.warning(f"press_key should be used with single characters, got: {key}") | ||
|
|
||
| client = await self._get_client() | ||
| await client.paste_text(key) | ||
| self.logger.debug(f"Pressed key: {repr(key)}") | ||
|
|
There was a problem hiding this comment.
[HIGH] press_key() delegates to paste_text(), which only supports printable characters. FR-005 requires modifier key combinations (e.g., Ctrl+Alt+Delete), but the upstream library lacks a raw HID API for this.
Suggested fix: either add a method using the NanoKVM HID endpoint directly, or at minimum document this limitation clearly.
AI-generated, human reviewed
| dependencies = [ | ||
| "anyio>=4.10.0", | ||
| "jumpstarter", | ||
| "jumpstarter-driver-composite", | ||
| "nanokvm>=1.0.0", | ||
| "aiohttp", | ||
| "pillow", | ||
| "pydantic", | ||
| "yarl", | ||
| "click", | ||
| ] |
There was a problem hiding this comment.
[HIGH] pyproject.toml declares nanokvm>=1.0.0 but has no [tool.uv.sources] section to specify the git repository. The lock file resolves version 0.0.1 from git, creating a version/source mismatch. Running uv lock fresh will fail.
Suggested fix: add a [tool.uv.sources] section with the git source for nanokvm and workspace sources for jumpstarter dependencies.
AI-generated, human reviewed
| @@ -0,0 +1,227 @@ | |||
| """Tests for NanoKVM driver""" | |||
|
|
|||
| from unittest.mock import AsyncMock, MagicMock, patch | |||
There was a problem hiding this comment.
[MEDIUM] Around 20 narration comments in the test file, plus redundant comments in driver.py and client.py. Per project conventions, comments are only acceptable as a last resort when the code itself cannot be made self-explanatory.
Suggested fix: remove the narration comments and let the code speak for itself.
AI-generated, human reviewed
| This uses paste_text under the hood, so it supports the same character set. | ||
| For special keys like Enter, use '\\n'. For Tab, use '\\t'. | ||
| """ | ||
| if len(key) > 2: # Allow for escape sequences like \n, \t |
There was a problem hiding this comment.
[LOW] press_key length validation uses if len(key) > 2, which allows two-character strings without warning. Since decoded escape sequences are single characters, the threshold should be > 1.
Suggested fix: change to if len(key) > 1:.
AI-generated, human reviewed
| def test_nanokvm_mouse_click(mock_nanokvm_client, mock_aiohttp_session): | ||
| """Test mouse click""" | ||
| from nanokvm.models import MouseButton | ||
|
|
||
| hid = NanoKVMHID(host="test.local", username="admin", password="admin") | ||
|
|
||
| with serve(hid) as client: | ||
| # Click left button | ||
| client.mouse_click("left") | ||
|
|
||
| # Verify the mock was called | ||
| mock_nanokvm_client.mouse_click.assert_called_once_with(MouseButton.LEFT, None, None) |
There was a problem hiding this comment.
[LOW] The mouse_click test only covers the no-coordinates branch with a left button. The coordinates branch and non-left button conversions are untested.
Suggested fix: add test cases that pass coordinates and use a non-left button value.
AI-generated, human reviewed
| def test_nanokvm_video_client_creation(): | ||
| """Test that NanoKVMVideo returns correct client class""" | ||
| assert NanoKVMVideo.client() == "jumpstarter_driver_nanokvm.client.NanoKVMVideoClient" | ||
|
|
||
|
|
||
| def test_nanokvm_hid_client_creation(): | ||
| """Test that NanoKVMHID returns correct client class""" | ||
| assert NanoKVMHID.client() == "jumpstarter_driver_nanokvm.client.NanoKVMHIDClient" | ||
|
|
||
|
|
||
| def test_nanokvm_client_creation(): | ||
| """Test that NanoKVM returns correct client class""" | ||
| assert NanoKVM.client() == "jumpstarter_driver_nanokvm.client.NanoKVMClient" |
There was a problem hiding this comment.
[LOW] The client class string tests are near-tautological, just verifying a constant equals itself. They inflate coverage without testing any real behavior.
Suggested fix: replace with import verification tests or tests that exercise actual functionality.
AI-generated, human reviewed
| def test_nanokvm_video_snapshot(mock_nanokvm_client, mock_aiohttp_session): | ||
| """Test video snapshot functionality""" | ||
| video = NanoKVMVideo(host="test.local", username="admin", password="admin") | ||
|
|
||
| with serve(video) as client: | ||
| # Get a snapshot | ||
| image = client.snapshot() | ||
|
|
||
| # Verify it's a PIL Image | ||
| assert isinstance(image, Image.Image) | ||
| assert image.size == (640, 480) | ||
|
|
There was a problem hiding this comment.
[LOW] The snapshot skip_frames test uses identical mock frames, so the off-by-one bug in skip logic (F012) can't be detected.
Suggested fix: use distinguishable frames in the mock and verify the correct frame is returned.
AI-generated, human reviewed
2 participants
Summary by CodeRabbit
New Features
Documentation
✏️ Tip: You can customize this high-level summary in your review settings.