happyflow-generator

name	happyflow-generator
description	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
• 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"
            :
                spec_type = 

    
     spec_type == :
         parse_openapi_spec(spec_source, **kwargs)
     spec_type == :
         parse_graphql_spec(spec_source, **kwargs)
    :
         ValueError()

 () -> [, ]:
    

    
     (spec_source, )  spec_source.startswith():
        response = requests.get(spec_source, headers=headers  {})
        response.raise_for_status()
        content = response.text
        :
            spec = json.loads(content)
         json.JSONDecodeError:
            spec = yaml.safe_load(content)
    :
         (spec_source, )  f:
            content = f.read()
            :
                spec = json.loads(content)
             json.JSONDecodeError:
                spec = yaml.safe_load(content)

    
    openapi_version = spec.get(, spec.get(, ))
    base_url = 

       spec  spec[]:
        base_url = spec[][][]
       spec:
        scheme = spec.get(, [])[]
        base_path = spec.get(, )
        base_url = 

    
    endpoints = []
    paths = spec.get(, {})

     path, path_item  paths.items():
         method  [, , , , ]:
             method   path_item:
                

            operation = path_item[method]

            
            parameters = []
             param  operation.get(, []):
                parameters.append({
                    : param.get(),
                    : param.get(),
                    : param.get(, ),
                    : param.get(, {}),
                    : param.get()
                })

            
            request_body = 
               operation:
                rb = operation[]
                content = rb.get(, {})

                   content:
                    json_content = content[]
                    request_body = {
                        : rb.get(, ),
                        : ,
                        : json_content.get(, {}),
                        : json_content.get()
                    }
                   content:
                    form_content = content[]
                    request_body = {
                        : rb.get(, ),
                        : ,
                        : form_content.get(, {}),
                        : form_content.get()
                    }

            
            responses = {}
             status_code, response_data  operation.get(, {}).items():
                 status_code.startswith():
                    content = response_data.get(, {})
                       content:
                        json_content = content[]
                        responses[status_code] = {
                            : response_data.get(, ),
                            : json_content.get(, {}),
                            : json_content.get()
                        }

            endpoint = {
                : operation.get(, ),
                : path,
                : method.upper(),
                : operation.get(, []),
                : operation.get(, ),
                : parameters,
                : request_body,
                : responses
            }

            endpoints.append(endpoint)

     {
        : openapi_version,
        : base_url,
        : endpoints,
        : spec.get(, {}).get(, {})
    }

 () -> [, ]:
    
    
    
    
    
    base_url = spec_source  (spec_source, )  spec_source.startswith()  
    
    
    endpoints = [
        {
            : ,
            : ,
            : ,
            : [],
            : ,
            : [],
            : {
                : ,
                : ,
                : {},
                : {: }
            },
            : {
                : {
                    : ,
                    : {},
                    : {}
                }
            }
        }
    ]
    
     {
        : ,
        : base_url,
        : endpoints,
        : {}
    }

---

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(, {})
                        properties = schema.get(, {})

                           properties  param  properties:
                             other_id != endpoint_id  other_id   dependencies[endpoint_id]:
                                dependencies[endpoint_id].append(other_id)
                                output_field =     properties  param
                                outputs[other_id][param] = 

    
    method_priority = {: , : , : , : , : }

     endpoint  endpoints:
        endpoint_id = 
        path_clean = re.sub(, , endpoint[])

         other_endpoint  endpoints:
            other_id = 
            other_path_clean = re.sub(, , other_endpoint[])

             path_clean == other_path_clean:
                 method_priority.get(endpoint[], ) > method_priority.get(other_endpoint[], ):
                     other_id   dependencies[endpoint_id]:
                        dependencies[endpoint_id].append(other_id)

    
     ():
        in_degree = {node:   node  deps}
         node  deps:
             dep  deps[node]:
                in_degree[dep] = in_degree.get(dep, ) + 

        queue = [node  node  deps  in_degree[node] == ]
        result = []

         queue:
            queue.sort(key= x: (x.split()[].count(), method_priority.get(x.split()[], )))
            node = queue.pop()
            result.append(node)

             other_node  deps:
                 node  deps[other_node]:
                    in_degree[other_node] -= 
                     in_degree[other_node] == :
                        queue.append(other_node)

         result

    execution_order_ids = topological_sort(dependencies)

    execution_plan = []
     step, endpoint_id  (execution_order_ids, ):
        endpoint = (e  e  endpoints   == endpoint_id)

        inputs = {}
         dep_id  dependencies[endpoint_id]:
             dep_id  outputs:
                 param_name, json_path  outputs[dep_id].items():
                    dep_step = execution_order_ids.index(dep_id) + 
                    inputs[param_name] = {
                        : ,
                        : json_path
                    }

        execution_plan.append({
            : step,
            : endpoint,
            : dependencies[endpoint_id],
            : inputs,
            : outputs[endpoint_id]
        })

     {
        : execution_plan,
        : dependencies
    }

 () -> [[]]:
    
    
    
    parallel_groups = []
    processed_steps = ()
    
    
    independent_steps = [step[]  step  execution_plan   step[]]
     independent_steps:
        parallel_groups.append(independent_steps)
        processed_steps.update(independent_steps)
    
    
    remaining_steps = [step  step  execution_plan  step[]   processed_steps]
    
    
    dependency_map = {}
     step  remaining_steps:
        dep_tuple = ((step[]))
         dep_tuple   dependency_map:
            dependency_map[dep_tuple] = []
        dependency_map[dep_tuple].append(step[])
    
     group  dependency_map.values():
        parallel_groups.append(group)
    
     parallel_groups

---

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 '[e-mail]'
        elif schema.get('format') == 'uuid':
            return '550e8400-e29b-41d4-a716-446655440000'
        elif 'email' in field_name.lower():
            return 
           field_name.lower():
             
           field_name.lower():
             
         
     schema_type == :
        minimum = schema.get(, )
        maximum = schema.get(, minimum + )
         (minimum, )  
     schema_type == :
         
     schema_type == :
         
     schema_type == :
        items_schema = schema.get(, {})
         [generate_value_from_schema(items_schema)]
     schema_type == :
        obj = {}
         prop, prop_schema  schema.get(, {}).items():
             prop  schema.get(, [])   schema.get():
                obj[prop] = generate_value_from_schema(prop_schema, prop)
         obj

     

 () -> :
    

    lines = []

    
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    
     parallel_execution:
        lines.append()
    
    lines.append()
    lines.append()

    
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()

    lines.append()
    lines.append()
    lines.append()

    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()

     parallel_execution  parallel_groups:
        lines.append()
        lines.append()
        lines.append()
        lines.append()
         group  parallel_groups:
             (group) > :  
                 step_num  group:
                    lines.append()
                
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()

    lines.append()
    lines.append()

    
     parallel_execution  parallel_groups:
        executed_steps = ()
         i, group  (parallel_groups):
             (group) > :
                
                lines.append()
                lines.append()
                lines.append()
                executed_steps.update(group)
            :
                
                step_num = group[]
                 step_num   executed_steps:
                    lines.append()
                    executed_steps.add(step_num)
        
        
         step_info  execution_plan:
            step_num = step_info[]
             step_num   executed_steps:
                lines.append()
    :
        
         step_info  execution_plan:
            lines.append()

    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()

    
     step_info  execution_plan:
        endpoint = step_info[]
        step_num = step_info[]
        method = endpoint[]
        path = endpoint[]

        lines.append()
        lines.append()
        lines.append()

        
        lines.append()
        lines.append()
        lines.append()
        lines.append( % method)
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()

        lines.append()
        
        url_expr = 
        
           path:
             param  re.findall(, path):
                url_expr = url_expr.replace(, )
        lines.append()
        lines.append()
        lines.append()
        lines.append()

        
         endpoint.get():
            schema = endpoint[].get(, {})
            example = endpoint[].get()
            content_type = endpoint[].get(, )

             example:
                payload = example
            :
                payload = generate_value_from_schema(schema)

            lines.append()
             content_type == :
                lines.append()
                lines.append()
                lines.append()
                lines.append()
                lines.append( % method.lower())
            :
                lines.append()
                lines.append()
                lines.append( % method.lower())
        :
            lines.append()
            lines.append( % method.lower())

        lines.append()
        lines.append()
        lines.append()

        
        lines.append()
        lines.append()
        lines.append()
        
        
        success_response = 
         status_code, resp_data  endpoint.get(, {}).items():
             status_code.startswith():
                success_response = resp_data
                
        
         success_response  success_response.get():
            schema = success_response[]
            lines.append()
            lines.append( % json.dumps(schema))
            lines.append()
            lines.append()
            lines.append()
            lines.append()
            lines.append()
            lines.append()

        
         step_info[]:
             output_name, json_path  step_info[].items():
                field = json_path.split()[-]
                lines.append()

        lines.append()
        lines.append()

        
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()

        
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()

        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()

        
        lines.append()
        lines.append()
        lines.append( % step_num)
        lines.append()
        lines.append( % method)
        lines.append( % path)
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()

    
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()

    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()

    
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()
    lines.append()

     .join(lines)

---

Phase 5: Execute and Iterate

Execute this code to run the script and fix errors:

import subprocess
import tempfile
import os
import re

def execute_script_with_retries(script_content: str, max_retries: int = 5, detailed_reporting: bool = False):
    """Execute script and retry with fixes"""

    for attempt in range(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_content,
                    : attempt
                }

            
            ()

            
               result.stdout    result.stdout:
                
                field_match = re.search(, result.stdout)
                 field_match:
                    field = field_match.group()
                    script_content = script_content.replace(
                        ,
                        
                    )
                    ()
                    

               result.stdout:
                
                script_content = script_content.replace(, )
                script_content = script_content.replace(, )
                ()
                

            

         subprocess.TimeoutExpired:
            ()
            
         Exception  e:
            ()
            
        :
             os.path.exists(script_path):
                os.unlink(script_path)

     {
        : ,
        : script_content,
        : max_retries
    }

---

Complete End-to-End Example

Here's how to execute the entire workflow:

# 1. Setup
auth_headers = setup_authentication("bearer", {"token": "YOUR_TOKEN"})

# 2. Parse specification (auto-detects OpenAPI/GraphQL)
parsed_spec = parse_specification("https://api.example.com/openapi.json")
print(f"Found {len(parsed_spec['endpoints'])} endpoints")

# 3. Analyze dependencies
dependency_analysis = analyze_dependencies(parsed_spec['endpoints'])
parallel_groups = identify_parallel_groups(dependency_analysis['execution_order'])
print(f"Execution order: {len(dependency_analysis['execution_order'])} steps")

# 4. Generate script with enhanced features
generated_script = generate_python_script(
    dependency_analysis['execution_order'],
    parsed_spec['base_url'],
    auth_headers,
    parallel_execution=True,  # Enable parallel execution
    parallel_groups=parallel_groups
)
print(f"Generated script: {len(generated_script)} characters")

# 5. Execute with retries and detailed reporting
final_result = execute_script_with_retries(generated_script, max_retries=5, detailed_reporting=True)

# 6. Output results
if final_result['success']:
    print("\n" + "="*60)
    print("✓ HAPPYFLOW SCRIPT GENERATED SUCCESSFULLY")
    print("="*60)
    ()
    ()
    (final_result[])
:
    ()
    ()

Usage Instructions

When invoked, execute this skill by:

1. Receive input from user (API spec URL + credentials)
2. Execute Phase 1 code with user's auth credentials
3. Execute Phase 2 code with spec URL
4. Execute Phase 3 code with parsed endpoints
5. Execute Phase 4 code to generate script with enhanced features
6. Execute Phase 5 code to test and fix script
7. Return final working script to user

Output Format

Return to user:

## ✓ HappyFlow Script Generated Successfully

**API**: [API name from spec]
**Total Endpoints**: [count]
**Execution Attempts**: [attempts]

### Generated Script
```python
[COMPLETE WORKING SCRIPT]

Usage

1. Save as test_api.py
2. Run: python test_api.py
3. All requests will return 2xx status codes

Enhanced Features Used

• Parallel Execution: Enabled for faster testing
• Detailed Reporting: Set DETAILED_REPORT=true for comprehensive metrics
• Rate Limiting Handling: Automatic retry with exponential backoff
• Response Validation: JSON Schema validation for responses

## Enhanced Features

### Multi-Format Support
- **OpenAPI 3.0+**: Full specification parsing with schema resolution
- **GraphQL**: Schema introspection and operation extraction

### Advanced Execution
- **Parallel Execution**: Concurrent execution of independent endpoints
- **Detailed Reporting**: Comprehensive execution metrics and timing
- **Connection Pooling**: HTTP connection reuse for improved performance
- **Caching**: Specification parsing cache for reduced processing time

### Enhanced Testing Capabilities
- **File Upload Support**: Multipart/form-data request handling
- **Response Schema Validation**: JSON Schema validation against specifications
- **Rate Limiting Handling**: Automatic retry with exponential backoff
- **Error Recovery**: Intelligent error handling and automatic fixes

### Improved Code Quality
- **Modular Architecture**: Well-organized components for maintainability
- **Type Hints**: Comprehensive type annotations throughout
- **Custom Exceptions**: Structured exception hierarchy
- **Proper Logging**: Structured logging instead of print statements

## Version History

- v2.0.0 (2026-01-08): Enhanced implementation with modular architecture
- v1.0.0 (2025-12-29): Self-contained implementation with embedded code