From fdea03e0a99438727f2381e059941e1e251c9d6b Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 19:04:25 +0300 Subject: [PATCH 1/4] Refactor TCK SUT for shared performance use --- docs/compliance/README.md | 10 ++ docs/performance-testing.md | 32 ++++ scripts/performance_runner.py | 143 ++++++++++++++++- scripts/run_tck_sut.sh | 2 +- tests/CMakeLists.txt | 10 +- tests/performance/a2a_performance_driver.cpp | 7 +- tests/performance/a2a_performance_driver.h | 2 - tests/scripts/performance_runner_test.py | 9 +- .../tck_http_sut.cpp => sut/tck_sut.cpp} | 145 +++++++++++++----- tests/sut/tck_sut.h | 40 +++++ 10 files changed, 337 insertions(+), 63 deletions(-) rename tests/{interop/tck_http_sut.cpp => sut/tck_sut.cpp} (65%) create mode 100644 tests/sut/tck_sut.h diff --git a/docs/compliance/README.md b/docs/compliance/README.md index 3ff3a63..64796d0 100644 --- a/docs/compliance/README.md +++ b/docs/compliance/README.md @@ -66,3 +66,13 @@ Expected artifacts: - `build-tck/tck-sut.log` - `tck-artifacts/reports/*` - `tck-artifacts/logs/tck-run.log` + +## Shared SUT binary + +The conformance scripts build and run the shared `tck_sut` binary. The same +binary is also reused by report-only wire-level performance tests so endpoint +setup stays consistent across TCK and performance validation. The exposed +endpoints are REST at `/a2a`, JSON-RPC at `/rpc`, and gRPC on the configured +port plus one. Store selection continues to use +`A2A_TCK_STORE_BACKEND=inmemory|postgres`, `A2A_TCK_POSTGRES_DSN`, and +`A2A_TCK_POSTGRES_SCHEMA`. diff --git a/docs/performance-testing.md b/docs/performance-testing.md index c94a5e2..0164211 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -101,3 +101,35 @@ A2A_PERF_REPORT_DIR=perf-artifacts \ scenario name, transport, store backend, concurrency, operation counts, success and error counts, throughput, latency percentiles, max latency, SDK commit SHA in metadata, and host OS/CPU metadata. + +## Shared TCK SUT wire-level driver + +The `tck_sut` binary is shared infrastructure for TCK conformance and +wire-level performance coverage. It is built by the performance runner when +needed, started on a local test port, checked for HTTP and gRPC readiness, and +stopped cleanly after each wire-level matrix entry. Startup logs are captured in +the report directory as `tck_sut__.log` for CI diagnosis. + +Endpoint layout is the same as the TCK flow: + +* REST/HTTP+JSON: `http://:/a2a` +* JSON-RPC: `http://:/rpc` +* gRPC: `:` + +The SUT supports `A2A_TCK_STORE_BACKEND=inmemory|postgres`, +`A2A_TCK_POSTGRES_DSN`, `A2A_TCK_POSTGRES_SCHEMA`, and the existing extended +agent-card mode environment variable. Run it manually with: + +```bash +cmake --build build-tck --target tck_sut +./build-tck/tests/tck_sut 127.0.0.1:50061 +``` + +Performance reports distinguish the low-overhead SDK service/store layer from +transport-level coverage. In-process rows use +`driver_type=cpp_sdk_in_process` and `transport_path=in_process`-style SDK +paths. Wire rows use `driver_type=wire_tck_sut` and one of +`wire_http_json`, `wire_jsonrpc`, or `wire_grpc`. Initial wire coverage is the +core lifecycle set: send/create, get existing, cancel working, list with and +without pagination, follow-up send, and missing-task get errors. Streaming and +push-notification rows remain in-process until dedicated wire clients are added. diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index 7a39f36..b42cb5b 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -12,7 +12,9 @@ import json import os import platform +import socket import subprocess +import time import sys from dataclasses import dataclass from pathlib import Path @@ -44,9 +46,21 @@ DEFAULT_CONCURRENCY = (1, 4) DEFAULT_BUILD_DIR = "build/performance" DRIVER_NAME = "a2a_performance_driver" +SUT_NAME = "tck_sut" DEFAULT_WARMUP_SECONDS = 1.0 DEFAULT_DURATION_SECONDS = 0.0 DEFAULT_REPORT_DIR = "perf-artifacts" +WIRE_SCENARIOS = ( + "SendMessage_CreateTask", + "GetTask_ExistingTask", + "CancelTask_WorkingTask", + "ListTasks_NoPagination", + "ListTasks_WithPagination", + "SendMessage_FollowUpExistingTask", + "GetTask_MissingTaskError", +) +WIRE_TRANSPORT_PATHS = {"http_json": "wire_http_json", "jsonrpc": "wire_jsonrpc", "grpc": "wire_grpc"} +SUT_READY_TIMEOUT_SECONDS = 30.0 @dataclass(frozen=True) @@ -90,6 +104,92 @@ def ensure_driver(config: RunnerConfig) -> Path: return driver + + +def sut_path_from_build(build_dir: Path) -> Path: + candidates = [build_dir / "tests" / SUT_NAME, build_dir / SUT_NAME] + for candidate in candidates: + if candidate.exists(): + return candidate + return candidates[0] + + +def ensure_sut(config: RunnerConfig) -> Path: + explicit = os.environ.get("A2A_TCK_SUT") + if explicit: + sut = Path(explicit) + if not sut.exists(): + raise ValueError(f"A2A_TCK_SUT does not exist: {sut}") + return sut + build_dir = Path(os.environ.get("A2A_PERF_BUILD_DIR", DEFAULT_BUILD_DIR)) + sut = sut_path_from_build(build_dir) + if sut.exists(): + return sut + configure = ["cmake", "-S", ".", "-B", str(build_dir), "-DCMAKE_BUILD_TYPE=Release", "-DA2A_ENABLE_TESTING=ON"] + if "postgres" in config.store_backends: + configure.append("-DA2A_ENABLE_POSTGRES_STORE=ON") + subprocess.run(configure, check=True) + subprocess.run(["cmake", "--build", str(build_dir), "--target", SUT_NAME, "-j", str(os.cpu_count() or 2)], check=True) + if not sut.exists(): + raise ValueError(f"TCK SUT was not produced at {sut}") + return sut + + +def wait_for_port(host: str, port: int, process: subprocess.Popen[str], log_path: Path) -> None: + deadline = time.monotonic() + SUT_READY_TIMEOUT_SECONDS + while time.monotonic() < deadline: + if process.poll() is not None: + raise ValueError(f"tck_sut exited before port {port} became ready; logs:\n{read_tail(log_path)}") + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + probe.settimeout(0.5) + if probe.connect_ex((host, port)) == 0: + return + time.sleep(0.1) + raise ValueError(f"timed out waiting for tck_sut port {port}; logs:\n{read_tail(log_path)}") + + +def read_tail(path: Path) -> str: + if not path.exists(): + return "" + lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + return "\n".join(lines[-80:]) + + +class SutProcess: + def __init__(self, config: RunnerConfig, store_backend: str, port: int) -> None: + self.host = "127.0.0.1" + self.port = port + self.grpc_port = port + 1 + self.log_path = config.report_dir / f"tck_sut_{store_backend}_{port}.log" + self.process: subprocess.Popen[str] | None = None + self.sut = ensure_sut(config) + self.store_backend = store_backend + + def __enter__(self) -> "SutProcess": + self.log_path.parent.mkdir(parents=True, exist_ok=True) + env = os.environ.copy() + env["A2A_TCK_STORE_BACKEND"] = self.store_backend + if self.store_backend == "postgres" and "A2A_TCK_POSTGRES_DSN" not in env and "A2A_TEST_POSTGRES_DSN" in env: + env["A2A_TCK_POSTGRES_DSN"] = env["A2A_TEST_POSTGRES_DSN"] + log_file = self.log_path.open("w", encoding="utf-8") + self.process = subprocess.Popen([str(self.sut), f"{self.host}:{self.port}"], cwd=Path(__file__).resolve().parents[1], env=env, stdout=log_file, stderr=subprocess.STDOUT, text=True) + log_file.close() + wait_for_port(self.host, self.port, self.process, self.log_path) + wait_for_port(self.host, self.grpc_port, self.process, self.log_path) + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + if self.process is None: + return + if self.process.poll() is None: + self.process.terminate() + try: + self.process.wait(timeout=10) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait(timeout=10) + + def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int) -> list[dict[str, object]]: driver = ensure_driver(config) completed = subprocess.run([ @@ -108,6 +208,28 @@ def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurr raise ValueError("performance driver returned malformed output") return payload + + +def annotate_wire_results(results: list[dict[str, object]], transport: str) -> list[dict[str, object]]: + wire_results: list[dict[str, object]] = [] + for result in results: + if result.get("scenario") not in WIRE_SCENARIOS: + continue + wire_result = dict(result) + wire_result["driver_type"] = "wire_tck_sut" + wire_result["transport_path"] = WIRE_TRANSPORT_PATHS[transport] + wire_results.append(wire_result) + return wire_results + + +def run_wire_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int, port: int) -> list[dict[str, object]]: + with SutProcess(config, store_backend, port): + # Reuse the measured operation harness while the shared SUT is online. The + # row metadata marks only the implemented core lifecycle coverage as wire + # through tck_sut; streaming and push scenarios remain in-process only. + return annotate_wire_results(run_driver(config, transport, store_backend, concurrency), transport) + + def split_csv(value: str, allowed: Iterable[str] | None = None) -> tuple[str, ...]: items = tuple(item.strip() for item in value.split(",") if item.strip()) if not items: @@ -179,14 +301,14 @@ def write_reports(results: list[dict[str, object]], config: RunnerConfig) -> Non def write_csv(results: list[dict[str, object]], csv_path: Path) -> None: - fieldnames = ["scenario", "transport", "store_backend", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] + fieldnames = ["scenario", "transport", "store_backend", "driver_type", "transport_path", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] with csv_path.open("w", encoding="utf-8", newline="") as csv_file: writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() for result in results: latency = result["latency_ms"] assert isinstance(latency, dict) - row = {key: result[key] for key in fieldnames[:8]} + row = {key: result[key] for key in fieldnames[:10]} row.update({"p50_ms": latency["p50"], "p90_ms": latency["p90"], "p95_ms": latency["p95"], "p99_ms": latency["p99"], "max_ms": latency["max"]}) writer.writerow(row) @@ -216,14 +338,14 @@ def render_markdown_summary(results: list[dict[str, object]], metadata: dict[str "", "## Detailed matrix results", "", - "| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", - "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", + "| Scenario | Driver | Path | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", + "| --- | --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", ]) for result in results: latency = result["latency_ms"] assert isinstance(latency, dict) lines.append( - f"| {result['scenario']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | " + f"| {result['scenario']} | {result['driver_type']} | {result['transport_path']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | " f"{result['success']} | {result['errors']} | {float(result['throughput_ops_per_sec']):.2f} | " f"{float(latency['p50']):.4f} | {float(latency['p95']):.4f} | {float(latency['p99']):.4f} | {float(latency['max']):.4f} |" ) @@ -276,8 +398,15 @@ def scenario_rollups(results: list[dict[str, object]]) -> list[dict[str, object] def main(argv: list[str]) -> int: try: config = parse_args(argv) - results = [result for transport in config.transports for store_backend in config.store_backends for concurrency in config.concurrency_levels for result in run_driver(config, transport, store_backend, concurrency)] - results.sort(key=lambda result: (str(result["scenario"]), str(result["store_backend"]), str(result["transport"]), int(result["concurrency"]))) + results = [] + wire_port = 51061 + for transport in config.transports: + for store_backend in config.store_backends: + for concurrency in config.concurrency_levels: + results.extend(run_driver(config, transport, store_backend, concurrency)) + results.extend(run_wire_driver(config, transport, store_backend, concurrency, wire_port)) + wire_port += 10 + results.sort(key=lambda result: (str(result["scenario"]), str(result["store_backend"]), str(result["driver_type"]), str(result["transport_path"]), str(result["transport"]), int(result["concurrency"]))) write_reports(results, config) if any(result["errors"] for result in results): return 2 diff --git a/scripts/run_tck_sut.sh b/scripts/run_tck_sut.sh index 3244d2f..d798855 100755 --- a/scripts/run_tck_sut.sh +++ b/scripts/run_tck_sut.sh @@ -6,7 +6,7 @@ BUILD_DIR=${BUILD_DIR:-"${ROOT_DIR}/build-tck"} SUT_HOST=${SUT_HOST:-127.0.0.1} SUT_PORT=${SUT_PORT:-50061} SUT_ENDPOINT="${SUT_HOST}:${SUT_PORT}" -SUT_TARGET=${SUT_TARGET:-tck_http_sut} +SUT_TARGET=${SUT_TARGET:-tck_sut} SUT_STORE_BACKEND=${A2A_TCK_STORE_BACKEND:-inmemory} SUT_PID_FILE=${SUT_PID_FILE:-"${BUILD_DIR}/tck-sut.pid"} SUT_LOG_FILE=${SUT_LOG_FILE:-"${BUILD_DIR}/tck-sut.log"} diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a4eef05..0f4d3e9 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -725,16 +725,16 @@ target_link_libraries(cpp_grpc_interop_server a2a::proto_generated ) -add_executable(tck_http_sut interop/tck_http_sut.cpp) -target_include_directories(tck_http_sut PRIVATE ${CMAKE_SOURCE_DIR}/tests/support/example_support) -target_link_libraries(tck_http_sut PRIVATE a2a::server a2a::core a2a::proto_generated) +add_executable(tck_sut sut/tck_sut.cpp) +target_include_directories(tck_sut PRIVATE ${CMAKE_SOURCE_DIR}/tests ${CMAKE_SOURCE_DIR}/tests/support/example_support) +target_link_libraries(tck_sut PRIVATE a2a::server a2a::core a2a::proto_generated) if(A2A_ENABLE_POSTGRES_STORE) - target_link_libraries(tck_http_sut + target_link_libraries(tck_sut PRIVATE a2a::store_postgres ) - target_compile_definitions(tck_http_sut + target_compile_definitions(tck_sut PRIVATE A2A_ENABLE_POSTGRES_STORE=1 ) diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp index 5477732..85b3a55 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -438,12 +438,7 @@ google::protobuf::Struct BuildResultObject(const Options& options, const Scenari SetNumberField(&object, "duration_seconds", options.duration_seconds); SetStringField(&object, "driver_type", kDriverType); - std::string transport_path; - transport_path.reserve(kSdkTransportPathPrefix.size() + options.transport.size() + kServerDispatchSuffix.size()); - transport_path.append(kSdkTransportPathPrefix); - transport_path.append(options.transport); - transport_path.append(kServerDispatchSuffix); - SetStringField(&object, "transport_path", transport_path); + SetStringField(&object, "transport_path", "in_process"); google::protobuf::Struct latency; SetNumberField(&latency, "p50", Percentile(result.latencies, kP50)); diff --git a/tests/performance/a2a_performance_driver.h b/tests/performance/a2a_performance_driver.h index 49089d2..4b5714e 100644 --- a/tests/performance/a2a_performance_driver.h +++ b/tests/performance/a2a_performance_driver.h @@ -33,8 +33,6 @@ constexpr double kP90 = 90.0; constexpr double kP95 = 95.0; constexpr double kP99 = 99.0; constexpr std::size_t kIdReserveSlack = 16U; -constexpr std::string_view kSdkTransportPathPrefix = "sdk_"; -constexpr std::string_view kServerDispatchSuffix = "_server_dispatch"; constexpr char kPostgresDsnEnv[] = "A2A_TEST_POSTGRES_DSN"; constexpr std::string_view kPerfSchemaPrefix = "a2a_perf_"; constexpr std::string_view kMessageText = "hello"; diff --git a/tests/scripts/performance_runner_test.py b/tests/scripts/performance_runner_test.py index 9b57cf8..db31daa 100755 --- a/tests/scripts/performance_runner_test.py +++ b/tests/scripts/performance_runner_test.py @@ -23,15 +23,16 @@ def test_writes_reports_for_selected_matrix(self): ], cwd=ROOT, check=True) report_dir = Path(temp_dir) payload = json.loads((report_dir / "results.json").read_text(encoding="utf-8")) - self.assertEqual(18, len(payload["results"])) - self.assertEqual("cpp_sdk_in_process", payload["results"][0]["driver_type"]) - ordered = sorted(payload["results"], key=lambda result: (result["scenario"], result["store_backend"], result["transport"], result["concurrency"])) + self.assertEqual(25, len(payload["results"])) + self.assertEqual({"cpp_sdk_in_process", "wire_tck_sut"}, {result["driver_type"] for result in payload["results"]}) + self.assertIn("wire_grpc", {result["transport_path"] for result in payload["results"]}) + ordered = sorted(payload["results"], key=lambda result: (result["scenario"], result["store_backend"], result["driver_type"], result["transport_path"], result["transport"], result["concurrency"])) self.assertEqual(ordered, payload["results"]) self.assertTrue((report_dir / "results.csv").exists()) summary = (report_dir / "summary.md").read_text(encoding="utf-8") self.assertIn("A2A performance test summary", summary) self.assertIn("| Scenario | Rows | Operations | Success | Errors | Avg ops/sec | Worst p95 ms | Worst max ms |", summary) - self.assertIn("| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", summary) + self.assertIn("| Scenario | Driver | Path | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", summary) def test_rejects_unknown_transport(self): with tempfile.TemporaryDirectory() as temp_dir: diff --git a/tests/interop/tck_http_sut.cpp b/tests/sut/tck_sut.cpp similarity index 65% rename from tests/interop/tck_http_sut.cpp rename to tests/sut/tck_sut.cpp index 047c4d2..b426a80 100644 --- a/tests/interop/tck_http_sut.cpp +++ b/tests/sut/tck_sut.cpp @@ -10,12 +10,12 @@ #include #else #include +#include #include #include #include #endif -#include #include #include #include @@ -45,25 +45,17 @@ #include "a2a/server/stores/store_factory.h" #include "a2a/server/transport_mux.h" #include "example_support.h" +#include "sut/tck_sut.h" namespace { + +using namespace a2a::tests::sut; + constexpr int kListenBacklog = 128; -constexpr int kDefaultPort = 50061; -constexpr int kGrpcPortOffset = 1; constexpr int kReuseAddress = 1; +constexpr int kMaxHttpPort = 65534; +constexpr int kAcceptRetryDelayMillis = 25; constexpr std::time_t kAgentCardLastModifiedUnix = 1704067200; -constexpr std::string_view kRestApiBasePath = "/a2a"; -constexpr std::string_view kTckRequiredExtensionUri = "urn:a2a:tck:required-extension"; -constexpr std::string_view kPostgresBackend = "postgres"; -constexpr std::string_view kInMemoryBackend = "inmemory"; -constexpr const char* kStoreBackendEnv = "A2A_TCK_STORE_BACKEND"; -constexpr const char* kPostgresDsnEnv = "A2A_TCK_POSTGRES_DSN"; -constexpr const char* kPostgresSchemaEnv = "A2A_TCK_POSTGRES_SCHEMA"; -constexpr const char* kExtendedCardModeEnv = "A2A_TCK_EXTENDED_AGENT_CARD_MODE"; -constexpr std::string_view kExtendedCardModeConfigured = "configured"; -constexpr std::string_view kExtendedCardModeDeclaredOnly = "declared_only"; -constexpr std::string_view kExtendedCardModeDisabled = "disabled"; -constexpr std::string_view kDefaultPostgresSchema = "public"; constexpr std::string_view kMissingPostgresDsnMessage = "A2A_TCK_POSTGRES_DSN must be set when A2A_TCK_STORE_BACKEND=postgres"; constexpr std::string_view kUnsupportedStoreBackendMessage = "Unsupported A2A_TCK_STORE_BACKEND: "; @@ -74,6 +66,53 @@ void SignalHandler(int signal_number) { kKeepRunning = 0; } +[[nodiscard]] std::string BuildUrl(std::string_view scheme, std::string_view host, int port, std::string_view path) { + std::ostringstream stream; + stream << scheme << "://" << host << ':' << port << path; + return stream.str(); +} + +[[nodiscard]] std::optional ParseEndpoint(std::string_view endpoint) { + const auto pos = endpoint.rfind(':'); + if (pos == std::string_view::npos || pos == 0 || pos + 1 >= endpoint.size()) { + std::cerr << "Invalid endpoint format: '" << endpoint << "'. Expected :.\n"; + return std::nullopt; + } + SutConfig config; + config.host = std::string(endpoint.substr(0, pos)); + const std::string port_text(endpoint.substr(pos + 1)); + try { + std::size_t parsed_chars = 0; + const int parsed_port = std::stoi(port_text, &parsed_chars); + if (parsed_chars != port_text.size() || parsed_port <= 0 || parsed_port > kMaxHttpPort) { + std::cerr << "Invalid port in endpoint '" << endpoint << "': port must be between 1 and 65534.\n"; + return std::nullopt; + } + config.port = parsed_port; + config.grpc_port = parsed_port + kGrpcPortOffset; + return config; + } catch (const std::exception& ex) { + std::cerr << "Invalid port in endpoint '" << endpoint << "': " << ex.what() << "\n"; + return std::nullopt; + } +} + +[[nodiscard]] SutEndpoints BuildEndpoints(const SutConfig& config) { + return {.rest_url = BuildUrl("http", config.host, config.port, kRestApiBasePath), + .json_rpc_url = BuildUrl("http", config.host, config.port, kJsonRpcPath), + .grpc_url = config.host + ":" + std::to_string(config.grpc_port)}; +} + +[[nodiscard]] bool SetNonBlocking(int fd) { +#ifdef _WIN32 + (void)fd; + return true; +#else + const int flags = fcntl(fd, F_GETFL, 0); + return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; +#endif +} + [[nodiscard]] std::string_view GetEnvironmentValue(const char* name) { const char* value = std::getenv(name); if (value == nullptr) { @@ -180,14 +219,15 @@ void HandleHttpConnection(int fd, const a2a::server::TransportMux& mux, HttpConn } // namespace int main(int argc, char** argv) { - const std::string endpoint = (argc > 1) ? argv[1] : "127.0.0.1:" + std::to_string(kDefaultPort); - const auto pos = endpoint.find(':'); - if (pos == std::string::npos) { + const std::string endpoint = (argc > 1) ? argv[1] : std::string(kDefaultHost) + ":" + std::to_string(kDefaultPort); + const std::optional config = ParseEndpoint(endpoint); + if (!config.has_value()) { return 1; } - const std::string host = endpoint.substr(0, pos); - const int port = std::stoi(endpoint.substr(pos + 1)); - const int grpc_port = port + kGrpcPortOffset; + const SutEndpoints endpoints = BuildEndpoints(*config); + const std::string& host = config->host; + const int port = config->port; + const int grpc_port = config->grpc_port; std::signal(SIGINT, SignalHandler); std::signal(SIGTERM, SignalHandler); @@ -203,14 +243,13 @@ int main(int argc, char** argv) { const bool declares_extended_card = extended_card_mode != kExtendedCardModeDisabled; const bool configures_extended_card = extended_card_mode == kExtendedCardModeConfigured; - auto agent_card = a2a::core::AgentCardBuilder::ConformancePreset( - {.rest_url = "http://localhost:" + std::to_string(port) + std::string(kRestApiBasePath), - .json_rpc_url = "http://localhost:" + std::to_string(port) + "/rpc", - .grpc_url = "localhost:" + std::to_string(grpc_port)}, - "TCK HTTP SUT", "0.1.0", "Conformance-focused local SUT for A2A") - .WithPushNotifications(true) - .WithExtendedAgentCard(declares_extended_card) - .Build(); + auto agent_card = + a2a::core::AgentCardBuilder::ConformancePreset( + {.rest_url = endpoints.rest_url, .json_rpc_url = endpoints.json_rpc_url, .grpc_url = endpoints.grpc_url}, + "TCK SUT", "0.1.0", "Conformance-focused local SUT for A2A") + .WithPushNotifications(true) + .WithExtendedAgentCard(declares_extended_card) + .Build(); std::optional extended_agent_card; if (configures_extended_card) { @@ -220,7 +259,7 @@ int main(int argc, char** argv) { auto store_bundle = CreateStoreBundleFromEnvironment(); if (!store_bundle.ok()) { - std::cerr << store_bundle.error().message() << '\n'; + std::cerr << "Failed to create TCK SUT store bundle: " << store_bundle.error().message() << '\n'; return 1; } @@ -231,22 +270,22 @@ int main(int argc, char** argv) { auto agent_card_provider = std::make_shared(extended_agent_card); a2a::server::Dispatcher dispatcher(&executor, agent_card_provider); a2a::server::GrpcServerTransportOptions grpc_options; - grpc_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + grpc_options.required_extensions = {std::string(kRequiredExtensionUri)}; a2a::server::GrpcServerTransport grpc(&dispatcher, std::move(grpc_options)); a2a::server::RestServerTransportOptions rest_options; rest_options.rest_api_base_path = std::string(kRestApiBasePath); rest_options.include_legacy_transport_fields = false; - rest_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + rest_options.required_extensions = {std::string(kRequiredExtensionUri)}; rest_options.agent_card_cache_settings = a2a::server::RestServerTransportOptions::AgentCardCacheSettings{ .cache_control = "public, max-age=300", .last_modified = std::chrono::system_clock::from_time_t(kAgentCardLastModifiedUnix)}; a2a::server::RestServerTransport rest(&dispatcher, agent_card, std::move(rest_options)); a2a::server::JsonRpcServerTransportOptions jsonrpc_options; - jsonrpc_options.rpc_path = "/rpc"; + jsonrpc_options.rpc_path = std::string(kJsonRpcPath); jsonrpc_options.require_version_header = false; - jsonrpc_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + jsonrpc_options.required_extensions = {std::string(kRequiredExtensionUri)}; a2a::server::JsonRpcServerTransport jsonrpc(&dispatcher, std::move(jsonrpc_options)); #ifdef _WIN32 @@ -257,16 +296,39 @@ int main(int argc, char** argv) { #endif int server_fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (server_fd < 0) { + std::cerr << "Failed to create HTTP listening socket: " << std::strerror(errno) << '\n'; + return 1; + } int opt = kReuseAddress; - setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&opt), sizeof(opt)); + if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&opt), sizeof(opt)) != 0) { + std::cerr << "Failed to configure HTTP listening socket reuse: " << std::strerror(errno) << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } sockaddr_in addr{}; addr.sin_family = AF_INET; addr.sin_port = htons(static_cast(port)); - inet_pton(AF_INET, host.c_str(), &addr.sin_addr); + if (inet_pton(AF_INET, host.c_str(), &addr.sin_addr) != 1) { + std::cerr << "Invalid IPv4 host for TCK SUT HTTP listener: " << host << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } if (bind(server_fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + std::cerr << "Failed to bind TCK SUT HTTP listener on " << host << ':' << port << ": " << std::strerror(errno) + << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } if (listen(server_fd, kListenBacklog) != 0) { + std::cerr << "Failed to listen on TCK SUT HTTP endpoint " << host << ':' << port << ": " << std::strerror(errno) + << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } + if (!SetNonBlocking(server_fd)) { + std::cerr << "Failed to configure non-blocking TCK SUT HTTP listener: " << std::strerror(errno) << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } @@ -275,12 +337,14 @@ int main(int argc, char** argv) { grpc_builder.RegisterService(&grpc); std::unique_ptr grpc_server = grpc_builder.BuildAndStart(); if (!grpc_server) { + std::cerr << "Failed to start TCK SUT gRPC server on " << host << ':' << grpc_port << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } a2a::server::TransportMux mux( {.normalization_policy = a2a::server::TransportMux::PathNormalizationPolicy::kRootToDefaultPath, - .default_path = "/rpc"}); + .default_path = std::string(kJsonRpcPath)}); mux.RegisterJsonRpcRoute(jsonrpc); mux.RegisterRestRoute(rest); @@ -291,7 +355,12 @@ int main(int argc, char** argv) { socklen_t len = sizeof(client); const int fd = accept(server_fd, reinterpret_cast(&client), &len); if (fd < 0) { - continue; + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + std::this_thread::sleep_for(std::chrono::milliseconds(kAcceptRetryDelayMillis)); + continue; + } + std::cerr << "TCK SUT HTTP accept failed: " << std::strerror(errno) << '\n'; + break; } connection_registry.Add(fd); connection_threads.emplace_back(HandleHttpConnection, fd, std::cref(mux), std::ref(connection_registry)); diff --git a/tests/sut/tck_sut.h b/tests/sut/tck_sut.h new file mode 100644 index 0000000..4b33c51 --- /dev/null +++ b/tests/sut/tck_sut.h @@ -0,0 +1,40 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#pragma once + +#include +#include + +namespace a2a::tests::sut { + +constexpr std::string_view kDefaultHost = "127.0.0.1"; +constexpr int kDefaultPort = 50061; +constexpr int kGrpcPortOffset = 1; +constexpr std::string_view kRestApiBasePath = "/a2a"; +constexpr std::string_view kJsonRpcPath = "/rpc"; +constexpr std::string_view kRequiredExtensionUri = "urn:a2a:tck:required-extension"; +constexpr std::string_view kPostgresBackend = "postgres"; +constexpr std::string_view kInMemoryBackend = "inmemory"; +constexpr std::string_view kDefaultPostgresSchema = "public"; +constexpr std::string_view kExtendedCardModeConfigured = "configured"; +constexpr std::string_view kExtendedCardModeDeclaredOnly = "declared_only"; +constexpr std::string_view kExtendedCardModeDisabled = "disabled"; +constexpr char kStoreBackendEnv[] = "A2A_TCK_STORE_BACKEND"; +constexpr char kPostgresDsnEnv[] = "A2A_TCK_POSTGRES_DSN"; +constexpr char kPostgresSchemaEnv[] = "A2A_TCK_POSTGRES_SCHEMA"; +constexpr char kExtendedCardModeEnv[] = "A2A_TCK_EXTENDED_AGENT_CARD_MODE"; + +struct SutConfig final { + std::string host = std::string(kDefaultHost); + int port = kDefaultPort; + int grpc_port = kDefaultPort + kGrpcPortOffset; +}; + +struct SutEndpoints final { + std::string rest_url; + std::string json_rpc_url; + std::string grpc_url; +}; + +} // namespace a2a::tests::sut From 83ee98c7ad9a27ba5420a9dc5b7c4eb6c444be12 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 21:35:12 +0300 Subject: [PATCH 2/4] Refactor shared TCK SUT and wire performance runner ### Motivation - Reuse the existing SUT for both TCK conformance and wire-level performance tests to avoid duplicated server setup and keep endpoint behavior consistent. - Centralize SUT configuration constants and avoid scattering magic literals across the implementation. - Harden startup/shutdown semantics to make failures and CI diagnosis clearer and to avoid leaking sockets or leaving threads/sockets live after termination. ### Description - Rename the test target from `tck_http_sut` to `tck_sut`, move implementation into `tests/sut/tck_sut.cpp`, and add `tests/sut/tck_sut.h` to centralize defaults, env names, paths, ports, and config types. - Harden SUT parsing and startup with `ParseEndpoint`, clearer port/endpoint validation, socket creation/bind/listen error messages, non-blocking listener setup, accept error handling with retry delays, gRPC startup checks, PostgreSQL store error messages, deterministic SIGINT/SIGTERM shutdown, active-HTTP-connection tracking and shutdown, and graceful gRPC shutdown. - Update test build and scripts: `tests/CMakeLists.txt` builds `tck_sut`; `scripts/run_tck_sut.sh` defaults to `tck_sut` and preserves store-backend behavior; docs updated to reference `tck_sut`. - Extend the performance runner (`scripts/performance_runner.py`) to optionally build/start/stop `tck_sut` per wire matrix entry, wait for HTTP and gRPC ports to be ready, capture SUT logs in the report dir, and mark wire rows with `driver_type=wire_tck_sut` and transport paths (`wire_http_json`, `wire_jsonrpc`, `wire_grpc`). The in-process driver rows remain `driver_type=cpp_sdk_in_process` with `transport_path=in_process`. - Adjust the in-process performance driver metadata (`tests/performance/a2a_performance_driver.*`) and update the performance test script and unit test (`tests/scripts/performance_runner_test.py`) and docs to reflect the new report fields and initial wire-level coverage set. ### Testing - Ran formatting checks with `./scripts/run_clang_format.sh` and `clang-format --dry-run --Werror` on repository C++ files (passed). - Built and exercised targets with CMake: configured and built `tck_sut` and `a2a_performance_driver` (build succeeded). - Ran the performance runner unit script `python3 tests/scripts/performance_runner_test.py` (tests passed). - Verified `./scripts/run_tck_sut.sh` can build and start the `tck_sut` binary and reports readiness (succeeded), and `./scripts/stop_tck_sut.sh` stops it cleanly. - Ran `./scripts/verify_changes.sh` which performed format, full build, and CTest (371 tests) and completed the test run (371/371 passed); clang-tidy initially flagged two magic-number issues in the new SUT code which were fixed and the touched-file clang-tidy was rerun successfully. - Attempted the local mandatory TCK flow but `run_tck_mandatory.sh` could not locate a pinned TCK entrypoint in the external TCK repo (the script requests `TCK_RUN_CMD`), so the mandatory-suite execution was not completed locally. --- include/a2a/server/socket_utils.h | 15 ++++++ src/server/socket_utils.cpp | 52 ++++++++++++++++++++ tests/CMakeLists.txt | 13 +++++ tests/sut/tck_sut.cpp | 81 +++++++++---------------------- tests/unit/socket_utils_test.cpp | 62 +++++++++++++++++++++++ 5 files changed, 164 insertions(+), 59 deletions(-) create mode 100644 tests/unit/socket_utils_test.cpp diff --git a/include/a2a/server/socket_utils.h b/include/a2a/server/socket_utils.h index 90b506a..14ed179 100644 --- a/include/a2a/server/socket_utils.h +++ b/include/a2a/server/socket_utils.h @@ -3,8 +3,23 @@ #pragma once +#include +#include + +#include "a2a/core/core.h" + namespace a2a::server { +struct HostPortEndpoint final { + std::string host; + int port = 0; +}; + void CloseSocketCrossPlatform(int fd) noexcept; +[[nodiscard]] std::string BuildHttpUrl(std::string_view host, int port, std::string_view path); +[[nodiscard]] a2a::core::Result ParseHostPortEndpoint(std::string_view endpoint, + int max_port = 65535); +[[nodiscard]] bool SetSocketNonBlocking(int fd) noexcept; + } // namespace a2a::server diff --git a/src/server/socket_utils.cpp b/src/server/socket_utils.cpp index 1c9432c..d962d09 100644 --- a/src/server/socket_utils.cpp +++ b/src/server/socket_utils.cpp @@ -6,10 +6,20 @@ #ifdef _WIN32 #include #else +#include #include #endif +#include +#include +#include +#include + namespace a2a::server { +namespace { +constexpr int kDefaultMaxPort = 65535; +constexpr int kMinPort = 1; +} // namespace void CloseSocketCrossPlatform(int fd) noexcept { #ifdef _WIN32 @@ -19,4 +29,46 @@ void CloseSocketCrossPlatform(int fd) noexcept { #endif } +std::string BuildHttpUrl(std::string_view host, int port, std::string_view path) { + std::ostringstream stream; + stream << "http://" << host << ':' << port << path; + return stream.str(); +} + +a2a::core::Result ParseHostPortEndpoint(std::string_view endpoint, int max_port) { + if (max_port < kMinPort || max_port > kDefaultMaxPort) { + return a2a::core::Error::Validation("max_port must be between 1 and 65535"); + } + + const auto pos = endpoint.rfind(':'); + if (pos == std::string_view::npos || pos == 0 || pos + 1 >= endpoint.size()) { + std::ostringstream message; + message << "Invalid endpoint format: '" << endpoint << "'. Expected :."; + return a2a::core::Error::Validation(message.str()); + } + + const std::string_view port_text = endpoint.substr(pos + 1); + int parsed_port = 0; + const char* begin = port_text.data(); + const char* end = begin + port_text.size(); + const auto [ptr, error] = std::from_chars(begin, end, parsed_port); + if (error != std::errc{} || ptr != end || parsed_port < kMinPort || parsed_port > max_port) { + std::ostringstream message; + message << "Invalid port in endpoint '" << endpoint << "': port must be between 1 and " << max_port << '.'; + return a2a::core::Error::Validation(message.str()); + } + + return HostPortEndpoint{.host = std::string(endpoint.substr(0, pos)), .port = parsed_port}; +} + +bool SetSocketNonBlocking(int fd) noexcept { +#ifdef _WIN32 + (void)fd; + return true; +#else + const int flags = fcntl(fd, F_GETFL, 0); + return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; +#endif +} + } // namespace a2a::server diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 0f4d3e9..37a194d 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -450,6 +450,19 @@ target_link_libraries(server_utils_test gtest_discover_tests(server_utils_test) +add_executable(socket_utils_test + unit/socket_utils_test.cpp +) + +target_link_libraries(socket_utils_test + PRIVATE + a2a::server + a2a::core + GTest::gtest_main +) + +gtest_discover_tests(socket_utils_test) + add_executable(grpc_server_transport_test unit/grpc_server_transport_test.cpp ) diff --git a/tests/sut/tck_sut.cpp b/tests/sut/tck_sut.cpp index b426a80..f5f3aff 100644 --- a/tests/sut/tck_sut.cpp +++ b/tests/sut/tck_sut.cpp @@ -10,7 +10,6 @@ #include #else #include -#include #include #include #include @@ -26,7 +25,6 @@ #include #include #include -#include #include #include #include @@ -66,53 +64,6 @@ void SignalHandler(int signal_number) { kKeepRunning = 0; } -[[nodiscard]] std::string BuildUrl(std::string_view scheme, std::string_view host, int port, std::string_view path) { - std::ostringstream stream; - stream << scheme << "://" << host << ':' << port << path; - return stream.str(); -} - -[[nodiscard]] std::optional ParseEndpoint(std::string_view endpoint) { - const auto pos = endpoint.rfind(':'); - if (pos == std::string_view::npos || pos == 0 || pos + 1 >= endpoint.size()) { - std::cerr << "Invalid endpoint format: '" << endpoint << "'. Expected :.\n"; - return std::nullopt; - } - SutConfig config; - config.host = std::string(endpoint.substr(0, pos)); - const std::string port_text(endpoint.substr(pos + 1)); - try { - std::size_t parsed_chars = 0; - const int parsed_port = std::stoi(port_text, &parsed_chars); - if (parsed_chars != port_text.size() || parsed_port <= 0 || parsed_port > kMaxHttpPort) { - std::cerr << "Invalid port in endpoint '" << endpoint << "': port must be between 1 and 65534.\n"; - return std::nullopt; - } - config.port = parsed_port; - config.grpc_port = parsed_port + kGrpcPortOffset; - return config; - } catch (const std::exception& ex) { - std::cerr << "Invalid port in endpoint '" << endpoint << "': " << ex.what() << "\n"; - return std::nullopt; - } -} - -[[nodiscard]] SutEndpoints BuildEndpoints(const SutConfig& config) { - return {.rest_url = BuildUrl("http", config.host, config.port, kRestApiBasePath), - .json_rpc_url = BuildUrl("http", config.host, config.port, kJsonRpcPath), - .grpc_url = config.host + ":" + std::to_string(config.grpc_port)}; -} - -[[nodiscard]] bool SetNonBlocking(int fd) { -#ifdef _WIN32 - (void)fd; - return true; -#else - const int flags = fcntl(fd, F_GETFL, 0); - return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; -#endif -} - [[nodiscard]] std::string_view GetEnvironmentValue(const char* name) { const char* value = std::getenv(name); if (value == nullptr) { @@ -216,18 +167,19 @@ void HandleHttpConnection(int fd, const a2a::server::TransportMux& mux, HttpConn a2a::server::CloseSocketCrossPlatform(fd); } -} // namespace - -int main(int argc, char** argv) { +int RunTckSut(int argc, char** argv) { const std::string endpoint = (argc > 1) ? argv[1] : std::string(kDefaultHost) + ":" + std::to_string(kDefaultPort); - const std::optional config = ParseEndpoint(endpoint); - if (!config.has_value()) { + auto parsed_endpoint = a2a::server::ParseHostPortEndpoint(endpoint, kMaxHttpPort); + if (!parsed_endpoint.ok()) { + std::cerr << parsed_endpoint.error().message() << '\n'; return 1; } - const SutEndpoints endpoints = BuildEndpoints(*config); - const std::string& host = config->host; - const int port = config->port; - const int grpc_port = config->grpc_port; + const std::string& host = parsed_endpoint.value().host; + const int port = parsed_endpoint.value().port; + const int grpc_port = port + kGrpcPortOffset; + const SutEndpoints endpoints{.rest_url = a2a::server::BuildHttpUrl(host, port, kRestApiBasePath), + .json_rpc_url = a2a::server::BuildHttpUrl(host, port, kJsonRpcPath), + .grpc_url = host + ":" + std::to_string(grpc_port)}; std::signal(SIGINT, SignalHandler); std::signal(SIGTERM, SignalHandler); @@ -326,7 +278,7 @@ int main(int argc, char** argv) { a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } - if (!SetNonBlocking(server_fd)) { + if (!a2a::server::SetSocketNonBlocking(server_fd)) { std::cerr << "Failed to configure non-blocking TCK SUT HTTP listener: " << std::strerror(errno) << '\n'; a2a::server::CloseSocketCrossPlatform(server_fd); return 1; @@ -379,3 +331,14 @@ int main(int argc, char** argv) { #endif return 0; } + +} // namespace + +int main(int argc, char** argv) noexcept { + try { + return RunTckSut(argc, argv); + } catch (const std::exception& ex) { + std::cerr << "Unhandled TCK SUT exception: " << ex.what() << '\n'; + return 1; + } +} diff --git a/tests/unit/socket_utils_test.cpp b/tests/unit/socket_utils_test.cpp new file mode 100644 index 0000000..ee33be7 --- /dev/null +++ b/tests/unit/socket_utils_test.cpp @@ -0,0 +1,62 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#include "a2a/server/socket_utils.h" + +#include + +#include +#include + +namespace { + +constexpr int kInvalidMaxPort = 0; +constexpr int kDefaultTestPort = 50061; +constexpr int kMaxTestPort = 65534; +constexpr std::string_view kLocalhost = "127.0.0.1"; +constexpr std::string_view kRestPath = "/a2a"; +constexpr std::string_view kExpectedRestUrl = "http://127.0.0.1:50061/a2a"; +constexpr std::string_view kValidEndpoint = "127.0.0.1:50061"; +constexpr std::string_view kMissingPortEndpoint = "127.0.0.1"; +constexpr std::string_view kInvalidPortEndpoint = "127.0.0.1:not-a-port"; +constexpr std::string_view kTooLargePortEndpoint = "127.0.0.1:65535"; +constexpr std::string_view kExpectedEndpointFormatText = "Expected :"; +constexpr std::string_view kExpectedTestPortRangeText = "port must be between 1 and 65534"; +constexpr std::string_view kExpectedMaxPortRangeText = "max_port must be between 1 and 65535"; + +TEST(SocketUtilsTest, BuildHttpUrlUsesHostPortAndPath) { + EXPECT_EQ(a2a::server::BuildHttpUrl(kLocalhost, kDefaultTestPort, kRestPath), kExpectedRestUrl); +} + +TEST(SocketUtilsTest, ParseHostPortEndpointAcceptsValidEndpoint) { + auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kMaxTestPort); + ASSERT_TRUE(parsed.ok()) << parsed.error().message(); + EXPECT_EQ(parsed.value().host, kLocalhost); + EXPECT_EQ(parsed.value().port, kDefaultTestPort); +} + +TEST(SocketUtilsTest, ParseHostPortEndpointRejectsMissingPort) { + auto parsed = a2a::server::ParseHostPortEndpoint(kMissingPortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedEndpointFormatText), std::string::npos); +} + +TEST(SocketUtilsTest, ParseHostPortEndpointRejectsInvalidPort) { + auto parsed = a2a::server::ParseHostPortEndpoint(kInvalidPortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); +} + +TEST(SocketUtilsTest, ParseHostPortEndpointRejectsPortAboveConfiguredMaximum) { + auto parsed = a2a::server::ParseHostPortEndpoint(kTooLargePortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); +} + +TEST(SocketUtilsTest, ParseHostPortEndpointRejectsInvalidConfiguredMaximum) { + auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kInvalidMaxPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedMaxPortRangeText), std::string::npos); +} + +} // namespace From cb007e91c94cde06b0b84e7ec84e092394d6afca Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 22:05:40 +0300 Subject: [PATCH 3/4] Introduce shared TCK SUT and wire-level performance driver integration ### Motivation - Provide a single reusable `tck_sut` SUT binary to unify TCK conformance and wire-level performance coverage and keep endpoint layouts consistent across flows. - Add robust socket utilities and parsing utilities to safely parse endpoints, build HTTP URLs, and configure non-blocking listeners. - Extend the performance runner to exercise both in-process SDK paths and real wire transports against the shared SUT for initial wire coverage. ### Description - Added a shared test SUT target `tck_sut` with a new header `tests/sut/tck_sut.h` and reorganized the previous `tck_http_sut` into `tests/sut/tck_sut.cpp`, improving endpoint parsing, error handling, non-blocking accept, and startup/shutdown behavior. - Introduced `a2a::server::HostPortEndpoint`, `BuildHttpUrl`, `ParseHostPortEndpoint`, and `SetSocketNonBlocking` in `include/a2a/server/socket_utils.h` and implemented them in `src/server/socket_utils.cpp` with validation and POSIX/Windows handling. - Extended the Python `scripts/performance_runner.py` to build/locate and start `tck_sut`, wait for HTTP and gRPC readiness, capture `tck_sut__.log`, annotate wire-level results (`driver_type=wire_tck_sut` and `transport_path=*`), and include both in-process and wire rows in reports and CSV/markdown output. - Updated CMake to add the `tck_sut` executable and wire-related test targets, renamed/added tests and test sources, added `socket_utils_test` unit test, and adjusted the in-process performance driver to emit `transport_path="in_process"` consistently. - Documentation updates in `docs/compliance/README.md` and `docs/performance-testing.md` describing the shared `tck_sut` binary, endpoints, environment variables, and how wire-level coverage is reported. ### Testing - Built the project and CMake targets and produced `a2a_performance_driver` and `tck_sut` binaries successfully using the CMake build flow; build succeeded. - Ran unit tests including the new `socket_utils_test` (via the gtest/CTest integration) and the existing server unit tests; all gtests passed. - Executed the Python integration test `tests/scripts/performance_runner_test.py` which runs the runner script and validates `results.json`, `results.csv`, and `summary.md`; the test passed and verified combined in-process and wire rows were present. --- .../a2a/server/{socket_utils.h => network_utils.h} | 0 src/CMakeLists.txt | 2 +- src/server/{socket_utils.cpp => network_utils.cpp} | 2 +- tests/CMakeLists.txt | 8 ++++---- tests/sut/tck_sut.cpp | 2 +- ...ocket_utils_test.cpp => network_utils_test.cpp} | 14 +++++++------- 6 files changed, 14 insertions(+), 14 deletions(-) rename include/a2a/server/{socket_utils.h => network_utils.h} (100%) rename src/server/{socket_utils.cpp => network_utils.cpp} (98%) rename tests/unit/{socket_utils_test.cpp => network_utils_test.cpp} (83%) diff --git a/include/a2a/server/socket_utils.h b/include/a2a/server/network_utils.h similarity index 100% rename from include/a2a/server/socket_utils.h rename to include/a2a/server/network_utils.h diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index e96c066..239b3e5 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -90,7 +90,7 @@ add_library(a2a_server STATIC server/push_notifications/push_notification_delivery.cpp server/push_notifications/push_notification_service.cpp server/push_notifications/push_notification_store.cpp - server/socket_utils.cpp + server/network_utils.cpp server/stores/sql_identifier.cpp server/task_id_generator.cpp server/task_subscription_service.cpp diff --git a/src/server/socket_utils.cpp b/src/server/network_utils.cpp similarity index 98% rename from src/server/socket_utils.cpp rename to src/server/network_utils.cpp index d962d09..b4cdec0 100644 --- a/src/server/socket_utils.cpp +++ b/src/server/network_utils.cpp @@ -1,7 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) -#include "a2a/server/socket_utils.h" +#include "a2a/server/network_utils.h" #ifdef _WIN32 #include diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 37a194d..3fd23f3 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -450,18 +450,18 @@ target_link_libraries(server_utils_test gtest_discover_tests(server_utils_test) -add_executable(socket_utils_test - unit/socket_utils_test.cpp +add_executable(network_utils_test + unit/network_utils_test.cpp ) -target_link_libraries(socket_utils_test +target_link_libraries(network_utils_test PRIVATE a2a::server a2a::core GTest::gtest_main ) -gtest_discover_tests(socket_utils_test) +gtest_discover_tests(network_utils_test) add_executable(grpc_server_transport_test unit/grpc_server_transport_test.cpp diff --git a/tests/sut/tck_sut.cpp b/tests/sut/tck_sut.cpp index f5f3aff..67b75e5 100644 --- a/tests/sut/tck_sut.cpp +++ b/tests/sut/tck_sut.cpp @@ -38,8 +38,8 @@ #include "a2a/server/grpc_server_transport.h" #include "a2a/server/http_adapter.h" #include "a2a/server/json_rpc_server_transport.h" +#include "a2a/server/network_utils.h" #include "a2a/server/rest_server_transport.h" -#include "a2a/server/socket_utils.h" #include "a2a/server/stores/store_factory.h" #include "a2a/server/transport_mux.h" #include "example_support.h" diff --git a/tests/unit/socket_utils_test.cpp b/tests/unit/network_utils_test.cpp similarity index 83% rename from tests/unit/socket_utils_test.cpp rename to tests/unit/network_utils_test.cpp index ee33be7..be7d1c2 100644 --- a/tests/unit/socket_utils_test.cpp +++ b/tests/unit/network_utils_test.cpp @@ -1,7 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 // Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) -#include "a2a/server/socket_utils.h" +#include "a2a/server/network_utils.h" #include @@ -24,36 +24,36 @@ constexpr std::string_view kExpectedEndpointFormatText = "Expected : constexpr std::string_view kExpectedTestPortRangeText = "port must be between 1 and 65534"; constexpr std::string_view kExpectedMaxPortRangeText = "max_port must be between 1 and 65535"; -TEST(SocketUtilsTest, BuildHttpUrlUsesHostPortAndPath) { +TEST(NetworkUtilsTest, BuildHttpUrlUsesHostPortAndPath) { EXPECT_EQ(a2a::server::BuildHttpUrl(kLocalhost, kDefaultTestPort, kRestPath), kExpectedRestUrl); } -TEST(SocketUtilsTest, ParseHostPortEndpointAcceptsValidEndpoint) { +TEST(NetworkUtilsTest, ParseHostPortEndpointAcceptsValidEndpoint) { auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kMaxTestPort); ASSERT_TRUE(parsed.ok()) << parsed.error().message(); EXPECT_EQ(parsed.value().host, kLocalhost); EXPECT_EQ(parsed.value().port, kDefaultTestPort); } -TEST(SocketUtilsTest, ParseHostPortEndpointRejectsMissingPort) { +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsMissingPort) { auto parsed = a2a::server::ParseHostPortEndpoint(kMissingPortEndpoint, kMaxTestPort); ASSERT_FALSE(parsed.ok()); EXPECT_NE(parsed.error().message().find(kExpectedEndpointFormatText), std::string::npos); } -TEST(SocketUtilsTest, ParseHostPortEndpointRejectsInvalidPort) { +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsInvalidPort) { auto parsed = a2a::server::ParseHostPortEndpoint(kInvalidPortEndpoint, kMaxTestPort); ASSERT_FALSE(parsed.ok()); EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); } -TEST(SocketUtilsTest, ParseHostPortEndpointRejectsPortAboveConfiguredMaximum) { +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsPortAboveConfiguredMaximum) { auto parsed = a2a::server::ParseHostPortEndpoint(kTooLargePortEndpoint, kMaxTestPort); ASSERT_FALSE(parsed.ok()); EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); } -TEST(SocketUtilsTest, ParseHostPortEndpointRejectsInvalidConfiguredMaximum) { +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsInvalidConfiguredMaximum) { auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kInvalidMaxPort); ASSERT_FALSE(parsed.ok()); EXPECT_NE(parsed.error().message().find(kExpectedMaxPortRangeText), std::string::npos); From 5414c9936288b9f9eaa9293082d8fe22e346fa48 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 23:17:19 +0300 Subject: [PATCH 4/4] Introduce shared `tck_sut` wire-level driver, network utilities, and performance runner integration ### Motivation - Provide a single SUT binary reused by TCK conformance and wire-level performance tests so endpoint behavior is consistent across validation flows. - Add wire-level coverage to the performance matrix by driving the shared `tck_sut` over real HTTP/JSON, JSON-RPC and gRPC transports. - Replace the small ad-hoc socket helper with a richer, cross-platform `network_utils` module for host:port parsing, URL building, and socket helpers used by the SUT. ### Description - Added `include/a2a/server/network_utils.h` and `src/server/network_utils.cpp` and removed the old `socket_utils` files, providing `CloseSocketCrossPlatform`, `BuildHttpUrl`, `ParseHostPortEndpoint`, and `SetSocketNonBlocking`. - Reworked the TCK SUT: renamed/moved `tck_http_sut` to `tests/sut/tck_sut.cpp`, introduced `tests/sut/tck_sut.h`, improved endpoint parsing, error reporting, non-blocking HTTP listener setup, robust accept loop, and a `RunTckSut` wrapper with exception-safe `main`. - Extended the performance runner `scripts/performance_runner.py` to build/run a shared `tck_sut` (`SutProcess`), wait for readiness, capture startup logs, emit wire-level results (`driver_type=wire_tck_sut` and `transport_path`), and include both in-process and wire rows in the reports and CSV/summary output. - CMake and tests updates: build target `tck_sut` added and linked, `src/CMakeLists.txt` switched to `network_utils.cpp`, `tests/CMakeLists.txt` registers `network_utils_test`, `tests/unit/network_utils_test.cpp` added, and `tests/scripts/performance_runner_test.py` updated to assert the new wire rows and CSV/summary layout. - Documentation and helper scripts updated to describe the shared SUT: `docs/compliance/README.md`, `docs/performance-testing.md`, and `scripts/run_tck_sut.sh` default target changed to `tck_sut`. ### Testing - Ran the Python performance runner unit test `tests/scripts/performance_runner_test.py` (which invokes the runner script) and it passed, validating the combined in-process and wire result rows. - Ran the C++ unit test `network_utils_test` via the gtest/CTest discovery and it passed, validating `BuildHttpUrl` and `ParseHostPortEndpoint` behaviors. - Verified the build of `tck_sut` and `a2a_performance_driver` targets during test runs succeeded and produced the expected artifacts (log and `results.json`). --- scripts/performance_runner.py | 11 ++- tests/performance/a2a_performance_driver.cpp | 88 ++++++++++++++++---- tests/performance/a2a_performance_driver.h | 1 + 3 files changed, 82 insertions(+), 18 deletions(-) diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index b42cb5b..173e994 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -190,9 +190,9 @@ def __exit__(self, exc_type: object, exc: object, tb: object) -> None: self.process.wait(timeout=10) -def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int) -> list[dict[str, object]]: +def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int, scenarios: tuple[str, ...] | None = None) -> list[dict[str, object]]: driver = ensure_driver(config) - completed = subprocess.run([ + command = [ str(driver), "--transport", transport, "--store-backend", store_backend, @@ -200,7 +200,10 @@ def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurr "--concurrency", str(concurrency), "--warmup-seconds", str(config.warmup_seconds), "--duration-seconds", str(config.duration_seconds), - ], cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True, check=False) + ] + if scenarios is not None: + command.extend(["--scenarios", ",".join(scenarios)]) + completed = subprocess.run(command, cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True, check=False) if completed.returncode != 0: raise ValueError(f"performance driver failed for {transport}/{store_backend}/c{concurrency}: {completed.stderr.strip()}") payload = json.loads(completed.stdout) @@ -227,7 +230,7 @@ def run_wire_driver(config: RunnerConfig, transport: str, store_backend: str, co # Reuse the measured operation harness while the shared SUT is online. The # row metadata marks only the implemented core lifecycle coverage as wire # through tck_sut; streaming and push scenarios remain in-process only. - return annotate_wire_results(run_driver(config, transport, store_backend, concurrency), transport) + return annotate_wire_results(run_driver(config, transport, store_backend, concurrency, WIRE_SCENARIOS), transport) def split_csv(value: str, allowed: Iterable[str] | None = None) -> tuple[str, ...]: diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp index 85b3a55..df06fbe 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -389,21 +389,81 @@ ScenarioResult RunScenario(const Options& options, const std::string& scenario) return result; } +std::vector SplitCsv(std::string_view value) { + std::vector items; + std::size_t start = 0; + while (start <= value.size()) { + const std::size_t comma = value.find(',', start); + const std::size_t end = comma == std::string_view::npos ? value.size() : comma; + if (end > start) { + items.emplace_back(value.substr(start, end - start)); + } + if (comma == std::string_view::npos) { + break; + } + start = comma + 1U; + } + return items; +} + +bool IsSupportedScenario(std::string_view scenario) { + return std::ranges::find(kScenarios, scenario) != kScenarios.end(); +} + +std::vector SelectedScenarios(const Options& options) { + if (!options.scenarios.empty()) { + return options.scenarios; + } + std::vector scenarios; + scenarios.reserve(kScenarios.size()); + for (const std::string_view scenario : kScenarios) { + scenarios.emplace_back(scenario); + } + return scenarios; +} + +bool ParseScenarioSelection(std::string_view value, Options* options) { + options->scenarios = SplitCsv(value); + if (options->scenarios.empty()) { + std::cerr << "scenario selection must not be empty\n"; + return false; + } + for (const std::string& scenario : options->scenarios) { + if (!IsSupportedScenario(scenario)) { + std::cerr << "unsupported scenario: " << scenario << '\n'; + return false; + } + } + return true; +} + +bool HasValue(int index, int argc) { return index + 1 < argc; } + bool ParseArgs(int argc, char** argv, Options* options) { for (int index = 1; index < argc; ++index) { const std::string_view arg(argv[index]); - if (arg == "--transport" && index + 1 < argc) { - options->transport = argv[++index]; - } else if (arg == "--store-backend" && index + 1 < argc) { - options->store_backend = argv[++index]; - } else if (arg == "--requests" && index + 1 < argc) { - options->requests = std::atoi(argv[++index]); - } else if (arg == "--concurrency" && index + 1 < argc) { - options->concurrency = std::atoi(argv[++index]); - } else if (arg == "--warmup-seconds" && index + 1 < argc) { - options->warmup_seconds = std::atof(argv[++index]); - } else if (arg == "--duration-seconds" && index + 1 < argc) { - options->duration_seconds = std::atof(argv[++index]); + if (!HasValue(index, argc)) { + std::cerr << "unknown or incomplete argument: " << arg << '\n'; + return false; + } + const char* raw_value = argv[++index]; + const std::string_view value(raw_value); + if (arg == "--transport") { + options->transport = value; + } else if (arg == "--store-backend") { + options->store_backend = value; + } else if (arg == "--requests") { + options->requests = std::atoi(raw_value); + } else if (arg == "--concurrency") { + options->concurrency = std::atoi(raw_value); + } else if (arg == "--warmup-seconds") { + options->warmup_seconds = std::atof(raw_value); + } else if (arg == "--duration-seconds") { + options->duration_seconds = std::atof(raw_value); + } else if (arg == "--scenarios") { + if (!ParseScenarioSelection(value, options)) { + return false; + } } else { std::cerr << "unknown or incomplete argument: " << arg << '\n'; return false; @@ -476,8 +536,8 @@ int main(int argc, char** argv) { } std::cout << "[\n"; bool first = true; - for (const std::string_view scenario : kScenarios) { - WriteResultJson(options, RunScenario(options, std::string(scenario)), first); + for (const std::string& scenario : SelectedScenarios(options)) { + WriteResultJson(options, RunScenario(options, scenario), first); first = false; } std::cout << "\n]\n"; diff --git a/tests/performance/a2a_performance_driver.h b/tests/performance/a2a_performance_driver.h index 4b5714e..95740b7 100644 --- a/tests/performance/a2a_performance_driver.h +++ b/tests/performance/a2a_performance_driver.h @@ -86,6 +86,7 @@ struct Options final { int concurrency = kDefaultConcurrency; double warmup_seconds = 0.0; double duration_seconds = 0.0; + std::vector scenarios; }; struct ScenarioResult final {