From e36c52cf26bdfd835ac04b0148fab8dc9faa619e Mon Sep 17 00:00:00 2001 From: Kai Chappell Date: Wed, 3 Dec 2025 00:33:36 +0000 Subject: [PATCH] Add test execution CLI commands --- src/py_dvt_ate/app/cli.py | 42 +++++++ src/py_dvt_ate/app/test_commands.py | 175 ++++++++++++++++++++++++++++ 2 files changed, 217 insertions(+) create mode 100644 src/py_dvt_ate/app/test_commands.py diff --git a/src/py_dvt_ate/app/cli.py b/src/py_dvt_ate/app/cli.py index 4baf5c6..0b5d16a 100644 --- a/src/py_dvt_ate/app/cli.py +++ b/src/py_dvt_ate/app/cli.py @@ -83,5 +83,47 @@ def serve( ) +@app.command(name="list-tests") +def list_tests_cmd() -> None: + """List all available DVT tests.""" + from py_dvt_ate.app.test_commands import list_tests + + list_tests() + + +@app.command(name="run-test") +def run_test_cmd( + test_name: Annotated[ + str, + typer.Argument(help="Name of the test to run (use list-tests to see available tests)."), + ], + config_file: Annotated[ + str | None, + typer.Option("--config", "-c", help="Path to configuration YAML file."), + ] = None, + operator: Annotated[ + str | None, + typer.Option("--operator", "-o", help="Operator identifier (e.g., email address)."), + ] = None, + description: Annotated[ + str | None, + typer.Option("--description", "-d", help="Test run description."), + ] = None, +) -> None: + """Run a specific DVT test. + + The test will connect to instruments based on the configuration file + (default: config/default.yaml). Results are stored in the data directory. + """ + from py_dvt_ate.app.test_commands import run_test + + run_test( + test_name=test_name, + config_file=config_file, + operator=operator, + description=description, + ) + + if __name__ == "__main__": app() diff --git a/src/py_dvt_ate/app/test_commands.py b/src/py_dvt_ate/app/test_commands.py new file mode 100644 index 0000000..9ec8a55 --- /dev/null +++ b/src/py_dvt_ate/app/test_commands.py @@ -0,0 +1,175 @@ +"""Test execution commands for CLI.""" + +import importlib +import inspect +from pathlib import Path + +import typer + +from py_dvt_ate.app.config import load_config +from py_dvt_ate.data.repository import SQLiteRepository +from py_dvt_ate.framework.context import ITest +from py_dvt_ate.framework.runner import TestRunner +from py_dvt_ate.instruments.factory import InstrumentConfig, InstrumentFactory + + +def _discover_tests() -> dict[str, type]: + """Discover all available tests by scanning the tests package. + + Returns: + Dictionary mapping test names to test classes. + """ + tests: dict[str, type] = {} + + # Find the tests package directory + import py_dvt_ate.tests + + tests_pkg_path = Path(py_dvt_ate.tests.__file__).parent + + # Scan all Python files in the tests package + for py_file in tests_pkg_path.rglob("*.py"): + if py_file.name.startswith("_"): + continue + + # Convert file path to module name + rel_path = py_file.relative_to(tests_pkg_path.parent) + module_name = "py_dvt_ate." + str(rel_path.with_suffix("")).replace("/", ".").replace("\\", ".") + + try: + module = importlib.import_module(module_name) + + # Find all classes that implement ITest + for _name, obj in inspect.getmembers(module, inspect.isclass): + if ( + obj is not ITest + and issubclass(obj, ITest) + and not inspect.isabstract(obj) + and hasattr(obj, "name") + ): + # Create instance to get the name property + instance = obj() + tests[instance.name] = obj + + except (ImportError, AttributeError): + continue + + return tests + + +def list_tests() -> None: + """List all available DVT tests.""" + tests = _discover_tests() + + if not tests: + typer.echo("No tests found.") + return + + typer.echo("Available DVT tests:") + typer.echo("") + + for test_name in sorted(tests.keys()): + test_class = tests[test_name] + instance = test_class() + typer.echo(f" {test_name:15s} {instance.description}") + + +def run_test( + test_name: str, + config_file: str | None = None, + operator: str | None = None, + description: str | None = None, +) -> None: + """Run a specific DVT test. + + Args: + test_name: Name of the test to run. + config_file: Path to configuration YAML file. + operator: Operator identifier (e.g., email address). + description: Test run description. + """ + # Discover available tests + tests = _discover_tests() + + if test_name not in tests: + typer.echo(f"Error: Test \'{test_name}\' not found.", err=True) + typer.echo("", err=True) + typer.echo("Available tests:", err=True) + for name in sorted(tests.keys()): + typer.echo(f" - {name}", err=True) + raise typer.Exit(code=1) + + # Load configuration + config_path = config_file or "config/default.yaml" + try: + config = load_config(config_path) + except FileNotFoundError as err: + typer.echo(f"Error: Configuration file not found: {config_path}", err=True) + typer.echo("Run with --config to specify a different config file.", err=True) + raise typer.Exit(code=1) from err + except Exception as e: + typer.echo(f"Error loading configuration: {e}", err=True) + raise typer.Exit(code=1) from e + + # Create repository + try: + repository = SQLiteRepository(config.data.database_path) + except Exception as e: + typer.echo(f"Error initialising repository: {e}", err=True) + raise typer.Exit(code=1) from e + + # Create instruments + typer.echo(f"Connecting to instruments ({config.instruments.backend})...") + try: + # Convert AppConfig to InstrumentConfig + inst_config = InstrumentConfig( + backend=config.instruments.backend, + simulator_host=config.instruments.simulator.host, + chamber_port=config.instruments.simulator.thermal_chamber_port, + psu_port=config.instruments.simulator.power_supply_port, + dmm_port=config.instruments.simulator.multimeter_port, + chamber_visa=config.instruments.pyvisa.thermal_chamber, + psu_visa=config.instruments.pyvisa.power_supply, + dmm_visa=config.instruments.pyvisa.multimeter, + ) + instruments = InstrumentFactory.create(inst_config) + except Exception as e: + typer.echo(f"Error connecting to instruments: {e}", err=True) + raise typer.Exit(code=1) from e + + # Create test instance + test_class = tests[test_name] + test = test_class() + + # Run test + typer.echo(f"Running test: {test.name}") + typer.echo(f"Description: {test.description}") + typer.echo("") + + try: + runner = TestRunner(repository) + run_id = runner.run_test( + test=test, + instruments=instruments, + operator=operator, + description=description, + ) + + # Retrieve final status + run = repository.get_run(run_id) + typer.echo("") + typer.echo(f"Test completed: {run.status.value}") + typer.echo(f"Run ID: {run_id}") + + # Exit with appropriate code + if run.status.value == "PASSED": + raise typer.Exit(code=0) + else: + raise typer.Exit(code=1) + + except KeyboardInterrupt: + typer.echo("") + typer.echo("Test interrupted by user.") + raise typer.Exit(code=130) from None + except Exception as e: + typer.echo(f"Error running test: {e}", err=True) + raise typer.Exit(code=1) from e