happyflow-generator

name	happyflow-generator
description	Automatically generate and execute Python test scripts from OpenAPI specifications

HappyFlow Generator Skill

Metadata

• Skill Name: HappyFlow Generator
• Version: 1.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 that successfully call all API endpoints in dependency-correct order, ensuring all requests return 2xx status codes.

Input: OpenAPI spec (URL/file) + authentication credentials
Output: Working Python script that executes complete API happy path flow

Key Difference: This skill contains ALL implementation code - no external MCP tools required. Everything executes using built-in code execution capabilities.

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: OpenAPI Parsing

Execute this code to parse OpenAPI specifications:

import requests
import yaml
import json
import re
from typing import Dict, List, Any

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

    # Fetch spec
    if spec_source.startswith('http'):
        response = requests.get(spec_source)
        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(, [])[]
        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()
                    }

            
            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(, {})
    }

---

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
    }

---

Phase 4: Script Generation

Execute this code to generate the Python test script:

import json
from typing import Dict, List, Any

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 '[e-mail]'
        elif 'name' in field_name.lower():
             
         
     schema_type == :
         schema.get(, )
     schema_type == :
         
     schema_type == :
         
     schema_type == :
         [generate_value_from_schema(schema.get(, {}))]
     schema_type == :
        obj = {}
         prop, prop_schema  schema.get(, {}).items():
             prop  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()

    
    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_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()

        
        url_expr = 
        url_expr = re.sub(, , url_expr)
        lines.append()

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

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

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

        lines.append()
        lines.append()

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

        lines.append( % step_num)
        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):
    """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:
            result = subprocess.run(
                ['python', script_path],
                capture_output=True,
                text=True,
                timeout=300
            )

            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 fixes
            print(f"✗ Exit code: {result.returncode}")

            # Simple fix patterns
               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(, )
                ()
                

            

        :
             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 OpenAPI
parsed_spec = parse_openapi_spec("https://api.example.com/openapi.json")
print(f"Found {len(parsed_spec['endpoints'])} endpoints")

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

# 4. Generate script
generated_script = generate_python_script(
    dependency_analysis['execution_order'],
    parsed_spec['base_url'],
    auth_headers
)
print(f"Generated script: {len(generated_script)} characters")

# 5. Execute with retries
final_result = execute_script_with_retries(generated_script, max_retries=5)

# 6. Output results
if final_result['success']:
    print("\n" + "="*60)
    print("✓ HAPPYFLOW SCRIPT GENERATED SUCCESSFULLY")
    print("="*60)
    print(f"Attempts required: {final_result['attempts']}")
    print("\nFinal Script:")
    print(final_result[])
:
    ()
    ()

Usage Instructions

When invoked, execute this skill by:

1. Receive input from user (OpenAPI 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
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

## Advantages of Self-Contained Approach

- **No external dependencies**: All logic embedded in skill
- **Portable**: Works anywhere with Python execution
- **Transparent**: User can see exact implementation
- **Customizable**: Easy to modify code for specific needs
- **Debuggable**: Can trace through each function

## Version History

- v1.0.0 (2025-12-29): Self-contained implementation with embedded code