From 41d951b184db895dc0e5480153bcfae9e0681ebb Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Wed, 22 Jul 2026 16:22:37 +0200 Subject: [PATCH 1/3] Enforce Python coverage by package lifecycle Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9ae5ad8e-6b66-41b3-a862-4e2a3fae1cd0 --- .github/scripts/python_check_coverage.py | 212 ++++++++++ .github/tests/test_python_check_coverage.py | 134 ++++++ .github/workflows/github-automation-tests.yml | 11 +- .github/workflows/python-check-coverage.py | 382 ------------------ .github/workflows/python-test-coverage.yml | 7 +- python/.github/skills/python-testing/SKILL.md | 5 +- python/DEV_SETUP.md | 4 + .../azurefunctions/tests/test_func_utils.py | 33 ++ .../tests/test_workflow_af_context.py | 142 +++++++ .../bedrock/tests/test_bedrock_client.py | 265 +++++++++++- .../packages/chatkit/tests/test_converter.py | 294 +++++++++++++- .../durabletask/tests/test_async_bridge.py | 75 ++++ .../tests/test_workflow_dt_context.py | 121 ++++++ .../test_workflow_orchestrator_helpers.py | 363 +++++++++++++++++ .../tests/test_workflow_runner_context.py | 103 +++++ .../hyperlight/test_hyperlight_codeact.py | 82 ++++ .../monty/tests/monty/test_monty_codeact.py | 157 +++++++ .../tests/test_context_provider_edges.py | 234 +++++++++++ .../tools/tests/test_docker_shell_tool.py | 272 +++++++++++++ .../tools/tests/test_local_shell_tool.py | 140 ++++++- .../tools/tests/test_shell_killtree.py | 147 +++++++ .../tools/tests/test_shell_resolve.py | 48 ++- 22 files changed, 2823 insertions(+), 408 deletions(-) create mode 100644 .github/scripts/python_check_coverage.py create mode 100644 .github/tests/test_python_check_coverage.py delete mode 100644 .github/workflows/python-check-coverage.py create mode 100644 python/packages/azurefunctions/tests/test_workflow_af_context.py create mode 100644 python/packages/durabletask/tests/test_async_bridge.py create mode 100644 python/packages/durabletask/tests/test_workflow_dt_context.py create mode 100644 python/packages/durabletask/tests/test_workflow_orchestrator_helpers.py create mode 100644 python/packages/durabletask/tests/test_workflow_runner_context.py create mode 100644 python/packages/redis/tests/test_context_provider_edges.py create mode 100644 python/packages/tools/tests/test_shell_killtree.py diff --git a/.github/scripts/python_check_coverage.py b/.github/scripts/python_check_coverage.py new file mode 100644 index 0000000000..e0ad15da83 --- /dev/null +++ b/.github/scripts/python_check_coverage.py @@ -0,0 +1,212 @@ +# Copyright (c) Microsoft. All rights reserved. +"""Enforce Python package coverage according to package lifecycle.""" + +# ruff:file-ignore[print] +# ruff:file-ignore[implicit-namespace-package] + +from __future__ import annotations + +import re +import sys +import xml.etree.ElementTree as ET # ruff:ignore[suspicious-xml-etree-import] +from dataclasses import dataclass +from pathlib import Path + +import tomllib + +DEVELOPMENT_STATUS_PREFIX = "Development Status :: " +ENFORCED_DEVELOPMENT_STATUS = 4 +EXEMPT_PACKAGES = {"devui", "lab"} + + +@dataclass(frozen=True) +class PackagePolicy: + """Coverage policy derived from a package's project metadata.""" + + directory: str + distribution_name: str + development_status: int + development_status_label: str + enforced: bool + exempt: bool + + +@dataclass +class CoverageStats: + """Line and branch coverage counters.""" + + lines_valid: int = 0 + lines_covered: int = 0 + branches_valid: int = 0 + branches_covered: int = 0 + + @property + def line_coverage_percent(self) -> float: + """Return line coverage as a percentage.""" + if not self.lines_valid: + return 0 + return self.lines_covered / self.lines_valid * 100 + + +def normalize_coverage_path(path: str) -> str: + """Normalize a coverage path for matching.""" + return path.replace("\\", "/").lstrip("./") + + +def load_package_policies(packages_dir: Path) -> list[PackagePolicy]: + """Load lifecycle-based coverage policies from package pyproject files.""" + policies: list[PackagePolicy] = [] + for pyproject_path in sorted(packages_dir.glob("*/pyproject.toml")): + with pyproject_path.open("rb") as pyproject_file: + pyproject = tomllib.load(pyproject_file) + + project = pyproject.get("project", {}) + distribution_name = str(project.get("name", "")).strip() + if not distribution_name: + raise ValueError(f"{pyproject_path}: project.name is required") + + status_classifiers = [ + classifier + for classifier in project.get("classifiers", []) + if classifier.startswith(DEVELOPMENT_STATUS_PREFIX) + ] + if len(status_classifiers) != 1: + raise ValueError( + f"{pyproject_path}: expected exactly one Development Status classifier, found {len(status_classifiers)}" + ) + + match = re.fullmatch(r"Development Status :: (\d+) - (.+)", status_classifiers[0]) + if match is None: + raise ValueError(f"{pyproject_path}: malformed Development Status classifier") + + directory = pyproject_path.parent.name + development_status = int(match.group(1)) + exempt = directory in EXEMPT_PACKAGES + policies.append( + PackagePolicy( + directory=directory, + distribution_name=distribution_name, + development_status=development_status, + development_status_label=match.group(2), + enforced=development_status >= ENFORCED_DEVELOPMENT_STATUS and not exempt, + exempt=exempt, + ) + ) + + if not policies: + raise ValueError(f"No package pyproject.toml files found below {packages_dir}") + return policies + + +def parse_coverage_xml(xml_path: Path) -> tuple[dict[str, CoverageStats], float, float]: + """Parse Cobertura XML and aggregate coverage by package directory.""" + root = ET.parse(xml_path).getroot() # ruff:ignore[suspicious-xml-element-tree-usage] # Trusted CI-generated coverage report. + package_stats: dict[str, CoverageStats] = {} + + for class_elem in root.findall(".//class"): + file_path = normalize_coverage_path(class_elem.get("filename", "")) + path_parts = file_path.split("/") + try: + packages_index = path_parts.index("packages") + package_directory = path_parts[packages_index + 1] + except (ValueError, IndexError): + continue + + stats = package_stats.setdefault(package_directory, CoverageStats()) + for line in class_elem.findall(".//line"): + stats.lines_valid += 1 + if int(line.get("hits", 0)) > 0: + stats.lines_covered += 1 + + if line.get("branch") != "true": + continue + condition_coverage = line.get("condition-coverage", "") + match = re.search(r"\((\d+)/(\d+)\)", condition_coverage) + if match is not None: + stats.branches_covered += int(match.group(1)) + stats.branches_valid += int(match.group(2)) + + return ( + package_stats, + float(root.get("line-rate", 0)) * 100, + float(root.get("branch-rate", 0)) * 100, + ) + + +def check_coverage(xml_path: Path, threshold: float, packages_dir: Path) -> bool: + """Check all lifecycle-enforced packages against the coverage threshold.""" + policies = load_package_policies(packages_dir) + package_stats, overall_line_coverage, overall_branch_coverage = parse_coverage_xml(xml_path) + + print("\n" + "=" * 110) + print("PYTHON PACKAGE TEST COVERAGE") + print("=" * 110) + print(f"Overall Line Coverage: {overall_line_coverage:.1f}%") + print(f"Overall Branch Coverage: {overall_branch_coverage:.1f}%") + print(f"Enforced Threshold: {threshold:.1f}%") + print("-" * 110) + print(f"{'Package':<48} {'Stage':<20} {'Policy':<14} {'Lines':<12} {'Line Cov':<10}") + print("-" * 110) + + failed_packages: list[str] = [] + for policy in sorted(policies, key=lambda item: (not item.enforced, item.distribution_name)): + stats = package_stats.get(policy.directory) + if policy.exempt: + policy_label = "EXEMPT" + elif policy.enforced: + policy_label = "ENFORCED" + else: + policy_label = "REPORT ONLY" + + if stats is None: + lines = "-" + coverage = "missing" + if policy.enforced: + failed_packages.append(f"{policy.distribution_name} (missing from coverage report)") + else: + lines = f"{stats.lines_covered}/{stats.lines_valid}" + coverage = f"{stats.line_coverage_percent:.1f}%" + if policy.enforced and stats.line_coverage_percent < threshold: + failed_packages.append(f"{policy.distribution_name} ({coverage})") + + stage = f"{policy.development_status} - {policy.development_status_label}" + print(f"{policy.distribution_name:<48} {stage:<20} {policy_label:<14} {lines:<12} {coverage:<10}") + + print("-" * 110) + if failed_packages: + print(f"\nFAILED: Enforced packages below {threshold:.1f}% or missing:") + for package in failed_packages: + print(f" - {package}") + return False + + print(f"\nPASSED: All non-exempt Beta-or-higher packages meet {threshold:.1f}% line coverage.") + return True + + +def main() -> int: + """Run the coverage policy check.""" + if len(sys.argv) != 3: + print(f"Usage: {sys.argv[0]} ") + return 1 + + try: + threshold = float(sys.argv[2]) + except ValueError: + print(f"Error: Invalid threshold value: {sys.argv[2]}") + return 1 + + repository_root = Path(__file__).resolve().parents[2] + try: + passed = check_coverage( + Path(sys.argv[1]), + threshold, + repository_root / "python" / "packages", + ) + except (FileNotFoundError, ET.ParseError, ValueError) as error: + print(f"Error: {error}") + return 1 + return 0 if passed else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/tests/test_python_check_coverage.py b/.github/tests/test_python_check_coverage.py new file mode 100644 index 0000000000..6af23ea79e --- /dev/null +++ b/.github/tests/test_python_check_coverage.py @@ -0,0 +1,134 @@ +# Copyright (c) Microsoft. All rights reserved. + +# ruff:file-ignore[implicit-namespace-package, undocumented-public-class, undocumented-public-method] + +from __future__ import annotations + +import importlib.util +import sys +import tempfile +import unittest +from pathlib import Path + +SCRIPT_PATH = Path(__file__).parents[1] / "scripts" / "python_check_coverage.py" +SPEC = importlib.util.spec_from_file_location("python_check_coverage", SCRIPT_PATH) +if SPEC is None or SPEC.loader is None: + raise RuntimeError(f"Unable to load {SCRIPT_PATH}") +coverage_checker = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = coverage_checker +SPEC.loader.exec_module(coverage_checker) + + +class CoveragePolicyTests(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + self.root = Path(self.temp_dir.name) + self.packages_dir = self.root / "packages" + self.packages_dir.mkdir() + + def tearDown(self) -> None: + self.temp_dir.cleanup() + + def write_package(self, directory: str, name: str, status: str) -> None: + package_dir = self.packages_dir / directory + package_dir.mkdir() + (package_dir / "pyproject.toml").write_text( + f""" +[project] +name = "{name}" +classifiers = ["Development Status :: {status}"] +""".strip() + ) + + def write_coverage(self, files: dict[str, list[int]]) -> Path: + classes = [] + total_lines = 0 + covered_lines = 0 + for file_path, hits in files.items(): + lines = [] + for line_number, hit_count in enumerate(hits, start=1): + total_lines += 1 + covered_lines += hit_count > 0 + lines.append(f'') + classes.append(f'{"".join(lines)}') + + line_rate = covered_lines / total_lines if total_lines else 0 + xml_path = self.root / "coverage.xml" + xml_path.write_text( + f""" + + + + {"".join(classes)} + + + +""".strip() + ) + return xml_path + + def test_load_package_policies_uses_lifecycle_exemptions(self) -> None: + self.write_package("alpha", "agent-framework-alpha", "3 - Alpha") + self.write_package("beta", "agent-framework-beta", "4 - Beta") + self.write_package("stable", "agent-framework-stable", "5 - Production/Stable") + self.write_package("devui", "agent-framework-devui", "4 - Beta") + self.write_package("lab", "agent-framework-lab", "4 - Beta") + + policies = {policy.directory: policy for policy in coverage_checker.load_package_policies(self.packages_dir)} + + self.assertFalse(policies["alpha"].enforced) + self.assertTrue(policies["beta"].enforced) + self.assertTrue(policies["stable"].enforced) + self.assertTrue(policies["devui"].exempt) + self.assertFalse(policies["devui"].enforced) + self.assertTrue(policies["lab"].exempt) + self.assertFalse(policies["lab"].enforced) + + def test_load_package_policies_rejects_missing_lifecycle(self) -> None: + package_dir = self.packages_dir / "missing" + package_dir.mkdir() + (package_dir / "pyproject.toml").write_text('[project]\nname = "agent-framework-missing"\n') + + with self.assertRaisesRegex(ValueError, "exactly one Development Status"): + coverage_checker.load_package_policies(self.packages_dir) + + def test_parse_coverage_aggregates_nested_modules_by_distribution(self) -> None: + xml_path = self.write_coverage({ + "packages/core/agent_framework/_agents.py": [1, 0], + "packages/core/agent_framework/_workflows/_workflow.py": [1, 1], + }) + + package_stats, _, _ = coverage_checker.parse_coverage_xml(xml_path) + + self.assertEqual(package_stats["core"].lines_valid, 4) + self.assertEqual(package_stats["core"].lines_covered, 3) + + def test_beta_package_below_threshold_fails(self) -> None: + self.write_package("beta", "agent-framework-beta", "4 - Beta") + xml_path = self.write_coverage({"packages/beta/agent_framework_beta/client.py": [1, 0]}) + + self.assertFalse(coverage_checker.check_coverage(xml_path, 85, self.packages_dir)) + + def test_missing_beta_package_fails(self) -> None: + self.write_package("beta", "agent-framework-beta", "4 - Beta") + xml_path = self.write_coverage({}) + + self.assertFalse(coverage_checker.check_coverage(xml_path, 85, self.packages_dir)) + + def test_alpha_and_exempt_packages_do_not_fail(self) -> None: + self.write_package("alpha", "agent-framework-alpha", "3 - Alpha") + self.write_package("devui", "agent-framework-devui", "4 - Beta") + self.write_package("lab", "agent-framework-lab", "4 - Beta") + xml_path = self.write_coverage({}) + + self.assertTrue(coverage_checker.check_coverage(xml_path, 85, self.packages_dir)) + + def test_beta_package_at_threshold_passes(self) -> None: + self.write_package("beta", "agent-framework-beta", "4 - Beta") + xml_path = self.write_coverage({"packages/beta/agent_framework_beta/client.py": [1] * 17 + [0] * 3}) + + self.assertTrue(coverage_checker.check_coverage(xml_path, 85, self.packages_dir)) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/workflows/github-automation-tests.yml b/.github/workflows/github-automation-tests.yml index 906fa83222..6c015d2520 100644 --- a/.github/workflows/github-automation-tests.yml +++ b/.github/workflows/github-automation-tests.yml @@ -6,6 +6,7 @@ on: - ".github/actions/**" - ".github/scripts/**" - ".github/tests/**" + - ".github/workflows/python-test-coverage.yml" - ".github/workflows/github-automation-tests.yml" push: branches: @@ -14,6 +15,7 @@ on: - ".github/actions/**" - ".github/scripts/**" - ".github/tests/**" + - ".github/workflows/python-test-coverage.yml" - ".github/workflows/github-automation-tests.yml" permissions: @@ -29,5 +31,12 @@ jobs: with: node-version: "22" - - name: Run tests + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.11" + + - name: Run JavaScript tests run: node --test .github/tests/*.js + + - name: Run Python tests + run: python .github/tests/test_python_check_coverage.py diff --git a/.github/workflows/python-check-coverage.py b/.github/workflows/python-check-coverage.py deleted file mode 100644 index c9694aa35e..0000000000 --- a/.github/workflows/python-check-coverage.py +++ /dev/null @@ -1,382 +0,0 @@ -#!/usr/bin/env python3 -# Copyright (c) Microsoft. All rights reserved. -"""Check Python test coverage against threshold for enforced targets. - -This script parses a Cobertura XML coverage report and enforces a minimum -coverage threshold on specific targets. Targets can be package names -(e.g., "packages.core.agent_framework") or individual Python file paths -(e.g., "packages/core/agent_framework/observability.py"). - -Non-enforced targets 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 TARGETS CONFIGURATION -# ============================================================================= -# Add or remove entries from this set to control which targets must meet -# the coverage threshold. Only these targets will fail the build if below -# threshold. Other targets are reported for visibility only. -# -# Target values can be: -# - Package paths as they appear in the coverage report -# (e.g., "packages.azure-ai.agent_framework_azure_ai") -# - Python source file paths as they appear in the coverage report -# (e.g., "packages/core/agent_framework/observability.py") -# ============================================================================= -ENFORCED_TARGETS: set[str] = { - # Packages (sorted alphabetically) - "packages.anthropic.agent_framework_anthropic", - "packages.azure-ai-search.agent_framework_azure_ai_search", - "packages.core.agent_framework", - "packages.core.agent_framework._workflows", - "packages.foundry.agent_framework_foundry", - "packages.openai.agent_framework_openai", - "packages.purview.agent_framework_purview", - # Individual files (if you want to enforce specific files instead of whole packages) - "packages/core/agent_framework/observability.py", - # Add more targets here as coverage improves -} - - -@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 normalize_coverage_path(path: str) -> str: - """Normalize coverage paths for reliable matching.""" - return path.replace("\\", "/").lstrip("./") - - -def parse_coverage_xml( - xml_path: str, -) -> tuple[dict[str, PackageCoverage], 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, files_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] = {} - file_stats: dict[str, dict[str, int]] = {} - - 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"): - file_path = normalize_coverage_path(class_elem.get("filename", "")) - if file_path and file_path not in file_stats: - file_stats[file_path] = { - "lines_valid": 0, - "lines_covered": 0, - "branches_valid": 0, - "branches_covered": 0, - } - - for line in class_elem.findall(".//line"): - lines_valid += 1 - if int(line.get("hits", 0)) > 0: - lines_covered += 1 - - if file_path: - file_stats[file_path]["lines_valid"] += 1 - if int(line.get("hits", 0)) > 0: - file_stats[file_path]["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]) - if file_path: - file_stats[file_path]["branches_covered"] += int( - coverage_parts[0] - ) - file_stats[file_path]["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, - ) - - files: dict[str, PackageCoverage] = {} - for file_path, stats in file_stats.items(): - lines_valid = stats["lines_valid"] - lines_covered = stats["lines_covered"] - branches_valid = stats["branches_valid"] - branches_covered = stats["branches_covered"] - - files[file_path] = PackageCoverage( - name=file_path, - line_rate=0 if lines_valid == 0 else lines_covered / lines_valid, - branch_rate=0 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, files, 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 target 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], - files: 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. - files: Dictionary of file path to coverage data, used for per-file enforcement. - 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}%") - - enforced_targets = {normalize_coverage_path(t) for t in ENFORCED_TARGETS} - - # Package table - print("\n" + "-" * 110) - print(f"{'Package':<80} {'Lines':<15} {'Line Cov':<15}") - print("-" * 110) - - # Sort: enforced package targets first, then alphabetically - sorted_packages = sorted( - packages.values(), - key=lambda p: (p.name not in ENFORCED_TARGETS, p.name), - ) - - for pkg in sorted_packages: - is_enforced = normalize_coverage_path(pkg.name) in enforced_targets - 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) - - # Enforced file/model entries (if configured) - enforced_files = [ - files[target] - for target in sorted(enforced_targets) - if target in files and target.endswith(".py") - ] - - if enforced_files: - print("\nEnforced Files/Models") - print("-" * 110) - print(f"{'File':<80} {'Lines':<15} {'Line Cov':<15}") - print("-" * 110) - - for file_cov in enforced_files: - line_cov = format_coverage_value( - file_cov.line_coverage_percent, threshold, True - ) - lines_info = f"{file_cov.lines_covered}/{file_cov.lines_valid}" - print(f"[ENFORCED] {file_cov.name:<69} {lines_info:<15} {line_cov:<15}") - - print("-" * 110) - - -def check_coverage(xml_path: str, threshold: float) -> bool: - """Check if all enforced targets meet the coverage threshold. - - Args: - xml_path: Path to the Cobertura XML coverage report. - threshold: Minimum required coverage percentage. - - Returns: - True if all enforced targets pass, False otherwise. - """ - packages, files, overall_line_rate, overall_branch_rate = parse_coverage_xml( - xml_path - ) - - print_coverage_table( - packages, files, threshold, overall_line_rate, overall_branch_rate - ) - - # Check enforced targets - failed_targets: list[str] = [] - missing_targets: list[str] = [] - - for target_name in ENFORCED_TARGETS: - normalized_target = normalize_coverage_path(target_name) - package_alias = normalized_target.replace("/", ".") - - target_coverage = None - if target_name in packages: - target_coverage = packages[target_name] - elif normalized_target in files: - target_coverage = files[normalized_target] - elif package_alias in packages: - target_coverage = packages[package_alias] - - if target_coverage is None: - missing_targets.append(target_name) - continue - - if target_coverage.line_coverage_percent < threshold: - failed_targets.append( - f"{target_name} ({target_coverage.line_coverage_percent:.1f}%)" - ) - - # Report results - if missing_targets: - print( - f"\n❌ FAILED: Enforced targets not found in coverage report: {', '.join(missing_targets)}" - ) - return False - - if failed_targets: - print( - f"\n❌ FAILED: The following enforced targets are below {threshold}% coverage threshold:" - ) - for target in failed_targets: - print(f" - {target}") - print("\nTo fix: Add more tests to improve coverage for the failing targets.") - return False - - if ENFORCED_TARGETS: - found_enforced = [ - target - for target in ENFORCED_TARGETS - if target in packages or normalize_coverage_path(target) in files - ] - if found_enforced: - print( - f"\n✅ PASSED: All enforced targets 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 16867fce09..011b98c6e0 100644 --- a/.github/workflows/python-test-coverage.yml +++ b/.github/workflows/python-test-coverage.yml @@ -6,6 +6,9 @@ on: paths: - "python/packages/**" - "python/tests/unit/**" + - "python/scripts/workspace_poe_tasks.py" + - ".github/scripts/python_check_coverage.py" + - ".github/workflows/python-test-coverage.yml" env: # Configure a constant location for the uv cache UV_CACHE_DIR: /tmp/.uv-cache @@ -37,10 +40,10 @@ jobs: env: # Configure a constant location for the uv cache UV_CACHE_DIR: /tmp/.uv-cache - - name: Run all tests with coverage report + - name: Run aggregate tests with coverage report run: uv run poe test -A -C --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 }} + run: python ${{ github.workspace }}/.github/scripts/python_check_coverage.py python-coverage.xml ${{ env.COVERAGE_THRESHOLD }} - name: Upload coverage report uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: diff --git a/python/.github/skills/python-testing/SKILL.md b/python/.github/skills/python-testing/SKILL.md index 1d8fb50cf1..d2b2fc9cae 100644 --- a/python/.github/skills/python-testing/SKILL.md +++ b/python/.github/skills/python-testing/SKILL.md @@ -7,7 +7,9 @@ description: > # Python Testing -We strive for at least 85% test coverage across the codebase, with a focus on core packages and critical paths. Tests should be fast, reliable, and maintainable. +CI enforces at least 85% line coverage for every package classified Beta or Production/Stable. +Alpha packages are report-only, and the DevUI and experimental Lab packages are excluded from +aggregate coverage enforcement. Tests should be fast, reliable, and maintainable. When adding new code, check that the relevant sections of the codebase are covered by tests, and add new tests as needed. When modifying existing code, update or add tests to cover the changes. We run tests in two stages, for a PR each commit is tested with unit tests only (using `-m "not integration"`), and the full suite including integration tests is run when merging. @@ -82,6 +84,7 @@ packages/core/ ## File Naming - Files starting with `test_` are test files — do not use this prefix for helpers +- Prefer extending an existing test file that already covers the same component or behavior; create a new file only for a distinct surface without an appropriate existing file - Use `conftest.py` for shared utilities ## Integration Tests diff --git a/python/DEV_SETUP.md b/python/DEV_SETUP.md index 4485cff2ad..74433ac858 100644 --- a/python/DEV_SETUP.md +++ b/python/DEV_SETUP.md @@ -178,6 +178,10 @@ uv run poe test -A -C This will show you which files are not covered by the tests, including the specific lines not covered. Make sure to consider the untested lines from the code you are working on, but feel free to add other tests as well, that is always welcome! +CI automatically enforces at least 85% line coverage for every package classified Beta or +Production/Stable. Alpha packages are reported without blocking, and the DevUI and experimental Lab +packages are excluded from aggregate coverage enforcement. + ## Catching up with the latest changes There are many people committing to Agent Framework, so it is important to keep your local repository up to date. To do this, you can run the following commands: diff --git a/python/packages/azurefunctions/tests/test_func_utils.py b/python/packages/azurefunctions/tests/test_func_utils.py index 80add1dcc6..902a216379 100644 --- a/python/packages/azurefunctions/tests/test_func_utils.py +++ b/python/packages/azurefunctions/tests/test_func_utils.py @@ -160,3 +160,36 @@ async def test_apply_checkpoint_raises_not_implemented(self, context: CapturingR """Test that apply_checkpoint raises NotImplementedError.""" with pytest.raises(NotImplementedError): await context.apply_checkpoint(Mock()) + + def test_checkpoint_storage_noops_are_safe(self, context: CapturingRunnerContext) -> None: + """Unsupported checkpoint-storage hooks remain harmless no-ops.""" + storage = Mock() + + context.set_runtime_checkpoint_storage(storage) + context.clear_runtime_checkpoint_storage() + + assert context.has_checkpointing() is False + + def test_yield_output_classifier_can_be_overridden(self, context: CapturingRunnerContext) -> None: + """Custom yield-output classification is delegated to the configured classifier.""" + context.set_yield_output_classifier(lambda executor_id: None if executor_id == "secret" else "output") + + assert context.classify_yielded_output("secret") is None + assert context.classify_yielded_output("visible") == "output" + + async def test_add_request_info_event_tracks_pending_requests(self, context: CapturingRunnerContext) -> None: + """Request-info events are both queued and retained for later correlation.""" + event = WorkflowEvent("request_info", executor_id="reviewer", data={"question": "approve?"}, request_id="req-1") + + await context.add_request_info_event(event) + + pending = await context.get_pending_request_info_events() + queued = await context.drain_events() + + assert pending == {"req-1": event} + assert queued == [event] + + async def test_send_request_info_response_raises_not_implemented(self, context: CapturingRunnerContext) -> None: + """Activity contexts cannot resolve HITL responses directly.""" + with pytest.raises(NotImplementedError, match="orchestrator level"): + await context.send_request_info_response("req-1", {"approved": True}) diff --git a/python/packages/azurefunctions/tests/test_workflow_af_context.py b/python/packages/azurefunctions/tests/test_workflow_af_context.py new file mode 100644 index 0000000000..4947d4782d --- /dev/null +++ b/python/packages/azurefunctions/tests/test_workflow_af_context.py @@ -0,0 +1,142 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for the Azure Functions workflow-context adapter.""" + +# pyright: reportPrivateUsage=false + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any +from unittest.mock import Mock + +import pytest + +from agent_framework_azurefunctions._workflow import run_workflow_orchestrator +from agent_framework_azurefunctions._workflow_af_context import AzureFunctionsWorkflowContext + + +class _FakeDurableAIAgent: + def __init__(self, executor: Any, name: str) -> None: + self.executor = executor + self.name = name + self.calls: list[tuple[str, Any]] = [] + + def run(self, message: str, *, session: Any) -> dict[str, Any]: + self.calls.append((message, session)) + return {"message": message, "session": session, "executor": self.executor, "name": self.name} + + +class TestAzureFunctionsWorkflowContext: + """Behavior of the Azure Functions orchestration-context adapter.""" + + @pytest.fixture + def orchestration_context(self) -> Mock: + context = Mock() + context.instance_id = "instance-123" + context.is_replaying = True + context.current_utc_datetime = datetime(2025, 1, 2, 3, 4, 5, tzinfo=timezone.utc) + context.call_activity.return_value = "activity-task" + context.call_sub_orchestrator.return_value = "sub-task" + context.task_all.return_value = "all-task" + context.task_any.return_value = "any-task" + context.wait_for_external_event.return_value = "event-task" + context.create_timer.return_value = "timer-task" + context.new_uuid.return_value = "uuid-123" + return context + + def test_exposes_basic_context_properties(self, orchestration_context: Mock) -> None: + workflow_context = AzureFunctionsWorkflowContext(orchestration_context) + + assert workflow_context.instance_id == "instance-123" + assert workflow_context.is_replaying is True + assert workflow_context.supports_event_streaming is False + assert workflow_context.current_utc_datetime == orchestration_context.current_utc_datetime + + def test_prepare_agent_task_wraps_session_and_executor( + self, + monkeypatch: pytest.MonkeyPatch, + orchestration_context: Mock, + ) -> None: + executor_sentinel = object() + monkeypatch.setattr( + "agent_framework_azurefunctions._workflow_af_context.AzureFunctionsAgentExecutor", + lambda context: executor_sentinel if context is orchestration_context else None, + ) + monkeypatch.setattr("agent_framework_azurefunctions._workflow_af_context.DurableAIAgent", _FakeDurableAIAgent) + + workflow_context = AzureFunctionsWorkflowContext(orchestration_context) + result = workflow_context.prepare_agent_task("reviewer", "please approve", "orch-9") + + assert result["message"] == "please approve" + assert result["executor"] is executor_sentinel + assert result["name"] == "reviewer" + assert result["session"].durable_session_id.name == "reviewer" + assert result["session"].durable_session_id.key == "orch-9" + + def test_delegates_activity_and_orchestrator_primitives(self, orchestration_context: Mock) -> None: + workflow_context = AzureFunctionsWorkflowContext(orchestration_context) + + assert workflow_context.prepare_activity_task("activity-name", '{"payload": 1}') == "activity-task" + orchestration_context.call_activity.assert_called_once_with("activity-name", '{"payload": 1}') + + assert workflow_context.call_sub_orchestrator("child", {"x": 1}, instance_id="child-1") == "sub-task" + orchestration_context.call_sub_orchestrator.assert_called_once_with( + "child", input_={"x": 1}, instance_id="child-1" + ) + + assert workflow_context.task_all(["a", "b"]) == "all-task" + orchestration_context.task_all.assert_called_once_with(["a", "b"]) + + assert workflow_context.task_any(["a", "b"]) == "any-task" + orchestration_context.task_any.assert_called_once_with(["a", "b"]) + + assert workflow_context.wait_for_external_event("approval") == "event-task" + orchestration_context.wait_for_external_event.assert_called_once_with("approval") + + assert workflow_context.create_timer(orchestration_context.current_utc_datetime) == "timer-task" + orchestration_context.create_timer.assert_called_once_with(orchestration_context.current_utc_datetime) + + def test_status_uuid_and_task_helpers_delegate(self, orchestration_context: Mock) -> None: + workflow_context = AzureFunctionsWorkflowContext(orchestration_context) + + workflow_context.set_custom_status({"state": "running"}) + orchestration_context.set_custom_status.assert_called_once_with({"state": "running"}) + assert workflow_context.new_uuid() == "uuid-123" + + cancellable = Mock() + workflow_context.cancel_task(cancellable) + cancellable.cancel.assert_called_once_with() + + non_cancellable = object() + workflow_context.cancel_task(non_cancellable) + + done_task = Mock() + done_task.result = {"answer": 42} + assert workflow_context.get_task_result(done_task) == {"answer": 42} + assert workflow_context.get_task_result(object()) is None + + +def test_run_workflow_orchestrator_wraps_context(monkeypatch: pytest.MonkeyPatch) -> None: + """The Azure Functions wrapper delegates to the shared durabletask orchestrator.""" + + def _shared_runner(context: Any, workflow: Any, initial_message: Any, shared_state: dict[str, Any] | None) -> Any: + return context, workflow, initial_message, shared_state + + monkeypatch.setattr("agent_framework_azurefunctions._workflow._run_workflow_orchestrator_shared", _shared_runner) + + df_context = Mock() + workflow = Mock() + + wrapped_context, passed_workflow, passed_message, passed_state = run_workflow_orchestrator( + df_context, + workflow, + "hello", + {"x": 1}, + ) + + assert isinstance(wrapped_context, AzureFunctionsWorkflowContext) + assert wrapped_context.instance_id == df_context.instance_id + assert passed_workflow is workflow + assert passed_message == "hello" + assert passed_state == {"x": 1} diff --git a/python/packages/bedrock/tests/test_bedrock_client.py b/python/packages/bedrock/tests/test_bedrock_client.py index 3be2e16583..a2655eb877 100644 --- a/python/packages/bedrock/tests/test_bedrock_client.py +++ b/python/packages/bedrock/tests/test_bedrock_client.py @@ -3,12 +3,19 @@ from __future__ import annotations import json -from typing import Any +from collections import deque +from collections.abc import MutableMapping +from typing import Any, cast +from unittest.mock import MagicMock, patch import pytest -from agent_framework import Agent, Content, Message +from agent_framework import Agent, Content, FunctionTool, Message +from agent_framework._settings import SecretString +from boto3.session import Session as Boto3Session +from botocore.client import BaseClient from agent_framework_bedrock import BedrockChatClient +from agent_framework_bedrock._chat_client import BedrockSettings class _StubBedrockRuntime: @@ -234,3 +241,257 @@ def test_parse_usage_returns_none_when_no_recognized_keys() -> None: assert client._parse_usage({"unexpected": 1}) is None assert client._parse_usage({}) is None assert client._parse_usage(None) is None + + +def test_init_uses_boto3_session_when_runtime_client_not_supplied() -> None: + """BedrockChatClient should build a runtime client from a provided boto3 session.""" + + class _FakeSession: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + self.region_name: str | None = None + + def client(self, service_name: str, *, region_name: str, config: Any) -> _StubBedrockRuntime: + self.calls.append({"service_name": service_name, "region_name": region_name, "config": config}) + return _StubBedrockRuntime() + + session = _FakeSession() + + client = BedrockChatClient( + model="amazon.titan-text", + region="us-west-2", + boto3_session=cast(Boto3Session, session), + ) + + assert isinstance(client._bedrock_client, _StubBedrockRuntime) + assert session.calls == [ + { + "service_name": "bedrock-runtime", + "region_name": "us-west-2", + "config": session.calls[0]["config"], + } + ] + + +def test_create_session_uses_secret_values() -> None: + """Bedrock session creation should unwrap configured secret values.""" + settings: BedrockSettings = { + "region": "eu-west-1", + "access_key": SecretString("access"), + "secret_key": SecretString("secret"), + "session_token": SecretString("token"), + } + + with patch("agent_framework_bedrock._chat_client.Boto3Session", return_value=MagicMock()) as session_cls: + BedrockChatClient._create_session(settings) + + session_cls.assert_called_once_with( + region_name="eu-west-1", + aws_access_key_id="access", + aws_secret_access_key="secret", + aws_session_token="token", + ) + + +def test_invoke_converse_requires_mapping_response() -> None: + """Non-mapping Bedrock responses should be rejected.""" + + class _BadRuntime: + def converse(self, **_: Any) -> list[str]: + return ["not", "a", "mapping"] + + from agent_framework.exceptions import ChatClientInvalidResponseException + + client = BedrockChatClient( + model="amazon.titan-text", + region="us-west-2", + client=cast(BaseClient, _BadRuntime()), + ) + + with pytest.raises(ChatClientInvalidResponseException, match="must be a mapping"): + client._invoke_converse({"modelId": "amazon.titan-text"}) + + +def test_prepare_options_requires_model_when_unset() -> None: + """Preparing options without a configured model should raise.""" + client = _make_client() + client.model = None # type: ignore[assignment] + + with pytest.raises(ValueError, match="Bedrock model is required"): + client._prepare_options([Message(role="user", contents=[Content.from_text(text="hello")])], {}) + + +def test_prepare_options_adds_instructions_and_sampling_settings() -> None: + """Instructions and inference settings should be translated into Bedrock request fields.""" + client = _make_client() + messages = [ + Message(role="system", contents=[Content.from_text(text="Original system prompt")]), + Message(role="user", contents=[Content.from_text(text="hello")]), + ] + + request = client._prepare_options( + messages, + { + "instructions": "Runtime instructions", + "temperature": 0.2, + "top_p": 0.9, + "stop": ["DONE"], + "max_tokens": 5, + }, + ) + + assert request["system"] == [{"text": "Runtime instructions"}, {"text": "Original system prompt"}] + assert request["inferenceConfig"] == { + "maxTokens": 5, + "temperature": 0.2, + "topP": 0.9, + "stopSequences": ["DONE"], + } + + +def test_prepare_options_unsupported_tool_mode_raises(monkeypatch: pytest.MonkeyPatch) -> None: + """Unexpected tool modes should raise a clear error.""" + from agent_framework_bedrock import _chat_client as chat_client_module + + client = _make_client() + monkeypatch.setattr(chat_client_module, "validate_tool_mode", lambda _: {"mode": "unexpected"}) + + with pytest.raises(ValueError, match="Unsupported tool mode for Bedrock: unexpected"): + client._prepare_options( + [Message(role="user", contents=[Content.from_text(text="hello")])], + {"tool_choice": "auto"}, + ) + + +def test_prepare_bedrock_messages_skips_unsupported_content_and_unmatched_tool_results() -> None: + """Unsupported user content and orphaned tool results should be dropped.""" + client = _make_client() + messages = [ + Message(role="user", contents=[Content.from_data(data=b"x", media_type="application/octet-stream")]), + Message(role="tool", contents=[Content.from_function_result(call_id="call-1", result={"answer": 42})]), + Message(role="user", contents=[Content.from_text(text="hello")]), + ] + + prompts, conversation = client._prepare_bedrock_messages(messages) + + assert prompts == [] + assert conversation == [{"role": "user", "content": [{"text": "hello"}]}] + + +def test_align_tool_results_handles_pending_edge_cases() -> None: + """Tool result alignment should preserve valid blocks and drop invalid or extra results.""" + client = _make_client() + mixed_blocks = cast( + list[dict[str, Any]], + [ + "keep-me", + {"text": "note"}, + {"toolResult": {"content": []}}, + {"toolResult": {"content": []}}, + ], + ) + + aligned = client._align_tool_results_with_pending( + mixed_blocks, + deque(["call-1"]), + ) + unmatched = client._align_tool_results_with_pending( + [{"toolResult": {"toolUseId": "other", "content": []}}], + deque(["call-1"]), + ) + + assert aligned[0] == "keep-me" + assert aligned[1] == {"text": "note"} + assert aligned[2]["toolResult"]["toolUseId"] == "call-1" + assert len(aligned) == 3 + assert unmatched == [] + + +def test_convert_content_to_bedrock_block_handles_errors_and_missing_items() -> None: + """Function result conversion should serialize items, rich content warnings, and fallback results.""" + client = _make_client() + rich_result = Content.from_function_result( + call_id="call-1", + result=[Content.from_text(text="summary"), Content.from_data(data=b"x", media_type="image/png")], + exception="tool failed", + ) + fallback_result = Content.from_function_result(call_id="call-2", result={"answer": 42}) + fallback_result.items = None + + rich_block = client._convert_content_to_bedrock_block(rich_result) + fallback_block = client._convert_content_to_bedrock_block(fallback_result) + + assert rich_block == { + "toolResult": { + "toolUseId": "call-1", + "content": [{"text": "summary"}, {"text": "tool failed"}], + "status": "error", + } + } + assert fallback_block == { + "toolResult": { + "toolUseId": "call-2", + "content": [{"json": {"answer": 42}}], + "status": "success", + } + } + assert client._convert_content_to_bedrock_block(Content.from_data(data=b"x", media_type="text/plain")) is None + + +def test_tool_result_helpers_cover_text_json_and_sequence_values() -> None: + """Tool result helpers should normalize text, JSON, sequences, and custom objects.""" + client = _make_client() + + class _Serializable: + def to_dict(self) -> dict[str, int]: + return {"value": 1} + + assert client._convert_tool_result_to_blocks("plain text") == [{"text": "plain text"}] + assert client._convert_prepared_tool_result_to_blocks([{"answer": 1}, "done"]) == [ + {"json": {"answer": 1}}, + {"text": "done"}, + ] + assert client._convert_prepared_tool_result_to_blocks([]) == [{"text": ""}] + assert client._normalize_tool_result_value(("a", 2)) == {"json": ["a", 2]} + assert client._normalize_tool_result_value(Content.from_text(text="hello")) == {"text": "hello"} + assert client._normalize_tool_result_value(_Serializable()) == {"json": {"value": 1}} + + +def test_prepare_tools_parse_message_contents_and_finish_reason_helpers() -> None: + """Helper methods should ignore unsupported values and preserve Bedrock response semantics.""" + client = _make_client() + mixed_tools = cast( + list[FunctionTool | MutableMapping[str, Any]], + [ + object(), + {"toolSpec": {"name": "keep", "description": "desc", "inputSchema": {"json": {}}}}, + ], + ) + + prepared_tools = client._prepare_tools(mixed_tools) + error_result = client._parse_message_contents([{"toolResult": {"status": "failure", "content": [{"text": "bad"}]}}]) + unsupported_result = client._parse_message_contents([{"image": "ignored"}]) + + assert prepared_tools == { + "tools": [{"toolSpec": {"name": "keep", "description": "desc", "inputSchema": {"json": {}}}}] + } + assert client._generate_tool_call_id().startswith("tool-call-") + assert error_result[0].exception == "Bedrock tool result status: failure" + assert error_result[0].result == "bad" + assert unsupported_result == [] + assert client._map_finish_reason(None) is None + assert client._convert_bedrock_tool_result_to_value(None) is None + assert client._convert_bedrock_tool_result_to_value([{"text": "ok"}]) == "ok" + assert client._convert_bedrock_tool_result_to_value([{"json": {"x": 1}}, 7]) == [{"x": 1}, 7] + assert client._convert_bedrock_tool_result_to_value({"json": {"x": 1}}) == {"x": 1} + assert client._convert_bedrock_tool_result_to_value({"text": "ok"}) == "ok" + + +def test_parse_message_contents_requires_tool_use_name() -> None: + """Malformed toolUse blocks should raise a client response error.""" + from agent_framework.exceptions import ChatClientInvalidResponseException + + client = _make_client() + + with pytest.raises(ChatClientInvalidResponseException, match="missing required tool name"): + client._parse_message_contents([{"toolUse": {"toolUseId": "call-1"}}]) diff --git a/python/packages/chatkit/tests/test_converter.py b/python/packages/chatkit/tests/test_converter.py index a630062d37..3d334cba7c 100644 --- a/python/packages/chatkit/tests/test_converter.py +++ b/python/packages/chatkit/tests/test_converter.py @@ -2,6 +2,8 @@ """Tests for ChatKit to Agent Framework converter utilities.""" +from datetime import datetime +from types import SimpleNamespace from unittest.mock import Mock import pytest @@ -88,8 +90,6 @@ async def test_to_agent_input_no_content(self, converter): async def test_to_agent_input_multiple_content_parts(self, converter): """Test converting user message with multiple text content parts.""" - from datetime import datetime - from chatkit.types import UserMessageItem input_item = UserMessageItem( @@ -110,6 +110,27 @@ async def test_to_agent_input_multiple_content_parts(self, converter): assert len(result) == 1 assert result[0].text == "Hello world!" + async def test_to_agent_input_with_quoted_text_for_last_message(self, converter): + """Test quoted text is prepended as context for the last user message.""" + from chatkit.types import UserMessageItem + + input_item = UserMessageItem( + id="msg_quoted", + thread_id="thread_1", + created_at=datetime.now(), + type="user_message", + content=[UserMessageTextContent(text="Please summarize this")], + attachments=[], + quoted_text="Important excerpt", + inference_options=InferenceOptions(), + ) + + result = await converter.to_agent_input(input_item) + + assert [message.role for message in result] == ["user", "user"] + assert result[0].text == "The user is referring to this in particular:\nImportant excerpt" + assert result[1].text == "Please summarize this" + def test_hidden_context_to_input(self, converter): """Test converting hidden context item to Message.""" hidden_item = Mock() @@ -154,6 +175,16 @@ def test_tag_to_message_content_no_name(self, converter): assert result.type == "text" assert result.text == "Name:jane" + def test_tag_to_message_content_prefers_name_attribute(self, converter): + """Test converting tag content when the backing object exposes a name attribute.""" + tag = Mock() + tag.data = SimpleNamespace(name="Jane Doe") + tag.text = "fallback" + + result = converter.tag_to_message_content(tag) + + assert result.text == "Name:Jane Doe" + async def test_attachment_to_message_content_file_without_fetcher(self, converter): """Test that FileAttachment without data fetcher returns None.""" from chatkit.types import FileAttachment @@ -207,10 +238,30 @@ async def fetch_data(attachment_id: str) -> bytes: assert result.type == "data" assert result.media_type == "application/pdf" + async def test_attachment_to_message_content_fetcher_failure_falls_back_to_preview_url(self) -> None: + """Test failed attachment fetch falls back to image preview URLs.""" + from chatkit.types import ImageAttachment + + async def fetch_data(_: str) -> bytes: + raise RuntimeError("storage unavailable") + + converter = ThreadItemConverter(attachment_data_fetcher=fetch_data) + attachment = ImageAttachment( + id="img_fallback", + name="photo.jpg", + mime_type="image/jpeg", + type="image", + preview_url=AnyUrl("https://example.com/fallback.jpg"), + ) + + result = await converter.attachment_to_message_content(attachment) + + assert result is not None + assert result.type == "uri" + assert result.uri == "https://example.com/fallback.jpg" + async def test_to_agent_input_with_image_attachment(self): """Test converting user message with text and image attachment.""" - from datetime import datetime - from chatkit.types import ImageAttachment, UserMessageItem attachment = ImageAttachment( @@ -250,8 +301,6 @@ async def test_to_agent_input_with_image_attachment(self): async def test_to_agent_input_with_file_attachment_and_fetcher(self): """Test converting user message with file attachment using data fetcher.""" - from datetime import datetime - from chatkit.types import FileAttachment, UserMessageItem attachment = FileAttachment( @@ -291,8 +340,6 @@ async def fetch_data(attachment_id: str) -> bytes: def test_task_to_input(self, converter): """Test converting TaskItem to Message.""" - from datetime import datetime - from chatkit.types import CustomTask, TaskItem task_item = TaskItem( @@ -311,8 +358,6 @@ def test_task_to_input(self, converter): def test_task_to_input_no_custom_task(self, converter): """Test that non-custom tasks return None.""" - from datetime import datetime - from chatkit.types import TaskItem, ThoughtTask task_item = TaskItem( @@ -328,8 +373,6 @@ def test_task_to_input_no_custom_task(self, converter): def test_workflow_to_input(self, converter): """Test converting WorkflowItem to ChatMessages.""" - from datetime import datetime - from chatkit.types import CustomTask, Workflow, WorkflowItem workflow_item = WorkflowItem( @@ -353,10 +396,34 @@ def test_workflow_to_input(self, converter): assert "Step 1: First step" in result[0].text assert "Step 2: Second step" in result[1].text + def test_workflow_to_input_skips_non_custom_tasks(self, converter): + """Test workflows ignore unsupported or empty tasks but keep valid custom tasks.""" + from chatkit.types import CustomTask, ThoughtTask, Workflow, WorkflowItem + + workflow_item = WorkflowItem( + id="wf_skip", + thread_id="thread_1", + created_at=datetime.now(), + type="workflow", + workflow=Workflow( + type="custom", + tasks=[ + ThoughtTask(type="thought", title="Thinking", content="Working"), + CustomTask(type="custom"), + CustomTask(type="custom", title="Step", content="Done"), + ], + ), + ) + + result = converter.workflow_to_input(workflow_item) + + assert isinstance(result, list) + assert len(result) == 1 + assert result[0].text is not None + assert "Step: Done" in result[0].text + def test_workflow_to_input_empty(self, converter): """Test that workflows with no custom tasks return None.""" - from datetime import datetime - from chatkit.types import Workflow, WorkflowItem workflow_item = WorkflowItem( @@ -372,8 +439,6 @@ def test_workflow_to_input_empty(self, converter): def test_widget_to_input(self, converter): """Test converting WidgetItem to Message.""" - from datetime import datetime - from chatkit.types import WidgetItem from chatkit.widgets import Card, Text # ty: ignore[deprecated] @@ -391,6 +456,201 @@ def test_widget_to_input(self, converter): assert "widget_1" in result.text assert "graphical UI widget" in result.text + def test_widget_to_input_serialization_failure_returns_none(self, converter): + """Test widget conversion skips widgets that cannot be serialized.""" + widget_item = Mock() + widget_item.id = "widget_broken" + widget_item.widget = Mock() + widget_item.widget.model_dump_json.side_effect = RuntimeError("boom") + + assert converter.widget_to_input(widget_item) is None + + async def test_assistant_message_to_input_handles_empty_and_text_content(self, converter): + """Test assistant messages convert text content and skip empty messages.""" + from chatkit.types import AssistantMessageContent, AssistantMessageItem + + assistant_item = AssistantMessageItem( + id="assistant_1", + thread_id="thread_1", + created_at=datetime.now(), + type="assistant_message", + content=[ + AssistantMessageContent(type="output_text", text="Hello", annotations=[]), + AssistantMessageContent(type="output_text", text=" world", annotations=[]), + ], + ) + empty_item = AssistantMessageItem( + id="assistant_2", + thread_id="thread_1", + created_at=datetime.now(), + type="assistant_message", + content=[], + ) + + result = await converter.assistant_message_to_input(assistant_item) + + assert isinstance(result, Message) + assert result.role == "assistant" + assert result.text == "Hello world" + assert await converter.assistant_message_to_input(empty_item) is None + + async def test_client_tool_call_to_input_handles_pending_and_completed(self, converter): + """Test client tool call conversion only emits completed tool calls.""" + import json + + from chatkit.types import ClientToolCallItem + + pending_item = ClientToolCallItem( + id="tool_pending", + thread_id="thread_1", + created_at=datetime.now(), + type="client_tool_call", + status="pending", + call_id="call_pending", + name="get_weather", + arguments={"location": "SEA"}, + ) + completed_item = ClientToolCallItem( + id="tool_done", + thread_id="thread_1", + created_at=datetime.now(), + type="client_tool_call", + status="completed", + call_id="call_done", + name="get_weather", + arguments={"location": "SEA"}, + output={"temperature": 72}, + ) + + assert await converter.client_tool_call_to_input(pending_item) is None + + result = await converter.client_tool_call_to_input(completed_item) + + assert isinstance(result, list) + assert len(result) == 2 + assert result[0].role == "assistant" + assert result[0].contents[0].parse_arguments() == {"location": "SEA"} + assert result[1].role == "tool" + assert json.loads(result[1].contents[0].result) == {"temperature": 72} + + async def test_end_of_turn_to_input_returns_none(self, converter): + """Test end-of-turn markers are skipped.""" + from chatkit.types import EndOfTurnItem + + end_item = EndOfTurnItem( + id="end_1", + thread_id="thread_1", + created_at=datetime.now(), + type="end_of_turn", + ) + + assert await converter.end_of_turn_to_input(end_item) is None + + async def test_to_agent_input_dispatches_supported_variants(self, converter): + """Test thread item dispatch converts supported items and skips unsupported variants.""" + from chatkit.types import ( + AssistantMessageContent, + AssistantMessageItem, + ClientToolCallItem, + CustomTask, + EndOfTurnItem, + GeneratedImageItem, + HiddenContextItem, + SDKHiddenContextItem, + StructuredInputItem, + TaskItem, + WidgetItem, + Workflow, + WorkflowItem, + ) + from chatkit.widgets import Card, Text # ty: ignore[deprecated] + + thread_items = [ + AssistantMessageItem( + id="assistant_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="assistant_message", + content=[AssistantMessageContent(type="output_text", text="Assistant", annotations=[])], + ), + ClientToolCallItem( + id="tool_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="client_tool_call", + status="completed", + call_id="dispatch_call", + name="search", + arguments={"query": "docs"}, + output={"result": "ok"}, + ), + EndOfTurnItem(id="end_dispatch", thread_id="thread_1", created_at=datetime.now(), type="end_of_turn"), + WidgetItem( + id="widget_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="widget", + widget=Card(key="card_dispatch", children=[Text(value="Dispatch")]), # ty: ignore[deprecated] + ), + WorkflowItem( + id="workflow_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="workflow", + workflow=Workflow(type="custom", tasks=[CustomTask(type="custom", title="Step", content="Done")]), + ), + TaskItem( + id="task_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="task", + task=CustomTask(type="custom", title="Analysis", content="Completed"), + ), + HiddenContextItem( + id="hidden_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="hidden_context_item", + content="secret", + ), + SDKHiddenContextItem( + id="sdk_hidden_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="sdk_hidden_context", + content="sdk secret", + ), + GeneratedImageItem(id="generated_dispatch", thread_id="thread_1", created_at=datetime.now()), + StructuredInputItem( + id="structured_dispatch", + thread_id="thread_1", + created_at=datetime.now(), + type="structured_input", + inputs=[], + ), + object(), + ] + + result = await converter.to_agent_input(thread_items) + + assert [message.role for message in result] == [ + "assistant", + "assistant", + "tool", + "user", + "user", + "user", + "system", + "system", + ] + assert result[0].text == "Assistant" + assert result[2].contents[0].result is not None + assert "widget_dispatch" in result[3].text + assert "Step: Done" in result[4].text + assert "Analysis: Completed" in result[5].text + assert result[6].text == "secret" + assert result[7].text == "sdk secret" + class TestSimpleToAgentInput: """Tests for simple_to_agent_input helper function.""" @@ -402,8 +662,6 @@ async def test_simple_to_agent_input_empty_list(self): async def test_simple_to_agent_input_with_text(self): """Test simple conversion with text content.""" - from datetime import datetime - from chatkit.types import UserMessageItem input_item = UserMessageItem( diff --git a/python/packages/durabletask/tests/test_async_bridge.py b/python/packages/durabletask/tests/test_async_bridge.py new file mode 100644 index 0000000000..af519e9015 --- /dev/null +++ b/python/packages/durabletask/tests/test_async_bridge.py @@ -0,0 +1,75 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for the persistent async bridge used by durable handlers.""" + +from __future__ import annotations + +import asyncio +import threading +from collections.abc import Iterator +from unittest.mock import Mock + +import pytest + +import agent_framework_durabletask._async_bridge as async_bridge + + +@pytest.fixture(autouse=True) +def _restore_bridge_globals() -> Iterator[None]: + old_loop = async_bridge._loop + old_thread = async_bridge._thread + old_lock = async_bridge._lock + + async_bridge._loop = None + async_bridge._thread = None + async_bridge._lock = threading.Lock() + + yield + + new_loop = async_bridge._loop + new_thread = async_bridge._thread + if new_loop is not None and not new_loop.is_closed(): + new_loop.call_soon_threadsafe(new_loop.stop) + if new_thread is not None and new_thread.is_alive(): + new_thread.join(timeout=1) + if new_loop is not None and not new_loop.is_closed(): + new_loop.close() + + async_bridge._loop = old_loop + async_bridge._thread = old_thread + async_bridge._lock = old_lock + + +def test_ensure_loop_reuses_existing_live_loop() -> None: + loop = asyncio.new_event_loop() + thread = Mock() + thread.is_alive.return_value = True + async_bridge._loop = loop + async_bridge._thread = thread + + assert async_bridge._ensure_loop() is loop + + +def test_ensure_loop_replaces_orphaned_loop() -> None: + orphaned_loop = asyncio.new_event_loop() + dead_thread = Mock() + dead_thread.is_alive.return_value = False + async_bridge._loop = orphaned_loop + async_bridge._thread = dead_thread + + new_loop = async_bridge._ensure_loop() + + assert new_loop is not orphaned_loop + assert orphaned_loop.is_closed() is True + assert async_bridge._thread is not None + assert async_bridge._thread.is_alive() is True + + +def test_run_agent_coroutine_executes_on_shared_loop() -> None: + async def _compute() -> str: + await asyncio.sleep(0) + return "done" + + assert async_bridge.run_agent_coroutine(_compute()) == "done" + assert async_bridge._loop is not None + assert async_bridge._thread is not None diff --git a/python/packages/durabletask/tests/test_workflow_dt_context.py b/python/packages/durabletask/tests/test_workflow_dt_context.py new file mode 100644 index 0000000000..105945e6d4 --- /dev/null +++ b/python/packages/durabletask/tests/test_workflow_dt_context.py @@ -0,0 +1,121 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for the standalone durabletask workflow-context adapter.""" + +# pyright: reportPrivateUsage=false + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any +from unittest.mock import Mock + +import pytest + +from agent_framework_durabletask._workflows.dt_context import DurableTaskWorkflowContext + + +class _FakeDurableAIAgent: + def __init__(self, executor: Any, name: str) -> None: + self.executor = executor + self.name = name + + def run(self, message: str, *, session: Any) -> dict[str, Any]: + return {"message": message, "session": session, "executor": self.executor, "name": self.name} + + +class _FakeTask: + def __init__(self, result: Any) -> None: + self._result = result + + def get_result(self) -> Any: + return self._result + + +class TestDurableTaskWorkflowContext: + """Behavior of the durabletask-host workflow-context adapter.""" + + @pytest.fixture + def orchestration_context(self) -> Mock: + context = Mock() + context.instance_id = "instance-456" + context.is_replaying = False + context.current_utc_datetime = datetime(2025, 2, 3, 4, 5, 6, tzinfo=timezone.utc) + context.call_activity.return_value = "activity-task" + context.call_sub_orchestrator.return_value = "sub-task" + context.wait_for_external_event.return_value = "event-task" + context.create_timer.return_value = "timer-task" + context.new_uuid.return_value = "uuid-456" + return context + + def test_exposes_basic_context_properties(self, orchestration_context: Mock) -> None: + workflow_context = DurableTaskWorkflowContext(orchestration_context) + + assert workflow_context.instance_id == "instance-456" + assert workflow_context.is_replaying is False + assert workflow_context.supports_event_streaming is True + assert workflow_context.current_utc_datetime == orchestration_context.current_utc_datetime + + def test_prepare_agent_task_wraps_session_and_executor( + self, + monkeypatch: pytest.MonkeyPatch, + orchestration_context: Mock, + ) -> None: + monkeypatch.setattr("agent_framework_durabletask._workflows.dt_context.DurableAIAgent", _FakeDurableAIAgent) + + workflow_context = DurableTaskWorkflowContext(orchestration_context) + result = workflow_context.prepare_agent_task("reviewer", "please approve", "orch-12") + + assert result["message"] == "please approve" + assert result["name"] == "reviewer" + assert result["session"].durable_session_id.name == "reviewer" + assert result["session"].durable_session_id.key == "orch-12" + assert result["executor"] is workflow_context._executor + + def test_delegates_activity_and_orchestrator_primitives( + self, + monkeypatch: pytest.MonkeyPatch, + orchestration_context: Mock, + ) -> None: + monkeypatch.setattr("agent_framework_durabletask._workflows.dt_context.when_all", lambda tasks: ("all", tasks)) + monkeypatch.setattr("agent_framework_durabletask._workflows.dt_context.when_any", lambda tasks: ("any", tasks)) + + workflow_context = DurableTaskWorkflowContext(orchestration_context) + + assert workflow_context.prepare_activity_task("activity-name", '{"payload": 1}') == "activity-task" + orchestration_context.call_activity.assert_called_once_with("activity-name", input='{"payload": 1}') + + assert workflow_context.call_sub_orchestrator("child", {"x": 1}, instance_id="child-2") == "sub-task" + orchestration_context.call_sub_orchestrator.assert_called_once_with( + "child", input={"x": 1}, instance_id="child-2" + ) + + assert workflow_context.task_all(["a", "b"]) == ("all", ["a", "b"]) + assert workflow_context.task_any(["a", "b"]) == ("any", ["a", "b"]) + + assert workflow_context.wait_for_external_event("approval") == "event-task" + orchestration_context.wait_for_external_event.assert_called_once_with("approval") + + assert workflow_context.create_timer(orchestration_context.current_utc_datetime) == "timer-task" + orchestration_context.create_timer.assert_called_once_with(orchestration_context.current_utc_datetime) + + def test_status_uuid_and_task_helpers_delegate( + self, + monkeypatch: pytest.MonkeyPatch, + orchestration_context: Mock, + ) -> None: + monkeypatch.setattr("agent_framework_durabletask._workflows.dt_context.Task", _FakeTask) + workflow_context = DurableTaskWorkflowContext(orchestration_context) + + workflow_context.set_custom_status({"state": "running"}) + orchestration_context.set_custom_status.assert_called_once_with({"state": "running"}) + assert workflow_context.new_uuid() == "uuid-456" + + cancellable = Mock() + workflow_context.cancel_task(cancellable) + cancellable.cancel.assert_called_once_with() + + workflow_context.cancel_task(object()) + + assert workflow_context.get_task_result(_FakeTask({"answer": 42})) == {"answer": 42} + assert workflow_context.get_task_result(Mock(result="fallback")) == "fallback" diff --git a/python/packages/durabletask/tests/test_workflow_orchestrator_helpers.py b/python/packages/durabletask/tests/test_workflow_orchestrator_helpers.py new file mode 100644 index 0000000000..05d3f8d942 --- /dev/null +++ b/python/packages/durabletask/tests/test_workflow_orchestrator_helpers.py @@ -0,0 +1,363 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for shared durable workflow-orchestrator helper functions.""" + +# pyright: reportPrivateUsage=false + +from __future__ import annotations + +import json +from collections import defaultdict +from typing import Any +from unittest.mock import AsyncMock, Mock + +from agent_framework import ( + AgentExecutor, + AgentExecutorRequest, + AgentExecutorResponse, + AgentResponse, + Executor, + Message, +) +from agent_framework._workflows._edge import FanInEdgeGroup, SingleEdgeGroup +from agent_framework._workflows._state import State +from pydantic import BaseModel + +from agent_framework_durabletask._workflows.orchestrator import ( + SOURCE_HITL_RESPONSE, + ExecutorResult, + PendingHITLRequest, + TaskType, + _check_fan_in_ready, + _collect_hitl_requests, + _deserialize_hitl_response, + _prepare_activity_task, + _prepare_agent_task, + _prepare_all_tasks, + _process_activity_result, + _route_hitl_response, + _route_result_messages, + _select_primary_input_type, + execute_hitl_response_handler, +) +from agent_framework_durabletask._workflows.serialization import serialize_value + + +class _ApprovalModel(BaseModel): + approved: bool + + +def _agent_response(text: str) -> AgentExecutorResponse: + assistant = Message(role="assistant", contents=[text]) + return AgentExecutorResponse( + executor_id="exec", + agent_response=AgentResponse(messages=[assistant]), + full_conversation=[assistant], + ) + + +class TestPrepareTaskHelpers: + """Preparation helpers scope names and package activity input correctly.""" + + def test_prepare_agent_task_scopes_executor_id_and_extracts_message_text(self) -> None: + ctx = Mock() + ctx.instance_id = "instance-1" + ctx.prepare_agent_task.return_value = "agent-task" + request = AgentExecutorRequest(messages=[Message(role="user", contents=["hello there"])]) + + task = _prepare_agent_task(ctx, "reviewer", request, "moderation") + + assert task == "agent-task" + ctx.prepare_agent_task.assert_called_once_with("moderation-reviewer", "hello there", "instance-1") + + def test_prepare_activity_task_serializes_message_state_and_host_context(self) -> None: + ctx = Mock() + ctx.prepare_activity_task.return_value = "activity-task" + + task = _prepare_activity_task( + ctx, + "router", + {"payload": 1}, + "start", + {"existing": True}, + "moderation", + { + "root_instance_id": "root-1", + "root_workflow_name": "outer-workflow", + "request_path_prefix": "review~0~", + }, + ) + + assert task == "activity-task" + activity_name, activity_input_json = ctx.prepare_activity_task.call_args[0] + assert activity_name == "dafx-moderation-router" + + activity_input = json.loads(activity_input_json) + assert activity_input["executor_id"] == "router" + assert activity_input["message"] == serialize_value({"payload": 1}) + assert activity_input["shared_state_snapshot"] == {"existing": True} + assert activity_input["source_executor_ids"] == ["start"] + assert activity_input["host_context"] == { + "instance_id": "root-1", + "workflow_name": "outer-workflow", + "request_path_prefix": "review~0~", + } + + def test_prepare_all_tasks_groups_agent_messages_for_sequential_followup(self) -> None: + ctx = Mock() + ctx.instance_id = "instance-9" + ctx.prepare_agent_task.return_value = "agent-task" + ctx.prepare_activity_task.return_value = "activity-task" + + agent_executor = Mock(spec=AgentExecutor) + agent_executor.id = "reviewer" + activity_executor = Mock(spec=Executor) + activity_executor.id = "router" + + workflow = Mock() + workflow.name = "moderation" + workflow.executors = { + "reviewer": agent_executor, + "router": activity_executor, + } + + tasks, metadata, remaining = _prepare_all_tasks( + ctx, + workflow, + { + "reviewer": [("first", "start"), ("second", "other")], + "router": [(False, "reviewer")], + }, + {"x": 1}, + [0], + { + "root_instance_id": "root-9", + "root_workflow_name": "moderation", + "request_path_prefix": "", + }, + ) + + assert tasks == ["activity-task", "agent-task"] + assert [item.task_type for item in metadata] == [TaskType.ACTIVITY, TaskType.AGENT] + assert remaining == [("reviewer", "second", "other")] + + +class TestHitlHelpers: + """HITL helper functions sanitize, reconstruct, and route responses.""" + + def test_deserialize_hitl_response_handles_none_scalar_and_marker_rejection(self) -> None: + assert _deserialize_hitl_response(None, None) is None + assert _deserialize_hitl_response("approved", None) == "approved" + assert _deserialize_hitl_response({"__pickled__": "evil"}, None) is None + + def test_deserialize_hitl_response_reconstructs_typed_payload(self) -> None: + result = _deserialize_hitl_response({"approved": True}, f"{__name__}:_ApprovalModel") + + assert isinstance(result, _ApprovalModel) + assert result.approved is True + + def test_deserialize_hitl_response_returns_sanitized_dict_when_type_unknown(self) -> None: + payload = {"approved": False} + + assert _deserialize_hitl_response(payload, "missing.module:Type") == payload + + async def test_execute_hitl_response_handler_invokes_selected_handler(self) -> None: + handler = AsyncMock() + executor = Mock() + executor.id = "reviewer" + executor._find_response_handler.return_value = handler + shared_state = State() + runner_context = Mock() + + await execute_hitl_response_handler( + executor, + { + "original_request": {"question": "approve?"}, + "response": {"approved": True}, + "response_type": f"{__name__}:_ApprovalModel", + }, + shared_state, + runner_context, + ) + + assert handler.await_args is not None + response, workflow_context = handler.await_args.args + assert isinstance(response, _ApprovalModel) + assert response.approved is True + assert workflow_context._executor is executor + assert workflow_context._runner_context is runner_context + assert workflow_context.state is shared_state + executor._find_response_handler.assert_called_once() + + async def test_execute_hitl_response_handler_returns_when_no_handler_exists(self) -> None: + executor = Mock() + executor.id = "reviewer" + executor._find_response_handler.return_value = None + + await execute_hitl_response_handler( + executor, + {"original_request": {"question": "approve?"}, "response": "yes", "response_type": None}, + State(), + Mock(), + ) + + executor._find_response_handler.assert_called_once() + + def test_collect_hitl_requests_records_pending_entries(self) -> None: + pending: dict[str, PendingHITLRequest] = {} + + _collect_hitl_requests( + ExecutorResult( + executor_id="reviewer", + output_message=None, + activity_result={ + "pending_request_info_events": [ + { + "request_id": "req-1", + "data": {"question": "approve?"}, + "request_type": "ApprovalRequest", + "response_type": "ApprovalResponse", + } + ] + }, + task_type=TaskType.ACTIVITY, + ), + pending, + ) + + assert pending["req-1"] == PendingHITLRequest( + request_id="req-1", + source_executor_id="reviewer", + request_data={"question": "approve?"}, + request_type="ApprovalRequest", + response_type="ApprovalResponse", + ) + + def test_route_hitl_response_enqueues_message_for_source_executor(self) -> None: + pending_messages: dict[str, list[tuple[Any, str]]] = {} + + _route_hitl_response( + PendingHITLRequest( + request_id="req-2", + source_executor_id="reviewer", + request_data={"question": "approve?"}, + request_type="ApprovalRequest", + response_type="ApprovalResponse", + ), + {"approved": True}, + pending_messages, + ) + + assert pending_messages == { + "reviewer": [ + ( + { + "request_id": "req-2", + "original_request": {"question": "approve?"}, + "response": {"approved": True}, + "response_type": "ApprovalResponse", + }, + f"{SOURCE_HITL_RESPONSE}_req-2", + ) + ] + } + + +class TestResultRoutingHelpers: + """Result-processing helpers update state and feed routing queues correctly.""" + + def test_process_activity_result_applies_state_updates_and_outputs(self) -> None: + shared_state = {"keep": 1, "drop": 2} + workflow_outputs: list[Any] = [] + + result = _process_activity_result( + json.dumps({ + "shared_state_updates": {"added": 3}, + "shared_state_deletes": ["drop"], + "outputs": ["out-1"], + }), + "router", + shared_state, + workflow_outputs, + ) + + assert result.task_type == TaskType.ACTIVITY + assert shared_state == {"keep": 1, "added": 3} + assert workflow_outputs == ["out-1"] + + def test_route_result_messages_handles_output_messages_explicit_targets_and_fanin(self) -> None: + fan_in_group = FanInEdgeGroup(source_ids=["router", "other"], target_id="joined") + edge_group = SingleEdgeGroup(source_id="router", target_id="next", condition=lambda _message: True) + workflow = Mock() + workflow.edge_groups = [fan_in_group, edge_group] + + next_pending_messages: dict[str, list[tuple[Any, str]]] = {} + fan_in_pending: dict[str, dict[str, list[tuple[Any, str]]]] = {fan_in_group.id: defaultdict(list)} + + _route_result_messages( + ExecutorResult( + executor_id="router", + output_message=_agent_response("assistant said hello"), + activity_result={ + "sent_messages": [ + { + "message": serialize_value(0), + "target_id": "explicit", + "source_id": "router", + } + ] + }, + task_type=TaskType.ACTIVITY, + ), + workflow, + next_pending_messages, + fan_in_pending, + ) + + assert next_pending_messages["next"][0][1] == "router" + assert next_pending_messages["explicit"] == [(0, "router")] + assert fan_in_pending[fan_in_group.id]["router"][0][1] == "router" + + def test_check_fan_in_ready_delivers_aggregated_messages(self) -> None: + fan_in_group = FanInEdgeGroup(source_ids=["a", "b"], target_id="joined") + workflow = Mock() + workflow.edge_groups = [fan_in_group] + fan_in_pending = { + fan_in_group.id: { + "a": [("from-a", "a")], + "b": [("from-b", "b")], + } + } + next_pending_messages: dict[str, list[tuple[Any, str]]] = {} + + _check_fan_in_ready(workflow, fan_in_pending, next_pending_messages) + + assert next_pending_messages == {"joined": [(["from-a", "from-b"], "a")]} + assert fan_in_pending[fan_in_group.id] == defaultdict(list) + + def test_check_fan_in_ready_waits_for_all_sources(self) -> None: + fan_in_group = FanInEdgeGroup(source_ids=["a", "b"], target_id="joined") + workflow = Mock() + workflow.edge_groups = [fan_in_group] + fan_in_pending = {fan_in_group.id: {"a": [("from-a", "a")]}} + next_pending_messages: dict[str, list[tuple[Any, str]]] = {} + + _check_fan_in_ready(workflow, fan_in_pending, next_pending_messages) + + assert next_pending_messages == {} + + +class TestPrimaryInputSelection: + """Primary-input type selection skips non-concrete declarations.""" + + def test_returns_first_concrete_type(self) -> None: + executor = Mock() + executor.input_types = ["not-a-type", dict, str] + + assert _select_primary_input_type(executor) is dict + + def test_returns_none_when_no_concrete_type_exists(self) -> None: + executor = Mock() + executor.input_types = ["not-a-type", Mock()] + + assert _select_primary_input_type(executor) is None diff --git a/python/packages/durabletask/tests/test_workflow_runner_context.py b/python/packages/durabletask/tests/test_workflow_runner_context.py new file mode 100644 index 0000000000..624d7053c6 --- /dev/null +++ b/python/packages/durabletask/tests/test_workflow_runner_context.py @@ -0,0 +1,103 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for the durabletask workflow runner context.""" + +# pyright: reportPrivateUsage=false + +from __future__ import annotations + +from unittest.mock import Mock + +import pytest +from agent_framework import WorkflowEvent, WorkflowMessage +from agent_framework._workflows._state import State + +from agent_framework_durabletask._workflows.runner_context import ( + HOST_METADATA_INSTANCE_ID, + HOST_METADATA_REQUEST_PATH_PREFIX, + HOST_METADATA_WORKFLOW_NAME, + CapturingRunnerContext, +) + + +@pytest.fixture +def context() -> CapturingRunnerContext: + return CapturingRunnerContext() + + +async def test_send_and_drain_messages(context: CapturingRunnerContext) -> None: + message = WorkflowMessage(data="hello", target_id="target", source_id="source") + + await context.send_message(message) + + assert await context.has_messages() is True + assert await context.drain_messages() == {"source": [message]} + assert await context.has_messages() is False + + +async def test_events_can_be_queued_and_read(context: CapturingRunnerContext) -> None: + event = WorkflowEvent("output", executor_id="exec", data="payload") + + await context.add_event(event) + + assert await context.has_events() is True + assert await context.next_event() == event + assert await context.has_events() is False + + +def test_checkpointing_is_unsupported(context: CapturingRunnerContext) -> None: + storage = Mock() + + context.set_runtime_checkpoint_storage(storage) + context.clear_runtime_checkpoint_storage() + + assert context.has_checkpointing() is False + + +async def test_checkpoint_methods_raise(context: CapturingRunnerContext) -> None: + with pytest.raises(NotImplementedError, match="Checkpointing is not supported"): + await context.create_checkpoint("workflow", "sig", State(), None, 1) + + with pytest.raises(NotImplementedError, match="Checkpointing is not supported"): + await context.load_checkpoint("checkpoint-1") + + with pytest.raises(NotImplementedError, match="Checkpointing is not supported"): + await context.apply_checkpoint(Mock()) + + +def test_workflow_configuration_can_be_reset(context: CapturingRunnerContext) -> None: + context.set_workflow_id("workflow-123") + context.set_streaming(True) + context.set_host_metadata({ + HOST_METADATA_INSTANCE_ID: "root-instance", + HOST_METADATA_WORKFLOW_NAME: "wf", + HOST_METADATA_REQUEST_PATH_PREFIX: "sub~0~", + }) + context.set_yield_output_classifier(lambda executor_id: None if executor_id == "secret" else "intermediate") + + assert context.is_streaming() is True + assert context.host_metadata == { + HOST_METADATA_INSTANCE_ID: "root-instance", + HOST_METADATA_WORKFLOW_NAME: "wf", + HOST_METADATA_REQUEST_PATH_PREFIX: "sub~0~", + } + assert context.classify_yielded_output("secret") is None + assert context.classify_yielded_output("visible") == "intermediate" + + context.reset_for_new_run() + + assert context.is_streaming() is False + + +async def test_request_info_events_are_tracked(context: CapturingRunnerContext) -> None: + event = WorkflowEvent("request_info", executor_id="review", data={"question": "approve?"}, request_id="req-9") + + await context.add_request_info_event(event) + + assert await context.get_pending_request_info_events() == {"req-9": event} + assert await context.drain_events() == [event] + + +async def test_request_info_response_is_not_supported(context: CapturingRunnerContext) -> None: + with pytest.raises(NotImplementedError, match="orchestrator level"): + await context.send_request_info_response("req-9", {"approved": True}) diff --git a/python/packages/hyperlight/tests/hyperlight/test_hyperlight_codeact.py b/python/packages/hyperlight/tests/hyperlight/test_hyperlight_codeact.py index 70c7778b8d..1a62bbeb5b 100644 --- a/python/packages/hyperlight/tests/hyperlight/test_hyperlight_codeact.py +++ b/python/packages/hyperlight/tests/hyperlight/test_hyperlight_codeact.py @@ -43,6 +43,7 @@ from agent_framework_hyperlight import AllowedDomain, FileMount, HyperlightCodeActProvider, HyperlightExecuteCodeTool from agent_framework_hyperlight import _execute_code_tool as execute_code_module +from agent_framework_hyperlight import _instructions as instructions_module def _hyperlight_integration_static_skip_reason() -> str | None: @@ -1052,6 +1053,61 @@ def test_execute_code_tool_allowed_domains_use_structured_entries_and_replace_by ] +def test_execute_code_tool_normalizers_reject_invalid_inputs() -> None: + with pytest.raises(ValueError, match="must not be empty"): + execute_code_module._normalize_domain(" ") + with pytest.raises(ValueError, match="Could not normalize allowed domain entry"): + execute_code_module._normalize_domain("https://") + with pytest.raises(ValueError, match="must not be empty"): + execute_code_module._normalize_http_method(" ") + assert execute_code_module._normalize_http_methods(None) is None + with pytest.raises(ValueError, match="must not be empty when provided"): + execute_code_module._normalize_http_methods([]) + with pytest.raises(ValueError, match="must not be empty"): + execute_code_module._normalize_mount_path(" ") + with pytest.raises(ValueError, match="must stay within /input"): + execute_code_module._normalize_mount_path("/input/../escape") + with pytest.raises(ValueError, match="must point to a concrete path under /input"): + execute_code_module._normalize_mount_path("/input") + + +def test_execute_code_tool_shape_guards_validate_pairs() -> None: + assert execute_code_module._is_file_mount_pair(("source.txt", "mount.txt")) is True + assert execute_code_module._is_file_mount_pair(("source.txt", "mount.txt", "extra")) is False + assert execute_code_module._is_file_mount_pair(("source.txt", 1)) is False + + assert execute_code_module._is_allowed_domain_pair(("example.com", "get")) is True + assert execute_code_module._is_allowed_domain_pair(("example.com", ["get", "post"])) is True + assert execute_code_module._is_allowed_domain_pair((123, ["get"])) is False + assert execute_code_module._is_allowed_domain_pair(("example.com", 123)) is False + + +def test_instruction_builders_cover_mounted_paths_and_workspace_free_filesystem_state() -> None: + description = instructions_module.build_execute_code_description( + tools=[compute], + filesystem_enabled=True, + workspace_enabled=False, + mounted_paths=["/input/data/report.txt"], + allowed_domains=[], + ) + instructions = instructions_module.build_codeact_instructions( + tools=[compute], + tools_visible_to_model=True, + filesystem_enabled=True, + ) + filesystem_text = instructions_module._format_filesystem_capabilities( + filesystem_enabled=True, + workspace_enabled=False, + mounted_paths=[], + ) + + assert "Additional mounted paths:" in description + assert "/input/data/report.txt" in description + assert "Some tools may also appear directly" in instructions + assert "For larger artifacts, write them to `/output/` instead" in instructions + assert "No workspace root or explicit file mounts are currently configured." in filesystem_text + + def test_execute_code_tool_description_contains_call_tool_guidance(tmp_path: Path) -> None: workspace_root = tmp_path / "workspace" workspace_root.mkdir() @@ -1253,6 +1309,32 @@ async def test_provider_injects_run_scoped_execute_code_tool() -> None: assert [tool_obj.name for tool_obj in run_tool.get_tools()] == ["compute"] +def test_provider_delegates_file_mounts_and_allowed_domains_to_internal_tool(tmp_path: Path) -> None: + provider = HyperlightCodeActProvider() + + provider.add_file_mounts((tmp_path, "reports/output.txt")) + assert provider.get_file_mounts() == [FileMount(tmp_path.resolve(), "/input/reports/output.txt")] + + provider.remove_file_mount("/input/reports/output.txt") + assert provider.get_file_mounts() == [] + + provider.add_file_mounts((tmp_path, "reports/output.txt")) + provider.clear_file_mounts() + assert provider.get_file_mounts() == [] + + provider.add_allowed_domains([("api.example.com", "get"), "github.com"]) + assert provider.get_allowed_domains() == [ + AllowedDomain("api.example.com", ("GET",)), + AllowedDomain("github.com", None), + ] + + provider.remove_allowed_domain("github.com") + assert provider.get_allowed_domains() == [AllowedDomain("api.example.com", ("GET",))] + + provider.clear_allowed_domains() + assert provider.get_allowed_domains() == [] + + async def test_agent_runs_hyperlight_codeact_end_to_end_with_fake_sandbox(monkeypatch: pytest.MonkeyPatch) -> None: _FakeSandbox.instances.clear() monkeypatch.setattr(execute_code_module, "_load_sandbox_class", lambda: _FakeSandbox) diff --git a/python/packages/monty/tests/monty/test_monty_codeact.py b/python/packages/monty/tests/monty/test_monty_codeact.py index 3a0428100c..b319428bf5 100644 --- a/python/packages/monty/tests/monty/test_monty_codeact.py +++ b/python/packages/monty/tests/monty/test_monty_codeact.py @@ -24,6 +24,7 @@ from agent_framework_monty import MontyCodeActProvider, MontyExecuteCodeTool from agent_framework_monty import _execute_code_tool as execute_code_module +from agent_framework_monty import _instructions as instructions_module from agent_framework_monty import _monty_bridge as bridge_module # --------------------------------------------------------------------------- @@ -184,6 +185,14 @@ def dangerous_tool(payload: Annotated[str, "Anything"]) -> str: return payload +def _decode_content_bytes(item: Content) -> bytes: + import base64 + + assert item.uri is not None + _, _, encoded = item.uri.partition("base64,") + return base64.b64decode(encoded) + + # --------------------------------------------------------------------------- # MontyExecuteCodeTool tests # --------------------------------------------------------------------------- @@ -335,6 +344,22 @@ def test_dynamic_description_default_mentions_no_filesystem() -> None: assert "Filesystem access is unavailable" in description +def test_instruction_builders_describe_write_caps_and_visible_tools(tmp_path: Path) -> None: + from agent_framework_monty import FileMount + + mount = FileMount(host_path=tmp_path, mount_path="/work", mode="read-write", write_bytes_limit=128) + description = instructions_module.build_execute_code_description(tools=[add_tool], mounts=[mount]) + instructions = instructions_module.build_codeact_instructions( + tools=[add_tool], + tools_visible_to_model=True, + mounts=[mount], + ) + + assert "write cap 128 bytes" in description + assert "Files written to `/work` are returned" in description + assert "Some tools may also appear directly" in instructions + + def test_resource_limits_round_trip() -> None: monty_tool = MontyExecuteCodeTool(resource_limits={"max_duration_secs": 5.0}) assert monty_tool.resource_limits == {"max_duration_secs": 5.0} @@ -360,6 +385,61 @@ def test_execute_code_filtered_out_when_added_as_tool() -> None: assert [t.name for t in monty_tool.get_tools()] == ["add_tool"] +def test_mount_helpers_validate_inputs_and_convert_mounts( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from agent_framework_monty import FileMount + + host_dir = tmp_path / "host" + host_dir.mkdir() + file_path = tmp_path / "file.txt" + file_path.write_text("x", encoding="utf-8") + + assert execute_code_module._is_file_mount_pair((host_dir, "/work")) is True + assert execute_code_module._is_file_mount_pair(FileMount(host_path=host_dir, mount_path="/work")) is False + assert execute_code_module._is_file_mount_pair((host_dir, "/work", "extra")) is False + assert execute_code_module._is_file_mount_pair((host_dir, 1)) is False + + with pytest.raises(ValueError, match="must not be empty"): + execute_code_module._normalize_mount_path(" ") + with pytest.raises(ValueError, match="must not contain '..' segments"): + execute_code_module._normalize_mount_path("/work/../escape") + with pytest.raises(ValueError, match="must point to a concrete absolute path"): + execute_code_module._normalize_mount_path("/") + with pytest.raises(ValueError, match="existing directory"): + execute_code_module._resolve_existing_directory(file_path) + + calls: list[dict[str, Any]] = [] + + class _FakeMountDir: + def __init__(self, **kwargs: Any) -> None: + calls.append(kwargs) + + monkeypatch.setattr(bridge_module, "load_monty", lambda: types.SimpleNamespace(MountDir=_FakeMountDir)) + execute_code_module._to_monty_mount( + FileMount(host_path=host_dir, mount_path="/work", mode="read-write", write_bytes_limit=12) + ) + + assert calls == [ + { + "virtual_path": "/work", + "host_path": str(host_dir), + "mode": "read-write", + "write_bytes_limit": 12, + } + ] + + +def test_to_dict_materializes_dynamic_description(tmp_path: Path) -> None: + monty_tool = MontyExecuteCodeTool(tools=[add_tool], workspace_root=tmp_path) + serialized = monty_tool.to_dict() + + assert monty_tool.workspace_root == tmp_path.resolve() + assert "description" in serialized + assert "add_tool" in serialized["description"] + + # --------------------------------------------------------------------------- # _run_code behavior with the fake Monty runtime # --------------------------------------------------------------------------- @@ -507,6 +587,69 @@ async def run(self, code: str) -> dict[str, Any]: assert "boom" in (result[0].error_details or "") +def test_build_execution_contents_handles_truncation_and_non_json_output() -> None: + truncated = execute_code_module._build_execution_contents( + result={"stdout": "hello", "truncated": True, "output": complex(1, 2)} + ) + assert [item.text for item in truncated] == ["hello\n\n[stdout truncated]", "(1+2j)"] + + truncated_only = execute_code_module._build_execution_contents( + result={"stdout": "", "truncated": True, "output": None} + ) + assert [item.text for item in truncated_only] == ["[stdout truncated]"] + + +def test_capture_written_files_returns_new_files_and_omits_large_ones( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from agent_framework_monty import FileMount + + writable = tmp_path / "writable" + writable.mkdir() + readonly = tmp_path / "readonly" + readonly.mkdir() + nested = writable / "nested" + nested.mkdir() + + existing = writable / "existing.txt" + existing.write_text("before", encoding="utf-8") + (nested / "report.txt").write_text("old", encoding="utf-8") + (readonly / "ignored.txt").write_text("unchanged", encoding="utf-8") + + mounts = [ + FileMount(host_path=writable, mount_path="/work", mode="read-write"), + FileMount(host_path=readonly, mount_path="/readonly", mode="read-only"), + ] + pre_state = execute_code_module._snapshot_writable_mounts(mounts) + + existing.write_text("after", encoding="utf-8") + (nested / "report.txt").write_text("updated", encoding="utf-8") + (writable / "artifact.bin").write_bytes(b"\x00\x01") + (writable / "large.txt").write_text("123456789", encoding="utf-8") + monkeypatch.setattr(execute_code_module, "MAX_CAPTURED_FILE_BYTES", 8) + + captured = execute_code_module._capture_written_files(mounts, pre_state) + data_items = [item for item in captured if item.type == "data"] + text_items = [item for item in captured if item.type == "text"] + + assert set(pre_state) == {"/work"} + assert "existing.txt" in pre_state["/work"] + assert "ignored.txt" not in pre_state["/work"] + assert {item.additional_properties["path"] for item in data_items} == { + "/work/artifact.bin", + "/work/existing.txt", + "/work/nested/report.txt", + } + assert any("large.txt" in (item.text or "") and "omitted" in (item.text or "") for item in text_items) + assert any( + _decode_content_bytes(item) == b"after" + for item in data_items + if item.additional_properties["path"] == "/work/existing.txt" + ) + assert all(not item.additional_properties["path"].startswith("/readonly/") for item in data_items) + + # --------------------------------------------------------------------------- # MontyCodeActProvider tests # --------------------------------------------------------------------------- @@ -539,6 +682,20 @@ def test_provider_delegates_tool_management_to_internal_tool() -> None: assert provider.get_tools() == [] +def test_provider_delegates_file_mount_management_to_internal_tool(tmp_path: Path) -> None: + provider = MontyCodeActProvider() + provider.add_file_mounts((tmp_path, "/work")) + + assert [mount.mount_path for mount in provider.get_file_mounts()] == ["/work"] + + provider.remove_file_mount("/work") + assert provider.get_file_mounts() == [] + + provider.add_file_mounts((tmp_path, "/again")) + provider.clear_file_mounts() + assert provider.get_file_mounts() == [] + + # --------------------------------------------------------------------------- # generate_type_stubs - signature smoke test # --------------------------------------------------------------------------- diff --git a/python/packages/redis/tests/test_context_provider_edges.py b/python/packages/redis/tests/test_context_provider_edges.py new file mode 100644 index 0000000000..36146f5201 --- /dev/null +++ b/python/packages/redis/tests/test_context_provider_edges.py @@ -0,0 +1,234 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Additional edge-case coverage for ``RedisContextProvider``.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Generator +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from agent_framework.exceptions import IntegrationInvalidRequestException +from redisvl.utils.vectorize import BaseVectorizer + +from agent_framework_redis._context_provider import RedisContextProvider + + +@pytest.fixture +def mock_index() -> AsyncMock: + index = AsyncMock() + index.create = AsyncMock() + index.exists = AsyncMock(return_value=False) + index.load = AsyncMock() + index.query = AsyncMock(return_value=[]) + return index + + +@pytest.fixture +def patch_index(mock_index: AsyncMock) -> Generator[MagicMock]: + with patch("agent_framework_redis._context_provider.AsyncSearchIndex") as mock_cls: + mock_cls.from_dict = MagicMock(return_value=mock_index) + mock_cls.from_existing = AsyncMock() + yield mock_cls + + +def test_build_filter_from_dict_combines_multiple_tags( + patch_index: MagicMock, # noqa: ARG001 +) -> None: + provider = RedisContextProvider(source_id="ctx", user_id="user-1") + combined = provider._build_filter_from_dict({ + "application_id": "app-1", + "agent_id": None, + "user_id": "user-1", + }) + + assert combined is not None + assert str(combined) == "(@application_id:{app\\-1} @user_id:{user\\-1})" + + +def test_schema_dict_includes_vector_configuration( + patch_index: MagicMock, # noqa: ARG001 +) -> None: + vectorizer = MagicMock(spec=BaseVectorizer) + vectorizer.dims = 3 + vectorizer.dtype = "float16" + + provider = RedisContextProvider( + source_id="ctx", + user_id="user-1", + redis_vectorizer=vectorizer, + vector_field_name="embedding", + vector_algorithm="flat", + vector_distance_metric="l2", + ) + + vector_field = next(field for field in provider.schema_dict["fields"] if field["name"] == "embedding") + + assert vector_field["type"] == "vector" + assert vector_field["attrs"] == { + "algorithm": "flat", + "dims": 3, + "distance_metric": "l2", + "datatype": "float16", + } + + +async def test_ensure_index_short_circuits_after_first_initialization( + mock_index: AsyncMock, + patch_index: MagicMock, # noqa: ARG001 +) -> None: + provider = RedisContextProvider(source_id="ctx", user_id="user-1") + provider._index_initialized = True + + await provider._ensure_index() + + mock_index.exists.assert_not_called() + mock_index.create.assert_not_called() + + +async def test_ensure_index_validates_existing_schema_before_create( + mock_index: AsyncMock, + patch_index: MagicMock, # noqa: ARG001 +) -> None: + mock_index.exists.return_value = True + provider = RedisContextProvider(source_id="ctx", user_id="user-1") + + with patch.object(provider, "_validate_schema_compatibility", AsyncMock()) as validate_schema: + await provider._ensure_index() + + validate_schema.assert_awaited_once() + mock_index.create.assert_awaited_once_with(overwrite=False, drop=False) + assert provider._index_initialized is True + + +async def test_validate_schema_compatibility_raises_for_significant_mismatch( + patch_index: MagicMock, +) -> None: + provider = RedisContextProvider(source_id="ctx", user_id="user-1") + existing_index = AsyncMock() + existing_index.schema.to_dict = MagicMock( + return_value={ + "index": {"name": "context", "prefix": "other", "key_separator": ":", "storage_type": "hash"}, + "fields": [{"name": "content", "type": "text"}], + } + ) + patch_index.from_existing = AsyncMock(return_value=existing_index) + + with pytest.raises(ValueError, match="overwrite_index=True"): + await provider._validate_schema_compatibility() + + +async def test_add_requires_content_field( + patch_index: MagicMock, # noqa: ARG001 +) -> None: + provider = RedisContextProvider(source_id="ctx", user_id="user-1") + + with ( + patch.object(provider, "_ensure_index", AsyncMock()), + pytest.raises(IntegrationInvalidRequestException, match="requires a 'content' field"), + ): + await provider._add(data={"role": "user"}, session_id="session-1") + + +async def test_add_vectorizes_documents_and_applies_defaults( + mock_index: AsyncMock, + patch_index: MagicMock, # noqa: ARG001 +) -> None: + vectorizer = MagicMock(spec=BaseVectorizer) + vectorizer.dims = 2 + vectorizer.dtype = "float32" + vectorizer.aembed_many = AsyncMock(return_value=[[1.0, 2.0], [3.0, 4.0]]) + + provider = RedisContextProvider( + source_id="ctx", + application_id="app-1", + agent_id="agent-1", + user_id="user-1", + redis_vectorizer=vectorizer, + vector_field_name="embedding", + ) + + with patch.object(provider, "_ensure_index", AsyncMock()): + await provider._add( + data=[ + {"content": "first"}, + {"content": "second", "conversation_id": "custom-conversation"}, + ], + session_id="session-1", + ) + + loaded_docs = mock_index.load.await_args.args[0] + assert [doc["content"] for doc in loaded_docs] == ["first", "second"] + assert loaded_docs[0]["application_id"] == "app-1" + assert loaded_docs[0]["agent_id"] == "agent-1" + assert loaded_docs[0]["user_id"] == "user-1" + assert loaded_docs[0]["thread_id"] == "session-1" + assert loaded_docs[0]["conversation_id"] == "session-1" + assert isinstance(loaded_docs[0]["embedding"], bytes) + assert isinstance(loaded_docs[1]["embedding"], bytes) + vectorizer.aembed_many.assert_awaited_once_with(["first", "second"], batch_size=2) + + +async def test_redis_search_requires_non_empty_text( + patch_index: MagicMock, # noqa: ARG001 +) -> None: + provider = RedisContextProvider(source_id="ctx", user_id="user-1") + + with ( + patch.object(provider, "_ensure_index", AsyncMock()), + pytest.raises(IntegrationInvalidRequestException, match="non-empty text"), + ): + await provider._redis_search(text=" ") + + +async def test_redis_search_combines_explicit_filter_expression( + mock_index: AsyncMock, + patch_index: MagicMock, # noqa: ARG001 +) -> None: + provider = RedisContextProvider(source_id="ctx", user_id="user-1", application_id="app-1") + base_filter = MagicMock() + merged_filter = object() + base_filter.__and__.return_value = merged_filter + explicit_filter = object() + + with ( + patch.object(provider, "_ensure_index", AsyncMock()), + patch.object(provider, "_build_filter_from_dict", return_value=base_filter), + patch("agent_framework_redis._context_provider.TextQuery") as text_query, + ): + text_query.return_value = MagicMock() + await provider._redis_search( + text="hello redis", + session_id="session-1", + filter_expression=explicit_filter, + return_fields=["content"], + num_results=3, + ) + + base_filter.__and__.assert_called_once_with(explicit_filter) + assert text_query.call_args.kwargs["filter_expression"] is merged_filter + assert text_query.call_args.kwargs["return_fields"] == ["content"] + assert text_query.call_args.kwargs["num_results"] == 3 + mock_index.query.assert_awaited_once() + + +async def test_search_all_collects_paginated_batches( + mock_index: AsyncMock, + patch_index: MagicMock, # noqa: ARG001 +) -> None: + provider = RedisContextProvider(source_id="ctx", user_id="user-1") + + async def paginate(*args: Any, **kwargs: Any) -> AsyncIterator[list[dict[str, str]]]: # noqa: ARG001 + yield [{"content": "first"}] + yield [{"content": "second"}, {"content": "third"}] + + mock_index.paginate = MagicMock(return_value=paginate()) + + results = await provider.search_all(page_size=2) + + assert results == [ + {"content": "first"}, + {"content": "second"}, + {"content": "third"}, + ] diff --git a/python/packages/tools/tests/test_docker_shell_tool.py b/python/packages/tools/tests/test_docker_shell_tool.py index 2e9bda07a1..da486f8cb3 100644 --- a/python/packages/tools/tests/test_docker_shell_tool.py +++ b/python/packages/tools/tests/test_docker_shell_tool.py @@ -10,14 +10,21 @@ from __future__ import annotations +import asyncio import subprocess import sys +from collections.abc import Sequence +from typing import TypeAlias +from unittest.mock import AsyncMock, MagicMock, patch import pytest from agent_framework_tools.shell import ( + DockerNotAvailableError, DockerShellTool, + ShellCommandError, ShellExecutor, + ShellResult, is_docker_available, ) from agent_framework_tools.shell._docker import ( @@ -25,6 +32,44 @@ build_run_argv, ) +_CommunicateOutcome: TypeAlias = tuple[bytes, bytes] | BaseException +_WaitOutcome: TypeAlias = int | None | BaseException + + +class _FakeProcess: + def __init__( + self, + *, + pid: int = 1234, + returncode: int | None = 0, + communicate_results: Sequence[_CommunicateOutcome] | None = None, + wait_results: Sequence[_WaitOutcome] | None = None, + ) -> None: + self.pid = pid + self.returncode = returncode + self.stdout = object() + self.stderr = object() + self.killed = False + self._communicate_results = list(communicate_results or [(b"", b"")]) + self._wait_results = list(wait_results or [returncode]) + + async def communicate(self) -> tuple[bytes, bytes]: + result = self._communicate_results.pop(0) + if isinstance(result, BaseException): + raise result + stdout, stderr = result + return stdout, stderr + + async def wait(self) -> int | None: + result = self._wait_results.pop(0) + if isinstance(result, BaseException): + raise result + self.returncode = result + return result + + def kill(self) -> None: + self.killed = True + def _docker_image_available(image: str) -> bool: if not is_docker_available(): @@ -236,6 +281,233 @@ def test_as_function_carries_shell_kind(): ) +async def test_start_and_close_are_noops_in_stateless_mode() -> None: + tool = DockerShellTool(mode="stateless") + + with ( + patch.object(tool, "_start_container", AsyncMock()) as start_container, + patch.object(tool, "_stop_container", AsyncMock()) as stop_container, + ): + await tool.start() + await tool.close() + + start_container.assert_not_called() + stop_container.assert_not_called() + + +async def test_start_creates_and_reuses_persistent_session() -> None: + tool = DockerShellTool(docker_binary="podman", shell="sh") + session = AsyncMock() + + with ( + patch.object(tool, "_start_container", AsyncMock()) as start_container, + patch("agent_framework_tools.shell._docker.ShellSession", return_value=session) as shell_session, + ): + await tool.start() + await tool.start() + + start_container.assert_awaited_once() + shell_session.assert_called_once_with( + ["podman", "exec", "-i", tool._container_name, "sh"], + workdir=None, + env=None, + max_output_bytes=tool._max_output_bytes, + ) + assert session.start.await_count == 2 + + +async def test_close_terminates_session_and_container() -> None: + tool = DockerShellTool() + tool._container_started = True + session = AsyncMock() + tool._session = session + + with patch.object(tool, "_stop_container", AsyncMock()) as stop_container: + await tool.close() + + session.close.assert_awaited_once() + stop_container.assert_awaited_once() + assert tool._session is None + assert tool._container_started is False + + +async def test_run_rejects_denied_commands() -> None: + tool = DockerShellTool( + policy=MagicMock(evaluate=MagicMock(return_value=MagicMock(decision="deny", reason="blocked"))) + ) + + with pytest.raises(ShellCommandError, match="blocked"): + await tool.run("danger") + + +async def test_run_logs_audit_hook_failures_and_executes_persistent_command( + caplog: pytest.LogCaptureFixture, +) -> None: + def broken_hook(command: str) -> None: + raise RuntimeError(f"boom:{command}") + + tool = DockerShellTool(on_command=broken_hook) + tool._session = AsyncMock(run=AsyncMock(return_value=ShellResult("", "", 0, 1))) + + result = await tool.run("echo hi") + + assert result.exit_code == 0 + assert "on_command hook raised" in caplog.text + tool._session.run.assert_awaited_once_with("echo hi", timeout=30.0) + + +async def test_run_raises_if_start_did_not_create_persistent_session() -> None: + tool = DockerShellTool() + + with patch.object(tool, "start", AsyncMock()), pytest.raises(RuntimeError, match="session failed to start"): + await tool.run("echo hi") + + +async def test_run_dispatches_to_private_stateless_runner() -> None: + tool = DockerShellTool(mode="stateless") + expected = ShellResult(stdout="ok", stderr="", exit_code=0, duration_ms=1) + + with patch.object(tool, "_run_stateless", AsyncMock(return_value=expected)) as run_stateless: + result = await tool.run("echo hi", timeout=9.0) + + assert result is expected + run_stateless.assert_awaited_once_with("echo hi", timeout=9.0) + + +async def test_run_stateless_builds_expected_argv() -> None: + tool = DockerShellTool( + mode="stateless", + docker_binary="podman", + image="alpine:3", + shell="sh", + host_workdir="/repo", + workdir="/workspace", + mount_readonly=False, + env={"AF_TEST": "1"}, + ) + proc = _FakeProcess(returncode=3, communicate_results=[(b"hello\n", b"warning\n")]) + + with patch( + "agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", + AsyncMock(return_value=proc), + ) as create_proc: + result = await tool._run_stateless("echo hi", timeout=12.0) + + assert create_proc.await_args is not None + argv = create_proc.await_args.args + assert argv[:4] == ("podman", "run", "--rm", "-i") + assert "-v" in argv + assert "/repo:/workspace:rw" in argv + assert "AF_TEST=1" in argv + assert argv[-4:] == ("alpine:3", "sh", "-c", "echo hi") + assert result.stdout == "hello\n" + assert result.stderr == "warning\n" + assert result.exit_code == 3 + assert result.timed_out is False + + +async def test_run_stateless_timeout_reaps_container_when_kill_fails() -> None: + tool = DockerShellTool(mode="stateless") + command_proc = _FakeProcess( + returncode=137, + communicate_results=[asyncio.TimeoutError(), (b"after-timeout", b"stderr")], + ) + killer = _FakeProcess(returncode=1) + reaper = _FakeProcess(returncode=9, communicate_results=[(b"", b"rm failed")]) + + with patch( + "agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", + AsyncMock(side_effect=[command_proc, killer, reaper]), + ): + result = await tool._run_stateless("sleep 5", timeout=0.01) + + assert result.timed_out is True + assert result.exit_code == 137 + assert result.stdout == "after-timeout" + assert result.stderr == "stderr" + + +async def test_run_stateless_timeout_handles_kill_and_reaper_timeouts() -> None: + tool = DockerShellTool(mode="stateless") + command_proc = _FakeProcess( + returncode=None, + communicate_results=[asyncio.TimeoutError(), RuntimeError("drain failed")], + ) + killer = _FakeProcess(returncode=None, wait_results=[asyncio.TimeoutError()]) + reaper = _FakeProcess(returncode=None, communicate_results=[asyncio.TimeoutError()]) + + with patch( + "agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", + AsyncMock(side_effect=[command_proc, killer, reaper]), + ): + result = await tool._run_stateless("sleep 5", timeout=0.01) + + assert killer.killed is True + assert reaper.killed is True + assert result.timed_out is True + assert result.exit_code == -1 + assert result.stdout == "" + assert result.stderr == "" + + +async def test_start_container_success_logs_container_id(caplog: pytest.LogCaptureFixture) -> None: + tool = DockerShellTool() + proc = _FakeProcess(returncode=0, communicate_results=[(b"abcdef1234567890\n", b"")]) + + with ( + caplog.at_level("INFO", logger="agent_framework_tools.shell._docker"), + patch("agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", AsyncMock(return_value=proc)), + ): + await tool._start_container() + + assert f"started docker container {tool._container_name}" in caplog.text + + +async def test_start_container_raises_when_runtime_fails() -> None: + tool = DockerShellTool() + proc = _FakeProcess(returncode=7, communicate_results=[(b"", b"daemon unavailable")]) + + with ( + patch("agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", AsyncMock(return_value=proc)), + pytest.raises(DockerNotAvailableError, match="daemon unavailable"), + ): + await tool._start_container() + + +async def test_stop_container_returns_after_first_success() -> None: + tool = DockerShellTool() + proc = _FakeProcess(returncode=0, communicate_results=[(b"", b"")]) + + with patch("agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", AsyncMock(return_value=proc)): + await tool._stop_container() + + +async def test_stop_container_retries_when_first_attempt_fails() -> None: + tool = DockerShellTool() + first = _FakeProcess(returncode=1, communicate_results=[(b"", b"still running")]) + second = _FakeProcess(returncode=2, communicate_results=[(b"", b"still running")]) + + with patch( + "agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", + AsyncMock(side_effect=[first, second]), + ) as create_proc: + await tool._stop_container() + + assert create_proc.await_count == 2 + + +async def test_as_function_surfaces_command_errors() -> None: + tool = DockerShellTool(mode="persistent") + + with patch.object(tool, "run", AsyncMock(side_effect=ShellCommandError("blocked"))): + function = tool.as_function() + assert function.func is not None + result = await function.func("pwd") + + assert result == "blocked" + assert "persistent session" in function.description + + # --------------------------------------------------------------------- integration diff --git a/python/packages/tools/tests/test_local_shell_tool.py b/python/packages/tools/tests/test_local_shell_tool.py index 61d28c6a7e..5f5488bb29 100644 --- a/python/packages/tools/tests/test_local_shell_tool.py +++ b/python/packages/tools/tests/test_local_shell_tool.py @@ -1,13 +1,34 @@ # Copyright (c) Microsoft. All rights reserved. +import asyncio import os import sys +from unittest.mock import AsyncMock, patch import pytest from agent_framework_tools.shell import LocalShellTool, ShellCommandError, ShellPolicy +from agent_framework_tools.shell._executor import _popen_kwargs_for_group, run_stateless -pytestmark = pytest.mark.asyncio + +class _FakeExecProcess: + def __init__( + self, + *, + returncode: int | None = 0, + communicate_results: list[tuple[bytes, bytes] | BaseException] | None = None, + ) -> None: + self.returncode = returncode + self.stdout = object() + self.stderr = object() + self._communicate_results = list(communicate_results or [(b"", b"")]) + + async def communicate(self) -> tuple[bytes, bytes]: + result = self._communicate_results.pop(0) + if isinstance(result, BaseException): + raise result + stdout, stderr = result + return stdout, stderr async def test_stateless_echo() -> None: @@ -71,6 +92,123 @@ async def test_audit_hook_fires_for_allowed_commands() -> None: assert seen == [cmd] +def test_local_shell_tool_handles_mode_and_environment_variants(monkeypatch: pytest.MonkeyPatch) -> None: + with pytest.raises(ValueError, match="mode must be"): + LocalShellTool(mode="bogus") # type: ignore[arg-type] # ty: ignore[invalid-argument-type] + + monkeypatch.setenv("INHERITED", "yes") + inherited = LocalShellTool( + mode="stateless", + approval_mode="never_require", + acknowledge_unsafe=True, + env={"EXTRA": "1"}, + ) + clean = LocalShellTool( + mode="stateless", + approval_mode="never_require", + acknowledge_unsafe=True, + env={"ONLY": "2"}, + clean_env=True, + ) + + assert inherited._env is not None + assert inherited._env["INHERITED"] == "yes" + assert inherited._env["EXTRA"] == "1" + assert clean._env == {"ONLY": "2"} + + +async def test_local_shell_tool_stateless_start_is_noop() -> None: + tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True) + await tool.start() + await tool.close() + + +async def test_local_shell_tool_raises_if_start_did_not_create_session() -> None: + tool = LocalShellTool(mode="persistent", approval_mode="never_require", acknowledge_unsafe=True) + + with patch.object(tool, "start", AsyncMock()), pytest.raises(RuntimeError, match="session failed to start"): + await tool.run("echo hi") + + +async def test_local_shell_tool_as_function_returns_policy_errors() -> None: + tool = LocalShellTool(mode="persistent", approval_mode="never_require", acknowledge_unsafe=True) + + with patch.object(tool, "run", AsyncMock(side_effect=ShellCommandError("blocked"))): + function = tool.as_function(description="custom shell") + assert function.func is not None + result = await function.func("pwd") + + assert result == "blocked" + assert function.description == "custom shell" + + +def test_local_shell_tool_reanchors_powershell_paths() -> None: + tool = LocalShellTool( + mode="persistent", + shell="pwsh", + workdir="C:\\repo", + approval_mode="never_require", + acknowledge_unsafe=True, + ) + + assert tool._maybe_reanchor("Get-ChildItem").startswith("Set-Location -LiteralPath 'C:\\repo'") + + +def test_popen_kwargs_for_group_covers_windows_branch(monkeypatch: pytest.MonkeyPatch) -> None: + import agent_framework_tools.shell._executor as executor_module + + monkeypatch.setattr(executor_module.sys, "platform", "win32") + monkeypatch.setattr(executor_module.subprocess, "CREATE_NEW_PROCESS_GROUP", 77, raising=False) + + assert _popen_kwargs_for_group() == {"creationflags": 77} + + +async def test_run_stateless_adds_powershell_encoding_preamble() -> None: + proc = _FakeExecProcess(returncode=0, communicate_results=[(b"ok", b"")]) + + with ( + patch("agent_framework_tools.shell._executor.is_powershell", return_value=True), + patch( + "agent_framework_tools.shell._executor.asyncio.create_subprocess_exec", + AsyncMock(return_value=proc), + ) as create_proc, + ): + result = await run_stateless( + ["pwsh", "-Command"], + "Write-Output hi", + workdir=None, + env=None, + timeout=1.0, + max_output_bytes=1024, + ) + + assert result.stdout == "ok" + assert create_proc.await_args is not None + assert create_proc.await_args.args[-1].startswith("$OutputEncoding = [Console]::OutputEncoding") + + +async def test_run_stateless_timeout_returns_empty_output_if_drain_fails() -> None: + proc = _FakeExecProcess(returncode=None, communicate_results=[asyncio.TimeoutError(), RuntimeError("drain failed")]) + + with ( + patch("agent_framework_tools.shell._executor.asyncio.create_subprocess_exec", AsyncMock(return_value=proc)), + patch("agent_framework_tools.shell._executor.kill_process_tree", AsyncMock()) as kill_tree, + ): + result = await run_stateless( + ["/bin/sh", "-c"], + "sleep 5", + workdir=None, + env=None, + timeout=0.01, + max_output_bytes=1024, + ) + + kill_tree.assert_awaited_once_with(proc) + assert result.timed_out is True + assert result.stdout == "" + assert result.stderr == "" + + @pytest.mark.skipif(sys.platform == "win32", reason="persistent-mode sentinel on POSIX") async def test_persistent_preserves_cwd_and_exports_across_calls(tmp_path: os.PathLike[str]) -> None: async with LocalShellTool( diff --git a/python/packages/tools/tests/test_shell_killtree.py b/python/packages/tools/tests/test_shell_killtree.py new file mode 100644 index 0000000000..1052902670 --- /dev/null +++ b/python/packages/tools/tests/test_shell_killtree.py @@ -0,0 +1,147 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from typing import Any, cast +from unittest.mock import AsyncMock, MagicMock, patch + +from agent_framework_tools.shell._killtree import ( + _kill_via_psutil, + _kill_via_stdlib, + _resolve_taskkill, + kill_process_tree, +) + + +class _FakeAsyncProcess: + def __init__(self, *, pid: int = 101, returncode: int | None = None) -> None: + self.pid = pid + self.returncode = returncode + self.killed = False + + async def wait(self) -> int | None: + return self.returncode + + def kill(self) -> None: + self.killed = True + + +class _FakeExecProcess(_FakeAsyncProcess): + def __init__( + self, + *, + returncode: int | None = 0, + communicate_results: list[tuple[bytes, bytes] | BaseException] | None = None, + ) -> None: + super().__init__(returncode=returncode) + self.stdout = object() + self.stderr = object() + self._communicate_results = list(communicate_results or [(b"", b"")]) + + async def communicate(self) -> tuple[bytes, bytes]: + result = self._communicate_results.pop(0) + if isinstance(result, BaseException): + raise result + stdout, stderr = result + return stdout, stderr + + +def test_resolve_taskkill_uses_systemroot_and_caches(monkeypatch) -> None: + import agent_framework_tools.shell._killtree as killtree_module + + monkeypatch.setattr(killtree_module, "_taskkill_path", None) + monkeypatch.setenv("SystemRoot", "C:\\Windows") + monkeypatch.setattr(killtree_module.os.path, "isfile", lambda path: path.endswith("taskkill.exe")) + + assert _resolve_taskkill() == "C:\\Windows/System32/taskkill.exe" + assert _resolve_taskkill() == "C:\\Windows/System32/taskkill.exe" + + +async def test_kill_process_tree_short_circuits_or_delegates() -> None: + proc = cast(asyncio.subprocess.Process, _FakeAsyncProcess(returncode=0)) + await kill_process_tree(proc) + + live = cast(asyncio.subprocess.Process, _FakeAsyncProcess(returncode=None)) + with ( + patch("agent_framework_tools.shell._killtree._kill_via_psutil", AsyncMock()) as via_psutil, + patch("agent_framework_tools.shell._killtree._has_psutil", True), + ): + await kill_process_tree(live) + + via_psutil.assert_awaited_once_with(live, grace=2.0) + + +async def test_kill_via_psutil_terminates_parent_and_children() -> None: + import agent_framework_tools.shell._killtree as killtree_module + + no_such_process = type("NoSuchProcess", (Exception,), {}) + access_denied = type("AccessDenied", (Exception,), {}) + child = MagicMock(is_running=MagicMock(return_value=True)) + parent = MagicMock(children=MagicMock(return_value=[child]), is_running=MagicMock(return_value=True)) + fake_psutil = MagicMock( + Process=MagicMock(return_value=parent), + NoSuchProcess=no_such_process, + AccessDenied=access_denied, + ) + proc = cast(asyncio.subprocess.Process, _FakeAsyncProcess(pid=4321, returncode=None)) + + with patch.object(killtree_module, "psutil", fake_psutil): + await _kill_via_psutil(proc, grace=0.01) + + parent.terminate.assert_called_once() + child.terminate.assert_called_once() + parent.kill.assert_called_once() + child.kill.assert_called_once() + + +async def test_kill_via_psutil_handles_missing_parent_process() -> None: + import agent_framework_tools.shell._killtree as killtree_module + + no_such_process = type("NoSuchProcess", (Exception,), {}) + fake_psutil = MagicMock(Process=MagicMock(side_effect=no_such_process()), NoSuchProcess=no_such_process) + proc = cast(asyncio.subprocess.Process, _FakeAsyncProcess(pid=9999, returncode=None)) + + with patch.object(killtree_module, "psutil", fake_psutil): + await _kill_via_psutil(proc, grace=0.01) + + +async def test_kill_via_stdlib_windows_uses_taskkill_and_proc_kill(monkeypatch) -> None: + import agent_framework_tools.shell._killtree as killtree_module + + monkeypatch.setattr(killtree_module.sys, "platform", "win32") + monkeypatch.setattr(killtree_module, "_resolve_taskkill", lambda: "C:\\Windows\\System32\\taskkill.exe") + killer = _FakeExecProcess(returncode=None) + raw_proc = _FakeAsyncProcess(pid=55, returncode=None) + proc = cast(asyncio.subprocess.Process, raw_proc) + + with patch("agent_framework_tools.shell._killtree.asyncio.create_subprocess_exec", AsyncMock(return_value=killer)): + await _kill_via_stdlib(proc, grace=0.01) + + assert killer.killed is True + assert raw_proc.killed is True + + +async def test_kill_via_stdlib_posix_escalates_to_sigkill(monkeypatch) -> None: + import agent_framework_tools.shell._killtree as killtree_module + + monkeypatch.setattr(killtree_module.sys, "platform", "darwin") + killpg = MagicMock() + monkeypatch.setattr(killtree_module.os, "getpgid", lambda pid: 99) + monkeypatch.setattr(killtree_module.os, "killpg", killpg) + + calls = {"count": 0} + + async def fake_wait_for(awaitable: Any, timeout: float) -> None: + del timeout + calls["count"] += 1 + if calls["count"] == 1: + awaitable.close() + raise asyncio.TimeoutError + await awaitable + + proc = cast(asyncio.subprocess.Process, _FakeAsyncProcess(pid=12, returncode=None)) + + with patch("agent_framework_tools.shell._killtree.asyncio.wait_for", side_effect=fake_wait_for): + await _kill_via_stdlib(proc, grace=0.01) + + assert killpg.call_args_list[0].args == (99, killtree_module.signal.SIGTERM) + assert killpg.call_args_list[1].args == (99, killtree_module.signal.SIGKILL) diff --git a/python/packages/tools/tests/test_shell_resolve.py b/python/packages/tools/tests/test_shell_resolve.py index 4342000c16..845a982e1b 100644 --- a/python/packages/tools/tests/test_shell_resolve.py +++ b/python/packages/tools/tests/test_shell_resolve.py @@ -3,7 +3,7 @@ import pytest from agent_framework_tools.shell import ShellExecutionError -from agent_framework_tools.shell._resolve import resolve_shell +from agent_framework_tools.shell._resolve import _ensure_command_flag, is_powershell, resolve_shell def test_empty_string_shell_override_rejected() -> None: @@ -21,6 +21,52 @@ def test_empty_sequence_shell_override_rejected() -> None: resolve_shell([], interactive=True) +def test_resolve_shell_prefers_environment_override(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("AGENT_FRAMEWORK_SHELL", "/custom/pwsh -NoProfile") + + assert resolve_shell(None, interactive=False) == ["/custom/pwsh", "-NoProfile", "-Command"] + + +def test_resolve_shell_windows_defaults_and_missing_binary(monkeypatch: pytest.MonkeyPatch) -> None: + import agent_framework_tools.shell._resolve as resolve_module + + monkeypatch.setattr(resolve_module.sys, "platform", "win32") + monkeypatch.setattr(resolve_module.shutil, "which", lambda name: "C:/pwsh.exe" if name == "pwsh" else None) + assert resolve_shell(None, interactive=True) == [ + "C:/pwsh.exe", + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + "-", + ] + + monkeypatch.setattr(resolve_module.shutil, "which", lambda name: None) + with pytest.raises(ShellExecutionError, match="Neither 'pwsh' nor 'powershell'"): + resolve_shell(None, interactive=False) + + +def test_resolve_shell_posix_fallbacks(monkeypatch: pytest.MonkeyPatch) -> None: + import agent_framework_tools.shell._resolve as resolve_module + + monkeypatch.setattr(resolve_module.sys, "platform", "darwin") + monkeypatch.setattr(resolve_module.os.path, "exists", lambda candidate: candidate == "/bin/sh") + assert resolve_shell(None, interactive=False) == ["/bin/sh", "-c"] + + monkeypatch.setattr(resolve_module.os.path, "exists", lambda candidate: False) + monkeypatch.setattr(resolve_module.shutil, "which", lambda name: "/usr/local/bin/sh" if name == "sh" else None) + assert resolve_shell(None, interactive=True) == ["/usr/local/bin/sh"] + + monkeypatch.setattr(resolve_module.shutil, "which", lambda name: None) + with pytest.raises(ShellExecutionError, match="No POSIX shell found"): + resolve_shell(None, interactive=False) + + +def test_is_powershell_and_command_flag_helpers() -> None: + assert is_powershell([]) is False + assert _ensure_command_flag([]) == [] + + def test_stateless_appends_dash_c_for_posix_shell_without_flag() -> None: argv = resolve_shell("/bin/bash", interactive=False) assert argv == ["/bin/bash", "-c"] From a1fed0dab5f5dafa6358d410b7484ea24390039f Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Wed, 22 Jul 2026 16:55:11 +0200 Subject: [PATCH 2/3] Fix Python CI and deprecation usage Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9ae5ad8e-6b66-41b3-a862-4e2a3fae1cd0 --- python/.github/skills/python-testing/SKILL.md | 4 ++++ python/DEV_SETUP.md | 6 +++++ .../ag-ui/tests/ag_ui/test_endpoint.py | 2 +- .../agent_framework_chatkit/_converter.py | 12 ++++++---- .../packages/chatkit/tests/test_converter.py | 21 ++++++++++++----- .../agent_framework/_workflows/_workflow.py | 13 +++++------ .../_workflows/_workflow_builder.py | 23 ++++--------------- .../test_output_executors_contract.py | 18 +++++---------- .../tools/tests/test_shell_killtree.py | 10 ++++---- 9 files changed, 57 insertions(+), 52 deletions(-) diff --git a/python/.github/skills/python-testing/SKILL.md b/python/.github/skills/python-testing/SKILL.md index d2b2fc9cae..639cedd33f 100644 --- a/python/.github/skills/python-testing/SKILL.md +++ b/python/.github/skills/python-testing/SKILL.md @@ -13,6 +13,10 @@ aggregate coverage enforcement. Tests should be fast, reliable, and maintainable When adding new code, check that the relevant sections of the codebase are covered by tests, and add new tests as needed. When modifying existing code, update or add tests to cover the changes. We run tests in two stages, for a PR each commit is tested with unit tests only (using `-m "not integration"`), and the full suite including integration tests is run when merging. +When an API is marked as deprecated, migrate ordinary tests to its replacement in the same change. Retain only +focused tests that validate the deprecated behavior and warning; integration tests, samples, and unrelated unit +tests should use the supported API. + ## Running Tests ```bash diff --git a/python/DEV_SETUP.md b/python/DEV_SETUP.md index 74433ac858..c19da35a64 100644 --- a/python/DEV_SETUP.md +++ b/python/DEV_SETUP.md @@ -157,6 +157,12 @@ uv run poe --directory packages/core test Large packages (core, ag-ui, orchestrations, anthropic) use `pytest-xdist` for parallel test execution within the package. The aggregate `test -A` sweep also uses `pytest-xdist` across the selected packages. +### Testing deprecations + +When an API is marked as deprecated, update the test suite to use its replacement at the same time. Keep only +focused tests that validate the deprecated API and its warning; ordinary behavior, integration, and sample tests +should exercise the supported API so deprecation warnings do not accumulate in test runs. + ## Code quality checks To run the same checks that run during a commit and the GitHub Action `Python Code Quality`, you can use this command, from the [python](../python) folder: diff --git a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py index ac5ba41256..eb060c9315 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_endpoint.py +++ b/python/packages/ag-ui/tests/ag_ui/test_endpoint.py @@ -3151,7 +3151,7 @@ async def test_endpoint_error_handling(build_chat_client): client = TestClient(app) # Send invalid JSON to trigger parsing error before streaming - response = client.post("/failing", data=b"invalid json", headers={"content-type": "application/json"}) # type: ignore + response = client.post("/failing", content=b"invalid json", headers={"content-type": "application/json"}) # Pydantic validation now returns 422 for invalid request body assert response.status_code == 422 diff --git a/python/packages/chatkit/agent_framework_chatkit/_converter.py b/python/packages/chatkit/agent_framework_chatkit/_converter.py index dc322377d3..c97b847944 100644 --- a/python/packages/chatkit/agent_framework_chatkit/_converter.py +++ b/python/packages/chatkit/agent_framework_chatkit/_converter.py @@ -5,7 +5,7 @@ from __future__ import annotations import logging -from collections.abc import Awaitable, Callable, Sequence +from collections.abc import Awaitable, Callable, Mapping, Sequence from agent_framework import ( Content, @@ -240,7 +240,7 @@ def tag_to_message_content(self, tag: UserMessageTagContent) -> Content: content = converter.tag_to_message_content(tag) # Returns: Content.from_text(text="Name:John Doe") """ - name = getattr(tag.data, "name", tag.text if hasattr(tag, "text") else "unknown") + name = tag.data.get("name", tag.text) if isinstance(tag.data, Mapping) else getattr(tag.data, "name", tag.text) return Content.from_text(text=f"Name:{name}") def task_to_input(self, item: TaskItem) -> Message | list[Message] | None: @@ -370,13 +370,17 @@ def widget_to_input(self, item: WidgetItem) -> Message | list[Message] | None: .. code-block:: python # Widget item - from chatkit.widgets import Card, Text + from chatkit.widgets import WidgetTemplate widget_item = WidgetItem( id="widget_1", thread_id="thread_1", created_at=datetime.now(), - widget=Card(children=[Text(value="Hello")]), + widget=WidgetTemplate({ + "version": "1.0", + "name": "greeting", + "template": '{"type":"Card","children":[{"type":"Text","value":"Hello"}]}', + }).build(), ) message = converter.widget_to_input(widget_item) # Returns message with JSON representation of the widget diff --git a/python/packages/chatkit/tests/test_converter.py b/python/packages/chatkit/tests/test_converter.py index 3d334cba7c..730fd6ec6e 100644 --- a/python/packages/chatkit/tests/test_converter.py +++ b/python/packages/chatkit/tests/test_converter.py @@ -156,8 +156,7 @@ def test_tag_to_message_content(self, converter): result = converter.tag_to_message_content(tag) assert result.type == "text" - # Since data is a dict, getattr won't work, so it will fall back to text - assert result.text == "Name:john" + assert result.text == "Name:John Doe" def test_tag_to_message_content_no_name(self, converter): """Test converting tag with no name to message content.""" @@ -440,14 +439,18 @@ def test_workflow_to_input_empty(self, converter): def test_widget_to_input(self, converter): """Test converting WidgetItem to Message.""" from chatkit.types import WidgetItem - from chatkit.widgets import Card, Text # ty: ignore[deprecated] + from chatkit.widgets import WidgetTemplate widget_item = WidgetItem( id="widget_1", thread_id="thread_1", created_at=datetime.now(), type="widget", - widget=Card(key="card1", children=[Text(value="Hello")]), # ty: ignore[deprecated] + widget=WidgetTemplate({ + "version": "1.0", + "name": "greeting", + "template": '{"type":"Card","key":"card1","children":[{"type":"Text","value":"Hello"}]}', + }).build(), ) result = converter.widget_to_input(widget_item) @@ -563,7 +566,7 @@ async def test_to_agent_input_dispatches_supported_variants(self, converter): Workflow, WorkflowItem, ) - from chatkit.widgets import Card, Text # ty: ignore[deprecated] + from chatkit.widgets import WidgetTemplate thread_items = [ AssistantMessageItem( @@ -590,7 +593,13 @@ async def test_to_agent_input_dispatches_supported_variants(self, converter): thread_id="thread_1", created_at=datetime.now(), type="widget", - widget=Card(key="card_dispatch", children=[Text(value="Dispatch")]), # ty: ignore[deprecated] + widget=WidgetTemplate({ + "version": "1.0", + "name": "dispatch", + "template": ( + '{"type":"Card","key":"card_dispatch","children":[{"type":"Text","value":"Dispatch"}]}' + ), + }).build(), ), WorkflowItem( id="workflow_dispatch", diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 753ef77c33..a3733e7ffd 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -173,7 +173,7 @@ def status_timeline(self) -> list[WorkflowEvent[Any]]: class OutputDesignation: """Immutable rule for labeling executor yields as terminal, intermediate, or hidden outputs. - ``outputs`` is ``None`` in omitted-selection compatibility mode (every yield is terminal). In explicit mode, + ``outputs`` is ``None`` in the default all-output mode (every yield is terminal). In explicit mode, ``outputs`` and ``intermediates`` are disjoint executor ID sets; unlisted executor yields are hidden from caller-facing output/intermediate events. Package-internal value type owned by ``Workflow``; not exported from ``agent_framework``. @@ -305,9 +305,8 @@ def __init__( better observability and management. description: Optional description of what the workflow does. If the workflow is built using WorkflowBuilder, this will be the description of the builder. - output_from: List of executor IDs designated as workflow outputs, or - ``None`` for omitted-selection compatibility behavior when ``intermediate_output_from`` is also - ``None``. + output_from: List of executor IDs designated as workflow outputs, or ``None`` for the default + all-output behavior when ``intermediate_output_from`` is also ``None``. intermediate_output_from: List of executor IDs designated as intermediate outputs. In explicit designation mode, unlisted executor yields are hidden from caller-facing output/intermediate events. @@ -334,7 +333,7 @@ def __init__( self.graph_signature = self._compute_graph_signature() self.graph_signature_hash = self._hash_graph_signature(self.graph_signature) - # Single value type encodes omitted-selection compatibility vs explicit output-designation policy. + # Single value type encodes default all-output vs explicit output-designation policy. output_designation_ids = ( frozenset(output_from) if output_from is not None @@ -433,8 +432,8 @@ def get_start_executor(self) -> Executor: def get_output_executors(self) -> list[Executor]: """Get the list of output executors in the workflow. - In omitted-selection compatibility mode (no explicit ``output_from``), returns every - executor in the workflow. In explicit mode, returns only the designated output executors. + In the default all-output mode, returns every executor in the workflow. In explicit mode, + returns only the designated output executors. """ designated = self._output_designation.outputs if designated is None: diff --git a/python/packages/core/agent_framework/_workflows/_workflow_builder.py b/python/packages/core/agent_framework/_workflows/_workflow_builder.py index 83d53a988c..e092710913 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_builder.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_builder.py @@ -3,7 +3,6 @@ import logging import sys import uuid -import warnings from collections.abc import Callable, Sequence from typing import Any, Literal @@ -120,14 +119,13 @@ def __init__( Pass ``"all_other"`` to select every executor with declared workflow output types that is not selected by ``output_from``. If neither ``output_from`` nor ``intermediate_output_from`` is provided, - omitted-selection compatibility behavior applies and every ``yield_output`` produces - ``type='output'``. If either is provided, explicit mode applies: listed + every ``yield_output`` produces ``type='output'``. If either is provided, + explicit mode applies: listed workflow-output executors emit ``output``, listed intermediate executors emit ``intermediate``, and unlisted executor yields are hidden. Output selection behavior: - - Omit both selections: every ``yield_output`` emits ``output`` for compatibility, - with a deprecation warning. + - Omit both selections: every ``yield_output`` emits ``output``. - ``output_from="all"``: every output-capable executor emits ``output``. - ``output_from=[A]``: only A emits ``output``; other executor payloads are hidden. - ``output_from=[A], intermediate_output_from="all_other"``: A emits ``output``; @@ -156,8 +154,7 @@ def __init__( # being created for the same agent. self._agent_wrappers: dict[str, Executor] = {} - # ``None`` for both means omitted-selection compatibility behavior - # (every yield_output produces type='output'). + # ``None`` for both means the default all-output behavior. # If either is provided, explicit mode applies and unlisted executor yields are hidden. self._output_from: _OutputSelection = self._coerce_output_from(output_from) self._intermediate_output_from: _IntermediateOutputSelection = self._coerce_intermediate_output_from( @@ -794,7 +791,7 @@ async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None: print(events.get_outputs()) # [] print(events.get_intermediate_outputs()) # outputs from planner and answerer - # Explicitly preserve all-output behavior without relying on omitted-selection compatibility. + # Explicitly select all output-capable executors. workflow = ( WorkflowBuilder(start_executor=planner, output_from="all").add_edge(planner, answerer).build() ) @@ -812,16 +809,6 @@ async def process(self, text: str, ctx: WorkflowContext[Never, str]) -> None: "Starting executor must be set via the start_executor constructor parameter before building." ) - if self._output_from is None and self._intermediate_output_from is None: - warnings.warn( - "WorkflowBuilder built without explicit output_from or intermediate_output_from; " - "every yield_output produces type='output' for compatibility. Pass output_from='all', " - "output_from=[...], or intermediate_output_from=[...] to opt into explicit designation - " - "explicit designation will be required in a future version.", - DeprecationWarning, - stacklevel=2, - ) - start_executor = self._start_executor executors = self._executors edge_groups = self._edge_groups diff --git a/python/packages/core/tests/workflow/test_output_executors_contract.py b/python/packages/core/tests/workflow/test_output_executors_contract.py index 74df7d00cd..48fd730e98 100644 --- a/python/packages/core/tests/workflow/test_output_executors_contract.py +++ b/python/packages/core/tests/workflow/test_output_executors_contract.py @@ -35,23 +35,17 @@ async def _downstream(message: str, ctx: WorkflowContext[Never, str]) -> None: await ctx.yield_output("from-downstream") -def test_designation_unset_emits_deprecation_warning() -> None: - """State A: WorkflowBuilder built without explicit designation warns.""" - with pytest.warns(DeprecationWarning, match="output_from or intermediate_output_from") as warning_info: +def test_designation_unset_does_not_warn() -> None: + """Omitted designation is the supported all-output default.""" + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) WorkflowBuilder(start_executor=_emit_one).build() - assert str(warning_info[0].message) == ( - "WorkflowBuilder built without explicit output_from or intermediate_output_from; " - "every yield_output produces type='output' for compatibility. Pass output_from='all', " - "output_from=[...], or intermediate_output_from=[...] to opt into explicit designation - " - "explicit designation will be required in a future version." - ) @pytest.mark.asyncio async def test_designation_unset_preserves_compatibility_all_output_behavior() -> None: - """Omitted designation keeps compatibility all-output behavior while warning.""" - with pytest.warns(DeprecationWarning, match="output_from or intermediate_output_from"): - workflow = WorkflowBuilder(start_executor=_start).add_edge(_start, _downstream).build() + """Omitted designation emits all workflow outputs.""" + workflow = WorkflowBuilder(start_executor=_start).add_edge(_start, _downstream).build() result = await workflow.run([Message(role="user", contents=["hi"])]) diff --git a/python/packages/tools/tests/test_shell_killtree.py b/python/packages/tools/tests/test_shell_killtree.py index 1052902670..c18062d5c9 100644 --- a/python/packages/tools/tests/test_shell_killtree.py +++ b/python/packages/tools/tests/test_shell_killtree.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. import asyncio +import os from typing import Any, cast from unittest.mock import AsyncMock, MagicMock, patch @@ -52,8 +53,9 @@ def test_resolve_taskkill_uses_systemroot_and_caches(monkeypatch) -> None: monkeypatch.setenv("SystemRoot", "C:\\Windows") monkeypatch.setattr(killtree_module.os.path, "isfile", lambda path: path.endswith("taskkill.exe")) - assert _resolve_taskkill() == "C:\\Windows/System32/taskkill.exe" - assert _resolve_taskkill() == "C:\\Windows/System32/taskkill.exe" + expected_path = os.path.join("C:\\Windows", "System32", "taskkill.exe") + assert _resolve_taskkill() == expected_path + assert _resolve_taskkill() == expected_path async def test_kill_process_tree_short_circuits_or_delegates() -> None: @@ -125,8 +127,8 @@ async def test_kill_via_stdlib_posix_escalates_to_sigkill(monkeypatch) -> None: monkeypatch.setattr(killtree_module.sys, "platform", "darwin") killpg = MagicMock() - monkeypatch.setattr(killtree_module.os, "getpgid", lambda pid: 99) - monkeypatch.setattr(killtree_module.os, "killpg", killpg) + monkeypatch.setattr(killtree_module.os, "getpgid", lambda pid: 99, raising=False) + monkeypatch.setattr(killtree_module.os, "killpg", killpg, raising=False) calls = {"count": 0} From bab7b194ff98556967d4eebe990085b575728a33 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Wed, 22 Jul 2026 17:04:36 +0200 Subject: [PATCH 3/3] Make POSIX kill-tree test portable Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9ae5ad8e-6b66-41b3-a862-4e2a3fae1cd0 --- python/packages/tools/tests/test_shell_killtree.py | 1 + 1 file changed, 1 insertion(+) diff --git a/python/packages/tools/tests/test_shell_killtree.py b/python/packages/tools/tests/test_shell_killtree.py index c18062d5c9..304b85ffe4 100644 --- a/python/packages/tools/tests/test_shell_killtree.py +++ b/python/packages/tools/tests/test_shell_killtree.py @@ -129,6 +129,7 @@ async def test_kill_via_stdlib_posix_escalates_to_sigkill(monkeypatch) -> None: killpg = MagicMock() monkeypatch.setattr(killtree_module.os, "getpgid", lambda pid: 99, raising=False) monkeypatch.setattr(killtree_module.os, "killpg", killpg, raising=False) + monkeypatch.setattr(killtree_module.signal, "SIGKILL", 9, raising=False) calls = {"count": 0}