diff --git a/.github/workflows/python-check-coverage.py b/.github/workflows/python-check-coverage.py new file mode 100644 index 0000000000..c8f96cd0ab --- /dev/null +++ b/.github/workflows/python-check-coverage.py @@ -0,0 +1,266 @@ +#!/usr/bin/env python3 +# Copyright (c) Microsoft. All rights reserved. +"""Check Python test coverage against threshold for enforced modules. + +This script parses a Cobertura XML coverage report and enforces a minimum +coverage threshold on specific modules. Non-enforced modules are reported +for visibility but don't block the build. + +Usage: + python python-check-coverage.py + +Example: + python python-check-coverage.py python-coverage.xml 85 +""" + +import sys +import xml.etree.ElementTree as ET +from dataclasses import dataclass + +# ============================================================================= +# ENFORCED MODULES CONFIGURATION +# ============================================================================= +# Add or remove modules from this set to control which packages must meet +# the coverage threshold. Only these modules will fail the build if below +# threshold. Other modules are reported for visibility only. +# +# Module paths should match the package paths as they appear in the coverage +# report (e.g., "packages.azure-ai.agent_framework_azure_ai" for packages/azure-ai). +# Sub-modules can be included by specifying their full path. +# ============================================================================= +ENFORCED_MODULES: set[str] = { + "packages.azure-ai.agent_framework_azure_ai", + # Add more modules here as coverage improves: + # "packages.core.agent_framework", + # "packages.core.agent_framework._workflows", + # "packages.anthropic.agent_framework_anthropic", +} + + +@dataclass +class PackageCoverage: + """Coverage data for a single package.""" + + name: str + line_rate: float + branch_rate: float + lines_valid: int + lines_covered: int + branches_valid: int + branches_covered: int + + @property + def line_coverage_percent(self) -> float: + """Return line coverage as a percentage.""" + return self.line_rate * 100 + + @property + def branch_coverage_percent(self) -> float: + """Return branch coverage as a percentage.""" + return self.branch_rate * 100 + + +def parse_coverage_xml(xml_path: str) -> tuple[dict[str, PackageCoverage], float, float]: + """Parse Cobertura XML and extract per-package coverage data. + + Args: + xml_path: Path to the Cobertura XML coverage report. + + Returns: + A tuple of (packages_dict, overall_line_rate, overall_branch_rate). + """ + tree = ET.parse(xml_path) + root = tree.getroot() + + # Get overall coverage from root element + overall_line_rate = float(root.get("line-rate", 0)) + overall_branch_rate = float(root.get("branch-rate", 0)) + + packages: dict[str, PackageCoverage] = {} + + for package in root.findall(".//package"): + package_path = package.get("name", "unknown") + + line_rate = float(package.get("line-rate", 0)) + branch_rate = float(package.get("branch-rate", 0)) + + # Count lines and branches from classes within this package + lines_valid = 0 + lines_covered = 0 + branches_valid = 0 + branches_covered = 0 + + for class_elem in package.findall(".//class"): + for line in class_elem.findall(".//line"): + lines_valid += 1 + if int(line.get("hits", 0)) > 0: + lines_covered += 1 + # Branch coverage from line elements + if line.get("branch") == "true": + condition_coverage = line.get("condition-coverage", "") + if condition_coverage: + # Parse "X% (covered/total)" format + try: + coverage_parts = condition_coverage.split("(")[1].rstrip(")").split("/") + branches_covered += int(coverage_parts[0]) + branches_valid += int(coverage_parts[1]) + except (IndexError, ValueError): + # Ignore malformed condition-coverage strings; treat this line as having no branch data. + pass + + # Use full package path as the key (no aggregation) + packages[package_path] = PackageCoverage( + name=package_path, + line_rate=line_rate if lines_valid == 0 else lines_covered / lines_valid, + branch_rate=branch_rate if branches_valid == 0 else branches_covered / branches_valid, + lines_valid=lines_valid, + lines_covered=lines_covered, + branches_valid=branches_valid, + branches_covered=branches_covered, + ) + + return packages, overall_line_rate, overall_branch_rate + + +def format_coverage_value(coverage: float, threshold: float, is_enforced: bool) -> str: + """Format a coverage value with optional pass/fail indicator. + + Args: + coverage: Coverage percentage (0-100). + threshold: Minimum required coverage percentage. + is_enforced: Whether this module is enforced. + + Returns: + Formatted string like "85.5%" or "85.5% ✅" or "75.0% ❌". + """ + formatted = f"{coverage:.1f}%" + if is_enforced: + icon = "✅" if coverage >= threshold else "❌" + formatted = f"{formatted} {icon}" + return formatted + + +def print_coverage_table( + packages: dict[str, PackageCoverage], + threshold: float, + overall_line_rate: float, + overall_branch_rate: float, +) -> None: + """Print a formatted coverage summary table. + + Args: + packages: Dictionary of package name to coverage data. + threshold: Minimum required coverage percentage. + overall_line_rate: Overall line coverage rate (0-1). + overall_branch_rate: Overall branch coverage rate (0-1). + """ + print("\n" + "=" * 80) + print("PYTHON TEST COVERAGE REPORT") + print("=" * 80) + + # Overall coverage + print(f"\nOverall Line Coverage: {overall_line_rate * 100:.1f}%") + print(f"Overall Branch Coverage: {overall_branch_rate * 100:.1f}%") + print(f"Threshold: {threshold}%") + + # Package table + print("\n" + "-" * 110) + print(f"{'Package':<80} {'Lines':<15} {'Line Cov':<15}") + print("-" * 110) + + # Sort: enforced modules first, then alphabetically + sorted_packages = sorted( + packages.values(), + key=lambda p: (p.name not in ENFORCED_MODULES, p.name), + ) + + for pkg in sorted_packages: + is_enforced = pkg.name in ENFORCED_MODULES + enforced_marker = "[ENFORCED] " if is_enforced else "" + line_cov = format_coverage_value(pkg.line_coverage_percent, threshold, is_enforced) + lines_info = f"{pkg.lines_covered}/{pkg.lines_valid}" + package_label = f"{enforced_marker}{pkg.name}" + + print(f"{package_label:<80} {lines_info:<15} {line_cov:<15}") + + print("-" * 110) + + +def check_coverage(xml_path: str, threshold: float) -> bool: + """Check if all enforced modules meet the coverage threshold. + + Args: + xml_path: Path to the Cobertura XML coverage report. + threshold: Minimum required coverage percentage. + + Returns: + True if all enforced modules pass, False otherwise. + """ + packages, overall_line_rate, overall_branch_rate = parse_coverage_xml(xml_path) + + print_coverage_table(packages, threshold, overall_line_rate, overall_branch_rate) + + # Check enforced modules + failed_modules: list[str] = [] + missing_modules: list[str] = [] + + for module_name in ENFORCED_MODULES: + if module_name not in packages: + missing_modules.append(module_name) + continue + + pkg = packages[module_name] + if pkg.line_coverage_percent < threshold: + failed_modules.append(f"{module_name} ({pkg.line_coverage_percent:.1f}%)") + + # Report results + if missing_modules: + print(f"\n❌ FAILED: Enforced modules not found in coverage report: {', '.join(missing_modules)}") + return False + + if failed_modules: + print(f"\n❌ FAILED: The following enforced modules are below {threshold}% coverage threshold:") + for module in failed_modules: + print(f" - {module}") + print("\nTo fix: Add more tests to improve coverage for the failing modules.") + return False + + if ENFORCED_MODULES: + found_enforced = [m for m in ENFORCED_MODULES if m in packages] + if found_enforced: + print(f"\n✅ PASSED: All enforced modules meet the {threshold}% coverage threshold.") + + return True + + +def main() -> int: + """Main entry point. + + Returns: + Exit code: 0 for success, 1 for failure. + """ + if len(sys.argv) != 3: + print(f"Usage: {sys.argv[0]} ") + print(f"Example: {sys.argv[0]} python-coverage.xml 85") + return 1 + + xml_path = sys.argv[1] + try: + threshold = float(sys.argv[2]) + except ValueError: + print(f"Error: Invalid threshold value: {sys.argv[2]}") + return 1 + + try: + success = check_coverage(xml_path, threshold) + return 0 if success else 1 + except FileNotFoundError: + print(f"Error: Coverage file not found: {xml_path}") + return 1 + except ET.ParseError as e: + print(f"Error: Failed to parse coverage XML: {e}") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/python-test-coverage.yml b/.github/workflows/python-test-coverage.yml index 03cca20e06..a9acfba0de 100644 --- a/.github/workflows/python-test-coverage.yml +++ b/.github/workflows/python-test-coverage.yml @@ -9,6 +9,8 @@ on: env: # Configure a constant location for the uv cache UV_CACHE_DIR: /tmp/.uv-cache + # Coverage threshold percentage for enforced modules + COVERAGE_THRESHOLD: 85 jobs: python-tests-coverage: @@ -37,6 +39,8 @@ jobs: UV_CACHE_DIR: /tmp/.uv-cache - name: Run all tests with coverage report run: uv run poe all-tests-cov --cov-report=xml:python-coverage.xml -q --junitxml=pytest.xml + - name: Check coverage threshold + run: python ${{ github.workspace }}/.github/workflows/python-check-coverage.py python-coverage.xml ${{ env.COVERAGE_THRESHOLD }} - name: Upload coverage report uses: actions/upload-artifact@v6 with: diff --git a/python/packages/core/tests/workflow/test_magentic.py b/python/packages/core/tests/workflow/test_magentic.py index ba0b899c1c..9c6a2521b1 100644 --- a/python/packages/core/tests/workflow/test_magentic.py +++ b/python/packages/core/tests/workflow/test_magentic.py @@ -1245,7 +1245,7 @@ def agent_factory() -> AgentProtocol: custom_final_prompt = "Custom final: {task}" # Create a custom task ledger - from agent_framework._workflows._magentic import _MagenticTaskLedger + from agent_framework._workflows._magentic import _MagenticTaskLedger # type: ignore custom_task_ledger = _MagenticTaskLedger( facts=ChatMessage(role=Role.ASSISTANT, text="Custom facts"), diff --git a/python/samples/getting_started/agents/azure_ai/azure_ai_with_thread.py b/python/samples/getting_started/agents/azure_ai/azure_ai_with_thread.py index 766ee5fa51..ff2fa72a52 100644 --- a/python/samples/getting_started/agents/azure_ai/azure_ai_with_thread.py +++ b/python/samples/getting_started/agents/azure_ai/azure_ai_with_thread.py @@ -4,10 +4,10 @@ from random import randint from typing import Annotated +from agent_framework import tool from agent_framework.azure import AzureAIProjectAgentProvider from azure.identity.aio import AzureCliCredential from pydantic import Field -from agent_framework import tool """ Azure AI Agent with Thread Management Example @@ -16,6 +16,7 @@ persistent conversation capabilities using service-managed threads as well as storing messages in-memory. """ + # NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/getting_started/tools/function_tool_with_approval.py and samples/getting_started/tools/function_tool_with_approval_and_threads.py. @tool(approval_mode="never_require") def get_weather(