happyflow-generator

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
• Enhanced execution: Parallel execution, detailed reporting, connection pooling
• Advanced testing: File upload support, response schema validation, rate limiting handling
• Modular architecture: Well-organized codebase with proper error handling

Complete Workflow

Phase 1: Authentication Setup

Execute this code to prepare authentication headers:

import base64
import requests
from typing import Dict, Any

def setup_authentication(auth_type: str, credentials: Dict[str, Any]) -> Dict[str, str]:
    """Prepare authentication headers based on auth type"""

    if auth_type == "bearer":
        return {"Authorization": f"Bearer {credentials['token']}"}

    elif auth_type == "api_key":
        header_name = credentials.get('header_name', 'X-API-Key')
        return {header_name: credentials['api_key']}

    elif auth_type == "basic":
        auth_string = f"{credentials['username']}:{credentials['password']}"
        encoded = base64.b64encode(auth_string.encode()).decode()
        return {"Authorization": f"Basic {encoded}"}

    elif auth_type == "oauth2_client_credentials":
        token_url = credentials['token_url']
        data = {
            'grant_type': 'client_credentials',
            'client_id': credentials['client_id'],
            'client_secret': credentials['client_secret']
        }
        if 'scopes' in credentials:
            data['scope'] = ' '.join(credentials['scopes'])

        response = requests.post(token_url, data=data)
        response.raise_for_status()
        token_data = response.json()

        return {"Authorization": f"Bearer {token_data['access_token']}"}

    return {}

# Example usage:
# auth_headers = setup_authentication("bearer", {"token": "abc123"})

---

Phase 2: Specification Parsing

Execute this code to parse API specifications (OpenAPI or GraphQL):

import requests
import yaml
import json
import re
from typing import Dict, List, Any, Union
from pathlib import Path

def parse_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 specified
    if spec_type == "auto":
        if isinstance(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 type
    if 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}")

def parse_openapi_spec(spec_source: Union[str, Path], headers: Dict[str, str] = None) -> Dict[str, Any]:
    """Parse OpenAPI specification and extract structured information"""

    # Fetch spec
    if isinstance(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:
        with open(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 not in 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 = None
            if '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', {})
    }

def parse_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 if isinstance(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 import List, Dict, Any

def analyze_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 dependencies
    for 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 not in 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 not in dependencies[endpoint_id]:
                        dependencies[endpoint_id].append(other_id)

    # Topological sort
    def topological_sort(deps):
        in_degree = {node: 0 for 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] -= 1
                    if in_degree[other_node] == 0:
                        queue.append(other_node)

        return result

    execution_order_ids = topological_sort(dependencies)

    execution_plan = []
    for step, endpoint_id in enumerate(execution_order_ids, 1):
        endpoint = next(e for e in endpoints if f"{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
    }

def identify_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 if not 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'] not in processed_steps]
    
    # Simple grouping by dependency sets
    dependency_map = {}
    for step in remaining_steps:
        dep_tuple = tuple(sorted(step['dependencies']))
        if dep_tuple not in 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 import Dict, List, Any
from jsonschema import validate, ValidationError

def generate_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)
        return max(minimum, 1)  # Ensure positive for IDs
    elif schema_type == 'number':
        return 10.5
    elif schema_type == 'boolean':
        return True
    elif 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', []) or not schema.get('required'):
                obj[prop] = generate_value_from_schema(prop_schema, prop)
        return obj

    return None

def generate_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:
            if len(group) > 1:  # Only create parallel execution for groups with multiple steps
                for 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 groups
    if parallel_execution and parallel_groups:
        executed_steps = set()
        for i, group in enumerate(parallel_groups):
            if len(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 not in executed_steps:
                    lines.append(f'            self.step_{step_num}()')
                    executed_steps.add(step_num)
        
        # Execute any remaining steps not covered by groups
        for step_info in execution_plan:
            step_num = step_info['step']
            if step_num not in executed_steps:
                lines.append(f'            self.step_{step_num}()')
    else:
        # Sequential execution
        for 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 steps
    for 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('

<!-- Content truncated for initial SEO render. Open the source file tab for the full file. -->