Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Automatically generate and execute Python test scripts from OpenAPI specifications and GraphQL schemas with enhanced features
HappyFlow Generator Skill
Metadata
Skill Name: HappyFlow Generator
Version: 2.0.0
Category: API Testing & Automation
Required Capabilities: Code execution, web requests, file operations
Estimated Duration: 2-5 minutes per API spec
Difficulty: Intermediate
Description
Automatically generate and execute Python test scripts from OpenAPI specifications and GraphQL schemas that successfully call all API endpoints in dependency-correct order, ensuring all requests return 2xx status codes.
Input: OpenAPI/GraphQL spec (URL/file) + authentication credentials Output: Working Python script that executes complete API happy path flow
Key Features:
Multi-format support: OpenAPI 3.0+ and GraphQL schemas
Execute this code to parse API specifications (OpenAPI or GraphQL):
import requests
import yaml
import json
import re
from typing importDict, List, Any, Unionfrom pathlib import Path
defparse_specification(spec_source: Union[str, Path], spec_type: str = "auto", **kwargs) -> Dict[str, Any]:
"""Parse API specification and extract structured information
Args:
spec_source: Path or URL to API specification
spec_type: Type of specification ('openapi', 'graphql', or 'auto')
**kwargs: Additional arguments for specific parsers
Returns:
Dictionary containing parsed specification data
"""# Auto-detect specification type if not specifiedif spec_type == "auto":
ifisinstance(spec_source, str):
if spec_source.endswith(".graphql") or"graphql"in spec_source.lower():
spec_type = "graphql"else:
spec_type = "openapi"else:
# For file paths, check extension
path = Path(spec_source)
if path.suffix.lower() in [".graphql", ".gql"]:
spec_type = "graphql"else:
spec_type = "openapi"# Parse based on detected typeif spec_type == "openapi":
return parse_openapi_spec(spec_source, **kwargs)
elif spec_type == "graphql":
return parse_graphql_spec(spec_source, **kwargs)
else:
raise ValueError(f"Unsupported specification type: {spec_type}")
defparse_openapi_spec(spec_source: Union[str, Path], headers: Dict[str, str] = None) -> Dict[str, Any]:
"""Parse OpenAPI specification and extract structured information"""# Fetch specifisinstance(spec_source, str) and spec_source.startswith('http'):
response = requests.get(spec_source, headers=headers or {})
response.raise_for_status()
content = response.text
try:
spec = json.loads(content)
except json.JSONDecodeError:
spec = yaml.safe_load(content)
else:
withopen(spec_source, 'r') as f:
content = f.read()
try:
spec = json.loads(content)
except json.JSONDecodeError:
spec = yaml.safe_load(content)
# Extract base information
openapi_version = spec.get('openapi', spec.get('swagger', 'unknown'))
base_url = ""if'servers'in spec and spec['servers']:
base_url = spec['servers'][0]['url']
elif'host'in spec:
scheme = spec.get('schemes', ['https'])[0]
base_path = spec.get('basePath', '')
base_url = f"{scheme}://{spec['host']}{base_path}"# Extract endpoints
endpoints = []
paths = spec.get('paths', {})
for path, path_item in paths.items():
for method in ['get', 'post', 'put', 'patch', 'delete']:
if method notin path_item:
continue
operation = path_item[method]
# Extract parameters
parameters = []
for param in operation.get('parameters', []):
parameters.append({
'name': param.get('name'),
'in': param.get('in'),
'required': param.get('required', False),
'schema': param.get('schema', {}),
'example': param.get('example')
})
# Extract request body
request_body = Noneif'requestBody'in operation:
rb = operation['requestBody']
content = rb.get('content', {})
if'application/json'in content:
json_content = content['application/json']
request_body = {
'required': rb.get('required', False),
'content_type': 'application/json',
'schema': json_content.get('schema', {}),
'example': json_content.get('example')
}
elif'multipart/form-data'in content:
form_content = content['multipart/form-data']
request_body = {
'required': rb.get('required', False),
'content_type': 'multipart/form-data',
'schema': form_content.get('schema', {}),
'example': form_content.get('example')
}
# Extract responses
responses = {}
for status_code, response_data in operation.get('responses', {}).items():
if status_code.startswith('2'):
content = response_data.get('content', {})
if'application/json'in content:
json_content = content['application/json']
responses[status_code] = {
'description': response_data.get('description', ''),
'schema': json_content.get('schema', {}),
'example': json_content.get('example')
}
endpoint = {
'operation_id': operation.get('operationId', f"{method}_{path}"),
'path': path,
'method': method.upper(),
'tags': operation.get('tags', []),
'summary': operation.get('summary', ''),
'parameters': parameters,
'request_body': request_body,
'responses': responses
}
endpoints.append(endpoint)
return {
'openapi_version': openapi_version,
'base_url': base_url,
'endpoints': endpoints,
'schemas': spec.get('components', {}).get('schemas', {})
}
defparse_graphql_spec(spec_source: str, headers: Dict[str, str] = None) -> Dict[str, Any]:
"""Parse GraphQL schema and extract operations"""# For GraphQL, we'll create a simplified representation# In practice, this would use graphql-core to parse the schema
base_url = spec_source ifisinstance(spec_source, str) and spec_source.startswith('http') else""# Placeholder for GraphQL endpoints - in reality, this would be derived from schema introspection
endpoints = [
{
'operation_id': 'graphql_query',
'path': '/graphql',
'method': 'POST',
'tags': ['GraphQL'],
'summary': 'GraphQL Query',
'parameters': [],
'request_body': {
'required': True,
'content_type': 'application/json',
'schema': {},
'example': {'query': 'query { __schema { types { name } } }'}
},
'responses': {
'200': {
'description': 'Successful GraphQL response',
'schema': {},
'example': {}
}
}
}
]
return {
'spec_type': 'graphql',
'base_url': base_url,
'endpoints': endpoints,
'schemas': {}
}
# Example usage:# parsed_spec = parse_specification("https://api.example.com/openapi.json")# parsed_spec = parse_specification("https://api.example.com/graphql", spec_type="graphql")
Phase 3: Dependency Analysis
Execute this code to analyze dependencies and determine execution order:
import re
from typing importList, Dict, Anydefanalyze_dependencies(endpoints: List[Dict]) -> Dict[str, Any]:
"""Analyze endpoint dependencies and create execution order"""
dependencies = {}
outputs = {}
for endpoint in endpoints:
endpoint_id = f"{endpoint['method']}{endpoint['path']}"
dependencies[endpoint_id] = []
outputs[endpoint_id] = {}
# Detect path parameter dependenciesfor endpoint in endpoints:
endpoint_id = f"{endpoint['method']}{endpoint['path']}"
path = endpoint['path']
path_params = re.findall(r'\{(\w+)\}', path)
for param in path_params:
for other_endpoint in endpoints:
other_id = f"{other_endpoint['method']}{other_endpoint['path']}"if other_endpoint['method'] in ['POST', 'PUT']:
for status, response in other_endpoint.get('responses', {}).items():
schema = response.get('schema', {})
properties = schema.get('properties', {})
if'id'in properties or param in properties:
if other_id != endpoint_id and other_id notin dependencies[endpoint_id]:
dependencies[endpoint_id].append(other_id)
output_field = 'id'if'id'in properties else param
outputs[other_id][param] = f"response.body.{output_field}"# HTTP method ordering
method_priority = {'POST': 1, 'GET': 2, 'PUT': 3, 'PATCH': 3, 'DELETE': 4}
for endpoint in endpoints:
endpoint_id = f"{endpoint['method']}{endpoint['path']}"
path_clean = re.sub(r'\{[^}]+\}', '', endpoint['path'])
for other_endpoint in endpoints:
other_id = f"{other_endpoint['method']}{other_endpoint['path']}"
other_path_clean = re.sub(r'\{[^}]+\}', '', other_endpoint['path'])
if path_clean == other_path_clean:
if method_priority.get(endpoint['method'], 5) > method_priority.get(other_endpoint['method'], 5):
if other_id notin dependencies[endpoint_id]:
dependencies[endpoint_id].append(other_id)
# Topological sortdeftopological_sort(deps):
in_degree = {node: 0for node in deps}
for node in deps:
for dep in deps[node]:
in_degree[dep] = in_degree.get(dep, 0) + 1
queue = [node for node in deps if in_degree[node] == 0]
result = []
while queue:
queue.sort(key=lambda x: (x.split()[1].count('/'), method_priority.get(x.split()[0], 5)))
node = queue.pop(0)
result.append(node)
for other_node in deps:
if node in deps[other_node]:
in_degree[other_node] -= 1if in_degree[other_node] == 0:
queue.append(other_node)
return result
execution_order_ids = topological_sort(dependencies)
execution_plan = []
for step, endpoint_id inenumerate(execution_order_ids, 1):
endpoint = next(e for e in endpoints iff"{e['method']}{e['path']}" == endpoint_id)
inputs = {}
for dep_id in dependencies[endpoint_id]:
if dep_id in outputs:
for param_name, json_path in outputs[dep_id].items():
dep_step = execution_order_ids.index(dep_id) + 1
inputs[param_name] = {
'source': f"step_{dep_step}",
'json_path': json_path
}
execution_plan.append({
'step': step,
'endpoint': endpoint,
'dependencies': dependencies[endpoint_id],
'inputs': inputs,
'outputs': outputs[endpoint_id]
})
return {
'execution_order': execution_plan,
'dependency_graph': dependencies
}
defidentify_parallel_groups(execution_plan: List[Dict]) -> List[List[int]]:
"""Identify groups of steps that can be executed in parallel"""# Group steps by their dependencies
parallel_groups = []
processed_steps = set()
# Find steps with no dependencies (can run in parallel)
independent_steps = [step['step'] for step in execution_plan ifnot step['dependencies']]
if independent_steps:
parallel_groups.append(independent_steps)
processed_steps.update(independent_steps)
# For remaining steps, group those with the same dependencies
remaining_steps = [step for step in execution_plan if step['step'] notin processed_steps]
# Simple grouping by dependency sets
dependency_map = {}
for step in remaining_steps:
dep_tuple = tuple(sorted(step['dependencies']))
if dep_tuple notin dependency_map:
dependency_map[dep_tuple] = []
dependency_map[dep_tuple].append(step['step'])
for group in dependency_map.values():
parallel_groups.append(group)
return parallel_groups
# Example usage:# dependency_analysis = analyze_dependencies(parsed_spec['endpoints'])# parallel_groups = identify_parallel_groups(dependency_analysis['execution_order'])
Phase 4: Script Generation
Execute this code to generate the Python test script:
import json
import time
from typing importDict, List, Anyfrom jsonschema import validate, ValidationError
defgenerate_value_from_schema(schema: Dict, field_name: str = "") -> Any:
"""Generate example value based on schema"""if'example'in schema:
return schema['example']
if'default'in schema:
return schema['default']
if'enum'in schema:
return schema['enum'][0]
schema_type = schema.get('type', 'string')
if schema_type == 'string':
if schema.get('format') == 'email':
return'test@example.com'elif schema.get('format') == 'uuid':
return'550e8400-e29b-41d4-a716-446655440000'elif'email'in field_name.lower():
return'test@example.com'elif'name'in field_name.lower():
return'Test User'elif'description'in field_name.lower():
return'Test description'return'test_value'elif schema_type == 'integer':
minimum = schema.get('minimum', 1)
maximum = schema.get('maximum', minimum + 100)
returnmax(minimum, 1) # Ensure positive for IDselif schema_type == 'number':
return10.5elif schema_type == 'boolean':
returnTrueelif schema_type == 'array':
items_schema = schema.get('items', {})
return [generate_value_from_schema(items_schema)]
elif schema_type == 'object':
obj = {}
for prop, prop_schema in schema.get('properties', {}).items():
if prop in schema.get('required', []) ornot schema.get('required'):
obj[prop] = generate_value_from_schema(prop_schema, prop)
return obj
returnNonedefgenerate_python_script(
execution_plan: List[Dict],
base_url: str,
auth_headers: Dict,
parallel_execution: bool = False,
parallel_groups: List[List[int]] = None) -> str:
"""Generate complete Python script"""
lines = []
# Header
lines.append('#!/usr/bin/env python3')
lines.append('"""HappyFlow Generator - Auto-generated API test script"""')
lines.append('')
lines.append('import requests')
lines.append('import json')
lines.append('import sys')
lines.append('import time')
lines.append('from datetime import datetime')
if parallel_execution:
lines.append('from concurrent.futures import ThreadPoolExecutor, as_completed')
lines.append('from jsonschema import validate, ValidationError')
lines.append('')
# Class
lines.append('class APIFlowExecutor:')
lines.append(' def __init__(self, base_url, auth_headers):')
lines.append(' self.base_url = base_url.rstrip("/")')
lines.append(' self.session = requests.Session()')
lines.append(' self.session.headers.update(auth_headers)')
lines.append(' self.context = {}')
lines.append(' self.results = []')
lines.append('')
lines.append(' def log(self, message, level="INFO"):')
lines.append(' print(f"[{datetime.utcnow().isoformat()}] [{level}] {message}")')
lines.append('')
lines.append(' def _make_request(self, method, url, **kwargs):')
lines.append(' """Make HTTP request with retry logic for rate limiting"""')
lines.append(' max_retries = 3')
lines.append(' for attempt in range(max_retries):')
lines.append(' try:')
lines.append(' response = self.session.request(method, url, **kwargs)')
lines.append(' # Handle rate limiting')
lines.append(' if response.status_code == 429:')
lines.append(' if attempt < max_retries - 1:')
lines.append(' delay = 2 ** attempt # Exponential backoff')
lines.append(' self.log(f"Rate limited. Waiting {delay}s before retry...", "WARN")')
lines.append(' time.sleep(delay)')
lines.append(' continue')
lines.append(' return response')
lines.append(' except Exception as e:')
lines.append(' if attempt < max_retries - 1:')
lines.append(' delay = 2 ** attempt')
lines.append(' self.log(f"Request failed: {e}. Retrying in {delay}s...", "WARN")')
lines.append(' time.sleep(delay)')
lines.append(' else:')
lines.append(' raise')
lines.append('')
if parallel_execution and parallel_groups:
lines.append(' def execute_parallel_group(self, step_numbers):')
lines.append(' """Execute a group of steps in parallel"""')
lines.append(' with ThreadPoolExecutor(max_workers=5) as executor:')
lines.append(' future_to_step = {')
for group in parallel_groups:
iflen(group) > 1: # Only create parallel execution for groups with multiple stepsfor step_num in group:
lines.append(f' executor.submit(self.step_{step_num}): {step_num},')
break
lines.append(' }')
lines.append(' ')
lines.append(' for future in as_completed(future_to_step):')
lines.append(' step_num = future_to_step[future]')
lines.append(' try:')
lines.append(' future.result()')
lines.append(' self.log(f"Step {step_num} completed successfully")')
lines.append(' except Exception as e:')
lines.append(' self.log(f"Step {step_num} failed: {e}", "ERROR")')
lines.append(' raise')
lines.append('')
lines.append(' def execute_flow(self):')
lines.append(' try:')
# If parallel execution is enabled, organize steps by groupsif parallel_execution and parallel_groups:
executed_steps = set()
for i, group inenumerate(parallel_groups):
iflen(group) > 1:
# Parallel group
lines.append(f' # Parallel Group {i+1}')
lines.append(f' self.log("Executing parallel group: {group}")')
lines.append(f' self.execute_parallel_group({group})')
executed_steps.update(group)
else:
# Sequential step
step_num = group[0]
if step_num notin executed_steps:
lines.append(f' self.step_{step_num}()')
executed_steps.add(step_num)
# Execute any remaining steps not covered by groupsfor step_info in execution_plan:
step_num = step_info['step']
if step_num notin executed_steps:
lines.append(f' self.step_{step_num}()')
else:
# Sequential executionfor step_info in execution_plan:
lines.append(f' self.step_{step_info["step"]}()')
lines.append(' self.log("✓ All requests completed", "SUCCESS")')
lines.append(' return True')
lines.append(' except Exception as e:')
lines.append(' self.log(f"✗ Failed: {e}", "ERROR")')
lines.append(' return False')
lines.append('')
# Generate stepsfor step_info in execution_plan:
endpoint = step_info['endpoint']
step_num = step_info['step']
method = endpoint['method']
path = endpoint['path']
lines.append(f' def step_{step_num}(self):')
lines.append(f' """Step {step_num}: {method}{path}"""')
lines.append(f' self.log("Step {step_num}: {method}{path}")')
# Initialize tracking variables
lines.append(' # Initialize tracking variables')
lines.append(' start_time = time.time()')
lines.append(' request_details = {')
lines.append(' "method": "%s",' % method)
lines.append(' "url": None,')
lines.append(' "headers": dict(self.session.headers),')
lines.append(' "payload": None')
lines.append(' }')
lines.append(' response_details = {')
lines.append(' "status_code": None,')
lines.append(' "headers": None,')
lines.append(' "body": None,')
lines.append(' "elapsed": None')
lines.append(' }')
lines.append(' error_details = None')
lines.append('')
lines.append(' try:')
# Build URL
url_expr = f'f"{{self.base_url}}{path}"'# Replace path parametersif'{'in path:
for param in re.findall(r'\{(\w+)\}', path):
url_expr = url_expr.replace(f'{{{param}}}', f'{{self.context.get("{param}", "UNKNOWN_{param}")}}')
lines.append(f' # Build URL with path parameters')
lines.append(f' url = {url_expr}')
lines.append(' request_details["url"] = url')
lines.append('')
# Handle request bodyif endpoint.get('request_body'):
schema = endpoint['request_body'].get('schema', {})
example = endpoint['request_body'].get('example')
content_type = endpoint['request_body'].get('content_type', 'application/json')
if example:
payload = example
else:
payload = generate_value_from_schema(schema)
lines.append(f' # Handle request body ({content_type})')
if content_type == 'multipart/form-data':
lines.append(' # Handle file uploads')
lines.append(' files = {}')
lines.append(f' payload = {json.dumps(payload) if payload else {}}')
lines.append(' request_details["payload"] = payload')
lines.append(' response = self._make_request("%s", url, data=payload, files=files)' % method.lower())
else:
lines.append(f' payload = {json.dumps(payload) if payload else {}}')
lines.append(' request_details["payload"] = payload')
lines.append(' response = self._make_request("%s", url, json=payload)' % method.lower())
else:
lines.append(' # No request body')
lines.append(' response = self._make_request("%s", url)' % method.lower())
lines.append(' self.log(f"Status: {response.status_code}")')
lines.append(' if response.status_code not in [200, 201, 202, 204]:')
lines.append(' raise Exception(f"Unexpected status code: {response.status_code}")')
# Process response
lines.append(' if response.text:')
lines.append(' try:')
lines.append(' data = response.json()')
# Add response validation if schema exists
success_response = Nonefor status_code, resp_data in endpoint.get('responses', {}).items():
if status_code.startswith('2'):
success_response = resp_data
breakif success_response and success_response.get('schema'):
schema = success_response['schema']
lines.append(' # Validate response against schema')
lines.append(' schema = %s' % json.dumps(schema))
lines.append(' try:')
lines.append(' validate(instance=data, schema=schema)')
lines.append(' self.log("Response validated successfully against schema")')
lines.append(' except ValidationError as e:')
lines.append(' self.log(f"Response validation failed: {e.message}", "ERROR")')
lines.append(' self.log(f"Validation path: {\' -> \'.join(str(x) for x in e.absolute_path)}", "ERROR")')
# Extract outputsif step_info['outputs']:
for output_name, json_path in step_info['outputs'].items():
field = json_path.split('.')[-1]
lines.append(f' self.context["{output_name}"] = data.get("{field}")')
lines.append(' except ValueError:')
lines.append(' self.log("Warning: Response is not valid JSON", "WARN")')
# Calculate execution time
lines.append('')
lines.append(' # Calculate execution time')
lines.append(' end_time = time.time()')
lines.append(' elapsed_time = end_time - start_time')
lines.append('')
# Capture response details
lines.append(' # Capture response details')
lines.append(' response_details.update({')
lines.append(' "status_code": response.status_code,')
lines.append(' "headers": dict(response.headers),')
lines.append(' "body": response.text[:1000] if response.text else "",')
lines.append(' "elapsed": elapsed_time')
lines.append(' })')
lines.append('')
lines.append(' except Exception as e:')
lines.append(' error_details = str(e)')
lines.append(' self.log(f"Error processing response: {e}", "ERROR")')
lines.append(' # Still capture timing info even on error')
lines.append(' end_time = time.time()')
lines.append(' elapsed_time = end_time - start_time if "start_time" in locals() else 0')
lines.append(' # Capture partial response details if available')
lines.append(' if "response" in locals():')
lines.append(' response_details.update({')
lines.append(' "status_code": getattr(response, "status_code", None),')
lines.append(' "headers": dict(getattr(response, "headers", {})),')
lines.append(' "body": getattr(response, "text", "")[:1000] if getattr(response, "text", "") else "",')
lines.append(' "elapsed": elapsed_time')
lines.append(' })')
lines.append(' raise')
lines.append('')
# Store detailed results
lines.append(' # Store detailed results')
lines.append(' result_entry = {')
lines.append(' "step": %d,' % step_num)
lines.append(' "status": response.status_code if "response" in locals() else None,')
lines.append(' "method": "%s",' % method)
lines.append(' "path": "%s",' % path)
lines.append(' "elapsed_time": elapsed_time,')
lines.append(' "request": request_details,')
lines.append(' "response": response_details,')
lines.append(' "error": error_details')
lines.append(' }')
lines.append(' self.results.append(result_entry)')
lines.append('')
# Summary methods
lines.append(' def print_summary(self):')
lines.append(' print("\\n" + "="*60)')
lines.append(' print("EXECUTION SUMMARY")')
lines.append(' print("="*60)')
lines.append(' for r in self.results:')
lines.append(' print(f"✓ Step {r[\'step\']}: {r[\'method\']} {r[\'path\']} - {r[\'status\']} ({r[\'elapsed_time\']:.3f}s)")')
lines.append(' print("="*60)')
lines.append('')
lines.append(' def print_detailed_report(self):')
lines.append(' """Print detailed execution report with metrics"""')
lines.append(' print("\\n" + "="*80)')
lines.append(' print("DETAILED EXECUTION REPORT")')
lines.append(' print("="*80)')
lines.append(' ')
lines.append(' total_time = 0')
lines.append(' successful_steps = 0')
lines.append(' failed_steps = 0')
lines.append(' ')
lines.append(' for r in self.results:')
lines.append(' print(f"\\n--- Step {r[\'step\']}: {r[\'method\']} {r[\'path\']} ---")')
lines.append(' print(f" Status: {r[\'status\']}")')
lines.append(' print(f" Elapsed Time: {r[\'elapsed_time\']:.3f}s")')
lines.append(' ')
lines.append(' if r[\'error\'] is not None:')
lines.append(' print(f" Error: {r[\'error\']}")')
lines.append(' failed_steps += 1')
lines.append(' else:')
lines.append(' successful_steps += 1')
lines.append(' ')
lines.append(' # Request details')
lines.append(' req = r[\'request\']')
lines.append(' if req[\'payload\'] is not None:')
lines.append(' print(f" Request Payload: {req[\'payload\']}")')
lines.append(' ')
lines.append(' # Response details')
lines.append(' resp = r[\'response\']')
lines.append(' if resp[\'headers\'] is not None:')
lines.append(' content_type = resp[\'headers\'].get(\'Content-Type\', \'Unknown\')')
lines.append(' print(f" Content-Type: {content_type}")')
lines.append(' ')
lines.append(' total_time += r[\'elapsed_time\']')
lines.append(' ')
lines.append(' print("\\n" + "-"*80)')
lines.append(' print("SUMMARY STATISTICS")')
lines.append(' print("-"*80)')
lines.append(' print(f" Total Steps: {len(self.results)}")')
lines.append(' print(f" Successful: {successful_steps}")')
lines.append(' print(f" Failed: {failed_steps}")')
lines.append(' print(f" Total Execution Time: {total_time:.3f}s")')
lines.append(' if len(self.results) > 0:')
lines.append(' avg_time = total_time / len(self.results)')
lines.append(' print(f" Average Time per Step: {avg_time:.3f}s")')
lines.append(' print("="*80)')
lines.append('')
# Main
lines.append('def main():')
lines.append(f' BASE_URL = "{base_url}"')
lines.append(f' AUTH_HEADERS = {json.dumps(auth_headers)}')
lines.append(' executor = APIFlowExecutor(BASE_URL, AUTH_HEADERS)')
lines.append(' success = executor.execute_flow()')
lines.append(' executor.print_summary()')
lines.append(' # Check if DETAILED_REPORT environment variable is set')
lines.append(' import os')
lines.append(' if os.environ.get("DETAILED_REPORT", "").lower() == "true":')
lines.append(' executor.print_detailed_report()')
lines.append(' sys.exit(0 if success else 1)')
lines.append('')
lines.append('if __name__ == "__main__":')
lines.append(' main()')
return'\n'.join(lines)
# Example usage:# script = generate_python_script(dependency_analysis['execution_order'], base_url, auth_headers)# script = generate_python_script(dependency_analysis['execution_order'], base_url, auth_headers, parallel_execution=True, parallel_groups=parallel_groups)
Phase 5: Execute and Iterate
Execute this code to run the script and fix errors:
import subprocess
import tempfile
import os
import re
defexecute_script_with_retries(script_content: str, max_retries: int = 5, detailed_reporting: bool = False):
"""Execute script and retry with fixes"""for attempt inrange(1, max_retries + 1):
print(f"\n=== Attempt {attempt}/{max_retries} ===")
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(script_content)
script_path = f.name
try:
# Set environment for detailed reporting if requested
env = os.environ.copy()
if detailed_reporting:
env["DETAILED_REPORT"] = "true"
result = subprocess.run(
['python', script_path],
capture_output=True,
text=True,
timeout=300,
env=env
)
print(result.stdout)
if result.returncode == 0:
print("\n✓ SUCCESS! All requests returned 2xx")
return {
'success': True,
'script': script_content,
'attempts': attempt
}
# Analyze errors and apply fixesprint(f"✗ Exit code: {result.returncode}")
# Simple fix patternsif'400'in result.stdout and'missing required field'in result.stdout:
# Add missing fields
field_match = re.search(r"field '(\w+)'", result.stdout)
if field_match:
field = field_match.group(1)
script_content = script_content.replace(
'payload = {',
f'payload = {{"{field}": "test_value", '
)
print(f"Applied fix: Added missing field '{field}'")
continueif'422'in result.stdout:
# Adjust constraint violations
script_content = script_content.replace('"quantity": 0', '"quantity": 1')
script_content = script_content.replace('"age": 0', '"age": 18')
print("Applied fix: Adjusted values to meet constraints")
continuebreakexcept subprocess.TimeoutExpired:
print("✗ Script execution timed out")
breakexcept Exception as e:
print(f"✗ Execution error: {e}")
breakfinally:
if os.path.exists(script_path):
os.unlink(script_path)
return {
'success': False,
'script': script_content,
'attempts': max_retries
}
# Example usage:# result = execute_script_with_retries(generated_script)# result = execute_script_with_retries(generated_script, detailed_reporting=True)
