feat(driver-ssh): add --user selection to j ssh

Allow `j ssh -u someuser` to be able to select the user with
which to connect to the board.

Signed-off-by: Albert Esteve <aesteve@redhat.com>

Author: aesteve-rh

packages/jumpstarter-driver-ssh/jumpstarter_driver_ssh/client.py

@@ -62,16 +62,17 @@ def cli(self):
             help="Run SSH command with arguments",
         )
         @click.option("--direct", is_flag=True, help="Use direct TCP address")
+        @click.option("-u", "--user", help="Username to use for SSH connection")
         @click.argument("args", nargs=-1)
-        def ssh(direct, args):
+        def ssh(direct, user, args):
             options = SSHCommandRunOptions(
                 direct=direct,
                 # For the CLI, we never capture output so that interactive shells
                 # and long-running commands stream their output directly.
                 capture_output=False,
             )
 
-            result = self.run(options, args)
+            result = self.run(options, args, user=user)
             self.logger.debug("SSH exit code: %s", result.return_code)
 
             if result.stdout:
@@ -118,7 +119,7 @@ def username(self) -> str:
         """Get the default SSH username"""
         return self.call("get_default_username")
 
-    def run(self, options: SSHCommandRunOptions, args) -> SSHCommandRunResult:
+    def run(self, options: SSHCommandRunOptions, args, user: str | None = None) -> SSHCommandRunResult:
         """Run SSH command with the given parameters and arguments"""
         # Get SSH command and default username from driver
         if options.direct:
@@ -131,7 +132,7 @@ def run(self, options: SSHCommandRunOptions, args) -> SSHCommandRunResult:
                 if not host or not port:
                     raise ValueError(f"Invalid address format: {address}")
                 self.logger.debug("Using direct TCP connection for SSH - host: %s, port: %s", host, port)
-                return self._run_ssh_local(host, port, options, args)
+                return self._run_ssh_local(host, port, options, args, user)
             except (DriverMethodNotImplemented, ValueError) as e:
                 self.logger.error("Direct address connection failed (%s), falling back to SSH port forwarding", e)
                 return self.run(SSHCommandRunOptions(
@@ -147,9 +148,9 @@ def run(self, options: SSHCommandRunOptions, args) -> SSHCommandRunResult:
             ) as addr:
                 host, port = addr
                 self.logger.debug("SSH port forward established - host: %s, port: %s", host, port)
-                return self._run_ssh_local(host, port, options, args)
+                return self._run_ssh_local(host, port, options, args, user)
 
-    def _run_ssh_local(self, host, port, options, args):
+    def _run_ssh_local(self, host, port, options, args, user: str | None = None):
         """Run SSH command with the given host, port, and arguments"""
         # Create temporary identity file if needed
         ssh_identity = self.identity
@@ -175,7 +176,7 @@ def _run_ssh_local(self, host, port, options, args):
 
         try:
             # Build SSH command arguments
-            ssh_args = self._build_ssh_command_args(port, identity_file, args)
+            ssh_args = self._build_ssh_command_args(port, identity_file, args, user)
 
             # Separate SSH options from command arguments
             ssh_options, command_args = self._separate_ssh_options_and_command_args(args)
@@ -194,11 +195,11 @@ def _run_ssh_local(self, host, port, options, args):
                 except Exception as e:
                     self.logger.warning("Failed to clean up temporary identity file %s: %s", identity_file, str(e))
 
-    def _build_ssh_command_args(self, port, identity_file, args):
+    def _build_ssh_command_args(self, port, identity_file, args, user: str | None = None):
         """Build initial SSH command arguments"""
         # Split the SSH command into individual arguments
         ssh_args = shlex.split(self.command)
-        default_username = self.username
+        default_username = user or self.username
 
         # Add identity file if provided
         if identity_file:

packages/jumpstarter-driver-ssh/jumpstarter_driver_ssh/driver_test.py

@@ -105,6 +105,38 @@ def test_ssh_command_without_default_username():
             assert result.stdout == "some stdout"
 
 
+def test_ssh_command_with_explicit_user_parameter():
+    """Test SSH command execution with the user parameter overriding the default."""
+    instance = SSHWrapper(
+        children={"tcp": TcpNetwork(host="127.0.0.1", port=22)},
+        default_username="testuser",
+    )
+
+    with serve(instance) as client:
+        with patch("subprocess.run") as mock_run:
+            mock_run.return_value = MagicMock(returncode=0, stdout="some stdout", stderr="")
+
+            # Call run with an explicit user.
+            result = client.run(SSHCommandRunOptions(direct=False), ["hostname"], user="overrideuser")
+            assert isinstance(result, SSHCommandRunResult)
+
+            # Verify subprocess.run was called.
+            assert mock_run.called
+            call_args = mock_run.call_args[0][0]
+
+            # Check that the override user is present.
+            assert "-l" in call_args
+            assert "overrideuser" in call_args
+            assert "testuser" not in call_args
+            assert call_args[call_args.index("-l") + 1] == "overrideuser"
+
+            assert "127.0.0.1" in call_args
+            assert "hostname" in call_args
+
+            assert result.return_code == 0
+            assert result.stdout == "some stdout"
+
+
 def test_ssh_command_with_user_override():
     """Test SSH command execution with -l flag overriding default username"""
     instance = SSHWrapper(