happyflow-generator

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"
            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', ''),