cleveragents-core/simulations/sim17/assertions.py

import json
import re
from typing import Any, Dict, List, Optional, Union
from jsonpath_ng import parse as jsonpath_parse


class AssertionResult:
    """Result of a single assertion."""
    
    def __init__(self, passed: bool, message: str, assertion_type: str):
        self.passed = passed
        self.message = message
        self.assertion_type = assertion_type


class ResponseValidator:
    """Validates HTTP responses against expected criteria."""
    
    def __init__(self):
        pass
    
    def validate_response(self, response, expected: Dict[str, Any]) -> List[AssertionResult]:
        """Validate response against all expected criteria.
        
        Args:
            response: HTTP response object
            expected: Dictionary containing expected response criteria
            
        Returns:
            List of AssertionResult objects
        """
        results = []
        
        # Validate status code
        if "status" in expected:
            results.append(self._validate_status_code(response, expected["status"]))
        
        # Validate body contains
        if "body_contains" in expected:
            body_contains = expected["body_contains"]
            if isinstance(body_contains, str):
                body_contains = [body_contains]
            for text in body_contains:
                results.append(self._validate_body_contains(response, text))
        
        # Validate JSON path assertions
        if "json_path" in expected:
            json_assertions = expected["json_path"]
            if isinstance(json_assertions, dict):
                json_assertions = [json_assertions]
            for assertion in json_assertions:
                results.append(self._validate_json_path(response, assertion))
        
        # Validate headers
        if "headers" in expected:
            for header_name, expected_value in expected["headers"].items():
                results.append(self._validate_header(response, header_name, expected_value))
        
        # Validate response time (if specified)
        if "max_response_time" in expected:
            results.append(self._validate_response_time(response, expected["max_response_time"]))
        
        return results
    
    def _validate_status_code(self, response, expected_status: Union[int, List[int]]) -> AssertionResult:
        """Validate HTTP status code."""
        if isinstance(expected_status, list):
            passed = response.status_code in expected_status
            expected_str = f"one of {expected_status}"
        else:
            passed = response.status_code == expected_status
            expected_str = str(expected_status)
        
        if passed:
            message = f"Status code {response.status_code} matches expected {expected_str}"
        else:
            message = f"Expected status {expected_str}, got {response.status_code}"
        
        return AssertionResult(passed, message, "status_code")
    
    def _validate_body_contains(self, response, expected_text: str) -> AssertionResult:
        """Validate that response body contains expected text."""
        response_text = response.text
        passed = expected_text in response_text
        
        if passed:
            message = f"Response body contains '{expected_text}'"
        else:
            message = f"Response body does not contain '{expected_text}'"
        
        return AssertionResult(passed, message, "body_contains")
    
    def _validate_json_path(self, response, assertion: Dict[str, Any]) -> AssertionResult:
        """Validate JSON path assertion."""
        path = assertion.get("path", "")
        expected_value = assertion.get("value")
        operator = assertion.get("operator", "equals")
        
        try:
            json_data = response.json()
            jsonpath_expr = jsonpath_parse(path)
            matches = jsonpath_expr.find(json_data)
            
            if not matches:
                return AssertionResult(
                    False, 
                    f"JSON path '{path}' not found in response",
                    "json_path"
                )
            
            actual_value = matches[0].value
            passed = self._compare_values(actual_value, expected_value, operator)
            
            if passed:
                message = f"JSON path '{path}' assertion passed: {actual_value} {operator} {expected_value}"
            else:
                message = f"JSON path '{path}' assertion failed: {actual_value} {operator} {expected_value}"
            
            return AssertionResult(passed, message, "json_path")
            
        except json.JSONDecodeError:
            return AssertionResult(
                False,
                f"Response is not valid JSON for path '{path}'",
                "json_path"
            )
        except Exception as e:
            return AssertionResult(
                False,
                f"JSON path assertion error: {str(e)}",
                "json_path"
            )
    
    def _validate_header(self, response, header_name: str, expected_value: str) -> AssertionResult:
        """Validate response header value."""
        actual_value = response.headers.get(header_name)
        
        if actual_value is None:
            return AssertionResult(
                False,
                f"Header '{header_name}' not found in response",
                "header"
            )
        
        passed = str(actual_value) == str(expected_value)
        
        if passed:
            message = f"Header '{header_name}' matches expected value '{expected_value}'"
        else:
            message = f"Header '{header_name}' expected '{expected_value}', got '{actual_value}'"
        
        return AssertionResult(passed, message, "header")
    
    def _validate_response_time(self, response, max_time: float) -> AssertionResult:
        """Validate response time is within acceptable limit."""
        response_time = response.elapsed.total_seconds()
        passed = response_time <= max_time
        
        if passed:
            message = f"Response time {response_time:.3f}s is within limit of {max_time}s"
        else:
            message = f"Response time {response_time:.3f}s exceeds limit of {max_time}s"
        
        return AssertionResult(passed, message, "response_time")
    
    def _compare_values(self, actual: Any, expected: Any, operator: str) -> bool:
        """Compare two values using specified operator."""
        if operator == "equals":
            return actual == expected
        elif operator == "not_equals":
            return actual != expected
        elif operator == "greater_than":
            return float(actual) > float(expected)
        elif operator == "less_than":
            return float(actual) < float(expected)
        elif operator == "greater_than_or_equal":
            return float(actual) >= float(expected)
        elif operator == "less_than_or_equal":
            return float(actual) <= float(expected)
        elif operator == "contains":
            return str(expected) in str(actual)
        elif operator == "regex":
            return bool(re.search(str(expected), str(actual)))
        else:
            raise ValueError(f"Unknown operator: {operator}")