aishastr.live

Agentic Architecture Pattern: Integration Testing

Written by

admin

in


    """
Pattern 262: Integration Testing

Purpose: Component interaction testing with API testing, end-to-end scenarios, and dependency validation.
Tests how multiple components work together in integrated systems.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import time


class IntegrationType(Enum):
    """Types of integration tests."""
    API = "api"  # API endpoint testing
    DATABASE = "database"  # Database integration
    SERVICE = "service"  # Service-to-service
    END_TO_END = "end_to_end"  # Full workflow
    CONTRACT = "contract"  # Contract testing


class TestStatus(Enum):
    """Integration test status."""
    PASSED = "passed"
    FAILED = "failed"
    TIMEOUT = "timeout"
    ERROR = "error"


@dataclass
class IntegrationTestCase:
    """Integration test case."""
    name: str
    description: str
    integration_type: IntegrationType
    components: List[str]  # Components involved
    test_steps: List[str]
    expected_behavior: str
    timeout: float = 10.0  # seconds
    dependencies: List[str] = field(default_factory=list)


@dataclass
class IntegrationTestResult:
    """Integration test result."""
    test_case: IntegrationTestCase
    status: TestStatus
    execution_time: float
    component_results: Dict[str, bool] = field(default_factory=dict)
    error_message: str = ""
    warnings: List[str] = field(default_factory=list)


class IntegrationTestingAgent:
    """
    Integration Testing Agent - Component interaction testing.
    
    Features:
    - API endpoint testing
    - Database integration tests
    - Service-to-service tests
    - End-to-end workflows
    - Contract testing
    - Dependency validation
    - Performance monitoring
    - AI-powered test generation
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Test registry
        self.test_cases: List[IntegrationTestCase] = []
        self.test_results: List[IntegrationTestResult] = []
        
        # Mock components for testing
        self.mock_components: Dict[str, Callable] = {}
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_tests": 0,
            "passed": 0,
            "failed": 0,
            "errors": 0,
            "timeouts": 0,
            "integration_types": {t.value: 0 for t in IntegrationType},
            "avg_execution_time": 0.0
        }
    
    def generate_integration_tests(
        self,
        system_description: str,
        components: List[str],
        integration_type: IntegrationType = IntegrationType.END_TO_END
    ) -> List[IntegrationTestCase]:
        """Generate integration test cases."""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an integration testing expert. Generate comprehensive test cases "
                      "for component interactions and system integration."),
            ("human", f"System: {system_description}\n"
                     f"Components: {', '.join(components)}\n"
                     f"Integration Type: {integration_type.value}\n\n"
                     f"Generate 5 integration test cases. For each test, specify:\n"
                     f"- Test name\n"
                     f"- Components involved\n"
                     f"- Test steps (numbered)\n"
                     f"- Expected behavior\n"
                     f"Separate tests with '---'")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse test cases
        test_cases = self._parse_integration_tests(
            str(response.content),
            components,
            integration_type
        )
        
        self.test_cases.extend(test_cases)
        
        return test_cases
    
    def _parse_integration_tests(
        self,
        content: str,
        components: List[str],
        integration_type: IntegrationType
    ) -> List[IntegrationTestCase]:
        """Parse integration tests from LLM response."""
        test_cases = []
        
        tests = content.split("---")
        
        for test_text in tests:
            if not test_text.strip():
                continue
            
            # Extract test name
            lines = test_text.strip().split('\n')
            name = lines[0].strip() if lines else f"Integration Test {len(test_cases) + 1}"
            
            # Extract test steps
            steps = []
            for line in lines:
                if line.strip().startswith(('1.', '2.', '3.', '4.', '5.')):
                    steps.append(line.strip())
            
            # Use first 2 components by default
            involved_components = components[:2] if len(components) >= 2 else components
            
            test_case = IntegrationTestCase(
                name=name,
                description=f"Integration test for {integration_type.value}",
                integration_type=integration_type,
                components=involved_components,
                test_steps=steps if steps else ["Execute integration test"],
                expected_behavior="Components should integrate correctly"
            )
            
            test_cases.append(test_case)
        
        return test_cases[:5]  # Limit to 5
    
    def register_mock_component(
        self,
        component_name: str,
        mock_function: Callable
    ):
        """Register a mock component for testing."""
        self.mock_components[component_name] = mock_function
    
    def execute_integration_test(
        self,
        test_case: IntegrationTestCase
    ) -> IntegrationTestResult:
        """Execute an integration test."""
        
        start_time = time.time()
        component_results = {}
        warnings = []
        
        try:
            # Check dependencies
            for dep in test_case.dependencies:
                if dep not in self.mock_components:
                    warnings.append(f"Missing dependency: {dep}")
            
            # Execute test steps
            for component in test_case.components:
                if component in self.mock_components:
                    try:
                        result = self.mock_components[component]()
                        component_results[component] = bool(result)
                    except Exception as e:
                        component_results[component] = False
                        warnings.append(f"{component} failed: {str(e)}")
                else:
                    # Mock success if no mock available
                    component_results[component] = True
                    warnings.append(f"No mock for {component}, assumed success")
            
            # Check timeout
            execution_time = time.time() - start_time
            if execution_time > test_case.timeout:
                status = TestStatus.TIMEOUT
                error_msg = f"Test exceeded timeout of {test_case.timeout}s"
            elif all(component_results.values()):
                status = TestStatus.PASSED
                error_msg = ""
            else:
                status = TestStatus.FAILED
                failed_components = [k for k, v in component_results.items() if not v]
                error_msg = f"Components failed: {', '.join(failed_components)}"
        
        except Exception as e:
            execution_time = time.time() - start_time
            status = TestStatus.ERROR
            error_msg = str(e)
        
        # Update statistics
        self.statistics["total_tests"] += 1
        self.statistics["integration_types"][test_case.integration_type.value] += 1
        
        if status == TestStatus.PASSED:
            self.statistics["passed"] += 1
        elif status == TestStatus.FAILED:
            self.statistics["failed"] += 1
        elif status == TestStatus.TIMEOUT:
            self.statistics["timeouts"] += 1
        else:
            self.statistics["errors"] += 1
        
        # Update average execution time
        total = self.statistics["total_tests"]
        prev_avg = self.statistics["avg_execution_time"]
        self.statistics["avg_execution_time"] = (prev_avg * (total - 1) + execution_time) / total
        
        result = IntegrationTestResult(
            test_case=test_case,
            status=status,
            execution_time=execution_time,
            component_results=component_results,
            error_message=error_msg,
            warnings=warnings
        )
        
        self.test_results.append(result)
        
        return result
    
    def execute_all_tests(self) -> List[IntegrationTestResult]:
        """Execute all registered integration tests."""
        results = []
        
        for test_case in self.test_cases:
            result = self.execute_integration_test(test_case)
            results.append(result)
        
        return results
    
    def test_api_contract(
        self,
        api_name: str,
        expected_input: Dict[str, Any],
        expected_output: Dict[str, Any]
    ) -> bool:
        """Test API contract compliance."""
        
        # Simplified contract testing
        if api_name in self.mock_components:
            try:
                result = self.mock_components[api_name](expected_input)
                # Simple validation
                return result is not None
            except Exception:
                return False
        
        return False
    
    def generate_test_report(self) -> Dict[str, Any]:
        """Generate comprehensive test report."""
        
        # Calculate success rate
        success_rate = 0.0
        if self.statistics["total_tests"] > 0:
            success_rate = (self.statistics["passed"] / self.statistics["total_tests"]) * 100
        
        # Group by integration type
        by_type = defaultdict(lambda: {"total": 0, "passed": 0})
        for result in self.test_results:
            itype = result.test_case.integration_type.value
            by_type[itype]["total"] += 1
            if result.status == TestStatus.PASSED:
                by_type[itype]["passed"] += 1
        
        # Failed tests
        failed_tests = [
            {
                "name": r.test_case.name,
                "type": r.test_case.integration_type.value,
                "error": r.error_message,
                "components": r.test_case.components
            }
            for r in self.test_results
            if r.status != TestStatus.PASSED
        ]
        
        return {
            "summary": {
                "total_tests": self.statistics["total_tests"],
                "passed": self.statistics["passed"],
                "failed": self.statistics["failed"],
                "errors": self.statistics["errors"],
                "timeouts": self.statistics["timeouts"],
                "success_rate": f"{success_rate:.1f}%",
                "avg_execution_time": f"{self.statistics['avg_execution_time']:.3f}s"
            },
            "by_type": dict(by_type),
            "failed_tests": failed_tests
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get integration testing statistics."""
        return {
            "name": self.name,
            **self.statistics
        }


class IntegrationTestingPattern:
    """
    Integration Testing Pattern - Manage integration testing agents.
    
    Coordinates testing across multiple system components.
    """
    
    def __init__(self):
        self.agents: Dict[str, IntegrationTestingAgent] = {}
    
    def create_agent(self, name: str) -> IntegrationTestingAgent:
        """Create a new integration testing agent."""
        agent = IntegrationTestingAgent(name)
        self.agents[name] = agent
        print(f"   Created integration testing agent: {name}")
        return agent
    
    def run(
        self,
        agent_name: str,
        action: str,
        **kwargs
    ) -> Dict[str, Any]:
        """Execute integration testing action."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        
        if action == "generate":
            tests = agent.generate_integration_tests(
                kwargs.get("system_description", ""),
                kwargs.get("components", []),
                kwargs.get("integration_type", IntegrationType.END_TO_END)
            )
            
            print(f"\n=== Integration Test Generation by {agent_name} ===")
            print(f"Generated {len(tests)} integration test cases\n")
            
            for i, test in enumerate(tests, 1):
                print(f"{i}. {test.name}")
                print(f"   Type: {test.integration_type.value}")
                print(f"   Components: {', '.join(test.components)}")
                print(f"   Steps: {len(test.test_steps)}")
            
            return {
                "agent": agent_name,
                "action": "generate",
                "tests_generated": len(tests)
            }
        
        elif action == "execute":
            results = agent.execute_all_tests()
            
            print(f"\n=== Integration Test Execution by {agent_name} ===")
            report = agent.generate_test_report()
            
            print(f"\nSummary:")
            for key, value in report["summary"].items():
                print(f"  {key}: {value}")
            
            if report["failed_tests"]:
                print(f"\nFailed Tests ({len(report['failed_tests'])}):")
                for test in report["failed_tests"][:5]:
                    print(f"  - {test['name']}: {test['error']}")
            
            return {
                "agent": agent_name,
                "action": "execute",
                "report": report,
                "statistics": agent.get_statistics()
            }
        
        return {"error": f"Unknown action: {action}"}


# Example usage
if __name__ == "__main__":
    print("=== Integration Testing Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = IntegrationTestingPattern()
    
    # Create agent
    print("1. Setting up integration testing agent...")
    tester = pattern.create_agent("IntegrationTester")
    
    # Register mock components
    print("\n2. Registering mock components...")
    
    def mock_auth_service():
        """Mock authentication service."""
        return {"authenticated": True, "user_id": "123"}
    
    def mock_database():
        """Mock database."""
        return {"status": "connected", "records": 100}
    
    def mock_api_gateway():
        """Mock API gateway."""
        return {"status": "ok", "latency": 50}
    
    tester.register_mock_component("AuthService", mock_auth_service)
    tester.register_mock_component("Database", mock_database)
    tester.register_mock_component("APIGateway", mock_api_gateway)
    
    # Generate integration tests
    print("\n3. Generating integration tests...")
    
    result1 = pattern.run(
        "IntegrationTester",
        "generate",
        system_description="E-commerce platform with microservices",
        components=["AuthService", "Database", "APIGateway", "PaymentService"],
        integration_type=IntegrationType.SERVICE
    )
    
    print("\n" + "="*60 + "\n")
    
    # Execute integration tests
    print("4. Executing integration tests...")
    
    result2 = pattern.run(
        "IntegrationTester",
        "execute"
    )
    
    # Print final statistics
    print("\n" + "="*60)
    print("\n5. Final Statistics:")
    stats = result2["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\nāœ“ Comprehensive integration testing with component validation complete!")

Agentic Architecture Pattern: Integration Testing

Pattern 262: 262 Integration Testing – Line-by-Line Explanation

Overview

Pattern 262: Integration Testing

Purpose: Component interaction testing with API testing, end-to-end scenarios, and dependency validation. Tests how multiple components work together in integrated systems.

Code Structure Analysis

File Information

  • Pattern Number: 262
  • Pattern Name: 262 Integration Testing
  • File Path: patterns\testing_validation\262_integration_testing.py
  • Total Lines: 474

Detailed Code Explanation

Module Docstring

"""
Pattern 262: Integration Testing

Purpose: Component interaction testing with API testing, end-to-end scenarios, and dependency validation.
Tests how multiple components work together in integrated systems.
"""

Explanation: This module docstring describes the purpose and key features of the 262 Integration Testing pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import time  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

IntegrationType (Line 18)

class IntegrationType(Enum):
    API
    DATABASE
    SERVICE
    END_TO_END
    CONTRACT

Purpose: Types of integration tests.

Values: – API: Represents api – DATABASE: Represents database – SERVICE: Represents service – END_TO_END: Represents end to end – CONTRACT: Represents contract

TestStatus (Line 27)

class TestStatus(Enum):
    PASSED
    FAILED
    TIMEOUT
    ERROR

Purpose: Integration test status.

Values: – PASSED: Represents passed – FAILED: Execution failed – TIMEOUT: Represents timeout – ERROR: Error severity level

Classes

IntegrationType (Line 18)

Description: Types of integration tests.

TestStatus (Line 27)

Description: Integration test status.

IntegrationTestCase (Line 36)

Description: Integration test case.

IntegrationTestResult (Line 49)

Description: Integration test result.

IntegrationTestingAgent (Line 59)

Description: Integration Testing Agent – Component interaction testing.

Features: – API endpoint testing – Database integration tests – Service-to-service tests – End-to-end workflows – Contract testing – Dependency validation – Performance monitoring – AI-powered test generation

Methods:

__init__ (Line 74)
def __init__(self, name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

generate_integration_tests (Line 96)
def generate_integration_tests(self, system_description, components, integration_type):
    """Generate integration test cases."""

Purpose: Generate integration test cases.

Parameters: – self: Instance reference – system_description: Parameter for system_description – components: Parameter for components – integration_type: Parameter for integration_type

_parse_integration_tests (Line 132)
def _parse_integration_tests(self, content, components, integration_type):
    """Parse integration tests from LLM response."""

Purpose: Parse integration tests from LLM response.

Parameters: – self: Instance reference – content: Parameter for content – components: Parameter for components – integration_type: Parameter for integration_type

register_mock_component (Line 173)
def register_mock_component(self, component_name, mock_function):
    """Register a mock component for testing."""

Purpose: Register a mock component for testing.

Parameters: – self: Instance reference – component_name: Parameter for component_name – mock_function: Parameter for mock_function

execute_integration_test (Line 181)
def execute_integration_test(self, test_case):
    """Execute an integration test."""

Purpose: Execute an integration test.

Parameters: – self: Instance reference – test_case: Parameter for test_case

execute_all_tests (Line 260)
def execute_all_tests(self):
    """Execute all registered integration tests."""

Purpose: Execute all registered integration tests.

Parameters: – self: Instance reference

test_api_contract (Line 270)
def test_api_contract(self, api_name, expected_input, expected_output):
    """Test API contract compliance."""

Purpose: Test API contract compliance.

Parameters: – self: Instance reference – api_name: Parameter for api_name – expected_input: Parameter for expected_input – expected_output: Parameter for expected_output

generate_test_report (Line 289)
def generate_test_report(self):
    """Generate comprehensive test report."""

Purpose: Generate comprehensive test report.

Parameters: – self: Instance reference

get_statistics (Line 331)
def get_statistics(self):
    """Get integration testing statistics."""

Purpose: Get integration testing statistics.

Parameters: – self: Instance reference

IntegrationTestingPattern (Line 339)

Description: Integration Testing Pattern – Manage integration testing agents.

Coordinates testing across multiple system components.

Methods:

__init__ (Line 346)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 349)
def create_agent(self, name):
    """Create a new integration testing agent."""

Purpose: Create a new integration testing agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 356)
def run(self, agent_name, action):
    """Execute integration testing action."""

Purpose: Execute integration testing action.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – action: Parameter for action

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = IntegrationType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts