# Copyright (c) Microsoft. All rights reserved.

"""

Sample Validation Script

Validates all Python samples in the samples directory using a workflow that:

1. Discovers all sample files

2. Builds a nested concurrent workflow with one GitHub agent per sample

3. Runs the nested workflow

4. Generates a validation report

Usage:

uv run python -m sample_validation

uv run python -m sample_validation --subdir 03-workflows

uv run python -m sample_validation --output-dir ./reports

"""

import argparse

import asyncio

import os

import sys

import time

from pathlib import Path

# Add the samples directory to the path for imports

sys.path.insert(0, str(Path(__file__).parent.parent))

from sample_validation.models import Report

from sample_validation.report import save_report

from sample_validation.workflow import ValidationConfig, create_validation_workflow

def parse_arguments() -> argparse.Namespace:

"""Parse command line arguments."""

parser = argparse.ArgumentParser(

description="Validate Python samples using a dynamic nested concurrent workflow",

formatter_class=argparse.RawDescriptionHelpFormatter,

epilog="""

Examples:

uv run python -m sample_validation # Validate all samples

uv run python -m sample_validation --subdir 03-workflows # Validate only workflows

uv run python -m sample_validation --output-dir ./reports # Save reports to custom dir

""",

)

parser.add_argument(

"--subdir",

type=str,

help="Validate samples only in the specified subdirectory (relative to samples/)",

)

parser.add_argument(

"--output-dir",

type=str,

default="./sample_validation/reports",

help="Directory to save validation reports (default: ./sample_validation/reports)",

)

parser.add_argument(

"--save-report",

action="store_true",

help="Save the validation report to files",

)

parser.add_argument(

"--max-parallel-workers",

type=int,

default=10,

help="Maximum number of samples to run in parallel per batch (default: 10)",

)

parser.add_argument(

"--report-name",

type=str,

help="Custom name for the report files (without extension). If not provided, uses timestamp.",

)

parser.add_argument(

"--exclude",

nargs="+",

type=str,

help="Subdirectory paths to exclude (relative to the search directory set by --subdir)",

)

return parser.parse_args()

async def main() -> int:

"""Main entry point."""

args = parse_arguments()

# Determine paths

# Script is at python/scripts/sample_validation/__main__.py

# python_root is python/, samples_dir is python/samples/

python_root = Path(__file__).parent.parent.parent

samples_dir = python_root / "samples"

print("=" * 80)

print("SAMPLE VALIDATION WORKFLOW")

print("=" * 80)

print(f"Samples directory: {samples_dir}")

print(f"Python root: {python_root}")

if os.environ.get("GITHUB_COPILOT_MODEL"):

print(

f"Using GitHub Copilot model override: {os.environ['GITHUB_COPILOT_MODEL']}"

)

# Create validation config

config = ValidationConfig(

samples_dir=samples_dir,

python_root=python_root,

subdir=args.subdir,

exclude=args.exclude,

max_parallel_workers=max(1, args.max_parallel_workers),

)

# Create and run the workflow

workflow = create_validation_workflow(config)

print("\nStarting validation workflow...")

print("-" * 80)

# Run the workflow

run_start = time.perf_counter()

try:

events = await workflow.run("start")

finally:

run_duration = time.perf_counter() - run_start

print(f"\nWorkflow run completed in {run_duration:.2f}s")

outputs = events.get_outputs()

if not outputs:

print("\n[ERROR] Workflow did not produce any output")

return 1

report: Report = outputs[0]

# Save report if requested

if args.save_report:

output_dir = samples_dir / args.output_dir

md_path, json_path = save_report(report, output_dir, name=args.report_name)

print("\nReports saved:")

print(f" Markdown: {md_path}")

print(f" JSON: {json_path}")

# Return appropriate exit code

failed = report.failure_count + report.missing_setup_count

return 1 if failed > 0 else 0

if __name__ == "__main__":

exit_code = asyncio.run(main())

sys.exit(exit_code)