aishastr.live

Agentic Architecture Pattern: Unit Testing

Written by

admin

in


    """
Pattern 261: Unit Testing

Purpose: Automated unit test generation and execution with comprehensive coverage analysis.
Generates test cases, executes tests, and provides detailed reporting.
"""

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 re
import traceback


class TestType(Enum):
    """Types of unit tests."""
    POSITIVE = "positive"  # Happy path testing
    NEGATIVE = "negative"  # Error handling testing
    BOUNDARY = "boundary"  # Edge case testing
    PARAMETRIC = "parametric"  # Multiple input variations
    PROPERTY = "property"  # Property-based testing


class TestStatus(Enum):
    """Test execution status."""
    PASSED = "passed"
    FAILED = "failed"
    SKIPPED = "skipped"
    ERROR = "error"


@dataclass
class TestCase:
    """Unit test case."""
    name: str
    description: str
    test_type: TestType
    function_name: str
    inputs: Dict[str, Any]
    expected_output: Any
    assertions: List[str] = field(default_factory=list)
    setup: Optional[str] = None
    teardown: Optional[str] = None


@dataclass
class TestResult:
    """Test execution result."""
    test_case: TestCase
    status: TestStatus
    actual_output: Any = None
    error_message: str = ""
    execution_time: float = 0.0
    stack_trace: str = ""


class UnitTestingAgent:
    """
    Unit Testing Agent - Automated test generation and execution.
    
    Features:
    - Test case generation from specifications
    - Multiple test types (positive, negative, boundary)
    - Test execution with assertions
    - Coverage analysis
    - Result reporting
    - AI-powered test suggestions
    - Assertion generation
    - Edge case identification
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Test registry
        self.test_cases: List[TestCase] = []
        self.test_results: List[TestResult] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_tests": 0,
            "tests_passed": 0,
            "tests_failed": 0,
            "tests_error": 0,
            "test_types": {t.value: 0 for t in TestType},
            "coverage": 0.0
        }
    
    def generate_tests(
        self,
        function_name: str,
        function_signature: str,
        description: str = ""
    ) -> List[TestCase]:
        """Generate test cases for a function."""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a test generation expert. Generate comprehensive unit test cases."),
            ("human", f"Function: {function_name}\n"
                     f"Signature: {function_signature}\n"
                     f"Description: {description}\n\n"
                     f"Generate 5 test cases covering:\n"
                     f"1. Positive case (happy path)\n"
                     f"2. Negative case (error handling)\n"
                     f"3. Boundary case (edge values)\n"
                     f"4. Parametric case (various inputs)\n"
                     f"5. Property case (invariants)\n\n"
                     f"Format each as:\n"
                     f"TEST: \n"
                     f"TYPE: \n"
                     f"INPUTS: \n"
                     f"EXPECTED: \n"
                     f"---")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse test cases
        test_cases = self._parse_test_cases(str(response.content), function_name)
        
        # Add to registry
        self.test_cases.extend(test_cases)
        
        return test_cases
    
    def _parse_test_cases(self, content: str, function_name: str) -> List[TestCase]:
        """Parse test cases from LLM response."""
        test_cases = []
        
        # Split by test separator
        tests = content.split("---")
        
        for test_text in tests:
            if not test_text.strip():
                continue
            
            # Extract test name
            name_match = re.search(r'TEST:\s*(.+)', test_text)
            name = name_match.group(1).strip() if name_match else f"test_{len(test_cases)}"
            
            # Extract test type
            type_match = re.search(r'TYPE:\s*(\w+)', test_text)
            test_type_str = type_match.group(1).lower() if type_match else "positive"
            
            try:
                test_type = TestType(test_type_str)
            except ValueError:
                test_type = TestType.POSITIVE
            
            # Extract inputs (simplified)
            inputs_match = re.search(r'INPUTS:\s*(.+?)(?:EXPECTED|$)', test_text, re.DOTALL)
            inputs_str = inputs_match.group(1).strip() if inputs_match else "{}"
            
            # Simple parsing - in production use ast.literal_eval
            inputs = {"data": inputs_str}
            
            # Extract expected
            expected_match = re.search(r'EXPECTED:\s*(.+?)(?:---|\Z)', test_text, re.DOTALL)
            expected = expected_match.group(1).strip() if expected_match else None
            
            test_case = TestCase(
                name=name,
                description=f"Test {test_type.value} case",
                test_type=test_type,
                function_name=function_name,
                inputs=inputs,
                expected_output=expected
            )
            
            test_cases.append(test_case)
        
        return test_cases
    
    def execute_test(
        self,
        test_case: TestCase,
        test_function: Callable
    ) -> TestResult:
        """Execute a single test case."""
        
        start_time = datetime.now()
        
        try:
            # Execute function
            actual_output = test_function(**test_case.inputs)
            
            # Check assertions
            if test_case.expected_output is not None:
                if str(actual_output).strip() == str(test_case.expected_output).strip():
                    status = TestStatus.PASSED
                    error_msg = ""
                else:
                    status = TestStatus.FAILED
                    error_msg = f"Expected {test_case.expected_output}, got {actual_output}"
            else:
                # No expected output, just check no error
                status = TestStatus.PASSED
                error_msg = ""
            
        except Exception as e:
            status = TestStatus.ERROR
            actual_output = None
            error_msg = str(e)
            stack_trace = traceback.format_exc()
        
        end_time = datetime.now()
        execution_time = (end_time - start_time).total_seconds()
        
        # Update statistics
        self.statistics["total_tests"] += 1
        self.statistics["test_types"][test_case.test_type.value] += 1
        
        if status == TestStatus.PASSED:
            self.statistics["tests_passed"] += 1
        elif status == TestStatus.FAILED:
            self.statistics["tests_failed"] += 1
        elif status == TestStatus.ERROR:
            self.statistics["tests_error"] += 1
        
        result = TestResult(
            test_case=test_case,
            status=status,
            actual_output=actual_output,
            error_message=error_msg,
            execution_time=execution_time,
            stack_trace=stack_trace if status == TestStatus.ERROR else ""
        )
        
        self.test_results.append(result)
        
        return result
    
    def execute_all_tests(
        self,
        test_functions: Dict[str, Callable]
    ) -> List[TestResult]:
        """Execute all registered test cases."""
        
        results = []
        
        for test_case in self.test_cases:
            if test_case.function_name in test_functions:
                result = self.execute_test(test_case, test_functions[test_case.function_name])
                results.append(result)
        
        # Update coverage
        self._calculate_coverage()
        
        return results
    
    def _calculate_coverage(self):
        """Calculate test coverage."""
        if self.statistics["total_tests"] > 0:
            # Simple coverage: passed / total
            self.statistics["coverage"] = (
                self.statistics["tests_passed"] / self.statistics["total_tests"]
            ) * 100
    
    def generate_assertions(
        self,
        function_name: str,
        output_type: str,
        constraints: List[str] = None
    ) -> List[str]:
        """Generate assertions for a function."""
        
        constraints_str = "\n".join(constraints) if constraints else "None specified"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Generate appropriate assertions for unit testing."),
            ("human", f"Function: {function_name}\n"
                     f"Output Type: {output_type}\n"
                     f"Constraints: {constraints_str}\n\n"
                     f"Generate 5 assertions to verify correctness:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse assertions
        assertions = [
            line.strip()
            for line in str(response.content).split('\n')
            if line.strip() and not line.startswith('#')
        ]
        
        return assertions[:5]
    
    def suggest_edge_cases(
        self,
        function_name: str,
        input_types: Dict[str, str]
    ) -> List[str]:
        """Suggest edge cases to test."""
        
        inputs_str = ", ".join(f"{k}: {v}" for k, v in input_types.items())
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Identify important edge cases for comprehensive testing."),
            ("human", f"Function: {function_name}\n"
                     f"Inputs: {inputs_str}\n\n"
                     f"List 5 critical edge cases to test:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse edge cases
        edge_cases = [
            line.strip()
            for line in str(response.content).split('\n')
            if line.strip() and not line.startswith('#')
        ]
        
        return edge_cases[:5]
    
    def get_report(self) -> Dict[str, Any]:
        """Generate test report."""
        
        # Group results by status
        by_status = defaultdict(list)
        for result in self.test_results:
            by_status[result.status.value].append(result)
        
        # Group by test type
        by_type = defaultdict(int)
        for test in self.test_cases:
            by_type[test.test_type.value] += 1
        
        return {
            "summary": {
                "total_tests": self.statistics["total_tests"],
                "passed": self.statistics["tests_passed"],
                "failed": self.statistics["tests_failed"],
                "errors": self.statistics["tests_error"],
                "coverage": f"{self.statistics['coverage']:.1f}%"
            },
            "by_status": {k: len(v) for k, v in by_status.items()},
            "by_type": dict(by_type),
            "failed_tests": [
                {
                    "name": r.test_case.name,
                    "error": r.error_message
                }
                for r in self.test_results
                if r.status == TestStatus.FAILED
            ]
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get testing statistics."""
        return {
            "name": self.name,
            **self.statistics
        }


class UnitTestingPattern:
    """
    Unit Testing Pattern - Manage unit testing agents.
    
    Coordinates multiple testing agents for comprehensive coverage.
    """
    
    def __init__(self):
        self.agents: Dict[str, UnitTestingAgent] = {}
    
    def create_agent(self, name: str) -> UnitTestingAgent:
        """Create a new unit testing agent."""
        agent = UnitTestingAgent(name)
        self.agents[name] = agent
        print(f"   Created unit testing agent: {name}")
        return agent
    
    def run(
        self,
        agent_name: str,
        action: str,
        **kwargs
    ) -> Dict[str, Any]:
        """Execute 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_tests(
                kwargs.get("function_name", ""),
                kwargs.get("signature", ""),
                kwargs.get("description", "")
            )
            
            print(f"\n=== Test Generation by {agent_name} ===")
            print(f"Generated {len(tests)} test cases\n")
            
            for i, test in enumerate(tests, 1):
                print(f"{i}. {test.name} ({test.test_type.value})")
                print(f"   Function: {test.function_name}")
                print(f"   Expected: {test.expected_output}")
            
            return {
                "agent": agent_name,
                "action": "generate",
                "tests_generated": len(tests)
            }
        
        elif action == "execute":
            results = agent.execute_all_tests(kwargs.get("test_functions", {}))
            
            print(f"\n=== Test Execution by {agent_name} ===")
            report = agent.get_report()
            
            print(f"\nSummary:")
            for key, value in report["summary"].items():
                print(f"  {key}: {value}")
            
            if report["failed_tests"]:
                print(f"\nFailed Tests:")
                for test in report["failed_tests"]:
                    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("=== Unit Testing Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = UnitTestingPattern()
    
    # Create agent
    print("1. Setting up unit testing agent...")
    tester = pattern.create_agent("UnitTester")
    
    # Generate tests for a sample function
    print("\n2. Generating test cases...")
    
    result1 = pattern.run(
        "UnitTester",
        "generate",
        function_name="calculate_discount",
        signature="calculate_discount(price: float, discount_pct: float) -> float",
        description="Calculates discounted price given original price and discount percentage"
    )
    
    print("\n" + "="*60 + "\n")
    
    # Define test function
    def calculate_discount(price: float = 100.0, discount_pct: float = 10.0, data: str = "") -> float:
        """Calculate discount."""
        if price < 0 or discount_pct  100:
            raise ValueError("Invalid input")
        return price * (1 - discount_pct / 100)
    
    # Execute tests
    print("3. Executing tests...")
    
    result2 = pattern.run(
        "UnitTester",
        "execute",
        test_functions={"calculate_discount": calculate_discount}
    )
    
    # Print final statistics
    print("\n" + "="*60)
    print("\n4. Final Statistics:")
    stats = result2["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\nāœ“ Comprehensive unit testing with automated test generation complete!")

Agentic Architecture Pattern: Unit Testing

Pattern 261: 261 Unit Testing – Line-by-Line Explanation

Overview

Pattern 261: Unit Testing

Purpose: Automated unit test generation and execution with comprehensive coverage analysis. Generates test cases, executes tests, and provides detailed reporting.

Code Structure Analysis

File Information

  • Pattern Number: 261
  • Pattern Name: 261 Unit Testing
  • File Path: patterns\testing_validation\261_unit_testing.py
  • Total Lines: 487

Detailed Code Explanation

Module Docstring

"""
Pattern 261: Unit Testing

Purpose: Automated unit test generation and execution with comprehensive coverage analysis.
Generates test cases, executes tests, and provides detailed reporting.
"""

Explanation: This module docstring describes the purpose and key features of the 261 Unit 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 re
import traceback  # 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

TestType (Line 19)

class TestType(Enum):
    POSITIVE
    NEGATIVE
    BOUNDARY
    PARAMETRIC
    PROPERTY

Purpose: Types of unit tests.

Values: – POSITIVE: Represents positive – NEGATIVE: Represents negative – BOUNDARY: Represents boundary – PARAMETRIC: Represents parametric – PROPERTY: Represents property

TestStatus (Line 28)

class TestStatus(Enum):
    PASSED
    FAILED
    SKIPPED
    ERROR

Purpose: Test execution status.

Values: – PASSED: Represents passed – FAILED: Execution failed – SKIPPED: Represents skipped – ERROR: Error severity level

Classes

TestType (Line 19)

Description: Types of unit tests.

TestStatus (Line 28)

Description: Test execution status.

TestCase (Line 37)

Description: Unit test case.

TestResult (Line 51)

Description: Test execution result.

UnitTestingAgent (Line 61)

Description: Unit Testing Agent – Automated test generation and execution.

Features: – Test case generation from specifications – Multiple test types (positive, negative, boundary) – Test execution with assertions – Coverage analysis – Result reporting – AI-powered test suggestions – Assertion generation – Edge case identification

Methods:

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

Purpose: No method docstring.

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

generate_tests (Line 94)
def generate_tests(self, function_name, function_signature, description):
    """Generate test cases for a function."""

Purpose: Generate test cases for a function.

Parameters: – self: Instance reference – function_name: Parameter for function_name – function_signature: Parameter for function_signature – description: Parameter for description

_parse_test_cases (Line 132)
def _parse_test_cases(self, content, function_name):
    """Parse test cases from LLM response."""

Purpose: Parse test cases from LLM response.

Parameters: – self: Instance reference – content: Parameter for content – function_name: Parameter for function_name

execute_test (Line 180)
def execute_test(self, test_case, test_function):
    """Execute a single test case."""

Purpose: Execute a single test case.

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

execute_all_tests (Line 239)
def execute_all_tests(self, test_functions):
    """Execute all registered test cases."""

Purpose: Execute all registered test cases.

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

_calculate_coverage (Line 257)
def _calculate_coverage(self):
    """Calculate test coverage."""

Purpose: Calculate test coverage.

Parameters: – self: Instance reference

generate_assertions (Line 265)
def generate_assertions(self, function_name, output_type, constraints):
    """Generate assertions for a function."""

Purpose: Generate assertions for a function.

Parameters: – self: Instance reference – function_name: Parameter for function_name – output_type: Parameter for output_type – constraints: Parameter for constraints

suggest_edge_cases (Line 295)
def suggest_edge_cases(self, function_name, input_types):
    """Suggest edge cases to test."""

Purpose: Suggest edge cases to test.

Parameters: – self: Instance reference – function_name: Parameter for function_name – input_types: Parameter for input_types

get_report (Line 323)
def get_report(self):
    """Generate test report."""

Purpose: Generate test report.

Parameters: – self: Instance reference

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

Purpose: Get testing statistics.

Parameters: – self: Instance reference

UnitTestingPattern (Line 364)

Description: Unit Testing Pattern – Manage unit testing agents.

Coordinates multiple testing agents for comprehensive coverage.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

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

Purpose: Create a new unit testing agent.

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

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

Purpose: Execute 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 = TestType(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