feat(conformance): add report generation to adk conformance test command

Added `generate_report` and `report_dir` CLI args to the command.

Co-authored-by: Liang Wu <wuliang@google.com>
PiperOrigin-RevId: 867689055
This commit is contained in:
Liang Wu
2026-02-09 11:15:11 -08:00
committed by Copybara-Service
parent d2dba27134
commit 43c437e38b
4 changed files with 140 additions and 7 deletions
+41 -3
View File
@@ -294,11 +294,28 @@ def cli_conformance_record(
" runs evaluation-based verification."
),
)
@click.option(
"--generate_report",
is_flag=True,
show_default=True,
default=False,
help="Optional. Whether to generate a Markdown report of the test results.",
)
@click.option(
"--report_dir",
type=click.Path(file_okay=False, dir_okay=True, resolve_path=True),
help=(
"Optional. Directory to store the generated report. Defaults to current"
" directory."
),
)
@click.pass_context
def cli_conformance_test(
ctx,
paths: tuple[str, ...],
mode: str,
generate_report: bool,
report_dir: Optional[str] = None,
):
"""Run conformance tests to verify agent behavior consistency.
@@ -309,7 +326,7 @@ def cli_conformance_test(
- Contain a spec.yaml file directly (single test case)
- Contain subdirectories with spec.yaml files (multiple test cases)
If no paths are provided, defaults to searching the 'tests' folder.
If no paths are provided, defaults to searching for the 'tests' folder.
TEST MODES:
@@ -329,6 +346,11 @@ def cli_conformance_test(
generated-recordings.yaml # Recorded interactions (replay mode)
generated-session.yaml # Session data (replay mode)
REPORT GENERATION:
Use --generate_report to create a Markdown report of test results.
Use --report_dir to specify where the report should be saved.
EXAMPLES:
\b
@@ -346,6 +368,14 @@ def cli_conformance_test(
\b
# Run in live mode (when available)
adk conformance test --mode=live tests/core
\b
# Generate a test report
adk conformance test --generate_report
\b
# Generate a test report in a specific directory
adk conformance test --generate_report --report_dir=reports
"""
try:
@@ -363,10 +393,18 @@ def cli_conformance_test(
)
ctx.exit(1)
# Convert to Path objects, use default if empty (paths are already resolved by Click)
# Convert to Path objects, use default if empty (paths are already resolved
# by Click)
test_paths = [Path(p) for p in paths] if paths else [Path("tests").resolve()]
asyncio.run(run_conformance_test(test_paths=test_paths, mode=mode.lower()))
asyncio.run(
run_conformance_test(
test_paths=test_paths,
mode=mode.lower(),
generate_report=generate_report,
report_dir=report_dir,
)
)
@main.command("create", cls=HelpfulCommand)
@@ -0,0 +1,83 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Utilities for generating Markdown reports for conformance tests."""
from __future__ import annotations
from pathlib import Path
from typing import Optional
from typing import TYPE_CHECKING
import click
from ... import version
if TYPE_CHECKING:
from .cli_test import _ConformanceTestSummary
def generate_markdown_report(
summary: _ConformanceTestSummary, report_dir: Optional[str]
) -> None:
"""Generates a Markdown report of the test results."""
report_name = f"python_{'_'.join(version.__version__.split('.'))}_report.md"
if not report_dir:
report_path = Path(report_name)
else:
report_path = Path(report_dir) / report_name
report_path.parent.mkdir(parents=True, exist_ok=True)
with open(report_path, "w") as f:
f.write("# ADK Python Conformance Test Report\n\n")
# Summary
f.write("## Summary\n\n")
f.write(f"- **ADK Version**: {version.__version__}\n")
f.write(f"- **Total Tests**: {summary.total_tests}\n")
f.write(f"- **Passed**: {summary.passed_tests}\n")
f.write(f"- **Failed**: {summary.failed_tests}\n")
f.write(f"- **Success Rate**: {summary.success_rate:.1f}%\n\n")
# Table
f.write("## Test Results\n\n")
f.write("| Status | Category | Test Name | Description |\n")
f.write("| :--- | :--- | :--- | :--- |\n")
for result in summary.results:
status_icon = "✅ PASS" if result.success else "❌ FAIL"
description = (
result.description.replace("\n", " ") if result.description else ""
)
f.write(
f"| {status_icon} | {result.category} | {result.name} |"
f" {description} |\n"
)
f.write("\n")
# Failed Tests Details
if summary.failed_tests > 0:
f.write("## Failed Tests Details\n\n")
for result in summary.results:
if not result.success:
f.write(f"### {result.category}/{result.name}\n\n")
if result.description:
f.write(f"**Description**: {result.description}\n\n")
f.write("**Error**:\n")
f.write("```\n")
f.write(f"{result.error_message}\n")
f.write("```\n\n")
click.secho(f"\nReport generated at: {report_path.resolve()}", fg="blue")
@@ -69,7 +69,7 @@ def _generate_diff_message(
return _generate_mismatch_message(context, actual_json, recorded_json)
def compare_event(
def _compare_event(
actual_event: Event, recorded_event: Event, index: int
) -> ComparisonResult:
"""Compare a single actual event with a recorded event."""
@@ -133,7 +133,7 @@ def compare_events(
)
for i, (actual, recorded) in enumerate(zip(actual_events, recorded_events)):
result = compare_event(actual, recorded, i)
result = _compare_event(actual, recorded, i)
if not result.success:
return result
+14 -2
View File
@@ -25,13 +25,13 @@ import click
from google.genai import types
from ..adk_web_server import RunAgentRequest
from ._generate_markdown_utils import generate_markdown_report
from ._generated_file_utils import load_recorded_session
from ._generated_file_utils import load_test_case
from ._replay_validators import compare_events
from ._replay_validators import compare_session
from .adk_web_server_client import AdkWebServerClient
from .test_case import TestCase
from .test_case import TestSpec
@dataclass
@@ -42,6 +42,7 @@ class _TestResult:
name: str
success: bool
error_message: Optional[str] = None
description: Optional[str] = None
@dataclass
@@ -62,7 +63,7 @@ class _ConformanceTestSummary:
class ConformanceTestRunner:
"""Runs conformance tests in replay mode."""
"""Runs conformance tests."""
def __init__(
self,
@@ -193,6 +194,7 @@ class ConformanceTestRunner:
name=test_case.name,
success=False,
error_message="No final session available for comparison",
description=test_case.test_spec.description,
)
# Load recorded session data for comparison
@@ -203,6 +205,7 @@ class ConformanceTestRunner:
name=test_case.name,
success=False,
error_message="No recorded session found for replay comparison",
description=test_case.test_spec.description,
)
# Compare events and session
@@ -224,6 +227,7 @@ class ConformanceTestRunner:
name=test_case.name,
success=success,
error_message="\n\n".join(error_messages) if error_messages else None,
description=test_case.test_spec.description,
)
async def _run_test_case_replay(self, test_case: TestCase) -> _TestResult:
@@ -245,6 +249,7 @@ class ConformanceTestRunner:
name=test_case.name,
success=False,
error_message=f"Replay verification failed: {e}",
description=test_case.test_spec.description,
)
# Validate results and return test result
@@ -265,6 +270,7 @@ class ConformanceTestRunner:
name=test_case.name,
success=False,
error_message=f"Test setup failed: {e}",
description=test_case.test_spec.description,
)
async def run_all_tests(self) -> _ConformanceTestSummary:
@@ -295,6 +301,7 @@ Found {len(test_cases)} test cases to run in {self.mode} mode
name=test_case.name,
success=False,
error_message="Live mode not yet implemented",
description=test_case.test_spec.description,
)
results.append(result)
_print_test_case_result(result)
@@ -311,6 +318,8 @@ Found {len(test_cases)} test cases to run in {self.mode} mode
async def run_conformance_test(
test_paths: list[Path],
mode: str = "replay",
generate_report: bool = False,
report_dir: Optional[str] = None,
) -> None:
"""Run conformance tests."""
_print_test_header(mode)
@@ -319,6 +328,9 @@ async def run_conformance_test(
runner = ConformanceTestRunner(test_paths, client, mode)
summary = await runner.run_all_tests()
if generate_report:
generate_markdown_report(summary, report_dir)
_print_test_summary(summary)