This is an automated email from the ASF dual-hosted git repository. kpvdr pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/qpid-interop-test.git
commit fcaf7330989142bbc60724bd8311ed4b1e9ee643 Author: QIT Development Team <[email protected]> AuthorDate: Wed Aug 5 13:02:26 2026 -0400 Add parallel test execution to orchestrator (-j/--workers) ThreadPoolExecutor runs test cases concurrently (each uses a unique queue name so they don't interfere). Default remains sequential (-j 1). Local benchmark: 25 tests in 13s parallel vs 39s sequential (3x speedup). Co-Authored-By: Claude Opus 4.6 <[email protected]> --- src/qit/cli/main.py | 9 +++++ src/qit/core/orchestrator.py | 90 +++++++++++++++++++++++++++++++++++++------- 2 files changed, 86 insertions(+), 13 deletions(-) diff --git a/src/qit/cli/main.py b/src/qit/cli/main.py index 266a8b3..04a4dc9 100644 --- a/src/qit/cli/main.py +++ b/src/qit/cli/main.py @@ -122,6 +122,13 @@ def test() -> None: is_flag=True, help="Treat known failures (xfail) as real failures", ) [email protected]( + "--workers", + "-j", + type=int, + default=1, + help="Number of parallel test workers (default: 1 = sequential)", +) def test_amqp_types( sender: tuple[str, ...], receiver: tuple[str, ...], @@ -132,6 +139,7 @@ def test_amqp_types( junit_xml: str | None, extended: bool, strict: bool, + workers: int, ) -> None: """Test AMQP primitive and complex types interoperability.""" from pathlib import Path @@ -260,6 +268,7 @@ def test_amqp_types( amqp_types=test_types, sender_shims=sender_shims, receiver_shims=receiver_shims, + workers=workers, ) # Print report diff --git a/src/qit/core/orchestrator.py b/src/qit/core/orchestrator.py index 2ebabab..fa8ca42 100644 --- a/src/qit/core/orchestrator.py +++ b/src/qit/core/orchestrator.py @@ -4,6 +4,7 @@ Test orchestration engine. Coordinates shim execution, message comparison, and result reporting. """ +from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from itertools import product from typing import Any @@ -54,6 +55,7 @@ class Orchestrator: amqp_types: dict[str, list[Any]], sender_shims: list[str] | None = None, receiver_shims: list[str] | None = None, + workers: int = 1, ) -> list[TestResult]: """ Run full test matrix: all sender × receiver × type combinations. @@ -62,6 +64,7 @@ class Orchestrator: amqp_types: Map of type name to list of test values sender_shims: List of sender shim names (default: all shims) receiver_shims: List of receiver shim names (default: all shims) + workers: Number of parallel workers (1 = sequential) Returns: List of test results @@ -86,34 +89,95 @@ class Orchestrator: ) ) - print(f"Running {len(test_cases)} test cases...") + total = len(test_cases) + print(f"Running {total} test cases (workers={workers})...") print(f" Senders: {', '.join(sender_names)}") print(f" Receivers: {', '.join(receiver_names)}") print(f" Types: {', '.join(amqp_types.keys())}") print() - # Run tests + if workers <= 1: + return self._run_sequential(test_cases) + return self._run_parallel(test_cases, workers) + + def _run_sequential(self, test_cases: list[TestCase]) -> list[TestResult]: results: list[TestResult] = [] + total = len(test_cases) for i, test_case in enumerate(test_cases, 1): - print(f"[{i}/{len(test_cases)}] Testing {test_case.sender_shim} → {test_case.receiver_shim} " + print(f"[{i}/{total}] Testing {test_case.sender_shim} → {test_case.receiver_shim} " f"({test_case.amqp_type})...", end=" ", flush=True) result = self.run_test_case(test_case) results.append(result) + self._print_result(result) - if result.success and not result.xfail_diffs: - print("✓") - elif result.success and result.xfail_diffs: - print(f"✓ ({len(result.xfail_diffs)} xfail)") - else: - print("✗") - if result.error: - print(f" Error: {result.error}") - if result.diffs: - print(f" {len(result.diffs)} difference(s) found") + return results + + def _run_parallel(self, test_cases: list[TestCase], workers: int) -> list[TestResult]: + import threading + + total = len(test_cases) + completed = 0 + failed = 0 + lock = threading.Lock() + result_map: dict[int, TestResult] = {} + + def run_one(index: int, tc: TestCase) -> tuple[int, TestResult]: + return index, self.run_test_case(tc) + + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = { + executor.submit(run_one, i, tc): i + for i, tc in enumerate(test_cases) + } + for future in as_completed(futures): + index, result = future.result() + result_map[index] = result + with lock: + completed += 1 + if not result.success: + failed += 1 + tc = test_cases[index] + status = self._result_symbol(result) + print(f"[{completed}/{total}] {tc.sender_shim} → {tc.receiver_shim} " + f"({tc.amqp_type}) {status}", flush=True) + + results = [result_map[i] for i in range(total)] + + if failed: + print(f"\nFailed tests:") + for i, result in enumerate(results): + if not result.success: + tc = test_cases[i] + print(f" {tc.sender_shim} → {tc.receiver_shim} ({tc.amqp_type})") + if result.error: + print(f" Error: {result.error}") + if result.diffs: + print(f" {len(result.diffs)} difference(s) found") return results + @staticmethod + def _result_symbol(result: "TestResult") -> str: + if result.success and not result.xfail_diffs: + return "✓" + if result.success and result.xfail_diffs: + return f"✓ ({len(result.xfail_diffs)} xfail)" + return "✗" + + @staticmethod + def _print_result(result: "TestResult") -> None: + if result.success and not result.xfail_diffs: + print("✓") + elif result.success and result.xfail_diffs: + print(f"✓ ({len(result.xfail_diffs)} xfail)") + else: + print("✗") + if result.error: + print(f" Error: {result.error}") + if result.diffs: + print(f" {len(result.diffs)} difference(s) found") + def run_test_case(self, test_case: TestCase) -> TestResult: """ Run a single test case. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
