From 6158046eaec544ca9540fafa79d7f02725f68ae9 Mon Sep 17 00:00:00 2001 From: Aditya Chhabra Date: Fri, 13 Mar 2026 18:52:24 +0530 Subject: [PATCH] =?UTF-8?q?test:=20sim12,=20sim14,=20sim17=20=E2=80=94=20v?= =?UTF-8?q?erify=20multi-file=20DoD=20compliance=20after=20prompt=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All 3 simulations tested with the fixed generate prompt that passes the original DoD to the LLM. Results: - SIM12 (Rate Limiter): 7 files — algorithms.py, store.py, middleware.py, config.py, example_app.py, requirements.txt, __init__.py - SIM14 (JSON Validator): 6 files — validator.py, types.py, errors.py, cli.py, requirements.txt, __init__.py - SIM17 (API Tester): 7 files — runner.py, assertions.py, variables.py, reporter.py, cli.py, requirements.txt, test_suite_example.yaml All Python files pass compile() syntax check. All DoD files present. Files written directly via --output-dir, no manual cleanup needed. --- actions/sim12-rate-limiter.yaml | 16 ++ actions/sim14-json-validator.yaml | 15 ++ actions/sim17-api-tester.yaml | 16 ++ simulations/sim12/__init__.py | 19 ++ simulations/sim12/algorithms.py | 151 ++++++++++++++++ simulations/sim12/config.py | 25 +++ simulations/sim12/example_app.py | 105 +++++++++++ simulations/sim12/middleware.py | 94 ++++++++++ simulations/sim12/requirements.txt | 1 + simulations/sim12/store.py | 86 +++++++++ simulations/sim14/__init__.py | 38 ++++ simulations/sim14/cli.py | 109 ++++++++++++ simulations/sim14/errors.py | 35 ++++ simulations/sim14/requirements.txt | 2 + simulations/sim14/types.py | 208 ++++++++++++++++++++++ simulations/sim14/validator.py | 152 ++++++++++++++++ simulations/sim17/assertions.py | 185 +++++++++++++++++++ simulations/sim17/cli.py | 98 ++++++++++ simulations/sim17/reporter.py | 137 ++++++++++++++ simulations/sim17/requirements.txt | 3 + simulations/sim17/runner.py | 186 +++++++++++++++++++ simulations/sim17/test_suite_example.yaml | 153 ++++++++++++++++ simulations/sim17/variables.py | 121 +++++++++++++ 23 files changed, 1955 insertions(+) create mode 100644 actions/sim12-rate-limiter.yaml create mode 100644 actions/sim14-json-validator.yaml create mode 100644 actions/sim17-api-tester.yaml create mode 100644 simulations/sim12/__init__.py create mode 100644 simulations/sim12/algorithms.py create mode 100644 simulations/sim12/config.py create mode 100644 simulations/sim12/example_app.py create mode 100644 simulations/sim12/middleware.py create mode 100644 simulations/sim12/requirements.txt create mode 100644 simulations/sim12/store.py create mode 100644 simulations/sim14/__init__.py create mode 100644 simulations/sim14/cli.py create mode 100644 simulations/sim14/errors.py create mode 100644 simulations/sim14/requirements.txt create mode 100644 simulations/sim14/types.py create mode 100644 simulations/sim14/validator.py create mode 100644 simulations/sim17/assertions.py create mode 100644 simulations/sim17/cli.py create mode 100644 simulations/sim17/reporter.py create mode 100644 simulations/sim17/requirements.txt create mode 100644 simulations/sim17/runner.py create mode 100644 simulations/sim17/test_suite_example.yaml create mode 100644 simulations/sim17/variables.py diff --git a/actions/sim12-rate-limiter.yaml b/actions/sim12-rate-limiter.yaml new file mode 100644 index 00000000..c6cb2ed6 --- /dev/null +++ b/actions/sim12-rate-limiter.yaml @@ -0,0 +1,16 @@ +name: local/rate-limiter +description: > + Build a rate limiting middleware library in Python for Flask APIs. + Implement token bucket, sliding window, and fixed window algorithms. + Support per-IP and per-API-key limits. Store counters in an in-memory + store with TTL expiry. Include a Flask decorator for easy route-level + rate limiting and configurable response headers (X-RateLimit-*). +strategy_actor: anthropic/claude-sonnet-4-20250514 +execution_actor: anthropic/claude-sonnet-4-20250514 +definition_of_done: > + An algorithms.py implements TokenBucket, SlidingWindow, and FixedWindow classes. + A store.py provides an in-memory counter store with TTL-based expiry. + A middleware.py provides a Flask decorator @rate_limit() for route protection. + A config.py defines rate limit configuration dataclasses. + An example_app.py shows a Flask app using all three algorithms on different routes. + A requirements.txt lists dependencies (flask). diff --git a/actions/sim14-json-validator.yaml b/actions/sim14-json-validator.yaml new file mode 100644 index 00000000..3b476a21 --- /dev/null +++ b/actions/sim14-json-validator.yaml @@ -0,0 +1,15 @@ +name: local/json-validator +description: > + Build a JSON schema validator in Python from scratch (no jsonschema library). + Support type validation (string, number, integer, boolean, array, object, null), + constraints (minLength, maxLength, minimum, maximum, pattern, enum), + nested object and array validation, required fields, and custom error messages. + Output validation errors with JSON path to the failing field. +strategy_actor: anthropic/claude-sonnet-4-20250514 +execution_actor: anthropic/claude-sonnet-4-20250514 +definition_of_done: > + A validator.py implements SchemaValidator with recursive type and constraint checking. + A types.py defines validation rules for each JSON type with constraint support. + An errors.py provides ValidationError with JSON path tracking and human-readable messages. + A cli.py accepts --schema and --data file paths via argparse and prints validation results. + A requirements.txt lists dependencies (none beyond stdlib expected). diff --git a/actions/sim17-api-tester.yaml b/actions/sim17-api-tester.yaml new file mode 100644 index 00000000..c209a64d --- /dev/null +++ b/actions/sim17-api-tester.yaml @@ -0,0 +1,16 @@ +name: local/api-tester +description: > + Build an HTTP API testing framework in Python. Define test cases in YAML files + specifying method, URL, headers, body, and expected response (status code, + body contains, JSON path assertions). Run tests sequentially, support variable + extraction from responses for chaining requests, and output results with + pass/fail counts and detailed failure messages. +strategy_actor: anthropic/claude-sonnet-4-20250514 +execution_actor: anthropic/claude-sonnet-4-20250514 +definition_of_done: > + A runner.py loads YAML test suites and executes HTTP requests sequentially. + An assertions.py validates responses against expected status, body, and JSON path rules. + A variables.py handles variable extraction from responses and substitution in subsequent requests. + A reporter.py outputs test results with pass/fail summary and failure details. + A cli.py provides argparse interface with --suite (YAML path) and --base-url flags. + A requirements.txt lists dependencies (requests, pyyaml). diff --git a/simulations/sim12/__init__.py b/simulations/sim12/__init__.py new file mode 100644 index 00000000..f252afff --- /dev/null +++ b/simulations/sim12/__init__.py @@ -0,0 +1,19 @@ +""" +Rate limiting middleware library for Flask APIs. +""" + +from .algorithms import TokenBucket, SlidingWindow, FixedWindow +from .config import RateLimitConfig, LimitType +from .middleware import rate_limit +from .store import InMemoryStore + +__version__ = "1.0.0" +__all__ = [ + "TokenBucket", + "SlidingWindow", + "FixedWindow", + "RateLimitConfig", + "LimitType", + "rate_limit", + "InMemoryStore" +] \ No newline at end of file diff --git a/simulations/sim12/algorithms.py b/simulations/sim12/algorithms.py new file mode 100644 index 00000000..264c194f --- /dev/null +++ b/simulations/sim12/algorithms.py @@ -0,0 +1,151 @@ +import time +import math +from abc import ABC, abstractmethod +from typing import Tuple, Dict, Any, List +from .store import store + + +class RateLimitAlgorithm(ABC): + """Abstract base class for rate limiting algorithms.""" + + def __init__(self, limit: int, window: int): + self.limit = limit + self.window = window + + @abstractmethod + def is_allowed(self, identifier: str) -> Tuple[bool, Dict[str, Any]]: + """ + Check if request is allowed. + Returns (allowed, headers_dict). + """ + pass + + def _get_key(self, identifier: str) -> str: + """Generate storage key for identifier.""" + return f"{self.__class__.__name__}:{identifier}" + + +class TokenBucket(RateLimitAlgorithm): + """Token bucket rate limiting algorithm.""" + + def is_allowed(self, identifier: str) -> Tuple[bool, Dict[str, Any]]: + key = self._get_key(identifier) + now = time.time() + + # Get current bucket state + bucket_data = store.get(key) + + if bucket_data is None: + # Initialize bucket + bucket_data = { + 'tokens': self.limit, + 'last_refill': now + } + else: + # Refill tokens based on elapsed time + elapsed = now - bucket_data['last_refill'] + tokens_to_add = elapsed * (self.limit / self.window) + bucket_data['tokens'] = min(self.limit, bucket_data['tokens'] + tokens_to_add) + bucket_data['last_refill'] = now + + # Check if request can be served + if bucket_data['tokens'] >= 1: + bucket_data['tokens'] -= 1 + allowed = True + remaining = int(bucket_data['tokens']) + else: + allowed = False + remaining = 0 + + # Store updated bucket state (TTL = 2 * window for safety) + store.set(key, bucket_data, ttl=self.window * 2) + + # Calculate reset time + if bucket_data['tokens'] < self.limit: + time_to_full = (self.limit - bucket_data['tokens']) / (self.limit / self.window) + reset_time = int(now + time_to_full) + else: + reset_time = int(now) + + headers = { + 'X-RateLimit-Limit': str(self.limit), + 'X-RateLimit-Remaining': str(remaining), + 'X-RateLimit-Reset': str(reset_time) + } + + return allowed, headers + + +class SlidingWindow(RateLimitAlgorithm): + """Sliding window rate limiting algorithm.""" + + def is_allowed(self, identifier: str) -> Tuple[bool, Dict[str, Any]]: + key = self._get_key(identifier) + now = time.time() + + # Get current request timestamps + timestamps = store.get(key) or [] + + # Remove timestamps outside the window + window_start = now - self.window + timestamps = [ts for ts in timestamps if ts > window_start] + + # Check if request is allowed + if len(timestamps) < self.limit: + timestamps.append(now) + allowed = True + remaining = self.limit - len(timestamps) + else: + allowed = False + remaining = 0 + + # Store updated timestamps + store.set(key, timestamps, ttl=self.window + 1) + + # Calculate reset time (when oldest request in window expires) + if timestamps: + reset_time = int(timestamps[0] + self.window) + else: + reset_time = int(now + self.window) + + headers = { + 'X-RateLimit-Limit': str(self.limit), + 'X-RateLimit-Remaining': str(remaining), + 'X-RateLimit-Reset': str(reset_time) + } + + return allowed, headers + + +class FixedWindow(RateLimitAlgorithm): + """Fixed window rate limiting algorithm.""" + + def is_allowed(self, identifier: str) -> Tuple[bool, Dict[str, Any]]: + now = time.time() + + # Calculate current window + window_start = int(now // self.window) * self.window + key = f"{self._get_key(identifier)}:{window_start}" + + # Get current count for this window + current_count = store.get(key) or 0 + + # Check if request is allowed + if current_count < self.limit: + new_count = store.increment(key, ttl=self.window + 1) + allowed = True + remaining = max(0, self.limit - new_count) + else: + allowed = False + remaining = 0 + + # Calculate reset time (start of next window) + reset_time = window_start + self.window + + headers = { + 'X-RateLimit-Limit': str(self.limit), + 'X-RateLimit-Remaining': str(remaining), + 'X-RateLimit-Reset': str(int(reset_time)) + } + + return allowed, headers \ No newline at end of file diff --git a/simulations/sim12/config.py b/simulations/sim12/config.py new file mode 100644 index 00000000..a6512f25 --- /dev/null +++ b/simulations/sim12/config.py @@ -0,0 +1,25 @@ +from dataclasses import dataclass +from enum import Enum +from typing import Optional + + +class LimitType(Enum): + IP = "ip" + API_KEY = "api_key" + + +@dataclass +class RateLimitConfig: + """Configuration for rate limiting.""" + limit: int + window: int # Time window in seconds + limit_type: LimitType = LimitType.IP + api_key_header: str = "X-API-Key" + error_message: str = "Rate limit exceeded" + error_code: int = 429 + + def __post_init__(self): + if self.limit <= 0: + raise ValueError("Limit must be positive") + if self.window <= 0: + raise ValueError("Window must be positive") \ No newline at end of file diff --git a/simulations/sim12/example_app.py b/simulations/sim12/example_app.py new file mode 100644 index 00000000..9cc45fcc --- /dev/null +++ b/simulations/sim12/example_app.py @@ -0,0 +1,105 @@ +from flask import Flask, jsonify +from config import RateLimitConfig, LimitType +from middleware import rate_limit + +app = Flask(__name__) + +# Rate limiting configurations +token_bucket_config = RateLimitConfig( + limit=10, + window=60, # 10 requests per minute + limit_type=LimitType.IP, + error_message="Token bucket rate limit exceeded" +) + +sliding_window_config = RateLimitConfig( + limit=5, + window=30, # 5 requests per 30 seconds + limit_type=LimitType.IP, + error_message="Sliding window rate limit exceeded" +) + +fixed_window_config = RateLimitConfig( + limit=20, + window=60, # 20 requests per minute + limit_type=LimitType.API_KEY, + api_key_header="X-API-Key", + error_message="Fixed window rate limit exceeded" +) + + +@app.route("/") +def index(): + """Home route without rate limiting.""" + return jsonify({ + "message": "Rate Limiting Example API", + "endpoints": { + "/token-bucket": "Token bucket algorithm (10 req/min by IP)", + "/sliding-window": "Sliding window algorithm (5 req/30s by IP)", + "/fixed-window": "Fixed window algorithm (20 req/min by API key)" + } + }) + + +@app.route("/token-bucket") +@rate_limit(token_bucket_config, algorithm="token_bucket") +def token_bucket_endpoint(): + """Endpoint protected by token bucket rate limiting.""" + return jsonify({ + "message": "Token bucket endpoint accessed successfully", + "algorithm": "token_bucket", + "limit": "10 requests per minute per IP" + }) + + +@app.route("/sliding-window") +@rate_limit(sliding_window_config, algorithm="sliding_window") +def sliding_window_endpoint(): + """Endpoint protected by sliding window rate limiting.""" + return jsonify({ + "message": "Sliding window endpoint accessed successfully", + "algorithm": "sliding_window", + "limit": "5 requests per 30 seconds per IP" + }) + + +@app.route("/fixed-window") +@rate_limit(fixed_window_config, algorithm="fixed_window") +def fixed_window_endpoint(): + """Endpoint protected by fixed window rate limiting.""" + return jsonify({ + "message": "Fixed window endpoint accessed successfully", + "algorithm": "fixed_window", + "limit": "20 requests per minute per API key" + }) + + +@app.route("/health") +def health(): + """Health check endpoint without rate limiting.""" + return jsonify({"status": "healthy"}) + + +@app.errorhandler(429) +def rate_limit_handler(e): + """Custom handler for rate limit errors.""" + return jsonify({ + "error": "Rate limit exceeded", + "message": "Please wait before making more requests" + }), 429 + + +if __name__ == "__main__": + print("Starting Flask app with rate limiting examples...") + print("\nEndpoints:") + print("- GET / : Home page with API information") + print("- GET /token-bucket : Token bucket (10 req/min by IP)") + print("- GET /sliding-window : Sliding window (5 req/30s by IP)") + print("- GET /fixed-window : Fixed window (20 req/min by API key - requires X-API-Key header)") + print("- GET /health : Health check (no rate limiting)") + print("\nTesting with curl:") + print("curl http://localhost:5000/token-bucket") + print("curl -H 'X-API-Key: test-key' http://localhost:5000/fixed-window") + print() + + app.run(debug=True, host="0.0.0.0", port=5000) \ No newline at end of file diff --git a/simulations/sim12/middleware.py b/simulations/sim12/middleware.py new file mode 100644 index 00000000..5985ae71 --- /dev/null +++ b/simulations/sim12/middleware.py @@ -0,0 +1,94 @@ +import functools +from typing import Union, Optional +from flask import request, jsonify, Response +from .algorithms import RateLimitAlgorithm, TokenBucket, SlidingWindow, FixedWindow +from .config import RateLimitConfig, LimitType + + +def rate_limit( + config: RateLimitConfig, + algorithm: str = "fixed_window" +) -> callable: + """ + Flask decorator for rate limiting routes. + + Args: + config: RateLimitConfig instance + algorithm: Algorithm type ("token_bucket", "sliding_window", "fixed_window") + """ + + # Create algorithm instance + algorithm_map = { + "token_bucket": TokenBucket, + "sliding_window": SlidingWindow, + "fixed_window": FixedWindow + } + + if algorithm not in algorithm_map: + raise ValueError(f"Unknown algorithm: {algorithm}") + + rate_limiter = algorithm_map[algorithm](config.limit, config.window) + + def decorator(f): + @functools.wraps(f) + def wrapper(*args, **kwargs): + # Get identifier based on limit type + identifier = _get_identifier(config) + + if identifier is None: + return jsonify({ + "error": "Missing API key" if config.limit_type == LimitType.API_KEY else "Unable to identify client" + }), 400 + + # Check rate limit + allowed, headers = rate_limiter.is_allowed(identifier) + + if not allowed: + response = jsonify({"error": config.error_message}) + response.status_code = config.error_code + + # Add rate limit headers + for header_name, header_value in headers.items(): + response.headers[header_name] = header_value + + return response + + # Call original function + result = f(*args, **kwargs) + + # Add rate limit headers to successful response + if isinstance(result, Response): + response = result + else: + # Handle tuple returns (response, status_code, headers) + if isinstance(result, tuple): + if len(result) == 2: + response = jsonify(result[0]) + response.status_code = result[1] + elif len(result) == 3: + response = jsonify(result[0]) + response.status_code = result[1] + response.headers.update(result[2]) + else: + response = jsonify(result) + else: + response = jsonify(result) if not isinstance(result, Response) else result + + # Add rate limit headers + for header_name, header_value in headers.items(): + response.headers[header_name] = header_value + + return response + + return wrapper + return decorator + + +def _get_identifier(config: RateLimitConfig) -> Optional[str]: + """Get client identifier based on configuration.""" + if config.limit_type == LimitType.IP: + return request.environ.get('REMOTE_ADDR') or request.remote_addr + elif config.limit_type == LimitType.API_KEY: + return request.headers.get(config.api_key_header) + else: + return None \ No newline at end of file diff --git a/simulations/sim12/requirements.txt b/simulations/sim12/requirements.txt new file mode 100644 index 00000000..1fadb158 --- /dev/null +++ b/simulations/sim12/requirements.txt @@ -0,0 +1 @@ +flask>=2.0.0 \ No newline at end of file diff --git a/simulations/sim12/store.py b/simulations/sim12/store.py new file mode 100644 index 00000000..f8e4a2b5 --- /dev/null +++ b/simulations/sim12/store.py @@ -0,0 +1,86 @@ +import time +import threading +from typing import Dict, Any, Optional +from collections import defaultdict + + +class InMemoryStore: + """Thread-safe in-memory store with TTL-based expiry.""" + + def __init__(self): + self._data: Dict[str, Any] = {} + self._expiry: Dict[str, float] = {} + self._lock = threading.RLock() + self._cleanup_interval = 60 # Cleanup every 60 seconds + self._last_cleanup = time.time() + + def get(self, key: str) -> Optional[Any]: + """Get value by key, returns None if expired or not found.""" + with self._lock: + self._cleanup_if_needed() + + if key not in self._data: + return None + + if key in self._expiry and time.time() > self._expiry[key]: + self._delete_key(key) + return None + + return self._data[key] + + def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: + """Set value with optional TTL in seconds.""" + with self._lock: + self._cleanup_if_needed() + self._data[key] = value + + if ttl is not None: + self._expiry[key] = time.time() + ttl + elif key in self._expiry: + del self._expiry[key] + + def delete(self, key: str) -> None: + """Delete key from store.""" + with self._lock: + self._delete_key(key) + + def increment(self, key: str, amount: int = 1, ttl: Optional[int] = None) -> int: + """Increment counter, creating if doesn't exist.""" + with self._lock: + current = self.get(key) or 0 + new_value = current + amount + self.set(key, new_value, ttl) + return new_value + + def _delete_key(self, key: str) -> None: + """Internal method to delete key without lock.""" + self._data.pop(key, None) + self._expiry.pop(key, None) + + def _cleanup_if_needed(self) -> None: + """Cleanup expired keys if interval has passed.""" + now = time.time() + if now - self._last_cleanup > self._cleanup_interval: + self._cleanup_expired() + self._last_cleanup = now + + def _cleanup_expired(self) -> None: + """Remove all expired keys.""" + now = time.time() + expired_keys = [ + key for key, expiry_time in self._expiry.items() + if now > expiry_time + ] + + for key in expired_keys: + self._delete_key(key) + + def clear(self) -> None: + """Clear all data.""" + with self._lock: + self._data.clear() + self._expiry.clear() + + +# Global store instance +store = InMemoryStore() \ No newline at end of file diff --git a/simulations/sim14/__init__.py b/simulations/sim14/__init__.py new file mode 100644 index 00000000..96b9ed83 --- /dev/null +++ b/simulations/sim14/__init__.py @@ -0,0 +1,38 @@ +"""JSON Schema Validator Package + +A from-scratch implementation of JSON schema validation supporting: +- Basic JSON types (string, number, integer, boolean, array, object, null) +- String constraints (minLength, maxLength, pattern) +- Numeric constraints (minimum, maximum) +- Enum constraints +- Nested object and array validation +- Required field validation +- Detailed error reporting with JSON path tracking +""" + +from .validator import SchemaValidator +from .errors import ValidationError +from .types import ( + validate_string, + validate_number, + validate_integer, + validate_boolean, + validate_array, + validate_object, + validate_null +) + +__version__ = "1.0.0" +__author__ = "JSON Schema Validator" + +__all__ = [ + "SchemaValidator", + "ValidationError", + "validate_string", + "validate_number", + "validate_integer", + "validate_boolean", + "validate_array", + "validate_object", + "validate_null" +] \ No newline at end of file diff --git a/simulations/sim14/cli.py b/simulations/sim14/cli.py new file mode 100644 index 00000000..a47fbac0 --- /dev/null +++ b/simulations/sim14/cli.py @@ -0,0 +1,109 @@ +import argparse +import json +import sys +from pathlib import Path +from typing import Any, Dict +from .validator import SchemaValidator +from .errors import ValidationError + + +def load_json_file(file_path: str) -> Dict[str, Any]: + """Load JSON data from file. + + Args: + file_path: Path to JSON file + + Returns: + Parsed JSON data + + Raises: + SystemExit: If file cannot be loaded or parsed + """ + try: + with open(file_path, 'r', encoding='utf-8') as f: + return json.load(f) + except FileNotFoundError: + print(f"Error: File '{file_path}' not found.", file=sys.stderr) + sys.exit(1) + except json.JSONDecodeError as e: + print(f"Error: Invalid JSON in file '{file_path}': {e}", file=sys.stderr) + sys.exit(1) + except Exception as e: + print(f"Error: Failed to read file '{file_path}': {e}", file=sys.stderr) + sys.exit(1) + + +def format_validation_results(errors: list, data_file: str) -> str: + """Format validation results for output. + + Args: + errors: List of ValidationError objects + data_file: Name of the data file being validated + + Returns: + Formatted validation results + """ + if not errors: + return f"✓ Validation successful: '{data_file}' is valid according to the schema." + + result = f"✗ Validation failed: '{data_file}' has {len(errors)} error(s):\n\n" + + for i, error in enumerate(errors, 1): + result += f"{i}. {error}\n" + + return result + + +def main(): + """Main CLI entry point.""" + parser = argparse.ArgumentParser( + description="JSON Schema Validator - Validate JSON data against a schema" + ) + parser.add_argument( + "--schema", + required=True, + help="Path to JSON schema file" + ) + parser.add_argument( + "--data", + required=True, + help="Path to JSON data file to validate" + ) + parser.add_argument( + "--format", + choices=["human", "json"], + default="human", + help="Output format (default: human)" + ) + + args = parser.parse_args() + + # Load schema and data files + schema = load_json_file(args.schema) + data = load_json_file(args.data) + + # Create validator and perform validation + try: + validator = SchemaValidator(schema) + errors = validator.validate(data) + + # Output results based on format + if args.format == "json": + result = { + "valid": len(errors) == 0, + "errors": [error.to_dict() for error in errors] + } + print(json.dumps(result, indent=2)) + else: + print(format_validation_results(errors, args.data)) + + # Exit with appropriate code + sys.exit(0 if len(errors) == 0 else 1) + + except Exception as e: + print(f"Error during validation: {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/simulations/sim14/errors.py b/simulations/sim14/errors.py new file mode 100644 index 00000000..1b77c81e --- /dev/null +++ b/simulations/sim14/errors.py @@ -0,0 +1,35 @@ +from typing import Optional + + +class ValidationError(Exception): + """Custom exception for JSON schema validation errors.""" + + def __init__(self, path: str, message: str): + """Initialize validation error. + + Args: + path: JSON path to the failing field + message: Human-readable error message + """ + self.path = path + self.message = message + super().__init__(f"Validation error at {path}: {message}") + + def __str__(self) -> str: + """Return formatted error message.""" + return f"Error at {self.path}: {self.message}" + + def __repr__(self) -> str: + """Return detailed representation.""" + return f"ValidationError(path='{self.path}', message='{self.message}')" + + def to_dict(self) -> dict: + """Convert error to dictionary representation. + + Returns: + Dictionary with path and message keys + """ + return { + "path": self.path, + "message": self.message + } \ No newline at end of file diff --git a/simulations/sim14/requirements.txt b/simulations/sim14/requirements.txt new file mode 100644 index 00000000..b96adadd --- /dev/null +++ b/simulations/sim14/requirements.txt @@ -0,0 +1,2 @@ +# No external dependencies required +# This JSON schema validator uses only Python standard library \ No newline at end of file diff --git a/simulations/sim14/types.py b/simulations/sim14/types.py new file mode 100644 index 00000000..f23c2e32 --- /dev/null +++ b/simulations/sim14/types.py @@ -0,0 +1,208 @@ +import re +from typing import Any, Dict +from .errors import ValidationError + + +def validate_string(value: Any, schema: Dict[str, Any], path: str) -> None: + """Validate string type and constraints. + + Args: + value: Value to validate + schema: Schema definition + path: JSON path to the value + + Raises: + ValidationError: If validation fails + """ + if not isinstance(value, str): + raise ValidationError( + path=path, + message=f"Expected string, got {type(value).__name__}" + ) + + # Check minLength constraint + min_length = schema.get("minLength") + if min_length is not None and len(value) < min_length: + raise ValidationError( + path=path, + message=f"String length {len(value)} is less than minimum {min_length}" + ) + + # Check maxLength constraint + max_length = schema.get("maxLength") + if max_length is not None and len(value) > max_length: + raise ValidationError( + path=path, + message=f"String length {len(value)} exceeds maximum {max_length}" + ) + + # Check pattern constraint + pattern = schema.get("pattern") + if pattern is not None: + try: + if not re.match(pattern, value): + raise ValidationError( + path=path, + message=f"String '{value}' does not match pattern '{pattern}'" + ) + except re.error as e: + raise ValidationError( + path=path, + message=f"Invalid regex pattern '{pattern}': {str(e)}" + ) + + +def validate_number(value: Any, schema: Dict[str, Any], path: str) -> None: + """Validate number type and constraints. + + Args: + value: Value to validate + schema: Schema definition + path: JSON path to the value + + Raises: + ValidationError: If validation fails + """ + if not isinstance(value, (int, float)) or isinstance(value, bool): + raise ValidationError( + path=path, + message=f"Expected number, got {type(value).__name__}" + ) + + # Check minimum constraint + minimum = schema.get("minimum") + if minimum is not None and value < minimum: + raise ValidationError( + path=path, + message=f"Value {value} is less than minimum {minimum}" + ) + + # Check maximum constraint + maximum = schema.get("maximum") + if maximum is not None and value > maximum: + raise ValidationError( + path=path, + message=f"Value {value} exceeds maximum {maximum}" + ) + + +def validate_integer(value: Any, schema: Dict[str, Any], path: str) -> None: + """Validate integer type and constraints. + + Args: + value: Value to validate + schema: Schema definition + path: JSON path to the value + + Raises: + ValidationError: If validation fails + """ + if not isinstance(value, int) or isinstance(value, bool): + raise ValidationError( + path=path, + message=f"Expected integer, got {type(value).__name__}" + ) + + # Check minimum constraint + minimum = schema.get("minimum") + if minimum is not None and value < minimum: + raise ValidationError( + path=path, + message=f"Value {value} is less than minimum {minimum}" + ) + + # Check maximum constraint + maximum = schema.get("maximum") + if maximum is not None and value > maximum: + raise ValidationError( + path=path, + message=f"Value {value} exceeds maximum {maximum}" + ) + + +def validate_boolean(value: Any, schema: Dict[str, Any], path: str) -> None: + """Validate boolean type. + + Args: + value: Value to validate + schema: Schema definition + path: JSON path to the value + + Raises: + ValidationError: If validation fails + """ + if not isinstance(value, bool): + raise ValidationError( + path=path, + message=f"Expected boolean, got {type(value).__name__}" + ) + + +def validate_array(value: Any, schema: Dict[str, Any], path: str) -> None: + """Validate array type and constraints. + + Args: + value: Value to validate + schema: Schema definition + path: JSON path to the value + + Raises: + ValidationError: If validation fails + """ + if not isinstance(value, list): + raise ValidationError( + path=path, + message=f"Expected array, got {type(value).__name__}" + ) + + # Check minLength constraint for arrays + min_length = schema.get("minLength") + if min_length is not None and len(value) < min_length: + raise ValidationError( + path=path, + message=f"Array length {len(value)} is less than minimum {min_length}" + ) + + # Check maxLength constraint for arrays + max_length = schema.get("maxLength") + if max_length is not None and len(value) > max_length: + raise ValidationError( + path=path, + message=f"Array length {len(value)} exceeds maximum {max_length}" + ) + + +def validate_object(value: Any, schema: Dict[str, Any], path: str) -> None: + """Validate object type. + + Args: + value: Value to validate + schema: Schema definition + path: JSON path to the value + + Raises: + ValidationError: If validation fails + """ + if not isinstance(value, dict): + raise ValidationError( + path=path, + message=f"Expected object, got {type(value).__name__}" + ) + + +def validate_null(value: Any, schema: Dict[str, Any], path: str) -> None: + """Validate null type. + + Args: + value: Value to validate + schema: Schema definition + path: JSON path to the value + + Raises: + ValidationError: If validation fails + """ + if value is not None: + raise ValidationError( + path=path, + message=f"Expected null, got {type(value).__name__}" + ) \ No newline at end of file diff --git a/simulations/sim14/validator.py b/simulations/sim14/validator.py new file mode 100644 index 00000000..1a0a081c --- /dev/null +++ b/simulations/sim14/validator.py @@ -0,0 +1,152 @@ +import json +from typing import Any, Dict, List, Optional, Union +from .errors import ValidationError +from .types import validate_string, validate_number, validate_integer, validate_boolean, validate_array, validate_object, validate_null + + +class SchemaValidator: + """Main JSON schema validator class.""" + + def __init__(self, schema: Dict[str, Any]): + """Initialize validator with a JSON schema. + + Args: + schema: JSON schema dictionary + """ + self.schema = schema + + def validate(self, data: Any) -> List[ValidationError]: + """Validate data against the schema. + + Args: + data: Data to validate + + Returns: + List of validation errors (empty if valid) + """ + errors = [] + self._validate_recursive(data, self.schema, "$", errors) + return errors + + def _validate_recursive(self, data: Any, schema: Dict[str, Any], path: str, errors: List[ValidationError]) -> None: + """Recursively validate data against schema. + + Args: + data: Current data being validated + schema: Current schema definition + path: JSON path to current data + errors: List to accumulate errors + """ + # Handle type validation + schema_type = schema.get("type") + if schema_type: + if schema_type == "string": + self._validate_type(validate_string, data, schema, path, errors) + elif schema_type == "number": + self._validate_type(validate_number, data, schema, path, errors) + elif schema_type == "integer": + self._validate_type(validate_integer, data, schema, path, errors) + elif schema_type == "boolean": + self._validate_type(validate_boolean, data, schema, path, errors) + elif schema_type == "array": + self._validate_array(data, schema, path, errors) + elif schema_type == "object": + self._validate_object(data, schema, path, errors) + elif schema_type == "null": + self._validate_type(validate_null, data, schema, path, errors) + else: + errors.append(ValidationError( + path=path, + message=f"Unknown type '{schema_type}' in schema" + )) + + # Handle enum constraint (applies to all types) + if "enum" in schema: + if data not in schema["enum"]: + errors.append(ValidationError( + path=path, + message=f"Value must be one of {schema['enum']}, got {data}" + )) + + def _validate_type(self, validator_func, data: Any, schema: Dict[str, Any], path: str, errors: List[ValidationError]) -> None: + """Validate data using a type-specific validator function. + + Args: + validator_func: Type-specific validation function + data: Data to validate + schema: Schema definition + path: JSON path + errors: List to accumulate errors + """ + try: + validator_func(data, schema, path) + except ValidationError as e: + errors.append(e) + + def _validate_array(self, data: Any, schema: Dict[str, Any], path: str, errors: List[ValidationError]) -> None: + """Validate array data and its items. + + Args: + data: Data to validate + schema: Schema definition + path: JSON path + errors: List to accumulate errors + """ + try: + validate_array(data, schema, path) + except ValidationError as e: + errors.append(e) + return + + # Validate array items if schema is provided + items_schema = schema.get("items") + if items_schema and isinstance(data, list): + for i, item in enumerate(data): + item_path = f"{path}[{i}]" + self._validate_recursive(item, items_schema, item_path, errors) + + def _validate_object(self, data: Any, schema: Dict[str, Any], path: str, errors: List[ValidationError]) -> None: + """Validate object data and its properties. + + Args: + data: Data to validate + schema: Schema definition + path: JSON path + errors: List to accumulate errors + """ + try: + validate_object(data, schema, path) + except ValidationError as e: + errors.append(e) + return + + if not isinstance(data, dict): + return + + # Check required fields + required = schema.get("required", []) + for field in required: + if field not in data: + field_path = f"{path}.{field}" if path != "$" else f"$.{field}" + errors.append(ValidationError( + path=field_path, + message=f"Required field '{field}' is missing" + )) + + # Validate object properties + properties = schema.get("properties", {}) + for field, value in data.items(): + if field in properties: + field_path = f"{path}.{field}" if path != "$" else f"$.{field}" + self._validate_recursive(value, properties[field], field_path, errors) + + def is_valid(self, data: Any) -> bool: + """Check if data is valid against the schema. + + Args: + data: Data to validate + + Returns: + True if valid, False otherwise + """ + return len(self.validate(data)) == 0 \ No newline at end of file diff --git a/simulations/sim17/assertions.py b/simulations/sim17/assertions.py new file mode 100644 index 00000000..6f4b6b1a --- /dev/null +++ b/simulations/sim17/assertions.py @@ -0,0 +1,185 @@ +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}") \ No newline at end of file diff --git a/simulations/sim17/cli.py b/simulations/sim17/cli.py new file mode 100644 index 00000000..30894a51 --- /dev/null +++ b/simulations/sim17/cli.py @@ -0,0 +1,98 @@ +import argparse +import sys +from runner import TestRunner + + +def main(): + """Command-line interface for HTTP API testing framework.""" + parser = argparse.ArgumentParser( + description="HTTP API Testing Framework", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + python cli.py --suite tests.yaml + python cli.py --suite tests.yaml --base-url https://api.example.com + python cli.py --suite tests.yaml --base-url https://api.example.com --verbose + python cli.py --suite tests.yaml --output report.txt + """ + ) + + parser.add_argument( + '--suite', + required=True, + help='Path to YAML test suite file' + ) + + parser.add_argument( + '--base-url', + help='Base URL for API endpoints (can be overridden by suite file)' + ) + + parser.add_argument( + '--verbose', '-v', + action='store_true', + help='Show detailed test results including passed tests' + ) + + parser.add_argument( + '--output', '-o', + help='Save report to specified file' + ) + + parser.add_argument( + '--quiet', '-q', + action='store_true', + help='Suppress console output except for final summary' + ) + + args = parser.parse_args() + + try: + # Initialize test runner + runner = TestRunner(base_url=args.base_url or "") + + if not args.quiet: + print(f"Loading test suite: {args.suite}") + if args.base_url: + print(f"Base URL: {args.base_url}") + print("Starting test execution...") + print() + + # Execute tests + reporter = runner.run_test_suite(args.suite) + + # Generate and display report + if not args.quiet: + print(reporter.generate_report(verbose=args.verbose)) + else: + reporter.print_summary() + + # Save report to file if requested + if args.output: + reporter.save_report(args.output, verbose=args.verbose) + if not args.quiet: + print(f"Report saved to: {args.output}") + + # Exit with appropriate code + failed_tests = len(reporter.get_failed_tests()) + if failed_tests > 0: + sys.exit(1) + else: + sys.exit(0) + + except FileNotFoundError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(2) + except ValueError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(2) + except KeyboardInterrupt: + print("\nTest execution interrupted by user", file=sys.stderr) + sys.exit(130) + except Exception as e: + print(f"Unexpected error: {e}", file=sys.stderr) + sys.exit(1) + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/simulations/sim17/reporter.py b/simulations/sim17/reporter.py new file mode 100644 index 00000000..900c97ce --- /dev/null +++ b/simulations/sim17/reporter.py @@ -0,0 +1,137 @@ +import time +from typing import List, Dict, Any +from dataclasses import dataclass + + +@dataclass +class TestResult: + """Result of a single test case execution.""" + name: str + method: str + url: str + status_code: int + response_time: float + passed: bool + assertion_results: List[Any] # List of AssertionResult objects + error: str = None + + +class TestReporter: + """Generates test execution reports with detailed pass/fail information.""" + + def __init__(self): + self.results: List[TestResult] = [] + self.start_time = None + self.end_time = None + + def start_execution(self): + """Mark the start of test execution.""" + self.start_time = time.time() + + def end_execution(self): + """Mark the end of test execution.""" + self.end_time = time.time() + + def add_result(self, result: TestResult): + """Add a test result to the report.""" + self.results.append(result) + + def generate_report(self, verbose: bool = True) -> str: + """Generate formatted test report. + + Args: + verbose: Whether to include detailed failure information + + Returns: + Formatted report string + """ + report_lines = [] + + # Header + report_lines.append("=" * 80) + report_lines.append("HTTP API TEST RESULTS") + report_lines.append("=" * 80) + report_lines.append("") + + # Summary statistics + total_tests = len(self.results) + passed_tests = sum(1 for r in self.results if r.passed) + failed_tests = total_tests - passed_tests + + execution_time = (self.end_time - self.start_time) if (self.end_time and self.start_time) else 0 + + report_lines.append(f"Total Tests: {total_tests}") + report_lines.append(f"Passed: {passed_tests}") + report_lines.append(f"Failed: {failed_tests}") + report_lines.append(f"Success Rate: {(passed_tests/total_tests*100):.1f}%" if total_tests > 0 else "Success Rate: 0%") + report_lines.append(f"Execution Time: {execution_time:.2f}s") + report_lines.append("") + + # Test results + if verbose: + for i, result in enumerate(self.results, 1): + status_icon = "✓" if result.passed else "✗" + report_lines.append(f"{i}. [{status_icon}] {result.name}") + report_lines.append(f" {result.method} {result.url}") + report_lines.append(f" Status: {result.status_code} | Response Time: {result.response_time:.3f}s") + + if result.error: + report_lines.append(f" ERROR: {result.error}") + + # Show assertion details for failed tests + if not result.passed and result.assertion_results: + report_lines.append(" Assertion Results:") + for assertion in result.assertion_results: + assertion_icon = "✓" if assertion.passed else "✗" + report_lines.append(f" [{assertion_icon}] {assertion.message}") + + report_lines.append("") + else: + # Compact view - only show failed tests + failed_results = [r for r in self.results if not r.passed] + if failed_results: + report_lines.append("FAILED TESTS:") + report_lines.append("-" * 40) + for result in failed_results: + report_lines.append(f"✗ {result.name}") + report_lines.append(f" {result.method} {result.url} -> {result.status_code}") + if result.error: + report_lines.append(f" ERROR: {result.error}") + report_lines.append("") + + # Footer + report_lines.append("=" * 80) + + return "\n".join(report_lines) + + def print_summary(self): + """Print a quick summary to console.""" + total = len(self.results) + passed = sum(1 for r in self.results if r.passed) + failed = total - passed + + print(f"\nTest Summary: {passed}/{total} passed, {failed} failed") + if failed > 0: + print("Failed tests:") + for result in self.results: + if not result.passed: + print(f" - {result.name}: {result.error or 'Assertion failures'}") + + def get_failed_tests(self) -> List[TestResult]: + """Get list of failed test results.""" + return [r for r in self.results if not r.passed] + + def get_passed_tests(self) -> List[TestResult]: + """Get list of passed test results.""" + return [r for r in self.results if r.passed] + + def get_success_rate(self) -> float: + """Get success rate as percentage.""" + if not self.results: + return 0.0 + return (len(self.get_passed_tests()) / len(self.results)) * 100 + + def save_report(self, filename: str, verbose: bool = True): + """Save report to file.""" + with open(filename, 'w') as f: + f.write(self.generate_report(verbose)) \ No newline at end of file diff --git a/simulations/sim17/requirements.txt b/simulations/sim17/requirements.txt new file mode 100644 index 00000000..df74fbd9 --- /dev/null +++ b/simulations/sim17/requirements.txt @@ -0,0 +1,3 @@ +requests>=2.28.0 +pyyaml>=6.0 +jsonpath-ng>=1.5.0 \ No newline at end of file diff --git a/simulations/sim17/runner.py b/simulations/sim17/runner.py new file mode 100644 index 00000000..8e624118 --- /dev/null +++ b/simulations/sim17/runner.py @@ -0,0 +1,186 @@ +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 \ No newline at end of file diff --git a/simulations/sim17/test_suite_example.yaml b/simulations/sim17/test_suite_example.yaml new file mode 100644 index 00000000..87f97c19 --- /dev/null +++ b/simulations/sim17/test_suite_example.yaml @@ -0,0 +1,153 @@ +# Example HTTP API Test Suite +# This file demonstrates the YAML format for defining test cases + +# Suite-level configuration +base_url: "https://jsonplaceholder.typicode.com" + +# Global variables available to all tests +variables: + api_version: "v1" + default_timeout: 30 + +# Test cases executed sequentially +tests: + - name: "Get all posts" + method: GET + url: "/posts" + expected: + status: 200 + body_contains: "userId" + json_path: + - path: "$[0].id" + value: 1 + operator: "equals" + - path: "$" + operator: "greater_than" + value: 0 + extract: + first_post_id: + type: "json_path" + path: "$[0].id" + user_id: + type: "json_path" + path: "$[0].userId" + + - name: "Get specific post using extracted ID" + method: GET + url: "/posts/${first_post_id}" + expected: + status: 200 + json_path: + - path: "$.id" + value: "${first_post_id}" + operator: "equals" + - path: "$.userId" + value: "${user_id}" + operator: "equals" + extract: + post_title: + type: "json_path" + path: "$.title" + + - name: "Create new post" + method: POST + url: "/posts" + headers: + Content-Type: "application/json" + json: + title: "Test Post - ${post_title}" + body: "This is a test post created by the API testing framework" + userId: "${user_id}" + expected: + status: 201 + json_path: + - path: "$.id" + operator: "greater_than" + value: 100 + - path: "$.userId" + value: "${user_id}" + operator: "equals" + extract: + new_post_id: + type: "json_path" + path: "$.id" + + - name: "Update created post" + method: PUT + url: "/posts/${new_post_id}" + headers: + Content-Type: "application/json" + json: + id: "${new_post_id}" + title: "Updated Test Post" + body: "This post has been updated" + userId: "${user_id}" + expected: + status: 200 + json_path: + - path: "$.title" + value: "Updated Test Post" + operator: "equals" + + - name: "Delete created post" + method: DELETE + url: "/posts/${new_post_id}" + expected: + status: 200 + + - name: "Verify post was deleted" + method: GET + url: "/posts/${new_post_id}" + expected: + status: 404 + + - name: "Test with query parameters" + method: GET + url: "/posts" + params: + userId: "${user_id}" + expected: + status: 200 + json_path: + - path: "$[*].userId" + value: "${user_id}" + operator: "equals" + + - name: "Test response time validation" + method: GET + url: "/posts/1" + expected: + status: 200 + max_response_time: 2.0 + json_path: + - path: "$.id" + value: 1 + operator: "equals" + + - name: "Test header validation" + method: GET + url: "/posts/1" + expected: + status: 200 + headers: + Content-Type: "application/json; charset=utf-8" + + - name: "Test multiple body contains assertions" + method: GET + url: "/posts/1" + expected: + status: 200 + body_contains: + - "userId" + - "title" + - "body" + + - name: "Test regex extraction from response" + method: GET + url: "/posts/1" + extract: + content_type: + type: "header" + path: "Content-Type" + expected: + status: 200 \ No newline at end of file diff --git a/simulations/sim17/variables.py b/simulations/sim17/variables.py new file mode 100644 index 00000000..3d4ce6fe --- /dev/null +++ b/simulations/sim17/variables.py @@ -0,0 +1,121 @@ +import re +from typing import Any, Dict, Optional +from jsonpath_ng import parse as jsonpath_parse + + +class VariableContext: + """Manages variable extraction and substitution for request chaining.""" + + def __init__(self): + self.variables: Dict[str, Any] = {} + + def extract_variables(self, response, extractions: Dict[str, Dict[str, str]]) -> None: + """Extract variables from HTTP response based on extraction rules. + + Args: + response: HTTP response object + extractions: Dict mapping variable names to extraction configs + Format: {var_name: {"type": "json_path|header|regex", "path": "..."}} + """ + if not extractions: + return + + for var_name, config in extractions.items(): + extraction_type = config.get("type", "json_path") + path = config.get("path", "") + + try: + if extraction_type == "json_path": + self._extract_from_json(var_name, response, path) + elif extraction_type == "header": + self._extract_from_header(var_name, response, path) + elif extraction_type == "regex": + self._extract_from_regex(var_name, response, path) + except Exception as e: + print(f"Warning: Failed to extract variable '{var_name}': {e}") + + def _extract_from_json(self, var_name: str, response, path: str) -> None: + """Extract variable from JSON response using JSONPath.""" + try: + json_data = response.json() + jsonpath_expr = jsonpath_parse(path) + matches = jsonpath_expr.find(json_data) + if matches: + self.variables[var_name] = matches[0].value + except Exception as e: + raise ValueError(f"JSON path extraction failed: {e}") + + def _extract_from_header(self, var_name: str, response, path: str) -> None: + """Extract variable from response header.""" + header_value = response.headers.get(path) + if header_value is not None: + self.variables[var_name] = header_value + else: + raise ValueError(f"Header '{path}' not found") + + def _extract_from_regex(self, var_name: str, response, path: str) -> None: + """Extract variable from response text using regex.""" + match = re.search(path, response.text) + if match: + self.variables[var_name] = match.group(1) if match.groups() else match.group(0) + else: + raise ValueError(f"Regex pattern '{path}' not found") + + def substitute_variables(self, text: str) -> str: + """Replace variable placeholders in text with actual values. + + Args: + text: String containing variable placeholders like ${variable_name} + + Returns: + String with variables substituted + """ + if not isinstance(text, str): + return text + + def replace_var(match): + var_name = match.group(1) + if var_name in self.variables: + return str(self.variables[var_name]) + return match.group(0) # Return original if variable not found + + return re.sub(r'\$\{([^}]+)\}', replace_var, text) + + def substitute_in_dict(self, data: Dict[str, Any]) -> Dict[str, Any]: + """Recursively substitute variables in dictionary values.""" + if not isinstance(data, dict): + return data + + result = {} + for key, value in data.items(): + if isinstance(value, str): + result[key] = self.substitute_variables(value) + elif isinstance(value, dict): + result[key] = self.substitute_in_dict(value) + elif isinstance(value, list): + result[key] = self.substitute_in_list(value) + else: + result[key] = value + return result + + def substitute_in_list(self, data: list) -> list: + """Recursively substitute variables in list items.""" + result = [] + for item in data: + if isinstance(item, str): + result.append(self.substitute_variables(item)) + elif isinstance(item, dict): + result.append(self.substitute_in_dict(item)) + elif isinstance(item, list): + result.append(self.substitute_in_list(item)) + else: + result.append(item) + return result + + def get_variable(self, name: str) -> Optional[Any]: + """Get variable value by name.""" + return self.variables.get(name) + + def set_variable(self, name: str, value: Any) -> None: + """Set variable value.""" + self.variables[name] = value \ No newline at end of file