validate.py

#!/usr/bin/env python3
"""
OpenEnv Validation Script
Validates that the environment meets all OpenEnv spec requirements.
Run this before submitting to ensure compliance.
"""

import sys
import os
import json
import yaml
from typing import Dict, Any, List, Tuple

# Add current directory to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))


def validate_openenv_yaml() -> Tuple[bool, List[str]]:
    """Validate openenv.yaml structure and content."""
    errors = []
    
    try:
        with open("openenv.yaml", "r") as f:
            config = yaml.safe_load(f)
    except FileNotFoundError:
        return False, ["openenv.yaml not found"]
    except yaml.YAMLError as e:
        return False, [f"Invalid YAML: {e}"]
    
    # Required top-level fields
    required_fields = ["name", "version", "description", "environment", "api", 
                       "action_schema", "observation_schema", "reward_schema", "tasks"]
    
    for field in required_fields:
        if field not in config:
            errors.append(f"Missing required field: {field}")
    
    # Validate tasks
    if "tasks" in config:
        tasks = config["tasks"]
        if len(tasks) < 3:
            errors.append(f"Need at least 3 tasks, found {len(tasks)}")
        
        difficulties = set()
        for task in tasks:
            if "id" not in task:
                errors.append("Task missing 'id' field")
            if "difficulty" not in task:
                errors.append("Task missing 'difficulty' field")
            else:
                difficulties.add(task["difficulty"])
        
        if "easy" not in difficulties:
            errors.append("Missing 'easy' difficulty task")
        if "medium" not in difficulties:
            errors.append("Missing 'medium' difficulty task")
        if "hard" not in difficulties:
            errors.append("Missing 'hard' difficulty task")
    
    # Validate API section
    if "api" in config:
        api = config["api"]
        required_methods = ["reset", "step", "state"]
        for method in required_methods:
            if method not in api:
                errors.append(f"Missing API method: {method}")
    
    return len(errors) == 0, errors


def validate_pydantic_models() -> Tuple[bool, List[str]]:
    """Validate that Pydantic models are properly defined."""
    errors = []
    
    try:
        from env.models import Observation, Action, Reward, State
        
        # Check Observation
        obs_fields = Observation.model_fields
        required_obs = ["dataset", "step_count", "task_difficulty"]
        for field in required_obs:
            if field not in obs_fields:
                errors.append(f"Observation missing field: {field}")
        
        # Check Action
        action_fields = Action.model_fields
        if "action_type" not in action_fields:
            errors.append("Action missing 'action_type' field")
        
        # Check Reward
        reward_fields = Reward.model_fields
        if "total" not in reward_fields:
            errors.append("Reward missing 'total' field")
        
        # Check State
        state_fields = State.model_fields
        required_state = ["task_id", "current_dataset", "step_count"]
        for field in required_state:
            if field not in state_fields:
                errors.append(f"State missing field: {field}")
        
    except ImportError as e:
        errors.append(f"Failed to import models: {e}")
    except Exception as e:
        errors.append(f"Error validating models: {e}")
    
    return len(errors) == 0, errors


def validate_environment_api() -> Tuple[bool, List[str]]:
    """Validate that environment implements required API methods."""
    errors = []
    
    try:
        from env.environment import DataCleaningEnvironment
        
        # Check methods exist
        env = DataCleaningEnvironment(task="easy", seed=42)
        
        if not hasattr(env, "reset"):
            errors.append("Environment missing 'reset' method")
        if not hasattr(env, "step"):
            errors.append("Environment missing 'step' method")
        if not hasattr(env, "state"):
            errors.append("Environment missing 'state' method")
        
        # Test reset
        obs = env.reset()
        if obs is None:
            errors.append("reset() returned None")
        
        # Test step
        result = env.step({"action_type": "remove_duplicates"})
        if result is None:
            errors.append("step() returned None")
        if not hasattr(result, "observation"):
            errors.append("step() result missing 'observation'")
        if not hasattr(result, "reward"):
            errors.append("step() result missing 'reward'")
        if not hasattr(result, "done"):
            errors.append("step() result missing 'done'")
        
        # Test state
        state = env.state()
        if state is None:
            errors.append("state() returned None")
        
    except Exception as e:
        errors.append(f"Error validating environment: {e}")
    
    return len(errors) == 0, errors


def validate_grader() -> Tuple[bool, List[str]]:
    """Validate that grader produces scores in [0.0, 1.0]."""
    errors = []
    
    try:
        from env.environment import DataCleaningEnvironment
        
        for task in ["easy", "medium", "hard"]:
            env = DataCleaningEnvironment(task=task, seed=42)
            env.reset()
            
            # Run a few actions
            env.step({"action_type": "remove_duplicates"})
            env.step({"action_type": "finish"})
            
            score = env.get_final_score()
            
            if not isinstance(score, (int, float)):
                errors.append(f"Grader returned non-numeric score for {task}: {type(score)}")
            elif score < 0.0 or score > 1.0:
                errors.append(f"Grader score out of range for {task}: {score}")
        
        # Test determinism
        env1 = DataCleaningEnvironment(task="easy", seed=42)
        env1.reset()
        env1.step({"action_type": "remove_duplicates"})
        score1 = env1.get_final_score()
        
        env2 = DataCleaningEnvironment(task="easy", seed=42)
        env2.reset()
        env2.step({"action_type": "remove_duplicates"})
        score2 = env2.get_final_score()
        
        if score1 != score2:
            errors.append(f"Grader not deterministic: {score1} != {score2}")
        
    except Exception as e:
        errors.append(f"Error validating grader: {e}")
    
    return len(errors) == 0, errors


def validate_inference_script() -> Tuple[bool, List[str]]:
    """Validate that inference.py exists and has correct structure."""
    errors = []
    
    if not os.path.exists("inference.py"):
        errors.append("inference.py not found in root directory")
        return False, errors
    
    try:
        with open("inference.py", "r") as f:
            content = f.read()
        
        # Check for required environment variables
        if "OPENAI_API_KEY" not in content:
            errors.append("inference.py should read OPENAI_API_KEY")
        if "API_BASE_URL" not in content:
            errors.append("inference.py should read API_BASE_URL")
        if "MODEL_NAME" not in content:
            errors.append("inference.py should read MODEL_NAME")
        
        # Check for OpenAI client usage
        if "OpenAI" not in content and "openai" not in content:
            errors.append("inference.py should use OpenAI client")
        
        # Check that it doesn't have hardcoded API keys
        if "gsk_" in content:
            errors.append("WARNING: Possible hardcoded Groq API key found!")
        if "sk-" in content and "OPENAI_API_KEY" not in content:
            errors.append("WARNING: Possible hardcoded OpenAI API key found!")
        
    except Exception as e:
        errors.append(f"Error reading inference.py: {e}")
    
    return len(errors) == 0, errors


def validate_dockerfile() -> Tuple[bool, List[str]]:
    """Validate that Dockerfile exists and has correct structure."""
    errors = []
    
    if not os.path.exists("Dockerfile"):
        errors.append("Dockerfile not found")
        return False, errors
    
    try:
        with open("Dockerfile", "r") as f:
            content = f.read()
        
        if "FROM" not in content:
            errors.append("Dockerfile missing FROM instruction")
        if "python" not in content.lower():
            errors.append("Dockerfile should use Python base image")
        if "requirements.txt" not in content:
            errors.append("Dockerfile should install requirements.txt")
        
    except Exception as e:
        errors.append(f"Error reading Dockerfile: {e}")
    
    return len(errors) == 0, errors


def validate_readme() -> Tuple[bool, List[str]]:
    """Validate that README.md exists and has required sections."""
    errors = []
    
    if not os.path.exists("README.md"):
        errors.append("README.md not found")
        return False, errors
    
    try:
        with open("README.md", "r") as f:
            content = f.read().lower()
        
        required_sections = [
            ("action", "action space"),
            ("observation", "observation space"),
            ("reward", "reward"),
            ("task", "tasks"),
            ("setup", "setup or installation"),
        ]
        
        for keyword, description in required_sections:
            if keyword not in content:
                errors.append(f"README should describe {description}")
        
    except Exception as e:
        errors.append(f"Error reading README.md: {e}")
    
    return len(errors) == 0, errors


def main():
    """Run all validations."""
    print("=" * 60)
    print("OpenEnv Validation")
    print("=" * 60)
    
    validations = [
        ("openenv.yaml", validate_openenv_yaml),
        ("Pydantic Models", validate_pydantic_models),
        ("Environment API", validate_environment_api),
        ("Grader", validate_grader),
        ("inference.py", validate_inference_script),
        ("Dockerfile", validate_dockerfile),
        ("README.md", validate_readme),
    ]
    
    all_passed = True
    
    for name, validator in validations:
        print(f"\nValidating {name}...")
        passed, errors = validator()
        
        if passed:
            print(f"  [PASS] {name}")
        else:
            print(f"  [FAIL] {name}")
            for error in errors:
                print(f"    - {error}")
            all_passed = False
    
    print("\n" + "=" * 60)
    if all_passed:
        print("ALL VALIDATIONS PASSED")
        print("Your environment is ready for submission!")
    else:
        print("SOME VALIDATIONS FAILED")
        print("Please fix the errors above before submitting.")
    print("=" * 60)
    
    return 0 if all_passed else 1


if __name__ == "__main__":
    sys.exit(main())