From b99802b937f7150cd52418e683fbaaff9e3258af Mon Sep 17 00:00:00 2001 From: Miguel Angel Ajo Pelayo Date: Thu, 18 Sep 2025 09:56:33 +0000 Subject: [PATCH 1/2] Support simple commands at the base of j For example: j command instead of: j command subcommand The code now detects if the cli implementation of the client receives click_group and passes it down, letting the client driver decide the structure --- .../jumpstarter_driver_composite/client.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py b/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py index bd7fbf6f9..acc83e705 100644 --- a/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py +++ b/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py @@ -1,4 +1,5 @@ from dataclasses import dataclass +import inspect import click @@ -23,6 +24,11 @@ def base(): 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 From cc120518b0103c99681cb7628c9196d9718e729d Mon Sep 17 00:00:00 2001 From: Miguel Angel Ajo Pelayo Date: Wed, 17 Sep 2025 08:38:37 +0000 Subject: [PATCH 2/2] Implement TMTLocal driver --- .../reference/package-apis/drivers/index.md | 2 + .../reference/package-apis/drivers/tmt.md | 1 + .../jumpstarter_driver_composite/client.py | 28 +- packages/jumpstarter-driver-tmt/.gitignore | 3 + packages/jumpstarter-driver-tmt/README.md | 109 +++++ .../examples/exporter.yaml | 27 ++ .../jumpstarter_driver_tmt/__init__.py | 0 .../jumpstarter_driver_tmt/client.py | 157 +++++++ .../jumpstarter_driver_tmt/driver.py | 34 ++ .../jumpstarter_driver_tmt/driver_test.py | 437 ++++++++++++++++++ .../jumpstarter-driver-tmt/pyproject.toml | 44 ++ uv.lock | 31 ++ 12 files changed, 872 insertions(+), 1 deletion(-) create mode 120000 docs/source/reference/package-apis/drivers/tmt.md create mode 100644 packages/jumpstarter-driver-tmt/.gitignore create mode 100644 packages/jumpstarter-driver-tmt/README.md create mode 100644 packages/jumpstarter-driver-tmt/examples/exporter.yaml create mode 100644 packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/__init__.py create mode 100644 packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/client.py create mode 100644 packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver.py create mode 100644 packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver_test.py create mode 100644 packages/jumpstarter-driver-tmt/pyproject.toml diff --git a/docs/source/reference/package-apis/drivers/index.md b/docs/source/reference/package-apis/drivers/index.md index 2238c15a2..661f56745 100644 --- a/docs/source/reference/package-apis/drivers/index.md +++ b/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 diff --git a/docs/source/reference/package-apis/drivers/tmt.md b/docs/source/reference/package-apis/drivers/tmt.md new file mode 120000 index 000000000..ee1227ef5 --- /dev/null +++ b/docs/source/reference/package-apis/drivers/tmt.md @@ -0,0 +1 @@ +../../../../../packages/jumpstarter-driver-tmt/README.md \ No newline at end of file diff --git a/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py b/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py index acc83e705..9ebd500a0 100644 --- a/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py +++ b/packages/jumpstarter-driver-composite/jumpstarter_driver_composite/client.py @@ -1,11 +1,29 @@ -from dataclasses import dataclass 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): @@ -18,6 +36,14 @@ 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 diff --git a/packages/jumpstarter-driver-tmt/.gitignore b/packages/jumpstarter-driver-tmt/.gitignore new file mode 100644 index 000000000..cbc5d672b --- /dev/null +++ b/packages/jumpstarter-driver-tmt/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +.coverage +coverage.xml diff --git a/packages/jumpstarter-driver-tmt/README.md b/packages/jumpstarter-driver-tmt/README.md new file mode 100644 index 000000000..15dd6e873 --- /dev/null +++ b/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 -P -u -p `) +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 -P -u root -p password +``` diff --git a/packages/jumpstarter-driver-tmt/examples/exporter.yaml b/packages/jumpstarter-driver-tmt/examples/exporter.yaml new file mode 100644 index 000000000..2cf6847a2 --- /dev/null +++ b/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: "" +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 + + diff --git a/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/__init__.py b/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/client.py b/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/client.py new file mode 100644 index 000000000..9517680af --- /dev/null +++ b/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 + diff --git a/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver.py b/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver.py new file mode 100644 index 000000000..bdc5f8e29 --- /dev/null +++ b/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver.py @@ -0,0 +1,34 @@ +from dataclasses import dataclass + +from jumpstarter.common.exceptions import ConfigurationError +from jumpstarter.driver import Driver, export + + +@dataclass(kw_only=True) +class TMT(Driver): + """ driver for Jumpstarter""" + + reboot_cmd: str = "j power cycle" + default_username: str = "" + default_password: str = "" + + def __post_init__(self): + if hasattr(super(), "__post_init__"): + super().__post_init__() + + if "ssh" not in self.children: + raise ConfigurationError("'ssh' child is required via ref, or directly as a TcpNetwork driver instance") + + + @classmethod + def client(cls) -> str: + return "jumpstarter_driver_tmt.client.TMTClient" + + @export + def get_reboot_cmd(self): + return self.reboot_cmd + + @export + def get_default_user_pass(self): + return self.default_username, self.default_password + diff --git a/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver_test.py b/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver_test.py new file mode 100644 index 000000000..eb3afddff --- /dev/null +++ b/packages/jumpstarter-driver-tmt/jumpstarter_driver_tmt/driver_test.py @@ -0,0 +1,437 @@ +from unittest.mock import MagicMock, patch + +import click +import pytest +from click.testing import CliRunner +from jumpstarter_driver_network.driver import TcpNetwork + +from .client import replace_provision_args +from .driver import TMT +from jumpstarter.common.utils import serve + + +def test_drivers_tmt(): + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + assert client.ssh.address() == "tcp://127.0.0.1:22" + + +def test_drivers_tmt_cli(): + """Test the CLI functionality with tmt command and arguments""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + # Test the CLI tmt command without arguments + runner = CliRunner() + cli = client.cli(click.Group()) + + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 # Success return code + result = runner.invoke(cli, ["tmt"]) + assert result.exit_code == 0 + mock_run_tmt.assert_called_once() + + # Test the CLI tmt command with arguments + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 # Success return code + result = runner.invoke(cli, ["tmt", "test", "arg1", "arg2"]) + assert result.exit_code == 0 + mock_run_tmt.assert_called_once() + + +def test_drivers_tmt_cli_with_options(): + """Test the CLI functionality with various options""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + runner = CliRunner() + cli = client.cli(click.Group()) + + # Test with --forward-ssh flag + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 # Success return code + result = runner.invoke(cli, ["tmt", "--forward-ssh", "test"]) + assert result.exit_code == 0 + mock_run_tmt.assert_called_once() + + # Test with custom username and password + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 # Success return code + result = runner.invoke( + cli, ["tmt", "--tmt-username", "custom_user", "--tmt-password", "custom_pass", "test"] + ) + assert result.exit_code == 0 + mock_run_tmt.assert_called_once() + + # Test with custom tmt command + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 # Success return code + result = runner.invoke(cli, ["tmt", "--tmt-cmd", "custom-tmt", "test"]) + assert result.exit_code == 0 + mock_run_tmt.assert_called_once() + + +def test_drivers_tmt_cli_error_handling(): + """Test CLI error handling when TMT command fails""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + runner = CliRunner() + cli = client.cli(click.Group()) + + # Test CLI with non-zero return code + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 1 # Error return code + result = runner.invoke(cli, ["tmt", "test"]) + assert result.exit_code == 1 + mock_run_tmt.assert_called_once() + + +def test_drivers_tmt_cli_tmt_on_exporter(): + """Test CLI with --tmt-on-exporter flag""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + runner = CliRunner() + cli = client.cli(click.Group()) + + # Test CLI with --tmt-on-exporter flag (should abort) + result = runner.invoke(cli, ["tmt", "--tmt-on-exporter", "test"]) + assert result.exit_code == 1 # click.Abort() returns exit code 1 + assert "TMT will be run on the exporter" in result.output + assert "Aborted!" in result.output + + +def test_drivers_tmt_client_methods(): + """Test the client methods directly""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + # Test run_tmt method with default parameters + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + result = client.run_tmt_local(False, "tmt", None, None, []) + assert result == 0 + mock_run_tmt.assert_called_once() + + # Test run_tmt method with custom parameters + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + result = client.run_tmt_local(True, "custom-tmt", "user", "pass", ["arg1", "arg2"]) + assert result == 0 + mock_run_tmt.assert_called_once() + + +def test_drivers_tmt_run_tmt_with_forward_ssh(): + """Test run_tmt method with SSH forwarding""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + with patch('jumpstarter_driver_tmt.client.TcpPortforwardAdapter') as mock_adapter: + mock_adapter.return_value.__enter__.return_value = ("localhost", 2222) + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + result = client.run_tmt_local(True, "tmt", "user", "pass", ["arg1"]) + assert result == 0 + mock_run_tmt.assert_called_once_with( + "localhost", 2222, "tmt", "user", "pass", "j power cycle", ["arg1"] + ) + + +def test_drivers_tmt_run_tmt_with_direct_address(): + """Test run_tmt method with direct address connection""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + result = client.run_tmt_local(False, "tmt", "user", "pass", ["arg1"]) + assert result == 0 + mock_run_tmt.assert_called_once_with( + "127.0.0.1", 22, "tmt", "user", "pass", "j power cycle", ["arg1"] + ) + + +def test_drivers_tmt_run_tmt_fallback_to_forwarding(): + """Test run_tmt method fallback to SSH forwarding when address() fails""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + # Mock ssh.address() to raise DriverMethodNotImplemented + from jumpstarter.client.core import DriverMethodNotImplemented + client.ssh.address = MagicMock(side_effect=DriverMethodNotImplemented("Method not implemented")) + + with patch('jumpstarter_driver_tmt.client.TcpPortforwardAdapter') as mock_adapter: + mock_adapter.return_value.__enter__.return_value = ("localhost", 2222) + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + result = client.run_tmt_local(False, "tmt", "user", "pass", ["arg1"]) + assert result == 0 + mock_run_tmt.assert_called_once_with( + "localhost", 2222, "tmt", "user", "pass", "j power cycle", ["arg1"] + ) + + +def test_drivers_tmt_run_tmt_internal(): + """Test the internal _run_tmt method""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + with patch('subprocess.run') as mock_subprocess: + mock_result = MagicMock() + mock_result.returncode = 0 + mock_subprocess.return_value = mock_result + + result = client._run_tmt_local("localhost", 2222, "tmt", "user", "pass", "j power cycle", ["test", "arg"]) + + assert result == 0 + mock_subprocess.assert_called_once() + # Verify the command and args passed to subprocess.run + call_args = mock_subprocess.call_args[0][0] + assert call_args[0] == "tmt" + assert "test" in call_args + assert "arg" in call_args + + +def test_drivers_tmt_run_tmt_internal_with_error(): + """Test the internal _run_tmt method with error return code""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + with patch('subprocess.run') as mock_subprocess: + mock_result = MagicMock() + mock_result.returncode = 1 + mock_subprocess.return_value = mock_result + + result = client._run_tmt_local("localhost", 2222, "tmt", "user", "pass", "j power cycle", ["test"]) + + assert result == 1 + mock_subprocess.assert_called_once() + + +def test_drivers_tmt_driver_exports(): + """Test the driver export methods""" + instance = TMT( + children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}, + reboot_cmd="custom reboot", + default_username="testuser", + default_password="testpass" + ) + + with serve(instance) as client: + # Test get_reboot_cmd + reboot_cmd = client.call("get_reboot_cmd") + assert reboot_cmd == "custom reboot" + + # Test get_default_user_pass + username, password = client.call("get_default_user_pass") + assert username == "testuser" + assert password == "testpass" + + +def test_drivers_tmt_driver_defaults(): + """Test the driver with default values""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + # Test default reboot_cmd + reboot_cmd = client.call("get_reboot_cmd") + assert reboot_cmd == "j power cycle" + + # Test default username and password + username, password = client.call("get_default_user_pass") + assert username == "" + assert password == "" + + +def test_drivers_tmt_configuration_error(): + """Test that ConfigurationError is raised when ssh child is missing""" + from jumpstarter.common.exceptions import ConfigurationError + + with pytest.raises(ConfigurationError, match="'ssh' child is required"): + TMT(children={}) + + +def test_replace_provision_args_no_provision(): + """Test replace_provision_args when no provision section exists""" + args = ["discover", "prepare", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, "user", "pass", "j power cycle") + + expected = ["discover", "prepare", "execute"] + assert result == expected + logger.debug.assert_called_with("Provision or run section not found, ignoring") + + +def test_replace_provision_args_with_provision(): + """Test replace_provision_args when provision section exists""" + args = ["discover", "provision", "-h", "old_host", "-g", "old_ip", "-P", "old_port", "prepare", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "new_host", 2222, "new_user", "new_pass", "j power cycle") + + expected = [ + "discover", "provision", "-h", "connect", "-g", "new_host", "-P", "2222", + "-u", "new_user", "-p", "new_pass", "--hard-reboot", "j power cycle", "prepare", "execute" + ] + assert result == expected + + +def test_replace_provision_args_without_username_password(): + """Test replace_provision_args without username and password""" + args = ["discover", "provision", "prepare", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, None, None, "j power cycle") + + expected = [ + "discover", "provision", "-h", "connect", "-g", "host", "-P", "22", + "--hard-reboot", "j power cycle", "prepare", "execute" + ] + assert result == expected + + +def test_replace_provision_args_complex(): + """Test replace_provision_args with complex provision section""" + args = [ + "--root", ".", "-c", "tracing=off", "provision", "-h", "connect", "-g", "192.168.1.1", + "-P", "22", "-u", "root", "-p", "password", "prepare", "execute" + ] + logger = MagicMock() + result = replace_provision_args(logger, args, "new_host", 2222, "new_user", "new_pass", "j power cycle") + + expected = [ + "--root", ".", "-c", "tracing=off", "provision", "-h", "connect", "-g", "new_host", + "-P", "2222", "-u", "new_user", "-p", "new_pass", "--hard-reboot", "j power cycle", "prepare", "execute" + ] + assert result == expected + + +def test_replace_provision_args_with_tmt_run_commands(): + """Test replace_provision_args with TMT run commands in provision section""" + args = ["provision", "plan", "test", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, "user", "pass", "j power cycle") + + expected = [ + "provision", "-h", "connect", "-g", "host", "-P", "22", "-u", "user", "-p", "pass", + "--hard-reboot", "j power cycle", "plan", "test", "execute" + ] + assert result == expected + + +def test_replace_provision_args_with_run_command(): + """Test replace_provision_args with 'run' command (should add provision)""" + args = ["discover", "run", "test", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, "user", "pass", "j power cycle") + + expected = [ + "discover", "run", "test", "execute", "provision", "-h", "connect", "-g", "host", + "-P", "22", "-u", "user", "-p", "pass", "--hard-reboot", "j power cycle" + ] + assert result == expected + logger.debug.assert_called_with("Run section found, adding provision arguments") + + +def test_replace_provision_args_no_provision_no_run(): + """Test replace_provision_args with no provision or run section (should ignore)""" + args = ["discover", "prepare", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, "user", "pass", "j power cycle") + + expected = ["discover", "prepare", "execute"] + assert result == expected + logger.debug.assert_called_with("Provision or run section not found, ignoring") + + +def test_replace_provision_args_with_hard_reboot(): + """Test replace_provision_args with hard reboot command""" + args = ["discover", "provision", "prepare", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, "user", "pass", "custom reboot command") + + expected = [ + "discover", "provision", "-h", "connect", "-g", "host", "-P", "22", + "-u", "user", "-p", "pass", "--hard-reboot", "custom reboot command", "prepare", "execute" + ] + assert result == expected + + +def test_replace_provision_args_without_hard_reboot(): + """Test replace_provision_args without hard reboot command (empty string)""" + args = ["discover", "provision", "prepare", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, "user", "pass", "") + + expected = [ + "discover", "provision", "-h", "connect", "-g", "host", "-P", "22", + "-u", "user", "-p", "pass", "prepare", "execute" + ] + assert result == expected + + +def test_replace_provision_args_without_hard_reboot_none(): + """Test replace_provision_args without hard reboot command (None)""" + args = ["discover", "provision", "prepare", "execute"] + logger = MagicMock() + result = replace_provision_args(logger, args, "host", 22, "user", "pass", None) + + expected = [ + "discover", "provision", "-h", "connect", "-g", "host", "-P", "22", + "-u", "user", "-p", "pass", "prepare", "execute" + ] + assert result == expected + + +def test_drivers_tmt_logging_functionality(): + """Test logging functionality in TMT client""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + with patch('jumpstarter_driver_tmt.client.TcpPortforwardAdapter') as mock_adapter: + mock_adapter.return_value.__enter__.return_value = ("localhost", 2222) + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + + # Test that debug logging is called for SSH port forwarding + with patch.object(client.logger, 'debug') as mock_debug: + client.run_tmt_local(True, "tmt", "user", "pass", ["arg1"]) + mock_debug.assert_any_call("Using SSH port forwarding for TMT connection") + mock_debug.assert_any_call("SSH port forward established - host: localhost, port: 2222") + + # Test that debug logging is called for direct connection + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + with patch.object(client.logger, 'debug') as mock_debug: + client.run_tmt_local(False, "tmt", "user", "pass", ["arg1"]) + mock_debug.assert_any_call("Using direct SSH connection for tmt - host: 127.0.0.1, port: 22") + + # Test that warning logging is called for fallback + from jumpstarter.client.core import DriverMethodNotImplemented + client.ssh.address = MagicMock(side_effect=DriverMethodNotImplemented("Method not implemented")) + + with patch('jumpstarter_driver_tmt.client.TcpPortforwardAdapter') as mock_adapter: + mock_adapter.return_value.__enter__.return_value = ("localhost", 2222) + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + with patch.object(client.logger, 'warning') as mock_warning: + client.run_tmt_local(False, "tmt", "user", "pass", ["arg1"]) + mock_warning.assert_called_once_with( + "Direct address connection failed (Method not implemented), falling back to SSH port forwarding" + ) + + +def test_drivers_tmt_cli_logging(): + """Test logging in CLI functionality""" + instance = TMT(children={"ssh": TcpNetwork(host="127.0.0.1", port=22)}) + + with serve(instance) as client: + runner = CliRunner() + cli = client.cli(click.Group()) + + with patch.object(client, '_run_tmt_local') as mock_run_tmt: + mock_run_tmt.return_value = 0 + with patch.object(client.logger, 'debug') as mock_debug: + result = runner.invoke(cli, ["tmt", "test"]) + assert result.exit_code == 0 + mock_debug.assert_called_with("TMT result: 0") diff --git a/packages/jumpstarter-driver-tmt/pyproject.toml b/packages/jumpstarter-driver-tmt/pyproject.toml new file mode 100644 index 000000000..30278746a --- /dev/null +++ b/packages/jumpstarter-driver-tmt/pyproject.toml @@ -0,0 +1,44 @@ +[project] +name = "jumpstarter-driver-tmt" +dynamic = ["version", "urls"] +description = "Add your description here" +readme = "README.md" +license = "Apache-2.0" +authors = [ + { name = "Miguel Angel Ajo Pelayo", email = "miguelangel@ajo.es" } +] +requires-python = ">=3.11" +dependencies = [ + "anyio>=4.10.0", + "jumpstarter", + "jumpstarter-driver-network", + "jumpstarter-driver-composite", +] + +[tool.hatch.version] +source = "vcs" +raw-options = { 'root' = '../../'} + +[tool.hatch.metadata.hooks.vcs.urls] +Homepage = "https://jumpstarter.dev" +source_archive = "https://github.com/jumpstarter-dev/repo/archive/{commit_hash}.zip" + +[tool.pytest.ini_options] +addopts = "" +log_cli = true +log_cli_level = "INFO" +testpaths = ["jumpstarter_driver_tmt"] +asyncio_mode = "auto" + +[build-system] +requires = ["hatchling", "hatch-vcs", "hatch-pin-jumpstarter"] +build-backend = "hatchling.build" + +[tool.hatch.build.hooks.pin_jumpstarter] +name = "pin_jumpstarter" + +[dependency-groups] +dev = [ + "pytest-cov>=6.0.0", + "pytest>=8.3.3", +] diff --git a/uv.lock b/uv.lock index 1564fc10a..49ec60cca 100644 --- a/uv.lock +++ b/uv.lock @@ -33,6 +33,7 @@ members = [ "jumpstarter-driver-snmp", "jumpstarter-driver-tasmota", "jumpstarter-driver-tftp", + "jumpstarter-driver-tmt", "jumpstarter-driver-uboot", "jumpstarter-driver-ustreamer", "jumpstarter-driver-yepkit", @@ -1987,6 +1988,36 @@ dev = [ { name = "pytest-cov", specifier = ">=6.0.0" }, ] +[[package]] +name = "jumpstarter-driver-tmt" +source = { editable = "packages/jumpstarter-driver-tmt" } +dependencies = [ + { name = "anyio" }, + { name = "jumpstarter" }, + { name = "jumpstarter-driver-composite" }, + { name = "jumpstarter-driver-network" }, +] + +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-cov" }, +] + +[package.metadata] +requires-dist = [ + { name = "anyio", specifier = ">=4.10.0" }, + { name = "jumpstarter", editable = "packages/jumpstarter" }, + { name = "jumpstarter-driver-composite", editable = "packages/jumpstarter-driver-composite" }, + { name = "jumpstarter-driver-network", editable = "packages/jumpstarter-driver-network" }, +] + +[package.metadata.requires-dev] +dev = [ + { name = "pytest", specifier = ">=8.3.3" }, + { name = "pytest-cov", specifier = ">=6.0.0" }, +] + [[package]] name = "jumpstarter-driver-uboot" source = { editable = "packages/jumpstarter-driver-uboot" }