Add test filtering to e2e tests

tests/infra/runner.py

@@ -14,8 +14,8 @@
 from typing import List
 import sys
 import better_exceptions
-import re
 import infra.bencher
+import infra.test_filter
 import os
 
 from loguru import logger as LOG
@@ -216,12 +216,6 @@ def add(parser):
                 help="List all sub-tests without executing",
                 action="store_true",
             )
-            parser.add_argument(
-                "-R",
-                "--regex",
-                help="Run sub-tests whose name includes this string",
-                metavar="<string>",
-            )
             if add_options:
                 add_options(parser)
 
@@ -245,12 +239,18 @@ def run(self, max_concurrent=None):
         }
         LOG.configure(**config)
 
-        if self.args.regex:
+        runner_filter = infra.test_filter.get_runner_filter()
+        if runner_filter:
             self.threads = [
                 thread
                 for thread in self.threads
-                if re.compile(self.args.regex).search(thread.name)
+                if runner_filter == thread.name
             ]
+            if not self.threads:
+                raise RuntimeError(
+                    f'{infra.test_filter._ENV_VAR}="{infra.test_filter.get_filter()}" '
+                    f"matched no runner threads. Check for typos."
+                )
 
         if self.args.show_only:
             for thread in self.threads:
@@ -279,5 +279,7 @@ def run(self, max_concurrent=None):
             for thread in group:
                 thread.join()
 
+        infra.test_filter.check_any_matched()
+
         if FAILURES:
             raise RuntimeError(FAILURES)

tests/infra/test_filter.py

@@ -0,0 +1,110 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the Apache 2.0 License.
+
+"""
+Hierarchical test filtering via the CCF_TEST_FILTER environment variable.
+
+The filter uses a slash-separated hierarchy:
+  CCF_TEST_FILTER=operations           - run all tests in the "operations" runner thread
+  CCF_TEST_FILTER=operations/snapshot  - run tests in "operations" whose name contains "snapshot"
+
+The first segment is matched exactly against ConcurrentRunner thread names (the
+prefix passed to cr.add()).  The second segment (if present) uses
+case-insensitive substring matching against individual test function names and
+is checked inside the @reqs.description decorator.
+
+When a test-level filter is active, at least one test must match.  Call
+check_any_matched() at the end of a run to raise if nothing ran (likely a
+typo).  It also logs the list of tests that were executed.
+"""
+
+import os
+import threading
+
+from loguru import logger as LOG
+
+_ENV_VAR = "CCF_TEST_FILTER"
+
+# Track which test functions were executed while filtering was active.
+# Access is guarded by a lock since tests may run on multiple threads.
+_lock = threading.Lock()
+_matched_tests = []
+
+
+def get_filter():
+    """Return the raw filter string, or None if unset."""
+    return os.getenv(_ENV_VAR)
+
+
+def get_runner_filter():
+    """Return the first segment of the filter (runner/thread level), or None."""
+    f = get_filter()
+    if f is None:
+        return None
+    return f.split("/")[0]
+
+
+def get_test_filter():
+    """Return the second segment of the filter (test-function level), or None."""
+    f = get_filter()
+    if f is None:
+        return None
+    parts = f.split("/", 1)
+    if len(parts) < 2:
+        return None
+    return parts[1]
+
+
+def should_skip_test(func_name):
+    """
+    Return True if the current test function should be skipped based on the
+    test-level (second segment) filter.  Returns False (don't skip) when no
+    filter is set or no second segment is present.
+    """
+    substring = get_test_filter()
+    if substring is None:
+        return False
+    return substring.lower() not in func_name.lower()
+
+
+def record_match(func_name):
+    """Record that a test function was executed."""
+    with _lock:
+        _matched_tests.append(func_name)
+
+
+def check_any_matched():
+    """
+    If a test-level filter was active (second segment of CCF_TEST_FILTER),
+    assert that at least one test ran.  Call this at the end of a test run.
+    Raises RuntimeError on mismatch (likely a typo in the filter).
+
+    A runner-only filter (e.g. CCF_TEST_FILTER=schema) does not require
+    decorated sub-tests to have matched, since the runner thread itself
+    is the unit being selected.
+
+    When tests did run, logs the list of executed tests.
+    """
+    with _lock:
+        matched = list(_matched_tests)
+
+    if matched:
+        LOG.info(f"Executed {len(matched)} test(s):")
+        for name in matched:
+            LOG.info(f"  - {name}")
+
+    if get_test_filter() is None:
+        return
+
+    f = get_filter()
+    if not matched:
+        raise RuntimeError(
+            f'{_ENV_VAR}="{f}" was set but no tests matched. '
+            f"Check for typos in the filter value."
+        )
+
+
+def reset():
+    """Reset match tracking (mainly useful for tests of the filter itself)."""
+    with _lock:
+        _matched_tests.clear()

tests/suite/test_requirements.py

@@ -3,6 +3,7 @@
 
 import functools
 
+import infra.test_filter
 from infra.snp import SNP_SUPPORT
 from loguru import logger as LOG
 from infra.member import RecoveryRole
@@ -16,6 +17,13 @@ def description(desc):
     def decorator(func):
         @functools.wraps(func)
         def wrapper(*args, **kwargs):
+            if infra.test_filter.should_skip_test(func.__name__):
+                LOG.info(
+                    f'Skipping "{func.__name__}" (filtered by {infra.test_filter._ENV_VAR})'
+                )
+                # Return first arg (network) unchanged to preserve chaining
+                return args[0] if args else None
+            infra.test_filter.record_match(func.__name__)
             LOG.opt(colors=True, depth=1).info(
                 f'<magenta>Test: {desc} {(kwargs or "")}</>'
             )