cleveragents-core/simulations/sim17/runner.py

import yaml
import requests
import json
import time
from typing import Dict, List, Any, Optional
from urllib.parse import urljoin

from variables import VariableContext
from assertions import ResponseValidator
from reporter import TestReporter, TestResult


class TestRunner:
    """Main test execution engine for HTTP API testing framework."""
    
    def __init__(self, base_url: str = ""):
        self.base_url = base_url.rstrip('/')
        self.variable_context = VariableContext()
        self.validator = ResponseValidator()
        self.reporter = TestReporter()
        self.session = requests.Session()
    
    def load_test_suite(self, suite_path: str) -> Dict[str, Any]:
        """Load test suite from YAML file.
        
        Args:
            suite_path: Path to YAML test suite file
            
        Returns:
            Parsed test suite dictionary
        """
        try:
            with open(suite_path, 'r', encoding='utf-8') as f:
                suite = yaml.safe_load(f)
            return suite
        except FileNotFoundError:
            raise FileNotFoundError(f"Test suite file not found: {suite_path}")
        except yaml.YAMLError as e:
            raise ValueError(f"Invalid YAML in test suite: {e}")
    
    def run_test_suite(self, suite_path: str) -> TestReporter:
        """Execute complete test suite.
        
        Args:
            suite_path: Path to YAML test suite file
            
        Returns:
            TestReporter with execution results
        """
        suite = self.load_test_suite(suite_path)
        
        # Set up suite-level variables
        suite_vars = suite.get('variables', {})
        for name, value in suite_vars.items():
            self.variable_context.set_variable(name, value)
        
        # Update base URL if specified in suite
        if 'base_url' in suite and not self.base_url:
            self.base_url = suite['base_url'].rstrip('/')
        
        # Execute tests
        self.reporter.start_execution()
        
        tests = suite.get('tests', [])
        for test_case in tests:
            try:
                result = self.run_single_test(test_case)
                self.reporter.add_result(result)
            except Exception as e:
                # Create failed result for test execution errors
                error_result = TestResult(
                    name=test_case.get('name', 'Unknown Test'),
                    method=test_case.get('method', 'GET'),
                    url=test_case.get('url', ''),
                    status_code=0,
                    response_time=0.0,
                    passed=False,
                    assertion_results=[],
                    error=str(e)
                )
                self.reporter.add_result(error_result)
        
        self.reporter.end_execution()
        return self.reporter
    
    def run_single_test(self, test_case: Dict[str, Any]) -> TestResult:
        """Execute a single test case.
        
        Args:
            test_case: Test case configuration dictionary
            
        Returns:
            TestResult object
        """
        test_name = test_case.get('name', 'Unnamed Test')
        
        # Build request parameters with variable substitution
        method = test_case.get('method', 'GET').upper()
        url = self._build_url(test_case.get('url', ''))
        headers = self._substitute_variables_in_data(test_case.get('headers', {}))
        
        # Handle request body
        body = test_case.get('body')
        json_body = test_case.get('json')
        
        request_kwargs = {
            'headers': headers,
            'timeout': test_case.get('timeout', 30)
        }
        
        if json_body is not None:
            request_kwargs['json'] = self._substitute_variables_in_data(json_body)
        elif body is not None:
            if isinstance(body, str):
                request_kwargs['data'] = self.variable_context.substitute_variables(body)
            else:
                request_kwargs['data'] = self._substitute_variables_in_data(body)
        
        # Add query parameters if specified
        params = test_case.get('params', {})
        if params:
            request_kwargs['params'] = self._substitute_variables_in_data(params)
        
        # Execute request
        start_time = time.time()
        try:
            response = self.session.request(method, url, **request_kwargs)
            response_time = time.time() - start_time
        except requests.RequestException as e:
            return TestResult(
                name=test_name,
                method=method,
                url=url,
                status_code=0,
                response_time=time.time() - start_time,
                passed=False,
                assertion_results=[],
                error=f"Request failed: {str(e)}"
            )
        
        # Extract variables from response
        extract_config = test_case.get('extract', {})
        if extract_config:
            self.variable_context.extract_variables(response, extract_config)
        
        # Validate response
        expected = test_case.get('expected', {})
        assertion_results = []
        if expected:
            assertion_results = self.validator.validate_response(response, expected)
        
        # Determine overall test result
        passed = all(assertion.passed for assertion in assertion_results)
        
        return TestResult(
            name=test_name,
            method=method,
            url=url,
            status_code=response.status_code,
            response_time=response_time,
            passed=passed,
            assertion_results=assertion_results
        )
    
    def _build_url(self, path: str) -> str:
        """Build complete URL from base URL and path."""
        path = self.variable_context.substitute_variables(path)
        
        if path.startswith(('http://', 'https://')):
            return path
        
        if self.base_url:
            return urljoin(self.base_url + '/', path.lstrip('/'))
        
        return path
    
    def _substitute_variables_in_data(self, data: Any) -> Any:
        """Substitute variables in request data."""
        if isinstance(data, dict):
            return self.variable_context.substitute_in_dict(data)
        elif isinstance(data, list):
            return self.variable_context.substitute_in_list(data)
        elif isinstance(data, str):
            return self.variable_context.substitute_variables(data)
        else:
            return data