code_execution_engine.py

#!/usr/bin/env python3
"""
Code Execution Engine for API Research Study
Safely executes generated code and tests against real APIs
"""

import asyncio
import subprocess
import tempfile
import os
import time
import re
from typing import Dict, Tuple, Optional
import logging

logger = logging.getLogger(__name__)

class CodeExecutionEngine:
    """Safely executes generated code and tests API integration"""
    
    def __init__(self):
        self.env_vars = self.load_env_variables()
        
    def load_env_variables(self) -> Dict[str, str]:
        """Load environment variables for API authentication"""
        env_vars = {}
        
        try:
            with open('.env', 'r') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#') and '=' in line:
                        key, value = line.split('=', 1)
                        env_vars[key.strip()] = value.strip()
        except Exception as e:
            logger.error(f"❌ Failed to load environment variables: {e}")
            
        return env_vars
        
    async def execute_code(self, api_name: str, generated_code: str, env_key: str) -> Tuple[str, int, Optional[int], str, Dict]:
        """
        Execute generated code and test API integration
        
        Returns:
            Tuple of (execution_status, execution_time_ms, api_response_code, error_message, quality_metrics)
        """
        
        start_time = time.time()
        
        try:
            # Analyze code quality first
            quality_metrics = self.analyze_code_quality(generated_code)
            
            # Check for basic syntax issues
            if not self.has_basic_syntax(generated_code):
                execution_time = int((time.time() - start_time) * 1000)
                return "syntax_error", execution_time, None, "Code has basic syntax issues", quality_metrics
                
            # Check if code attempts API integration
            if not self.attempts_api_call(generated_code):
                execution_time = int((time.time() - start_time) * 1000)
                return "logic_error", execution_time, None, "Code doesn't attempt API call", quality_metrics
                
            # Execute code safely
            success, response_code, error_msg = await self.safe_execute(api_name, generated_code, env_key)
            
            execution_time = int((time.time() - start_time) * 1000)
            
            if success:
                if response_code and 200 <= response_code < 300:
                    return "full_success", execution_time, response_code, "", quality_metrics
                else:
                    return "partial_success", execution_time, response_code, error_msg, quality_metrics
            else:
                if "syntax" in error_msg.lower() or "indentation" in error_msg.lower():
                    return "syntax_error", execution_time, None, error_msg, quality_metrics
                else:
                    return "api_error", execution_time, response_code, error_msg, quality_metrics
                    
        except Exception as e:
            execution_time = int((time.time() - start_time) * 1000)
            logger.error(f"❌ Code execution failed for {api_name}: {e}")
            return "execution_error", execution_time, None, str(e), {}
            
    def analyze_code_quality(self, code: str) -> Dict:
        """Analyze code quality metrics"""
        
        metrics = {
            'has_error_handling': False,
            'has_authentication': False,
            'has_proper_imports': False,
            'complexity_score': 0
        }
        
        code_lower = code.lower()
        
        # Check for error handling
        error_patterns = ['try:', 'except:', 'raise', 'error', 'exception']
        metrics['has_error_handling'] = any(pattern in code_lower for pattern in error_patterns)
        
        # Check for authentication
        auth_patterns = ['api_key', 'token', 'auth', 'header', 'bearer', 'key=']
        metrics['has_authentication'] = any(pattern in code_lower for pattern in auth_patterns)
        
        # Check for proper imports
        import_patterns = ['import requests', 'import os', 'import json']
        metrics['has_proper_imports'] = any(pattern in code_lower for pattern in import_patterns)
        
        # Calculate complexity score (simple heuristic)
        lines = code.split('\n')
        non_empty_lines = [line for line in lines if line.strip()]
        functions = len([line for line in lines if 'def ' in line])
        classes = len([line for line in lines if 'class ' in line])
        
        metrics['complexity_score'] = len(non_empty_lines) + (functions * 2) + (classes * 3)
        
        return metrics
        
    def has_basic_syntax(self, code: str) -> bool:
        """Check if code has basic Python syntax"""
        try:
            compile(code, '<string>', 'exec')
            return True
        except SyntaxError:
            return False
        except Exception:
            # Other compilation errors might still be valid syntax
            return True
            
    def attempts_api_call(self, code: str) -> bool:
        """Check if code attempts to make an API call"""
        code_lower = code.lower()
        
        api_patterns = [
            'requests.get',
            'requests.post',
            'requests.put',
            'requests.delete',
            'http',
            'api',
            'url',
            '.get(',
            '.post('
        ]
        
        return any(pattern in code_lower for pattern in api_patterns)
        
    async def safe_execute(self, api_name: str, code: str, env_key: str) -> Tuple[bool, Optional[int], str]:
        """Safely execute code in isolated environment"""
        
        try:
            # Create temporary file with the code
            with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
                # Add environment variable setup
                env_setup = f"""
import os
import sys
import requests
import json

# Set up environment variables
"""
                
                # Add specific environment variables needed
                for key, value in self.env_vars.items():
                    if key.endswith('_API_KEY') or key.endswith('_KEY') or 'API' in key:
                        env_setup += f'os.environ["{key}"] = "{value}"\n'
                
                # Add timeout and safety measures
                safety_code = """
# Safety measures
import signal
def timeout_handler(signum, frame):
    raise TimeoutError("Code execution timeout")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(30)  # 30 second timeout

try:
"""
                
                # Add the generated code with indentation
                indented_code = '\n'.join('    ' + line for line in code.split('\n'))
                
                cleanup_code = """
except Exception as e:
    print(f"EXECUTION_ERROR: {e}")
    sys.exit(1)
finally:
    signal.alarm(0)  # Cancel timeout
"""
                
                full_code = env_setup + safety_code + indented_code + cleanup_code
                f.write(full_code)
                f.flush()
                
                temp_file = f.name
                
            # Execute the code
            try:
                result = subprocess.run(
                    ['python', temp_file],
                    capture_output=True,
                    text=True,
                    timeout=35,  # Slightly longer than internal timeout
                    cwd=os.getcwd()
                )
                
                # Parse output for API response information
                stdout = result.stdout
                stderr = result.stderr
                
                # Look for HTTP status codes in output
                response_code = self.extract_response_code(stdout + stderr)
                
                if result.returncode == 0:
                    # Success - look for positive indicators
                    if any(indicator in stdout.lower() for indicator in ['200', 'success', 'ok', 'data']):
                        return True, response_code, ""
                    else:
                        return True, response_code, "Code executed but no clear success indicator"
                else:
                    # Execution failed
                    error_msg = stderr if stderr else stdout
                    return False, response_code, error_msg[:500]  # Limit error message length
                    
            except subprocess.TimeoutExpired:
                return False, None, "Code execution timeout (30s)"
                
            finally:
                # Clean up temporary file
                try:
                    os.unlink(temp_file)
                except:
                    pass
                    
        except Exception as e:
            return False, None, f"Execution setup failed: {str(e)}"
            
    def extract_response_code(self, output: str) -> Optional[int]:
        """Extract HTTP response code from output"""
        
        # Look for common HTTP status code patterns
        patterns = [
            r'status.?code.?:?\s*(\d{3})',
            r'response.?code.?:?\s*(\d{3})',
            r'http.?(\d{3})',
            r'status.?\s*(\d{3})',
            r'(\d{3})\s*ok',
            r'(\d{3})\s*error'
        ]
        
        for pattern in patterns:
            match = re.search(pattern, output.lower())
            if match:
                try:
                    return int(match.group(1))
                except:
                    continue
                    
        return None

# Test function
async def test_code_execution():
    """Test code execution with sample code"""
    engine = CodeExecutionEngine()
    
    sample_code = """
import requests
import os

api_key = os.environ.get('OPENWEATHER_API_KEY')
url = f"https://api.openweathermap.org/data/2.5/weather?q=London&appid={api_key}"

try:
    response = requests.get(url)
    print(f"Status code: {response.status_code}")
    if response.status_code == 200:
        data = response.json()
        print(f"Weather in London: {data['weather'][0]['description']}")
    else:
        print(f"Error: {response.text}")
except Exception as e:
    print(f"Request failed: {e}")
"""
    
    status, time_ms, response_code, error, metrics = await engine.execute_code(
        "OpenWeatherMap", sample_code, "OPENWEATHER_API_KEY"
    )
    
    print(f"Status: {status}")
    print(f"Time: {time_ms}ms")
    print(f"Response code: {response_code}")
    print(f"Error: {error}")
    print(f"Metrics: {metrics}")

if __name__ == "__main__":
    asyncio.run(test_code_execution())