Test Automation with Claude and GPT-4: Real Integration Cases and Practical Implementation

import anthropic
import json
from typing import List, Dict

class ClaudeAPITestGenerator:
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)

    def generate_api_tests(self, openapi_spec: Dict, endpoint: str) -> str:
        """Generate pytest test suite from OpenAPI specification"""

        prompt = f"""
        You are an expert API test automation engineer. Generate a comprehensive pytest
        test suite for the following endpoint from this OpenAPI specification:

        Endpoint: {endpoint}
        Full OpenAPI Spec: {json.dumps(openapi_spec, indent=2)}

        Generate tests that include:
        1. Happy path scenarios with valid data
        2. Edge cases (boundary values, empty strings, null values)
        3. Error scenarios (401, 403, 404, 422, 500)
        4. Security tests (SQL injection, XSS in parameters)
        5. Schema validation using jsonschema
        6. Performance assertions (response time < 2s)

        Use pytest fixtures for authentication and reusable test data.
        Include detailed docstrings and assertions with meaningful error messages.
        """

        message = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4000,
            messages=[{"role": "user", "content": prompt}]
        )

        return message.content[0].text

# Usage example
generator = ClaudeAPITestGenerator(api_key="your-api-key")

openapi_spec = {
    "paths": {
        "/api/v1/transactions": {
            "post": {
                "summary": "Create transaction",
                "requestBody": {
                    "content": {
                        "application/json": {
                            "schema": {
                                "properties": {
                                    "amount": {"type": "number", "minimum": 0.01},
                                    "currency": {"type": "string", "enum": ["USD", "EUR"]},
                                    "recipient_id": {"type": "string", "format": "uuid"}
                                },
                                "required": ["amount", "currency", "recipient_id"]
                            }
                        }
                    }
                },
                "responses": {
                    "201": {"description": "Transaction created"},
                    "400": {"description": "Invalid request"},
                    "401": {"description": "Unauthorized"}
                }
            }
        }
    }
}

test_code = generator.generate_api_tests(openapi_spec, "/api/v1/transactions")
print(test_code)

import pytest
import requests
from jsonschema import validate
import time

BASE_URL = "https://api.example.com"

@pytest.fixture
def auth_token():
    """Fixture to obtain authentication token"""
    response = requests.post(
        f"{BASE_URL}/auth/login",
        json={"username": "test_user", "password": "test_pass"}
    )
    return response.json()["access_token"]

@pytest.fixture
def valid_transaction_data():
    """Fixture providing valid transaction data"""
    return {
        "amount": 100.50,
        "currency": "USD",
        "recipient_id": "550e8400-e29b-41d4-a716-446655440000"
    }

class TestTransactionCreation:

    def test_create_transaction_success(self, auth_token, valid_transaction_data):
        """Test successful transaction creation with valid data"""
        start_time = time.time()

        response = requests.post(
            f"{BASE_URL}/api/v1/transactions",
            headers={"Authorization": f"Bearer {auth_token}"},
            json=valid_transaction_data
        )

        response_time = time.time() - start_time

        # Assert HTTP status
        assert response.status_code == 201, f"Expected 201, got {response.status_code}"

        # Assert response time
        assert response_time < 2, f"Response time {response_time}s exceeds 2s threshold"

        # Assert response schema
        schema = {
            "type": "object",
            "properties": {
                "transaction_id": {"type": "string"},
                "status": {"type": "string"},
                "created_at": {"type": "string"}
            },
            "required": ["transaction_id", "status", "created_at"]
        }
        validate(instance=response.json(), schema=schema)

        # Assert business logic
        assert response.json()["status"] == "pending"

    def test_create_transaction_minimum_amount(self, auth_token):
        """Test transaction with minimum valid amount (boundary test)"""
        data = {
            "amount": 0.01,
            "currency": "USD",
            "recipient_id": "550e8400-e29b-41d4-a716-446655440000"
        }

        response = requests.post(
            f"{BASE_URL}/api/v1/transactions",
            headers={"Authorization": f"Bearer {auth_token}"},
            json=data
        )

        assert response.status_code == 201

    def test_create_transaction_invalid_amount(self, auth_token):
        """Test transaction with negative amount (validation error)"""
        data = {
            "amount": -10.00,
            "currency": "USD",
            "recipient_id": "550e8400-e29b-41d4-a716-446655440000"
        }

        response = requests.post(
            f"{BASE_URL}/api/v1/transactions",
            headers={"Authorization": f"Bearer {auth_token}"},
            json=data
        )

        assert response.status_code == 400
        assert "amount" in response.json()["errors"]

    def test_create_transaction_sql_injection(self, auth_token):
        """Security test: SQL injection attempt in recipient_id"""
        data = {
            "amount": 100.00,
            "currency": "USD",
            "recipient_id": "550e8400' OR '1'='1"
        }

        response = requests.post(
            f"{BASE_URL}/api/v1/transactions",
            headers={"Authorization": f"Bearer {auth_token}"},
            json=data
        )

        # Should reject with 400, not 500 (which would indicate SQL error)
        assert response.status_code == 400
        assert "Invalid UUID format" in response.json()["message"]

    def test_create_transaction_unauthorized(self, valid_transaction_data):
        """Test transaction creation without authentication"""
        response = requests.post(
            f"{BASE_URL}/api/v1/transactions",
            json=valid_transaction_data
        )

        assert response.status_code == 401

from openai import OpenAI
import json

class GPT4TestDataGenerator:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)

    def generate_test_users(self, count: int, requirements: str) -> List[Dict]:
        """Generate realistic test user data"""

        prompt = f"""
        Generate {count} realistic test user records for an e-commerce platform as JSON array.

        Requirements:
        {requirements}

        Each user should have:
        - id (UUID format)
        - first_name, last_name (diverse nationalities, include edge cases like O'Brien, María José)
        - email (valid format, varied domains)
        - phone (international formats: US, UK, India, Brazil)
        - address (include international addresses with proper formatting)
        - birth_date (ages 18-85, ISO format)
        - registration_date (within last 2 years)
        - total_orders (0-500, weighted toward 1-10)
        - total_spent (0-50000, realistic distribution)
        - account_status (active: 80%, suspended: 15%, closed: 5%)
        - preferences (random realistic shopping preferences)

        Include edge cases:
        - 5% with very long names (>50 characters)
        - 3% with special Unicode characters (Chinese, Arabic, Emoji)
        - 2% with minimal data (newly registered)

        Return only valid JSON array, no markdown formatting.
        """

        response = self.client.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=[
                {"role": "system", "content": "You are a test data generation expert."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.8,  # Higher temperature for variety
            max_tokens=4000
        )

        return json.loads(response.choices[0].message.content)

# Usage
generator = GPT4TestDataGenerator(api_key="your-openai-key")

requirements = """
- 30% US users, 20% UK, 15% India, 15% Brazil, 20% mixed
- 40% frequent buyers (>20 orders), 30% occasional (5-20), 30% new (<5)
- Include 10 users with purchase history exceeding $10,000 (VIP segment)
- Include 5 users with disputed transactions (for fraud testing)
"""

users = generator.generate_test_users(count=100, requirements=requirements)

# Save to file
with open('test_users.json', 'w', encoding='utf-8') as f:
    json.dump(users, f, indent=2, ensure_ascii=False)

[
  {
    "id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
    "first_name": "Anastasia",
    "last_name": "O'Connor-Fitzgerald",
    "email": "anastasia.oconnor@protonmail.com",
    "phone": "+353 86 123 4567",
    "address": {
      "street": "42 Grafton Street, Apartment 3B",
      "city": "Dublin",
      "postal_code": "D02 VK65",
      "country": "Ireland"
    },
    "birth_date": "1987-03-15",
    "registration_date": "2023-06-20",
    "total_orders": 47,
    "total_spent": 3421.50,
    "account_status": "active",
    "preferences": {
      "categories": ["electronics", "books", "home_decor"],
      "notification_channel": "email",
      "language": "en-IE"
    }
  },
  {
    "id": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
    "first_name": "李",
    "last_name": "明",
    "email": "liming@qq.com",
    "phone": "+86 138 0013 8000",
    "address": {
      "street": "望京SOHO T1 1508室",
      "city": "北京市",
      "postal_code": "100102",
      "country": "China"
    },
    "birth_date": "1992-11-08",
    "registration_date": "2024-01-12",
    "total_orders": 156,
    "total_spent": 12847.00,
    "account_status": "active",
    "preferences": {
      "categories": ["fashion", "beauty", "tech"],
      "notification_channel": "wechat",
      "language": "zh-CN"
    },
    "vip_status": true
  }
]

import anthropic
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
import base64

class SelfHealingTestRunner:
    def __init__(self, claude_api_key: str):
        self.claude = anthropic.Anthropic(api_key=claude_api_key)
        self.driver = webdriver.Chrome()

    def find_element_with_healing(self, original_selector: str, element_description: str):
        """Attempt to find element, use Claude to heal if selector fails"""

        try:
            return self.driver.find_element(By.CSS_SELECTOR, original_selector)
        except NoSuchElementException:
            print(f"⚠️  Selector failed: {original_selector}")
            print(f"🔧 Attempting self-healing...")

            # Capture screenshot and page source for context
            screenshot = self.driver.get_screenshot_as_base64()
            page_source = self.driver.page_source

            # Ask Claude to suggest new selector
            new_selector = self._heal_selector(
                original_selector,
                element_description,
                page_source,
                screenshot
            )

            print(f"✨ Healed selector: {new_selector}")

            # Try new selector
            element = self.driver.find_element(By.CSS_SELECTOR, new_selector)

            # Log for human review
            self._log_heal_event(original_selector, new_selector, element_description)

            return element

    def _heal_selector(self, old_selector: str, description: str, html: str, screenshot: str) -> str:
        """Use Claude to analyze page and suggest new selector"""

        prompt = f"""
        A UI test selector has broken. Analyze the current page and suggest a new CSS selector.

        Original (broken) selector: {old_selector}
        Element description: {description}

        Current page HTML (truncated):
        {html[:5000]}

        Task: Suggest the most robust CSS selector for the element matching: "{description}"

        Prefer selectors in this order:
        1. data-testid attributes
        2. aria-label attributes
        3. Stable ID attributes
        4. Specific class combinations
        5. XPath as last resort

        Return ONLY the selector string, no explanation.
        """

        message = self.claude.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=200,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": prompt},
                        {
                            "type": "image",
                            "source": {
                                "type": "base64",
                                "media_type": "image/png",
                                "data": screenshot
                            }
                        }
                    ]
                }
            ]
        )

        return message.content[0].text.strip()

    def _log_heal_event(self, old: str, new: str, description: str):
        """Log healing event for human review"""
        with open('selector_healing_log.txt', 'a') as f:
            f.write(f"{datetime.now()} | {description}\n")
            f.write(f"  Old: {old}\n")
            f.write(f"  New: {new}\n\n")

# Usage in test
runner = SelfHealingTestRunner(claude_api_key="your-key")
runner.driver.get("https://example.com/login")

# This will auto-heal if selector breaks
login_button = runner.find_element_with_healing(
    original_selector="button#login-submit",
    element_description="primary login submit button with text 'Sign In'"
)
login_button.click()

# After 1 month of usage
{
    "total_selector_failures": 127,
    "successful_auto_heals": 98,  # 77% success rate
    "manual_intervention_needed": 29,
    "time_saved": "~11 hours/week",
    "maintenance_cost_reduction": "73%"
}

from openai import OpenAI
import json

class TestPrioritizer:
    def __init__(self, openai_key: str):
        self.client = OpenAI(api_key=openai_key)

    def prioritize_tests(self, changed_files: List[str], test_catalog: List[Dict]) -> List[Dict]:
        """Use GPT-4 to intelligently prioritize tests based on code changes"""

        prompt = f"""
        Analyze these code changes and prioritize test execution.

        Changed files:
        {json.dumps(changed_files, indent=2)}

        Available tests:
        {json.dumps(test_catalog[:50], indent=2)}  # Send subset for token efficiency

        Prioritize tests based on:
        1. Direct impact (tests covering changed files)
        2. Blast radius (dependent modules)
        3. Historical failure rate (flaky tests later)
        4. Business criticality (payment/auth tests first)

        Return JSON array of test IDs in priority order with reasoning.
        Format: [{{"test_id": "...", "priority_score": 1-10, "reason": "..."}}]
        """

        response = self.client.chat.completions.create(
            model="gpt-4-turbo-preview",
            messages=[
                {"role": "system", "content": "You are a test strategy expert."},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"},
            temperature=0.3  # Lower temperature for consistent logic
        )

        return json.loads(response.choices[0].message.content)

# Example usage
prioritizer = TestPrioritizer(openai_key="your-key")

changed_files = [
    "src/payment/stripe_integration.py",
    "src/payment/payment_processor.py",
    "src/models/transaction.py"
]

test_catalog = [
    {"test_id": "test_payment_success", "file": "tests/test_payment.py", "avg_duration": 2.3, "failure_rate": 0.02},
    {"test_id": "test_ui_homepage", "file": "tests/test_ui.py", "avg_duration": 5.1, "failure_rate": 0.15},
    # ... more tests
]

prioritized = prioritizer.prioritize_tests(changed_files, test_catalog)

# Output:
# [
#   {"test_id": "test_payment_success", "priority_score": 10, "reason": "Direct impact on changed payment module"},
#   {"test_id": "test_transaction_creation", "priority_score": 9, "reason": "Transaction model changed, core business flow"},
#   {"test_id": "test_stripe_webhook", "priority_score": 8, "reason": "Integration test for modified Stripe code"},
#   ...
# ]

# .github/workflows/smart-testing.yml
name: AI-Prioritized Testing

on: [pull_request]

jobs:
  smart_test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Get changed files
        id: changes
        run: |
          echo "files=$(git diff --name-only origin/main | jq -R -s -c 'split("\n")[:-1]')" >> $GITHUB_OUTPUT

      - name: AI Test Prioritization
        run: |
          python prioritize_tests.py \
            --changed-files '${{ steps.changes.outputs.files }}' \
            --output prioritized_tests.json

      - name: Run Priority Tests (fast fail)
        run: |
          pytest $(head -20 prioritized_tests.json | jq -r '.[] | .test_id') \
            --maxfail=3 \
            --tb=short

      - name: Run Remaining Tests (if priority tests pass)
        if: success()
        run: pytest $(tail -n +21 prioritized_tests.json | jq -r '.[] | .test_id')

# Bad: Sending huge context every time
def generate_tests_bad(api_spec):  # 50K tokens
    response = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        messages=[{"role": "user", "content": f"Generate tests for {api_spec}"}]
    )
    # Cost: ~$0.60 per call

# Good: Use prompt caching (Claude) or embeddings (GPT-4)
def generate_tests_good(api_spec):
    # Cache the large spec
    response = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2000,
        system=[
            {
                "type": "text",
                "text": "You are an API test expert.",
                "cache_control": {"type": "ephemeral"}
            },
            {
                "type": "text",
                "text": f"OpenAPI Spec:\n{api_spec}",
                "cache_control": {"type": "ephemeral"}
            }
        ],
        messages=[{"role": "user", "content": "Generate tests for POST /users endpoint"}]
    )
    # Cost: ~$0.06 per call (90% savings on cached portion)

def generate_structured_test(endpoint_spec):
    schema = {
        "test_suite": {
            "setup_fixtures": ["list of fixture names"],
            "test_cases": [
                {
                    "name": "string",
                    "description": "string",
                    "test_code": "string",
                    "assertions": ["list of assertion descriptions"]
                }
            ]
        }
    }

    prompt = f"""
    Generate test suite for: {endpoint_spec}

    CRITICAL: Return ONLY valid JSON matching this exact schema:
    {json.dumps(schema, indent=2)}

    No markdown, no explanations, just JSON.
    """

    response = openai_client.chat.completions.create(
        model="gpt-4-turbo-preview",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},  # Enforced JSON mode
        temperature=0.2  # Lower variance
    )

    result = json.loads(response.choices[0].message.content)

    # Validate before using
    try:
        validate_test_suite_schema(result)
        return result
    except ValidationError as e:
        # Retry with error feedback
        return generate_structured_test(f"{endpoint_spec}\n\nPrevious attempt failed: {e}")

class ReviewableTestGenerator:
    def generate_with_review(self, spec, criticality="medium"):
        test_code = self.llm_generate(spec)

        if criticality in ["high", "critical"]:
            # Save for human review
            review_queue.add({
                "code": test_code,
                "spec": spec,
                "status": "pending_review",
                "reviewers": ["senior_qa_lead"]
            })
            return None  # Block until reviewed

        # Auto-approve for low/medium criticality
        return test_code