forked from cleveragents/cleveragents-core
aditya
2f69358dcf
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.
121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
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 |