run_tests.py

#!/usr/bin/env python3
"""
run_tests.py - Run PVS verification tests for all examples
"""

import os
import sys
import subprocess
import shutil
import argparse
import yaml
from pathlib import Path
from typing import List, Tuple, Dict, Optional

# Colors for terminal output
class Colors:
    GREEN = '\033[0;32m'
    RED = '\033[0;31m'
    YELLOW = '\033[1;33m'
    BLUE = '\033[0;34m'
    NC = '\033[0m'  # No Color

def log(msg: str, color: str = ''):
    """Print colored message."""
    if color:
        print(f"{color}{msg}{Colors.NC}")
    else:
        print(msg)

def clean_example_caches(examples_dir: Path):
    """Clean PVS cache files from all example directories."""
    log("Cleaning PVS cache files from examples...", Colors.YELLOW)
    
    cleaned_count = 0
    for example_dir in examples_dir.iterdir():
        if not example_dir.is_dir():
            continue
        
        # Remove .pvscontext files
        pvscontext = example_dir / ".pvscontext"
        if pvscontext.exists():
            pvscontext.unlink()
            cleaned_count += 1
        
        # Remove pvsbin directories
        pvsbin = example_dir / "pvsbin"
        if pvsbin.exists() and pvsbin.is_dir():
            shutil.rmtree(pvsbin)
            cleaned_count += 1
        
        # Remove test_output directories
        test_output = example_dir / "test_output"
        if test_output.exists() and test_output.is_dir():
            shutil.rmtree(test_output)
        
        # Remove output directories
        output = example_dir / "output"
        if output.exists() and output.is_dir():
            shutil.rmtree(output)
        
        # Remove auto-generated ADT files
        for adt_file in example_dir.glob("*_adt.pvs"):
            adt_file.unlink()
            cleaned_count += 1
        for adt_reduce_file in example_dir.glob("*_adt_reduce.pvs"):
            adt_reduce_file.unlink()
            cleaned_count += 1
    
    log(f"Cleaned {cleaned_count} cache files/directories", Colors.GREEN)
    print()

def clean_test_cache(example_dir: Path):
    """Clean PVS cache files from a single test directory."""
    pvscontext = example_dir / ".pvscontext"
    if pvscontext.exists():
        pvscontext.unlink()
    
    pvsbin = example_dir / "pvsbin"
    if pvsbin.exists() and pvsbin.is_dir():
        shutil.rmtree(pvsbin)

def run_test(name: str, example_dir: Path, transform: Path, constraints: Path, 
             input_schema: Path, output_schema: Path, pvs_path: Path) -> bool:
    """Run a single verification test."""
    
    log(f"Testing: {name}", Colors.YELLOW)
    print(f"  Directory: {example_dir}")
    
    output_dir = example_dir / "test_output"
    
    # Build command
    cmd = [
        sys.executable,
        "pvs_main.py",
        "verify",
        "--transform", str(transform),
        "--constraints", str(constraints),
        "--input-schema", str(input_schema),
        "--output-schema", str(output_schema),
        "--pvs", str(pvs_path),
        "--output", str(output_dir),
        "--instance-name", f"{name}_test"
    ]
    
    try:
        # Run verification (suppress output)
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=300  # 5 minute timeout
        )
        
        if result.returncode == 0:
            log("  ✓ PASSED", Colors.GREEN)
            # Clean up test output
            if output_dir.exists():
                shutil.rmtree(output_dir)
            # Clean up cache files from example directory
            clean_test_cache(example_dir)
            return True
        else:
            log("  ✗ FAILED", Colors.RED)
            if result.stderr:
                print(f"  Error: {result.stderr[:200]}")
            return False
            
    except subprocess.TimeoutExpired:
        log("  ✗ TIMEOUT", Colors.RED)
        return False
    except Exception as e:
        log(f"  ✗ ERROR: {e}", Colors.RED)
        return False
    finally:
        print()

def load_test_config(example_dir: Path) -> Optional[Dict]:
    """Load test configuration from YAML file."""
    config_file = example_dir / "test_config.yaml"
    if not config_file.exists():
        return None
    
    try:
        with open(config_file, 'r') as f:
            config = yaml.safe_load(f)
        
        # Validate required fields
        required = ['input_schema', 'output_schema', 'transform', 'constraints']
        if not all(field in config for field in required):
            return None
        
        return config
    except Exception as e:
        return None

def discover_tests(examples_dir: Path) -> List[Tuple[str, Path, Path, Path, Path, Path]]:
    """Discover all test cases in examples directory."""
    tests = []
    
    for example_subdir in sorted(examples_dir.iterdir()):
        if not example_subdir.is_dir():
            continue
        
        # Skip ex1 (JSLT-based, not for pvs_main)
        if example_subdir.name == 'ex1':
            continue
        
        # Load test configuration
        config = load_test_config(example_subdir)
        if not config:
            continue
        
        # Build paths from config
        transform = example_subdir / config['transform']
        constraints = example_subdir / config['constraints']
        input_schema = example_subdir / config['input_schema']
        output_schema = example_subdir / config['output_schema']
        
        # Verify all files exist
        if all(f.exists() for f in [transform, constraints, input_schema, output_schema]):
            tests.append((
                example_subdir.name,
                example_subdir,
                transform,
                constraints,
                input_schema,
                output_schema
            ))
    
    return tests

def main():
    """Main test runner."""
    
    # Parse arguments
    parser = argparse.ArgumentParser(description="Run PVS verification tests")
    parser.add_argument('--clean', action='store_true', 
                       help='Clean PVS cache files before running tests')
    args = parser.parse_args()
    
    # Configuration
    script_dir = Path(__file__).parent.resolve()
    examples_dir = script_dir / "examples"
    
    # Detect Docker environment
    if os.environ.get("FHIRFLY_DOCKER") == "true":
        default_pvs = "/home/fhirfly/PVS/pvs"
    else:
        default_pvs = os.path.expanduser("~/git/PVS/pvs")
    
    pvs_path = Path(os.environ.get("PVS_PATH", default_pvs))
    
    # Print header
    log("=" * 50, Colors.BLUE)
    log("FHIR-Fly PVS Verification Test Suite", Colors.BLUE)
    log("=" * 50, Colors.BLUE)
    print(f"PVS Path: {pvs_path}")
    print()
    
    # Clean caches if requested
    if args.clean:
        clean_example_caches(examples_dir)
    
    # Check PVS exists
    if not pvs_path.exists():
        log(f"Error: PVS not found at {pvs_path}", Colors.RED)
        log("Set PVS_PATH environment variable to the correct path", Colors.RED)
        return 1
    
    # Discover tests
    tests = discover_tests(examples_dir)
    
    if not tests:
        log("No tests found!", Colors.YELLOW)
        return 1
    
    log(f"Found {len(tests)} test(s)", Colors.BLUE)
    print()
    
    # Run tests
    results = []
    for name, ex_dir, transform, constraints, in_schema, out_schema in tests:
        passed = run_test(name, ex_dir, transform, constraints, in_schema, out_schema, pvs_path)
        results.append((name, passed))
    
    # Print summary
    log("=" * 50, Colors.BLUE)
    log("Test Summary", Colors.BLUE)
    log("=" * 50, Colors.BLUE)
    
    total = len(results)
    passed = sum(1 for _, p in results if p)
    failed = total - passed
    
    print(f"Total tests: {total}")
    log(f"Passed: {passed}", Colors.GREEN)
    log(f"Failed: {failed}", Colors.RED if failed > 0 else Colors.GREEN)
    log("=" * 50, Colors.BLUE)
    
    if failed == 0:
        log("All tests passed!", Colors.GREEN)
        return 0
    else:
        log("Some tests failed!", Colors.RED)
        print("\nFailed tests:")
        for name, passed in results:
            if not passed:
                log(f"  - {name}", Colors.RED)
        return 1

if __name__ == '__main__':
    os.chdir(Path(__file__).parent)
    sys.exit(main())