fix: small fixes batch (#516, #525, #652, #517)

python/packages/jumpstarter-cli-common/jumpstarter_cli_common/exceptions.py

@@ -135,6 +135,13 @@ def _map_grpc_exception(exc: BaseException) -> click.ClickException | None:
                 details,
             )
         )
+    if code == "FAILED_PRECONDITION":
+        return ClickExceptionRed(
+            _append_details(
+                "A precondition for the requested operation was not met. Check resource state and retry.",
+                details,
+            )
+        )
     return None
 
 

python/packages/jumpstarter-cli-common/jumpstarter_cli_common/exceptions_test.py

@@ -151,3 +151,35 @@ def grpc_invalid_arg_fn():
 
     with pytest.raises(click.ClickException, match="Invalid request arguments"):
         grpc_invalid_arg_fn()
+
+
+def test_handle_exceptions_maps_grpc_failed_precondition() -> None:
+    class MockGrpcError(Exception):
+        def code(self):
+            return type("Code", (), {"name": "FAILED_PRECONDITION"})()
+
+        def details(self):
+            return "exporter is not ready"
+
+    @handle_exceptions
+    def grpc_precondition_fn():
+        raise MockGrpcError()
+
+    with pytest.raises(click.ClickException, match="precondition"):
+        grpc_precondition_fn()
+
+
+def test_handle_exceptions_maps_grpc_failed_precondition_without_details() -> None:
+    class MockGrpcError(Exception):
+        def code(self):
+            return type("Code", (), {"name": "FAILED_PRECONDITION"})()
+
+        def details(self):
+            return ""
+
+    @handle_exceptions
+    def grpc_precondition_no_details_fn():
+        raise MockGrpcError()
+
+    with pytest.raises(click.ClickException, match="precondition"):
+        grpc_precondition_no_details_fn()

python/packages/jumpstarter-cli/jumpstarter_cli/login.py

@@ -118,14 +118,20 @@ def parse_login_argument(login_arg: str) -> tuple[str | None, str]:
     return None, login_arg
 
 
+def _warn_exporter_client_only_flags(config_kind: str | None, allow: str, unsafe: bool | None) -> None:
+    if config_kind is not None and config_kind.startswith("exporter"):
+        if allow:
+            click.echo("Warning: --allow is ignored for exporter configs (only applies to client configs).")
+        if unsafe:
+            click.echo("Warning: --unsafe is ignored for exporter configs (only applies to client configs).")
+
+
 @click.command("login", short_help="Login")
 @click.argument("login_target", required=False, default=None)
 @click.option("-e", "--endpoint", type=str, help="Enter the Jumpstarter service endpoint.", default=None)
 @click.option("--namespace", type=str, help="Enter the Jumpstarter exporter namespace.", default=None)
 @click.option("--name", type=str, help="Enter the Jumpstarter exporter name.", default=None)
 @opt_oidc
-# client specific
-# TODO: warn if used with exporter
 @click.option(
     "--allow",
     type=str,
@@ -135,7 +141,6 @@ def parse_login_argument(login_arg: str) -> tuple[str | None, str]:
 @click.option(
     "--unsafe", is_flag=True, help="Should all driver client packages be allowed to load (UNSAFE!).", default=None
 )
-# end client specific
 @opt_insecure_tls
 @opt_nointeractive
 @opt_config(allow_missing=True)
@@ -290,6 +295,8 @@ async def login(  # noqa: C901
                     token="",
                 )
 
+    _warn_exporter_client_only_flags(config_kind, allow, unsafe)
+
     if issuer is None:
         if nointeractive:
             raise click.UsageError("Issuer is required in non-interactive mode.")

python/packages/jumpstarter-cli/jumpstarter_cli/login_test.py

@@ -11,6 +11,7 @@
 from jumpstarter_cli.login import (
     _validate_auth_config_payload,
     _validate_login_endpoint_url,
+    _warn_exporter_client_only_flags,
     fetch_auth_config,
     parse_login_argument,
 )
@@ -209,6 +210,39 @@ async def test_fetch_auth_config_defaults_to_https_with_insecure_tls():
     assert result["grpcEndpoint"] == "grpc.example.com"
 
 
+def test_warn_exporter_client_only_flags_warns_on_allow(capsys) -> None:
+    _warn_exporter_client_only_flags("exporter", "some-driver", None)
+    captured = capsys.readouterr()
+    assert "--allow" in captured.out
+    assert "ignored" in captured.out.lower()
+
+
+def test_warn_exporter_client_only_flags_warns_on_unsafe(capsys) -> None:
+    _warn_exporter_client_only_flags("exporter", "", True)
+    captured = capsys.readouterr()
+    assert "--unsafe" in captured.out
+    assert "ignored" in captured.out.lower()
+
+
+def test_warn_exporter_client_only_flags_warns_on_exporter_config_kind(capsys) -> None:
+    _warn_exporter_client_only_flags("exporter_config", "pkg", True)
+    captured = capsys.readouterr()
+    assert "--allow" in captured.out
+    assert "--unsafe" in captured.out
+
+
+def test_warn_exporter_client_only_flags_silent_for_client(capsys) -> None:
+    _warn_exporter_client_only_flags("client", "some-driver", True)
+    captured = capsys.readouterr()
+    assert captured.out == ""
+
+
+def test_warn_exporter_client_only_flags_silent_when_no_flags(capsys) -> None:
+    _warn_exporter_client_only_flags("exporter", "", None)
+    captured = capsys.readouterr()
+    assert captured.out == ""
+
+
 def test_login_maps_ssl_cert_error_during_oidc_to_friendly_message(monkeypatch) -> None:
     auth_config = {
         "grpcEndpoint": "grpc.example.com:443",

python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py

@@ -85,7 +85,6 @@ def __post_init__(self):
     def set_console_debug(self, debug: bool):
         """Set console debug mode"""
         self._console_debug = debug
-        # TODO: also set console debug on uboot client
 
     @contextmanager
     def busybox_shell(self):

python/packages/jumpstarter-driver-uboot/jumpstarter_driver_uboot/client.py

@@ -19,7 +19,7 @@ def prompt(self) -> str:
         return self.call("get_prompt")
 
     @contextmanager
-    def reboot_to_console(self, *, debug=False) -> Generator[None]:
+    def reboot_to_console(self, *, debug: bool = False, retries: int = 100) -> Generator[None]:
         """
         Reboot to U-Boot console
 
@@ -43,11 +43,9 @@ def reboot_to_console(self, *, debug=False) -> Generator[None]:
             if debug:
                 p.logfile_read = sys.stdout.buffer
 
-            for _ in range(100):  # TODO: configurable retries
+            for _ in range(retries):
                 try:
                     p.send(ESC)
-                    # in case of "bootmenu" there are all sort of escape sequences in the output so try to
-                    # catch prompt without any leading newlines, hoping it's not in the menu text somewhere
                     p.expect_exact(self.prompt.lstrip("\n"), timeout=0.1)
                 except pexpect.TIMEOUT:
                     continue

python/packages/jumpstarter-driver-uboot/jumpstarter_driver_uboot/client_test.py

@@ -0,0 +1,101 @@
+import inspect
+import logging
+from contextlib import contextmanager
+from unittest.mock import MagicMock, call, patch
+
+import pexpect
+import pytest
+
+from .client import UbootConsoleClient
+from .common import ESC
+
+
+def test_reboot_to_console_accepts_retries_kwarg() -> None:
+    sig = inspect.signature(UbootConsoleClient.reboot_to_console)
+    retries_param = sig.parameters.get("retries")
+    assert retries_param is not None
+    assert retries_param.default == 100
+    assert retries_param.kind == inspect.Parameter.KEYWORD_ONLY
+
+
+def test_reboot_to_console_retries_limits_attempts() -> None:
+    mock_power = MagicMock()
+    mock_pexpect_process = MagicMock()
+    mock_pexpect_process.send = MagicMock()
+    mock_pexpect_process.expect_exact = MagicMock(side_effect=pexpect.TIMEOUT("timeout"))
+
+    @contextmanager
+    def fake_pexpect():
+        yield mock_pexpect_process
+
+    mock_serial = MagicMock()
+    mock_serial.pexpect = fake_pexpect
+
+    client = object.__new__(UbootConsoleClient)
+    client.children = {"power": mock_power, "serial": mock_serial}
+    client.logger = logging.getLogger("test_uboot")
+
+    prompt_value = "=> "
+    with patch.object(type(client), "prompt", new_callable=lambda: property(lambda self: prompt_value)):
+        with pytest.raises(RuntimeError, match="Failed to get U-Boot prompt"):
+            with client.reboot_to_console(retries=3):
+                pass
+
+    assert mock_pexpect_process.send.call_count == 3
+    mock_pexpect_process.send.assert_has_calls([call(ESC)] * 3)
+    mock_power.cycle.assert_called_once()
+
+
+def test_reboot_to_console_yields_on_prompt_match() -> None:
+    mock_power = MagicMock()
+    mock_pexpect_process = MagicMock()
+    mock_pexpect_process.send = MagicMock()
+    mock_pexpect_process.expect_exact = MagicMock(
+        side_effect=[pexpect.TIMEOUT("timeout"), pexpect.TIMEOUT("timeout"), None]
+    )
+
+    @contextmanager
+    def fake_pexpect():
+        yield mock_pexpect_process
+
+    mock_serial = MagicMock()
+    mock_serial.pexpect = fake_pexpect
+
+    client = object.__new__(UbootConsoleClient)
+    client.children = {"power": mock_power, "serial": mock_serial}
+    client.logger = logging.getLogger("test_uboot")
+
+    prompt_value = "=> "
+    with patch.object(type(client), "prompt", new_callable=lambda: property(lambda self: prompt_value)):
+        entered = False
+        with client.reboot_to_console(retries=5):
+            entered = True
+
+    assert entered
+    assert mock_pexpect_process.send.call_count == 3
+    mock_power.cycle.assert_called_once()
+
+
+def test_reboot_to_console_retries_zero_raises_immediately() -> None:
+    mock_power = MagicMock()
+    mock_pexpect_process = MagicMock()
+
+    @contextmanager
+    def fake_pexpect():
+        yield mock_pexpect_process
+
+    mock_serial = MagicMock()
+    mock_serial.pexpect = fake_pexpect
+
+    client = object.__new__(UbootConsoleClient)
+    client.children = {"power": mock_power, "serial": mock_serial}
+    client.logger = logging.getLogger("test_uboot")
+
+    prompt_value = "=> "
+    with patch.object(type(client), "prompt", new_callable=lambda: property(lambda self: prompt_value)):
+        with pytest.raises(RuntimeError, match="Failed to get U-Boot prompt"):
+            with client.reboot_to_console(retries=0):
+                pass
+
+    mock_pexpect_process.send.assert_not_called()
+    mock_power.cycle.assert_called_once()