Support replay-only AuTest YAML tests

tests/README.md

@@ -34,8 +34,9 @@ To run autest again, or to run individual tests, the cmake build generates a hel
 
 To run a single test, you can use the `--filter` flag to name
 which test to run. The tests are in files whose names are the test name
-, and are suffixed with `.test.py`. Thus, the `something_descriptive`
-test will be specified in a file named `something_descriptive.test.py`.
+, and are typically suffixed with `.test.py` or `.test.yaml`. Thus, the
+`something_descriptive` test will be specified in a file named
+`something_descriptive.test.py` or `something_descriptive.test.yaml`.
 The corresponding `autest.sh` command is:
 
     $ ./autest.sh --filter=something_descriptive

tests/autest-parallel.py.in

@@ -45,6 +45,7 @@ from typing import Dict, List, Optional, Tuple
 DEFAULT_SERIAL_TESTS_FILE = Path("${CMAKE_CURRENT_SOURCE_DIR}") / "serial_tests.txt"
 # Default estimate for unknown tests (seconds)
 DEFAULT_TEST_TIME = 15.0
+KNOWN_TEST_SUFFIXES = ('.test.yaml', '.test.yml', '.test.py', '.test')
 
 
 @dataclass
@@ -66,40 +67,87 @@ class TestResult:
     is_serial: bool = False
 
 
-def discover_tests(test_dir: Path, filter_patterns: Optional[List[str]] = None) -> List[str]:
+def normalize_test_name(test_name: str) -> str:
+    """Normalize a test path or file name to the autest test name."""
+    name = Path(test_name).name
+    for suffix in KNOWN_TEST_SUFFIXES:
+        if name.endswith(suffix):
+            return name[:-len(suffix)]
+    return Path(name).stem.replace('.test', '')
+
+
+def parse_autest_list_output(output: str) -> List[str]:
+    """Extract the discovered test names from `autest list --json` output."""
+    clean_output = strip_ansi(output)
+    match = re.search(r'(\[[\s\S]*\])', clean_output)
+    if match is None:
+        raise ValueError(f"Could not find JSON in autest list output:\n{clean_output}")
+
+    entries = json.loads(match.group(1))
+    return sorted(entry['name'] for entry in entries if 'name' in entry)
+
+
+def discover_tests(
+        test_dir: Path,
+        filter_patterns: Optional[List[str]] = None,
+        script_dir: Optional[Path] = None,
+        ats_bin: Optional[str] = None,
+        build_root: Optional[str] = None,
+        extra_args: Optional[List[str]] = None) -> List[str]:
     """
-    Discover all .test.py files in the test directory.
+    Discover tests via `autest list --json`.
 
     Args:
         test_dir: Path to gold_tests directory
         filter_patterns: Optional list of glob patterns to filter tests
+        script_dir: Directory in which to invoke autest
+        ats_bin: Optional ATS bin directory to pass through to autest list
+        build_root: Optional build root to pass through to autest list
+        extra_args: Additional autest CLI arguments to pass through
 
     Returns:
-        List of test names (without .test.py extension)
+        List of discovered test names
     """
-    tests = []
-    for test_file in test_dir.rglob("*.test.py"):
-        # Extract test name (filename without .test.py)
-        test_name = test_file.stem.replace('.test', '')
-
-        # Apply filters if provided
-        if filter_patterns:
-            if any(fnmatch.fnmatch(test_name, pattern) for pattern in filter_patterns):
-                tests.append(test_name)
-        else:
-            tests.append(test_name)
+    cmd = [
+        'uv',
+        'run',
+        'autest',
+        'list',
+        '--directory',
+        str(test_dir),
+        '--json',
+    ]
+
+    if ats_bin:
+        cmd.extend(['--ats-bin', ats_bin])
+    if build_root:
+        cmd.extend(['--build-root', build_root])
+    if filter_patterns:
+        cmd.append('--filters')
+        cmd.extend(filter_patterns)
+    if extra_args:
+        cmd.extend(extra_args)
+
+    proc = subprocess.run(
+        cmd,
+        cwd=script_dir,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
 
-    return sorted(tests)
+    output = proc.stdout + proc.stderr
+    return parse_autest_list_output(output)
 
 
 def load_serial_tests(serial_file: Path) -> set:
     """
     Load list of tests that must run serially from a file.
 
     The file format is one test name per line, with # for comments.
-    Test names can be full paths like ``subdir/test_name.test.py``.
-    The .test.py extension is stripped, and only the basename (stem) is
-    used for matching against discovered test names.
+    Test names can be full paths like ``subdir/test_name.test.py`` or
+    ``subdir/test_name.test.yaml``. Known test suffixes are stripped before
+    matching against discovered test names.
 
     Returns:
         Set of test base names that must run serially
@@ -115,12 +163,7 @@ def load_serial_tests(serial_file: Path) -> set:
                 # Skip empty lines and comments
                 if not line or line.startswith('#'):
                     continue
-                # Remove .test.py extension if present
-                if line.endswith('.test.py'):
-                    line = line[:-8]  # Remove .test.py
-                # Extract just the test name from path
-                test_name = Path(line).stem.replace('.test', '')
-                serial_tests.add(test_name)
+                serial_tests.add(normalize_test_name(line))
     except IOError:
         pass  # File is optional; missing file means no serial tests
 
@@ -801,7 +844,7 @@ Examples:
     %(prog)s -j 2 --filter "cache-*" --filter "tls-*" --ats-bin /opt/ats/bin --sandbox /tmp/autest
 
     # List tests without running
-    %(prog)s --list --ats-bin /opt/ats/bin
+    %(prog)s --list
 
     # Collect timing data (runs tests one at a time for accurate timing)
     %(prog)s -j 4 --collect-timings --ats-bin /opt/ats/bin --sandbox /tmp/autest
@@ -873,7 +916,7 @@ Examples:
         print(f"Loaded {len(serial_tests)} serial tests from {args.serial_tests_file}")
 
     # Discover tests
-    all_tests = discover_tests(test_dir, args.filters)
+    all_tests = discover_tests(test_dir, args.filters, script_dir, args.ats_bin, build_root, args.extra_args)
 
     if not all_tests:
         print("No tests found matching the specified filters.", file=sys.stderr)

tests/gold_tests/autest-site/ats_replay.test.ext

@@ -22,6 +22,8 @@ import os
 import re
 import yaml
 
+ATS_REPLAY_TEST_EXTENSIONS = ['.test.yaml', '.test.yml']
+
 
 def configure_ats(obj: 'TestRun', server: 'Process', ats_config: dict, dns: Optional['Process'] = None):
     '''Configure ATS per the configuration in the replay file.
@@ -205,24 +207,43 @@ def _requires_persistent_ats(ats_config: dict) -> bool:
     return bool(ats_config.get('metric_checks'))
 
 
-def ATSReplayTest(obj, replay_file: str):
-    '''Create a TestRun that configures ATS and runs HTTP traffic using the replay file.
+def _is_list_action(obj) -> bool:
+    return obj.Variables.Autest.Action == 'list'
+
+
+def _get_replay_path(obj, replay_file: str) -> str:
+    return replay_file if os.path.isabs(replay_file) else os.path.join(obj.TestDirectory, replay_file)
 
-    :param obj: The Test object to add the test run to.
-    :param replay_file: Replay file specifying the test configuration and test traffic.
-    :returns: The TestRun object.
-    '''
 
-    replay_path = replay_file if os.path.isabs(replay_file) else os.path.join(obj.TestDirectory, replay_file)
+def _load_replay_config(obj, replay_file: str):
+    replay_path = _get_replay_path(obj, replay_file)
     with open(replay_path, 'r') as f:
         replay_config = yaml.safe_load(f)
 
+    if not isinstance(replay_config, dict):
+        raise ValueError(f"Replay file {replay_file} does not contain a YAML mapping")
+
     # The user must specify the 'autest' node.
     if not 'autest' in replay_config:
         raise ValueError(f"Replay file {replay_file} does not contain 'autest' section")
     autest_config = replay_config['autest']
 
-    tr = obj.AddTestRun(autest_config['description'])
+    if not isinstance(autest_config, dict):
+        raise ValueError(f"Replay file {replay_file} does not contain a mapping in 'autest' section")
+
+    return replay_config, autest_config
+
+
+def _get_replay_summary(autest_config: dict):
+    return autest_config.get('summary', autest_config.get('description'))
+
+
+def _build_ats_replay_test(obj, replay_file: str, autest_config: dict):
+    description = autest_config.get('description', _get_replay_summary(autest_config))
+    if description is None:
+        raise ValueError(f"Replay file {replay_file} does not contain 'autest.description' or 'autest.summary'")
+
+    tr = obj.AddTestRun(description)
 
     # Copy the specified files and directories down.
     tr.Setup.Copy(replay_file, tr.RunDirectory)
@@ -346,4 +367,33 @@ def ATSReplayTest(obj, replay_file: str):
     return tr
 
 
+def _load_ats_replay(obj, replay_file: str):
+    replay_config, autest_config = _load_replay_config(obj, replay_file)
+
+    summary = _get_replay_summary(autest_config)
+    if summary and not obj.Summary:
+        obj.Summary = summary
+
+    if _is_list_action(obj):
+        return None
+
+    return _build_ats_replay_test(obj, replay_file, autest_config)
+
+
+def _load_ats_replay_test_file(obj):
+    return _load_ats_replay(obj, obj.TestFile)
+
+
+def ATSReplayTest(obj, replay_file: str):
+    '''Create a TestRun that configures ATS and runs HTTP traffic using the replay file.
+
+    :param obj: The Test object to add the test run to.
+    :param replay_file: Replay file specifying the test configuration and test traffic.
+    :returns: The TestRun object.
+    '''
+
+    return _load_ats_replay(obj, replay_file)
+
+
 ExtendTest(ATSReplayTest, name="ATSReplayTest")
+RegisterTestFormat(_load_ats_replay_test_file, "ATSReplayYAMLTest", ext=ATS_REPLAY_TEST_EXTENSIONS)

tests/gold_tests/autest-site/init.cli.ext

@@ -38,7 +38,9 @@ if found_microserver_version < needed_microserver_version:
         "Please update MicroServer:\n  rm -rf .venv && uv sync\n",
         show_stack=False)
 
-Settings.path_argument(["--ats-bin"], required=True, help="A user provided directory to ATS bin")
+# --ats-bin is needed for running the tests, but if the user passes
+# `autest list` to list the tests, it is not required for that.
+Settings.path_argument(["--ats-bin"], required=False, help="A user provided directory to ATS bin")
 
 Settings.path_argument(["--build-root"], required=False, help="The location of the build root for out of source builds")
 

tests/gold_tests/autest-site/setup.cli.ext

@@ -27,10 +27,11 @@ PROXY_VERIFIER_VERSION_FILENAME = 'proxy-verifier-version.txt'
 
 test_root = dirname(dirname(AutestSitePath))
 repo_root = dirname(test_root)
+is_list_action = Arguments.subcommand == 'list'
+ats_bin = Arguments.ats_bin
 
-if Arguments.ats_bin is not None:
-    # Add environment variables
-    ENV['ATS_BIN'] = Arguments.ats_bin
+if ats_bin is not None:
+    ENV['ATS_BIN'] = ats_bin
 
 if Arguments.build_root is not None:
     ENV['BUILD_ROOT'] = Arguments.build_root
@@ -59,34 +60,35 @@ else:
         if path_search is not None:
             ENV['VERIFIER_BIN'] = dirname(path_search)
             host.WriteVerbose(['ats'], "Using Proxy Verifier found in PATH: ", ENV['VERIFIER_BIN'])
-        else:
+        elif not is_list_action:
             prepare_proxy_verifier_path = os.path.join(test_root, "prepare_proxy_verifier.sh")
             host.WriteError("Could not find Proxy Verifier binaries. "
                             "Try running: ", prepare_proxy_verifier_path)
 
-required_pv_version = Version(proxy_verifer_version[1:])
-verifier_client = os.path.join(ENV['VERIFIER_BIN'], 'verifier-client')
-pv_version_out = subprocess.check_output([verifier_client, "--version"])
-pv_version = Version(pv_version_out.decode("utf-8").split()[1])
-if pv_version < required_pv_version:
-    host.WriteError(
-        f"Proxy Verifier at {verifier_client} is too old. "
-        f"Version required: {required_pv_version}, version found: {pv_version}")
-else:
-    host.WriteVerbose(['ats'], f"Proxy Verifier at {verifier_client} has version: {pv_version}")
+if ENV.get('VERIFIER_BIN') is not None and not is_list_action:
+    required_pv_version = Version(proxy_verifer_version[1:])
+    verifier_client = os.path.join(ENV['VERIFIER_BIN'], 'verifier-client')
+    pv_version_out = subprocess.check_output([verifier_client, "--version"])
+    pv_version = Version(pv_version_out.decode("utf-8").split()[1])
+    if pv_version < required_pv_version:
+        host.WriteError(
+            f"Proxy Verifier at {verifier_client} is too old. "
+            f"Version required: {required_pv_version}, version found: {pv_version}")
+    else:
+        host.WriteVerbose(['ats'], f"Proxy Verifier at {verifier_client} has version: {pv_version}")
 
-if ENV['ATS_BIN'] is not None:
+if ats_bin is not None:
     # Add variables for Tests
-    traffic_layout = os.path.join(ENV['ATS_BIN'], "traffic_layout")
-    if not os.path.isdir(ENV['ATS_BIN']):
+    traffic_layout = os.path.join(ats_bin, "traffic_layout")
+    if not os.path.isdir(ats_bin):
         host.WriteError("--ats-bin requires a directory", show_stack=False)
     # setting up data from traffic_layout
     # this is getting layout structure
     if not os.path.isfile(traffic_layout):
         hint = ''
-        if os.path.isfile(os.path.join(ENV['ATS_BIN'], 'bin', 'traffic_layout')):
+        if os.path.isfile(os.path.join(ats_bin, 'bin', 'traffic_layout')):
             hint = "\nDid you mean '--ats-bin {}'?".\
-                format(os.path.join(ENV['ATS_BIN'], 'bin'))
+                format(os.path.join(ats_bin, 'bin'))
         host.WriteError("traffic_layout is not found. Aborting tests - Bad build or install.{}".format(hint), show_stack=False)
     try:
         out = subprocess.check_output([traffic_layout, "--json"])
@@ -109,11 +111,13 @@ if ENV['ATS_BIN'] is not None:
     out = Version(out.decode("utf-8").split("-")[2].strip())
     Variables.trafficserver_version = out
     host.WriteVerbose(['ats'], "Traffic server version:", out)
+elif not is_list_action:
+    host.WriteError("--ats-bin is required to run tests", show_stack=False)
 
 Variables.AtsExampleDir = os.path.join(AutestSitePath, '..', '..', '..', 'example')
 Variables.AtsToolsDir = os.path.join(AutestSitePath, '..', '..', '..', 'tools')
 Variables.AtsTestToolsDir = os.path.join(AutestSitePath, '..', '..', 'tools')
-Variables.VerifierBinPath = ENV['VERIFIER_BIN']
+Variables.VerifierBinPath = ENV.get('VERIFIER_BIN')
 Variables.BuildRoot = ENV['BUILD_ROOT']
 Variables.RepoDir = repo_root
 Variables.AtsTestPluginsDir = os.path.join(Variables.BuildRoot, 'tests', 'tools', 'plugins', '.libs')