Merge pull request #1166 from cloudbees-oss/AIENG-263

[AIENG-263] prototyping the command line argument/option processing framework that works for us

Author: kohsuke

pyproject.toml

@@ -11,13 +11,13 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = [
-    "typer>=0.19.1",
     "requests>=2.32.5",
     "urllib3>=2.5",
     "junitparser>=4.0.0",
     "more-itertools>=7.1.0",
     "python-dateutil>=2.9",
     "tabulate>=0.9",
+    "click>=8.1,<8.2",
 ]
 dynamic = ["version"]
 

smart_tests/__main__.py

@@ -1,179 +1,60 @@
 import importlib
 import importlib.util
-import logging
-import os
+import sys
 from glob import glob
 from os.path import basename, dirname, join
-from typing import Annotated
-
-import typer
 
 from smart_tests.app import Application
-from smart_tests.commands.detect_flakes import create_nested_command_app as create_detect_flakes_commands
-from smart_tests.commands.record.tests import create_nested_commands as create_record_target_commands
-from smart_tests.commands.subset import create_nested_commands as create_subset_target_commands
-from smart_tests.utils.test_runner_registry import get_registry
-
-from .commands import compare, detect_flakes, inspect, record, stats, subset, verify
-from .utils import logger
-from .utils.env_keys import SKIP_CERT_VERIFICATION
-from .version import __version__
-
-# Load all test runners at module level so they register their commands
-for f in glob(join(dirname(__file__), 'test_runners', "*.py")):
-    f = basename(f)[:-3]
-    if f == '__init__':
-        continue
-    importlib.import_module('smart_tests.test_runners.%s' % f)
-
-# Create initial NestedCommand commands with built-in test runners
-try:
-    create_subset_target_commands()
-    create_record_target_commands()
-    create_detect_flakes_commands()
-except Exception as e:
-    # If NestedCommand creation fails, continue with legacy commands
-    # This ensures backward compatibility
-    logging.warning(f"Failed to create NestedCommand commands at import time: {e}")
-    pass
-
-# Global flag to track if plugins have been loaded and commands need rebuilding
-_plugins_loaded = False
-
-
-def _rebuild_nested_commands_with_plugins():
-    """Rebuild NestedCommand apps after plugins are loaded."""
-    global _plugins_loaded
-    if _plugins_loaded:
-        return  # Already rebuilt
-
-    try:
-        # Clear existing commands from nested apps and rebuild
-        for module_name in ['smart_tests.commands.subset', 'smart_tests.commands.record.tests',
-                            'smart_tests.commands.detect_flakes']:
-            module = importlib.import_module(module_name)
-            if hasattr(module, 'nested_command_app'):
-                nested_app = module.nested_command_app
-                nested_app.registered_commands.clear()
-                nested_app.registered_groups.clear()
-            if hasattr(module, 'create_nested_commands'):
-                module.create_nested_commands()
-
-        _plugins_loaded = True
-        logging.info("Successfully rebuilt NestedCommand apps with plugins")
-
-    except Exception as e:
-        logging.warning(f"Failed to rebuild NestedCommand apps with plugins: {e}")
-        import traceback
-        logging.warning(f"Traceback: {traceback.format_exc()}")
-
-
-# Set up automatic rebuilding when new test runners are registered
-
-
-def _on_test_runner_registered():
-    """Callback triggered when new test runners are registered."""
-    _rebuild_nested_commands_with_plugins()
+from smart_tests.args4p.command import Group
+from smart_tests.commands.compare import compare
+from smart_tests.commands.detect_flakes import detect_flakes
+from smart_tests.commands.inspect import inspect
+from smart_tests.commands.record import record
+from smart_tests.commands.stats import stats
+from smart_tests.commands.subset import subset
+from smart_tests.commands.verify import verify
+
+cli = Group(name="cli", callback=Application)
+cli.add_command(record)
+cli.add_command(subset)
+# TODO: main.add_command(split_subset)
+cli.add_command(verify)
+cli.add_command(inspect)
+cli.add_command(stats)
+cli.add_command(compare)
+cli.add_command(detect_flakes)
+
+
+def _load_test_runners():
+    # load all test runners
+    for f in glob(join(dirname(__file__), 'test_runners', "*.py")):
+        f = basename(f)[:-3]
+        if f == '__init__':
+            continue
+        importlib.import_module(f'smart_tests.test_runners.{f}')
+
+    # load all plugins. Here we do a bit of command line parsing ourselves,
+    # because the command line could look something like `smart-tests record tests myprofile --plugins ...
+    plugin_dir = None
+    if "--plugins" in sys.argv:
+        idx = sys.argv.index("--plugins")
+        if idx + 1 < len(sys.argv):
+            plugin_dir = sys.argv[idx + 1]
 
-
-get_registry().set_on_register_callback(_on_test_runner_registered)
-
-app = typer.Typer()
-
-
-def version_callback(value: bool):
-    if value:
-        typer.echo(f"smart-tests-cli {__version__}")
-        raise typer.Exit()
-
-
-def main(
-    ctx: typer.Context,
-    log_level: Annotated[str, typer.Option(
-        help="Set logger's log level (CRITICAL, ERROR, WARNING, AUDIT, INFO, DEBUG)."
-    )] = logger.LOG_LEVEL_DEFAULT_STR,
-    plugin_dir: Annotated[str | None, typer.Option(
-        "--plugin-dir", "--plugins",
-        help="Directory to load plugins from"
-    )] = None,
-    dry_run: Annotated[bool, typer.Option(
-        help="Dry-run mode. No data is sent to the server. However, sometimes "
-             "GET requests without payload data or side effects could be sent."
-             "note: Since the dry run log is output together with the AUDIT log, "
-             "even if the log-level is set to warning or higher, the log level will "
-             "be forced to be set to AUDIT."
-    )] = False,
-    skip_cert_verification: Annotated[bool, typer.Option(
-        help="Skip the SSL certificate check. This lets you bypass system setup issues "
-             "like CERTIFICATE_VERIFY_FAILED, at the expense of vulnerability against "
-             "a possible man-in-the-middle attack. Use it as an escape hatch, but with caution."
-    )] = False,
-    version: Annotated[bool | None, typer.Option(
-        "--version", help="Show version and exit", callback=version_callback, is_eager=True
-    )] = None,
-):
-    level = logger.get_log_level(log_level)
-    # In the case of dry-run, it is forced to set the level below the AUDIT.
-    # This is because the dry-run log will be output along with the audit log.
-    if dry_run and level > logger.LOG_LEVEL_AUDIT:
-        level = logger.LOG_LEVEL_AUDIT
-
-    if not skip_cert_verification:
-        skip_cert_verification = (os.environ.get(SKIP_CERT_VERIFICATION) is not None)
-
-    logging.basicConfig(level=level)
-
-    # load all plugins
     if plugin_dir:
         for f in glob(join(plugin_dir, '*.py')):
             spec = importlib.util.spec_from_file_location(
                 f"smart_tests.plugins.{basename(f)[:-3]}", f)
-            if spec is None:
-                raise ImportError(f"Failed to create module spec for plugin: {f}")
-            if spec.loader is None:
-                raise ImportError(f"Plugin spec has no loader: {f}")
             plugin = importlib.util.module_from_spec(spec)
             spec.loader.exec_module(plugin)
 
-    # After loading plugins, rebuild NestedCommand apps to include plugin commands
-    if plugin_dir:
-        _rebuild_nested_commands_with_plugins()
-
-    ctx.obj = Application(dry_run=dry_run, skip_cert_verification=skip_cert_verification)
-
-
-# Use NestedCommand apps if available, otherwise fall back to legacy
-try:
-    from smart_tests.commands.detect_flakes import nested_command_app as detect_flakes_target_app
-    from smart_tests.commands.record.tests import nested_command_app as record_target_app
-    from smart_tests.commands.subset import nested_command_app as subset_target_app
-
-    app.add_typer(record.app, name="record")
-    app.add_typer(subset_target_app, name="subset")  # Use NestedCommand version
-    app.add_typer(verify.app, name="verify")
-    app.add_typer(inspect.app, name="inspect")
-    app.add_typer(stats.app, name="stats")
-    app.add_typer(compare.app, name="compare")
-    app.add_typer(detect_flakes_target_app, name="detect-flakes")
 
-    # Add record-target as a sub-app to record command
-    record.app.add_typer(record_target_app, name="tests")
+_load_test_runners()
 
-except Exception as e:
-    logging.warning(f"Failed to use NestedCommand apps at init: {e}")
-    # Fallback to original structure
-    app.add_typer(record.app, name="record")
-    app.add_typer(subset.app, name="subset")
-    app.add_typer(verify.app, name="verify")
-    app.add_typer(inspect.app, name="inspect")
-    app.add_typer(stats.app, name="stats")
-    app.add_typer(detect_flakes.app, name="detect-flakes")
 
-app.callback()(main)
+def main():
+    cli.main()
 
-# For backward compatibility with tests that expect a Click CLI
-# We'll need to use Typer's testing utilities instead
-main = app
 
 if __name__ == '__main__':
-    app()
+    main()

smart_tests/app.py

@@ -2,10 +2,66 @@
 # Currently it's used to keep global configurations.
 #
 # From command implementations, this is available via dependency injection
+
+import logging
+import os
+from typing import Annotated
+
+import click
+
+import smart_tests.args4p.typer as typer
+from smart_tests.utils import logger
+from smart_tests.utils.env_keys import SKIP_CERT_VERIFICATION
+from smart_tests.version import __version__
+
+
 class Application:
-    def __init__(self, dry_run: bool = False, skip_cert_verification: bool = False):
+    # Group commands that take the CLI profile as a sub-command shall set this parameter
+    test_runner: str | None = None
+
+    # this maps to the main entry point of the CLI command
+    def __init__(
+            self,
+            log_level: Annotated[str, typer.Option(
+                help="Set logger's log level (CRITICAL, ERROR, WARNING, AUDIT, INFO, DEBUG)."
+            )] = logger.LOG_LEVEL_DEFAULT_STR,
+            plugin_dir: Annotated[str | None, typer.Option(
+                "--plugins",
+                help="Directory to load plugins from"
+            )] = None,
+            dry_run: Annotated[bool, typer.Option(
+                help="Dry-run mode. No data is sent to the server. However, sometimes "
+                     "GET requests without payload data or side effects could be sent."
+                     "note: Since the dry run log is output together with the AUDIT log, "
+                     "even if the log-level is set to warning or higher, the log level will "
+                     "be forced to be set to AUDIT."
+            )] = False,
+            skip_cert_verification: Annotated[bool, typer.Option(
+                help="Skip the SSL certificate check. This lets you bypass system setup issues "
+                     "like CERTIFICATE_VERIFY_FAILED, at the expense of vulnerability against "
+                     "a possible man-in-the-middle attack. Use it as an escape hatch, but with caution."
+            )] = False,
+            version: Annotated[bool, typer.Option(
+                "--version", help="Show version and exit"
+            )] = False,
+    ):
+        if version:
+            click.echo(f"smart-tests-cli {__version__}")
+            raise typer.Exit(0)
+
+        level = logger.get_log_level(log_level)
+        # In the case of dry-run, it is forced to set the level below the AUDIT.
+        # This is because the dry-run log will be output along with the audit log.
+        if dry_run and level > logger.LOG_LEVEL_AUDIT:
+            level = logger.LOG_LEVEL_AUDIT
+        logging.basicConfig(level=level)
+
+        # plugin_dir is processed earlier. If we do it here, it's too late
+
         # Dry run mode. This command is used by customers to inspect data we'd send to our server,
         # but without actually doing so.
         self.dry_run = dry_run
-        # Skip SSL certificate validation
+
+        if not skip_cert_verification:
+            skip_cert_verification = (os.environ.get(SKIP_CERT_VERIFICATION) is not None)
         self.skip_cert_verification = skip_cert_verification