forked from cleveragents/cleveragents-core
test: sim12, sim14, sim17 — verify multi-file DoD compliance after prompt fix
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.
This commit is contained in:
@@ -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).
|
||||
@@ -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).
|
||||
@@ -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).
|
||||
@@ -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"
|
||||
]
|
||||
@@ -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
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -0,0 +1 @@
|
||||
flask>=2.0.0
|
||||
@@ -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()
|
||||
@@ -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"
|
||||
]
|
||||
@@ -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()
|
||||
@@ -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
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
# No external dependencies required
|
||||
# This JSON schema validator uses only Python standard library
|
||||
@@ -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__}"
|
||||
)
|
||||
@@ -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
|
||||
@@ -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}")
|
||||
@@ -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()
|
||||
@@ -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))
|
||||
@@ -0,0 +1,3 @@
|
||||
requests>=2.28.0
|
||||
pyyaml>=6.0
|
||||
jsonpath-ng>=1.5.0
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user