You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2026/08/10 21:12:33 UTC
(qpid-interop-test) 33/41: Add parallel test execution to orchestrator (-j/--workers)
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 <qi...@example.com>
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 <no...@anthropic.com>
---
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",
)
+@click.option(
+ "--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: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org