Tmtee helper driver

docs/source/reference/package-apis/drivers/index.md

@@ -79,6 +79,7 @@ Drivers for debugging and programming devices:
 General-purpose utility drivers:
 
 * **[Shell](shell.md)** (`jumpstarter-driver-shell`) - Shell command execution
+* **[TMT](tmt.md)** (`jumpstarter-driver-tmt`) - TMT (Test Management Tool) wrapper driver
 
 ```{toctree}
 :hidden:
@@ -102,6 +103,7 @@ sdwire.md
 shell.md
 snmp.md
 tasmota.md
+tmt.md
 tftp.md
 uboot.md
 ustreamer.md

docs/source/reference/package-apis/drivers/tmt.md

@@ -0,0 +1 @@
+../../../../../packages/jumpstarter-driver-tmt/README.md

packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py

@@ -1,10 +1,29 @@
+import inspect
+import logging
 from dataclasses import dataclass
 
 import click
+from rich import traceback
 
 from jumpstarter.client import DriverClient
 
 
+def _opt_log_level_callback(ctx, param, value):
+    traceback.install()
+
+    # Set the log level
+    log_level = value.upper() if value else "INFO"
+
+    # Update the root logger level to ensure all loggers inherit it
+    root_logger = logging.getLogger()
+    root_logger.setLevel(log_level)
+
+    # Update all existing loggers to use the new level
+    for logger_name in logging.Logger.manager.loggerDict:
+        logger = logging.getLogger(logger_name)
+        logger.setLevel(log_level)
+
+
 @dataclass(kw_only=True)
 class CompositeClient(DriverClient):
     def __getattr__(self, name):
@@ -17,12 +36,25 @@ def close(self):
 
     def cli(self):
         @click.group
+        @click.option(
+            "--log-level",
+            "log_level",
+            type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]),
+            help="Set the log level",
+            expose_value=False,
+            callback=_opt_log_level_callback,
+        )
         def base():
             """Generic composite device"""
             pass
 
         for k, v in self.children.items():
             if hasattr(v, "cli"):
-                base.add_command(v.cli(), k)
+                # Check if the cli method accepts a click_group parameter
+                sig = inspect.signature(v.cli)
+                if "click_group" in sig.parameters:
+                    base.add_command(v.cli(click_group=base), k)
+                else:
+                    base.add_command(v.cli(), k)
 
         return base

packages/jumpstarter-driver-tmt/.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+.coverage
+coverage.xml

packages/jumpstarter-driver-tmt/README.md

@@ -0,0 +1,109 @@
+# TMT Driver
+
+`jumpstarter-driver-tmt` provides functionality for running TMT (Test Management Tool) commands locally while connecting to remote devices via SSH network connections. This driver allows you to execute TMT test plans and commands that provision and test remote hardware through SSH connections.
+
+## Installation
+
+```shell
+pip3 install --extra-index-url https://pkg.jumpstarter.dev/simple/ jumpstarter-driver-tmt
+```
+
+## Configuration
+
+Example configuration:
+
+```yaml
+export:
+  tmt:
+    type: jumpstarter_driver_tmt.driver.TMT
+    config:
+      reboot_cmd: "j power cycle"
+      default_username: "root"
+      default_password: "somePassword"
+    children:
+      ssh:
+        type: jumpstarter_driver_network.driver.TcpNetwork
+        config:
+          host: "192.168.1.100"
+          port: 22
+          enable_address: true
+```
+
+### Config parameters
+
+| Parameter | Description | Type | Required | Default |
+|-----------|-------------|------|----------|---------|
+| reboot_cmd | Command to reboot the target device | str | no | "j power cycle" |
+| default_username | Default username for SSH connections | str | no | "" |
+| default_password | Default password for SSH connections | str | no | "" |
+
+### Required Children
+
+| Child | Description | Required |
+|-------|-------------|----------|
+| ssh | Network TCP driver instance for SSH connection | yes |
+
+## Usage
+
+### CLI Commands
+
+The TMT driver provides a CLI command `tmt` that allows you to run TMT commands locally while connecting to remote devices:
+
+```bash
+# assuming that your DUT has a power and storage driver
+j power on
+j storage flash ....
+
+# Running part of your plan with tmt
+j tmt --root . -c tracing=off -c arch=aarch64 -c distro=rhel-9 -c hw_target=rcar_s4 run --workdir-root /tmp/ -a -vv provision .. some other provisioning... plan -vv --name ^/podman/plans/fusa/tests$
+
+# Use SSH port forwarding (if no direct connection to the DUT is possible)
+j tmt --forward-ssh ....
+
+# Specify custom username and password
+j tmt --tmt-username root --tmt-password mypassword ...
+
+# Raise log level of the tmt wrapper driver
+j --log-level DEBUG tmt --root . -c tracing=off -c arch=aarch64 -c distro=rhel-9 -c hw_target=r
+car_s4 run --workdir-root /tmp/ -a -vv provision .. some other provisioning... plan -vv --name ^/podman/plans/fusa/te
+sts$
+[09/22/25 13:27:18] DEBUG    Using direct SSH connection for tmt - host: 127.0.0.1, port: 2222           client.py:64
+                    DEBUG    Provision to be replaced: ('provision', '..', 'some', 'other',             client.py:117
+                             'provisioning...')                                                                      
+                    DEBUG    Will be replaced with: ['provision', '-h', 'connect', '-g', '127.0.0.1',   client.py:118
+                             '-P', '2222', '-u', 'root', '-p', '******']                                           
+                    DEBUG    Running TMT command: ['tmt', '--root', '.', '-c', 'tracing=off', '-c',      client.py:74
+                             'arch=aarch64', '-c', 'distro=rhel-9', '-c', 'hw_target=rcar_s4', 'run',                
+                             '--workdir-root', '/tmp/', '-a', '-vv', 'provision', '-h', 'connect', '-g',             
+                             '127.0.0.1', '-P', '2222', '-u', 'root', '-p', '******', 'plan', '-vv',               
+                             '--name', '^/podman/plans/fusa/tests$']                                                 
+/tmp/run-018
+...
+```
+
+### CLI Options
+
+| Option | Description | Default |
+|--------|-------------|---------|
+| `--forward-ssh` | Use SSH port forwarding for connection | false |
+| `--tmt-username` | Username for SSH connections | from config |
+| `--tmt-password` | Password for SSH connections | from config |
+| `--tmt-cmd` | TMT command to execute | "tmt" |
+| `--tmt-on-exporter` | Run TMT on the exporter (not implemented) | false |
+
+### Provision Arguments Handling
+
+The driver automatically handles TMT provision arguments by:
+
+1. **Detecting provision sections**: Looks for `provision` or `run` commands in the TMT arguments
+2. **Replacing connection details**: Automatically replaces or adds SSH connection parameters (`-h connect -g <host> -P <port> -u <username> -p <password>`)
+3. **Preserving other arguments**: Keeps all other TMT arguments intact
+
+Example of how arguments are transformed:
+```bash
+# Input command
+j tmt run --name /my/test/plan provision -h connect -g 192.168.1.100 -P 22
+
+# Automatically transformed to use SSH connection
+# TMT receives: run --name /my/test/plan provision -h connect -g <forwarded_host> -P <forwarded_port> -u root -p password
+```

packages/jumpstarter-driver-tmt/examples/exporter.yaml

@@ -0,0 +1,27 @@
+apiVersion: jumpstarter.dev/v1alpha1
+kind: ExporterConfig
+metadata:
+  namespace: default
+  name: demo
+endpoint: grpc.jumpstarter.192.168.0.203.nip.io:8082
+token: "<token>"
+export:
+  tmt:
+    type: jumpstarter_driver_tmt.driver.TMT
+    config:
+      reboot_cmd: "j power cycle"
+      default_username: "root"
+      default_password: "somePass"
+    children:
+      ssh:
+        ref: ssh
+  ssh:
+    type: jumpstarter_driver_network.driver.TcpNetwork
+    config:
+      host: 127.0.0.1
+      port: 2222
+      enable_address: true
+  power:
+    type: jumpstarter_driver_power.driver.MockPower
+
+

packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/client.py

@@ -0,0 +1,157 @@
+import subprocess
+from dataclasses import dataclass
+from urllib.parse import urlparse
+
+import click
+from jumpstarter_driver_composite.client import CompositeClient
+from jumpstarter_driver_network.adapters import TcpPortforwardAdapter
+
+from jumpstarter.client.core import DriverMethodNotImplemented
+
+
+def redact_password_in_args(args):
+    """Redact password arguments in a list for safe logging"""
+    redacted = list(args)
+    try:
+        # Find -p flag and redact the next argument (password)
+        p_index = redacted.index("-p")
+        if p_index + 1 < len(redacted):
+            redacted[p_index + 1] = "******"
+    except ValueError:
+        # -p flag not found, nothing to redact
+        pass
+    return redacted
+
+
+@dataclass(kw_only=True)
+class TMTClient(CompositeClient):
+    """
+    Client interface for LocalTMT driver
+
+    This client provides methods to interact with LocalTMT devices via SSH
+    """
+
+    def cli(self, click_group):
+
+        @click_group.command(context_settings={"ignore_unknown_options": True})
+        @click.option("--forward-ssh", is_flag=True)
+        @click.option("--tmt-username", default=None)
+        @click.option("--tmt-password", default=None)
+        @click.option("--tmt-cmd", default="tmt")
+        @click.option("--tmt-on-exporter", is_flag=True)
+        @click.argument("args", nargs=-1)
+        def tmt(forward_ssh, tmt_username, tmt_password, tmt_cmd, tmt_on_exporter, args):
+            """Run TMT command with arguments"""
+            if tmt_on_exporter:
+                click.echo("TMT will be run on the exporter")
+                raise click.Abort("Still not implemented")
+            else:
+                result = self.run_tmt_local(forward_ssh, tmt_cmd, tmt_username, tmt_password, args)
+                self.logger.debug(f"TMT result: {result}")
+                if result != 0:
+                    click.get_current_context().exit(result)
+                return result
+
+        return tmt
+
+    def run_tmt_local(self, forward_ssh, tmt_cmd, username, password, args):
+        # if we are asked to forward the ssh connection, or we have to fallback, we do that
+        def_user, def_pass = self.call("get_default_user_pass")
+        username = username or def_user
+        password = password or def_pass
+        hard_reboot_cmd = self.call("get_reboot_cmd")
+        if forward_ssh:
+            self.logger.debug("Using SSH port forwarding for TMT connection")
+            with TcpPortforwardAdapter(
+                client=self.ssh,
+            ) as addr:
+                host = addr[0]
+                port = addr[1]
+                self.logger.debug(f"SSH port forward established - host: {host}, port: {port}")
+                return self._run_tmt_local(host, port, tmt_cmd, username, password, hard_reboot_cmd, args)
+        else:
+            # if we are not asked to forward the ssh connection, we try to get the address from the ssh driver
+            try:
+                address = self.ssh.address() # (format: "tcp://host:port")
+                parsed = urlparse(address)
+                host = parsed.hostname
+                port = parsed.port
+                if not host or not port:
+                    raise ValueError(f"Invalid address format: {address}")
+                self.logger.debug(f"Using direct SSH connection for tmt - host: {host}, port: {port}")
+                return self._run_tmt_local(host, port, tmt_cmd, username, password, hard_reboot_cmd, args)
+            except (DriverMethodNotImplemented, ValueError) as e:
+                self.logger.warning(f"Direct address connection failed ({e}), falling back to SSH port forwarding")
+                return self.run_tmt_local(True, tmt_cmd, username, password, args)
+
+    def _run_tmt_local(self, host, port, tmt_cmd, username, password, hard_reboot_cmd, args):
+        """Run TMT command with the given host, port, and arguments"""
+        # This is a placeholder implementation - replace with actual TMT command execution
+        args = replace_provision_args(self.logger, args, host, port, username, password, hard_reboot_cmd)
+        # Redact password for safe logging
+        safe_args = redact_password_in_args(args)
+        self.logger.debug(f"Running TMT command: {[tmt_cmd] + safe_args}")
+        # execute the command on the local machine
+        try:
+            result = subprocess.run([tmt_cmd] + args)
+            return result.returncode
+        except FileNotFoundError:
+            self.logger.error(
+                f"TMT command '{tmt_cmd}' not found. Please ensure TMT is installed and available in PATH."
+            )
+            return 127  # Standard exit code for "command not found"
+
+# the tmt commands are executed locally, but we need to identify
+# the connection part of the commandline and replace it or insert our own
+# this is a possible set of args that we may receive:
+# --root . -c tracing=off -c arch=aarch64 -c distro=rhel-9 -c hw_target=rcar_s4 run
+# --workdir-root /tmp/ -a -vv provision -h connect -g $IP -P $PORT -u root -p password --help plan -vv
+# --name ^/podman/plans/fusa/tests$
+#
+# in this case we need to identify the provision section and replace it with our own,
+# and reuse the -u root -p password if provided
+# provision can have any number of -flag arguments, so we need to identify them and replace them with our own
+def replace_provision_args(logger, args, host, port, username, password, hard_reboot_cmd):
+    """Replace or add provision arguments for TMT command"""
+    # this list is used to identify the end of the provision section
+    TMT_RUN_CMDS = [
+        "discover", "provision", "prepare", "execute", "report", "finish", "cleanup",
+        "login", "reboot", "plan", "plans", "test", "run"
+    ]
+    # find the provision section
+    provision_args = ["provision","-h", "connect", "-g", host, "-P", str(port)]
+    if username:
+        provision_args.append("-u")
+        provision_args.append(username)
+    if password:
+        provision_args.append("-p")
+        provision_args.append(password)
+    if hard_reboot_cmd:
+        provision_args.append("--hard-reboot")
+        provision_args.append(hard_reboot_cmd)
+    try:
+        provision_index = args.index("provision")
+    except ValueError:
+        # "provision" not found in args
+        if "run" in args:
+            logger.debug("Run section found, adding provision arguments")
+            return list(args) + provision_args
+        else:
+            logger.debug("Provision or run section not found, ignoring")
+            return list(args)
+
+    next_cmd_index = provision_index + 1
+    while next_cmd_index < len(args) and args[next_cmd_index] not in TMT_RUN_CMDS:
+        next_cmd_index += 1
+    # Redact passwords for safe logging
+    safe_provision_section = redact_password_in_args(args[provision_index:next_cmd_index])
+    safe_provision_args = redact_password_in_args(provision_args)
+    logger.debug(f"Provision to be replaced: {safe_provision_section}")
+    logger.debug(f"Will be replaced with: {safe_provision_args}")
+    # get the provision section
+    before_provision = args[:provision_index]
+    after_provision = args[next_cmd_index:]
+
+    args = list(before_provision) + provision_args + list(after_provision)
+    return args
+