aishastr.live

Agentic Architecture Pattern: Test Double

Written by

admin

in


    """
Pattern 267: Test Double Pattern with LangChain/LangGraph

Purpose: Comprehensive test double implementations (dummy, fake, spy, mock, stub).
Provides intelligent test double orchestration with LLM-powered behavior analysis.
"""

from typing import Dict, Any, List, Optional, Callable, TypedDict
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
from langgraph.graph import StateGraph, END
import json


class DoubleType(Enum):
    """Types of test doubles."""
    DUMMY = "dummy"  # No logic, fills parameter
    FAKE = "fake"  # Working implementation, simplified
    SPY = "spy"  # Records calls to real object
    MOCK = "mock"  # Pre-programmed expectations
    STUB = "stub"  # Pre-programmed responses


class VerificationStatus(Enum):
    """Status of expectation verification."""
    NOT_VERIFIED = "not_verified"
    PASSED = "passed"
    FAILED = "failed"


@dataclass
class Expectation:
    """Mock expectation."""
    method: str
    expected_calls: int
    expected_args: Optional[tuple] = None
    expected_kwargs: Optional[Dict[str, Any]] = None
    return_value: Any = None


@dataclass
class CallRecord:
    """Record of method call."""
    method: str
    args: tuple
    kwargs: Dict[str, Any]
    result: Any
    double_type: DoubleType
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class DoubleAnalysis:
    """Analysis of test double usage."""
    recommended_type: DoubleType
    reasoning: str
    complexity_score: float
    usage_patterns: List[str]


class TestDoubleState(TypedDict):
    """State for test double workflow."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    double_type: DoubleType
    analysis: Optional[DoubleAnalysis]
    result: Any
    verification_status: VerificationStatus
    expectations_met: bool
    insights: List[str]
    call_count: int


class IntelligentTestDouble:
    """
    Intelligent Test Double with LangGraph workflow integration.
    
    Features:
    - All test double types (dummy, fake, spy, mock, stub)
    - LLM-powered behavior analysis
    - Expectation verification
    - Call recording and replay
    - Pattern detection
    - Intelligent recommendations
    """
    
    def __init__(self, name: str, double_type: DoubleType, model_name: str = "gpt-4", temperature: float = 0.5):
        self.name = name
        self.double_type = double_type
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        
        # Configuration
        self.methods: Dict[str, Any] = {}
        self.expectations: Dict[str, List[Expectation]] = defaultdict(list)
        self.fake_implementations: Dict[str, Callable] = {}
        
        # Tracking
        self.call_records: List[CallRecord] = []
        self.call_counts: Dict[str, int] = defaultdict(int)
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_calls": 0,
            "verifications_passed": 0,
            "verifications_failed": 0,
            "double_type": double_type.value
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(TestDoubleState)
        
        workflow.add_node("analyze_double_type", self._analyze_double_type_node)
        workflow.add_node("configure_behavior", self._configure_behavior_node)
        workflow.add_node("execute_call", self._execute_call_node)
        workflow.add_node("verify_expectations", self._verify_expectations_node)
        workflow.add_node("generate_report", self._generate_report_node)
        
        workflow.set_entry_point("analyze_double_type")
        workflow.add_edge("analyze_double_type", "configure_behavior")
        workflow.add_edge("configure_behavior", "execute_call")
        workflow.add_edge("execute_call", "verify_expectations")
        workflow.add_edge("verify_expectations", "generate_report")
        workflow.add_edge("generate_report", END)
        
        return workflow.compile()
    
    def _analyze_double_type_node(self, state: TestDoubleState) -> TestDoubleState:
        """Analyze test double type and usage."""
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        
        # Increment call count
        self.call_counts[method_name] += 1
        state["call_count"] = self.call_counts[method_name]
        
        # Use LLM to analyze if this is first call to method
        if self.call_counts[method_name] == 1 and not self.methods.get(method_name):
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are an expert in test double patterns and testing strategies."),
                ("user", """Analyze this method call for test double usage:

Method: {method_name}
Arguments: {args}
Keyword Arguments: {kwargs}
Current Double Type: {double_type}

Provide analysis as JSON with:
- recommended_type: Best double type for this method
- reasoning: Why this type is recommended
- complexity_score: 0.0-1.0 score for implementation complexity
- usage_patterns: List of observed usage patterns

Double types: dummy (parameter filler), fake (simplified working), spy (record calls), mock (verify expectations), stub (fixed responses)""")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "method_name": method_name,
                "args": str(args),
                "kwargs": str(kwargs),
                "double_type": self.double_type.value
            })
            
            try:
                analysis_data = json.loads(response.content)
                state["analysis"] = DoubleAnalysis(
                    recommended_type=DoubleType(analysis_data.get("recommended_type", self.double_type.value)),
                    reasoning=analysis_data.get("reasoning", "Default reasoning"),
                    complexity_score=analysis_data.get("complexity_score", 0.5),
                    usage_patterns=analysis_data.get("usage_patterns", [])
                )
            except:
                state["analysis"] = DoubleAnalysis(
                    recommended_type=self.double_type,
                    reasoning="Using configured double type",
                    complexity_score=0.5,
                    usage_patterns=[]
                )
        
        return state
    
    def _configure_behavior_node(self, state: TestDoubleState) -> TestDoubleState:
        """Configure double behavior."""
        method_name = state["method_name"]
        
        # Store method configuration if not exists
        if method_name not in self.methods:
            self.methods[method_name] = {
                "return_value": None,
                "configured": False
            }
        
        return state
    
    def _execute_call_node(self, state: TestDoubleState) -> TestDoubleState:
        """Execute the test double call."""
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        double_type = state["double_type"]
        
        result = None
        
        if double_type == DoubleType.DUMMY:
            # Dummy: Do nothing, return None
            result = None
        
        elif double_type == DoubleType.STUB:
            # Stub: Return configured value
            method_config = self.methods.get(method_name, {})
            result = method_config.get("return_value")
        
        elif double_type == DoubleType.FAKE:
            # Fake: Use simplified implementation
            if method_name in self.fake_implementations:
                try:
                    result = self.fake_implementations[method_name](*args, **kwargs)
                except Exception as e:
                    result = {"error": str(e)}
            else:
                # Default fake behavior
                result = {"status": "fake_success", "method": method_name, "data": "fake_data"}
        
        elif double_type == DoubleType.SPY:
            # Spy: Record call and execute real implementation if available
            if method_name in self.fake_implementations:
                try:
                    result = self.fake_implementations[method_name](*args, **kwargs)
                except Exception as e:
                    result = {"error": str(e)}
            else:
                result = {"status": "spy_recorded", "method": method_name}
        
        elif double_type == DoubleType.MOCK:
            # Mock: Return expected value if configured
            expectations = self.expectations.get(method_name, [])
            if expectations:
                # Use first expectation's return value
                result = expectations[0].return_value
            else:
                result = {"status": "mock_success", "method": method_name}
        
        state["result"] = result
        
        # Record call
        call_record = CallRecord(
            method=method_name,
            args=args,
            kwargs=kwargs,
            result=result,
            double_type=double_type
        )
        self.call_records.append(call_record)
        self.stats["total_calls"] += 1
        
        return state
    
    def _verify_expectations_node(self, state: TestDoubleState) -> TestDoubleState:
        """Verify expectations for mocks."""
        method_name = state["method_name"]
        args = state["args"]
        kwargs = state["kwargs"]
        
        expectations_met = True
        
        if state["double_type"] == DoubleType.MOCK:
            expectations = self.expectations.get(method_name, [])
            
            for expectation in expectations:
                # Check call count
                actual_calls = self.call_counts[method_name]
                if actual_calls != expectation.expected_calls:
                    expectations_met = False
                    state["insights"] = state.get("insights", [])
                    state["insights"].append(
                        f"Expected {expectation.expected_calls} calls to {method_name}, got {actual_calls}"
                    )
                
                # Check arguments if specified
                if expectation.expected_args is not None and args != expectation.expected_args:
                    expectations_met = False
                    state["insights"] = state.get("insights", [])
                    state["insights"].append(
                        f"Expected args {expectation.expected_args}, got {args}"
                    )
                
                if expectation.expected_kwargs is not None and kwargs != expectation.expected_kwargs:
                    expectations_met = False
                    state["insights"] = state.get("insights", [])
                    state["insights"].append(
                        f"Expected kwargs {expectation.expected_kwargs}, got {kwargs}"
                    )
        
        state["expectations_met"] = expectations_met
        state["verification_status"] = VerificationStatus.PASSED if expectations_met else VerificationStatus.FAILED
        
        if expectations_met:
            self.stats["verifications_passed"] += 1
        else:
            self.stats["verifications_failed"] += 1
        
        return state
    
    def _generate_report_node(self, state: TestDoubleState) -> TestDoubleState:
        """Generate insights report."""
        if not state.get("insights"):
            state["insights"] = []
        
        # Add summary insight
        state["insights"].append(
            f"Test double '{self.name}' of type {state['double_type'].value} executed {state['method_name']}"
        )
        
        return state
    
    def configure_method(
        self,
        method_name: str,
        return_value: Any = None,
        implementation: Optional[Callable] = None
    ):
        """Configure method behavior."""
        self.methods[method_name] = {
            "return_value": return_value,
            "configured": True
        }
        
        if implementation:
            self.fake_implementations[method_name] = implementation
    
    def expect_call(
        self,
        method_name: str,
        times: int = 1,
        with_args: Optional[tuple] = None,
        with_kwargs: Optional[Dict[str, Any]] = None,
        returns: Any = None
    ):
        """Set expectation for mock."""
        expectation = Expectation(
            method=method_name,
            expected_calls=times,
            expected_args=with_args,
            expected_kwargs=with_kwargs,
            return_value=returns
        )
        self.expectations[method_name].append(expectation)
    
    def call_method(self, method_name: str, *args, **kwargs) -> Any:
        """Execute method call through workflow."""
        initial_state: TestDoubleState = {
            "method_name": method_name,
            "args": args,
            "kwargs": kwargs,
            "double_type": self.double_type,
            "analysis": None,
            "result": None,
            "verification_status": VerificationStatus.NOT_VERIFIED,
            "expectations_met": True,
            "insights": [],
            "call_count": 0
        }
        
        final_state = self.workflow.invoke(initial_state)
        return final_state["result"]
    
    def verify(self, method_name: str, times: Optional[int] = None) -> bool:
        """Verify method was called expected times."""
        actual_calls = self.call_counts[method_name]
        
        if times is None:
            return actual_calls > 0
        
        return actual_calls == times
    
    def get_calls(self, method_name: Optional[str] = None) -> List[CallRecord]:
        """Get call records."""
        if method_name:
            return [call for call in self.call_records if call.method == method_name]
        return self.call_records
    
    def reset(self):
        """Reset double state."""
        self.call_records.clear()
        self.call_counts.clear()
        self.stats["total_calls"] = 0
        self.stats["verifications_passed"] = 0
        self.stats["verifications_failed"] = 0
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get test double statistics."""
        return {
            "name": self.name,
            **self.stats,
            "methods_configured": len(self.methods),
            "expectations_set": sum(len(exps) for exps in self.expectations.values())
        }


class TestDoublePattern:
    """
    Comprehensive Test Double Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.doubles: Dict[str, IntelligentTestDouble] = {}
        self.model_name = model_name
    
    def create_dummy(self, name: str) -> IntelligentTestDouble:
        """Create a dummy double."""
        double = IntelligentTestDouble(name, DoubleType.DUMMY, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_stub(self, name: str) -> IntelligentTestDouble:
        """Create a stub double."""
        double = IntelligentTestDouble(name, DoubleType.STUB, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_fake(self, name: str) -> IntelligentTestDouble:
        """Create a fake double."""
        double = IntelligentTestDouble(name, DoubleType.FAKE, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_spy(self, name: str) -> IntelligentTestDouble:
        """Create a spy double."""
        double = IntelligentTestDouble(name, DoubleType.SPY, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def create_mock(self, name: str) -> IntelligentTestDouble:
        """Create a mock double."""
        double = IntelligentTestDouble(name, DoubleType.MOCK, model_name=self.model_name)
        self.doubles[name] = double
        return double
    
    def call(self, double_name: str, method_name: str, *args, **kwargs) -> Dict[str, Any]:
        """Execute method on double."""
        if double_name not in self.doubles:
            return {"error": f"Double {double_name} not found"}
        
        double = self.doubles[double_name]
        result = double.call_method(method_name, *args, **kwargs)
        
        return {
            "double": double_name,
            "type": double.double_type.value,
            "method": method_name,
            "result": result,
            "call_count": double.call_counts[method_name]
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_calls = sum(d.stats["total_calls"] for d in self.doubles.values())
        total_verifications_passed = sum(d.stats["verifications_passed"] for d in self.doubles.values())
        total_verifications_failed = sum(d.stats["verifications_failed"] for d in self.doubles.values())
        
        return {
            "total_doubles": len(self.doubles),
            "total_calls": total_calls,
            "verifications_passed": total_verifications_passed,
            "verifications_failed": total_verifications_failed
        }


if __name__ == "__main__":
    print("=== Test Double Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Dummy Double (Parameter Filler)")
    print("-" * 50)
    pattern = TestDoublePattern(model_name="gpt-4")
    
    dummy = pattern.create_dummy("DummyLogger")
    
    print("\nCalling dummy methods (no functionality)...")
    result1 = pattern.call("DummyLogger", "log", message="Test message")
    print(f"Result: {result1['result']}")
    
    result2 = pattern.call("DummyLogger", "debug", message="Debug info")
    print(f"Result: {result2['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 2: Stub Double (Fixed Responses)")
    print("-" * 50)
    
    stub = pattern.create_stub("DatabaseStub")
    
    # Configure stub responses
    stub.configure_method("get_user", return_value={"id": 1, "name": "Test User", "email": "test@example.com"})
    stub.configure_method("get_count", return_value=42)
    
    print("\nCalling stub methods...")
    result3 = pattern.call("DatabaseStub", "get_user", user_id=1)
    print(f"get_user result: {result3['result']}")
    
    result4 = pattern.call("DatabaseStub", "get_count")
    print(f"get_count result: {result4['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Fake Double (Simplified Implementation)")
    print("-" * 50)
    
    fake = pattern.create_fake("FakeEmailService")
    
    # Configure fake implementation
    def fake_send_email(to: str, subject: str, body: str = "") -> Dict[str, Any]:
        return {
            "status": "sent",
            "message_id": f"FAKE_{hash(to + subject)}",
            "to": to,
            "subject": subject
        }
    
    fake.configure_method("send_email", implementation=fake_send_email)
    
    print("\nCalling fake email service...")
    result5 = pattern.call("FakeEmailService", "send_email", 
                          to="user@example.com", 
                          subject="Test Email",
                          body="This is a test")
    print(f"Result: {result5['result']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 4: Spy Double (Record Calls)")
    print("-" * 50)
    
    spy = pattern.create_spy("MetricsSpy")
    
    # Configure spy implementation
    def track_metric(metric_name: str, value: float) -> Dict[str, Any]:
        return {
            "tracked": True,
            "metric": metric_name,
            "value": value,
            "timestamp": datetime.now().isoformat()
        }
    
    spy.configure_method("track", implementation=track_metric)
    
    print("\nCalling spy methods (recording calls)...")
    result6 = pattern.call("MetricsSpy", "track", metric_name="page_views", value=100.0)
    print(f"Result: {result6['result']}")
    
    result7 = pattern.call("MetricsSpy", "track", metric_name="conversions", value=5.0)
    print(f"Result: {result7['result']}")
    
    # Show recorded calls
    print(f"\nRecorded calls to 'track': {spy.call_counts['track']}")
    calls = spy.get_calls("track")
    for i, call in enumerate(calls, 1):
        print(f"  Call {i}: args={call.args}, kwargs={call.kwargs}")
    
    print("\n" + "=" * 50)
    print("\nScenario 5: Mock Double (Expectation Verification)")
    print("-" * 50)
    
    mock = pattern.create_mock("PaymentMock")
    
    # Set expectations
    mock.expect_call(
        "process_payment",
        times=2,
        returns={"status": "success", "transaction_id": "TXN123"}
    )
    
    print("\nCalling mock methods (with expectations)...")
    result8 = pattern.call("PaymentMock", "process_payment", amount=100.0)
    print(f"Result: {result8['result']}")
    
    result9 = pattern.call("PaymentMock", "process_payment", amount=50.0)
    print(f"Result: {result9['result']}")
    
    # Verify expectations
    print(f"\nVerifying 'process_payment' called 2 times: {mock.verify('process_payment', times=2)}")
    
    print("\n" + "=" * 50)
    print("\nScenario 6: Multiple Calls and Pattern Detection")
    print("-" * 50)
    
    multi_stub = pattern.create_stub("ConfigStub")
    multi_stub.configure_method("get_setting", return_value={"timeout": 30})
    
    print("\nMaking multiple calls to detect patterns...")
    for i in range(3):
        result = pattern.call("ConfigStub", "get_setting", key="timeout")
        print(f"Call {i+1}: {result['result']}")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"\nOverall Statistics:")
    print(f"  Total Doubles: {overall_stats['total_doubles']}")
    print(f"  Total Calls: {overall_stats['total_calls']}")
    print(f"  Verifications Passed: {overall_stats['verifications_passed']}")
    print(f"  Verifications Failed: {overall_stats['verifications_failed']}")
    
    print(f"\nIndividual Double Statistics:")
    for name, double in pattern.doubles.items():
        stats = double.get_statistics()
        print(f"\n{name} ({stats['double_type']}):")
        print(f"  Total Calls: {stats['total_calls']}")
        print(f"  Methods Configured: {stats['methods_configured']}")
        print(f"  Expectations Set: {stats['expectations_set']}")
    
    print("\n" + "=" * 50)
    print("Test double pattern demonstration complete!")
    print("=" * 50)

Agentic Architecture Pattern: Test Double

Pattern 267: Test Double Pattern – Line-by-Line Explanation

Overview

This pattern implements comprehensive test double implementations (dummy, fake, spy, mock, stub) with intelligent orchestration and LLM-powered behavior analysis. Test doubles are objects that stand in for real objects during testing.

Code Explanation

Lines 1-17: Imports

"""
Pattern 267: Test Double Pattern with LangChain/LangGraph

Purpose: Comprehensive test double implementations (dummy, fake, spy, mock, stub).
Provides intelligent test double orchestration with LLM-powered behavior analysis.
"""

Explanation: Module docstring – this pattern provides all 5 types of test doubles from Martin Fowler’s testing taxonomy.

from typing import Dict, Any, List, Optional, Callable, TypedDict
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
from langgraph.graph import StateGraph, END
import json

Explanation: Standard imports plus json for parsing LLM responses.

Lines 19-30: Enum Definitions

class DoubleType(Enum):
    """Types of test doubles."""
    DUMMY = "dummy"  # No logic, fills parameter
    FAKE = "fake"  # Working implementation, simplified
    SPY = "spy"  # Records calls to real object
    MOCK = "mock"  # Pre-programmed expectations
    STUB = "stub"  # Pre-programmed responses

Explanation: Five test double types: – DUMMY: Fills parameter slots, never actually used (e.g., null object) – FAKE: Simplified working implementation (e.g., in-memory database) – SPY: Wraps real object and records calls for verification – MOCK: Has expectations about how it should be called – STUB: Returns canned responses (like pattern 266)

class VerificationStatus(Enum):
    """Status of expectation verification."""
    NOT_VERIFIED = "not_verified"
    PASSED = "passed"
    FAILED = "failed"

Explanation: Tracks whether mock expectations were verified.

Lines 32-71: Dataclasses

@dataclass
class Expectation:
    """Mock expectation."""
    method: str
    expected_calls: int
    expected_args: Optional[tuple] = None
    expected_kwargs: Optional[Dict[str, Any]] = None
    return_value: Any = None

Explanation: Defines what a mock expects: – How many times method should be called – What arguments it should receive – What it should return

Example: Expect process_payment called exactly 2 times with amount > 0.

@dataclass
class CallRecord:
    """Record of method call."""
    method: str
    args: tuple
    kwargs: Dict[str, Any]
    result: Any
    double_type: DoubleType
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Records each method invocation for spies and verification.

@dataclass
class DoubleAnalysis:
    """Analysis of test double usage."""
    recommended_type: DoubleType
    reasoning: str
    complexity_score: float
    usage_patterns: List[str]

Explanation: LLM analysis recommending best double type based on usage patterns.

Lines 73-85: TypedDict State

class TestDoubleState(TypedDict):
    """State for test double workflow."""
    method_name: str
    args: tuple
    kwargs: Dict[str, Any]
    double_type: DoubleType
    analysis: Optional[DoubleAnalysis]
    result: Any
    verification_status: VerificationStatus
    expectations_met: bool
    insights: List[str]
    call_count: int

Explanation: Workflow state tracking method call, analysis, result, and verification status.

Lines 87-137: IntelligentTestDouble Initialization

class IntelligentTestDouble:
    """
    Intelligent Test Double with LangGraph workflow integration.

    Features:
    - All test double types (dummy, fake, spy, mock, stub)
    - LLM-powered behavior analysis
    - Expectation verification
    - Call recording and replay
    - Pattern detection
    - Intelligent recommendations
    """

Explanation: Main class implementing all 5 double types with AI analysis.

def __init__(self, name: str, double_type: DoubleType, model_name: str = "gpt-4", temperature: float = 0.5):
    self.name = name
    self.double_type = double_type
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Configuration
    self.methods: Dict[str, Any] = {}
    self.expectations: Dict[str, List[Expectation]] = defaultdict(list)
    self.fake_implementations: Dict[str, Callable] = {}

Explanation: Initialize with: – Name and type of double – LLM for analysis – Method configurations – Mock expectations – Fake implementations (actual functions)

    # Tracking
    self.call_records: List[CallRecord] = []
    self.call_counts: Dict[str, int] = defaultdict(int)

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_calls": 0,
        "verifications_passed": 0,
        "verifications_failed": 0,
        "double_type": double_type.value
    }

Explanation: Initialize call tracking, workflow, and statistics.

Lines 139-157: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(TestDoubleState)

    workflow.add_node("analyze_double_type", self._analyze_double_type_node)
    workflow.add_node("configure_behavior", self._configure_behavior_node)
    workflow.add_node("execute_call", self._execute_call_node)
    workflow.add_node("verify_expectations", self._verify_expectations_node)
    workflow.add_node("generate_report", self._generate_report_node)

Explanation: Create 5-node workflow: 1. Analyze what type of double is needed 2. Configure behavior based on type 3. Execute the call 4. Verify expectations (for mocks) 5. Generate report

    workflow.set_entry_point("analyze_double_type")
    workflow.add_edge("analyze_double_type", "configure_behavior")
    workflow.add_edge("configure_behavior", "execute_call")
    workflow.add_edge("execute_call", "verify_expectations")
    workflow.add_edge("verify_expectations", "generate_report")
    workflow.add_edge("generate_report", END)

    return workflow.compile()

Explanation: Define linear flow through all nodes.

Lines 159-216: Analyze Double Type Node (LLM-Powered)

def _analyze_double_type_node(self, state: TestDoubleState) -> TestDoubleState:
    """Analyze test double type and usage."""
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]

    # Increment call count
    self.call_counts[method_name] += 1
    state["call_count"] = self.call_counts[method_name]

Explanation: Extract parameters and update call counter.

    # Use LLM to analyze if this is first call to method
    if self.call_counts[method_name] == 1 and not self.methods.get(method_name):
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in test double patterns and testing strategies."),
            ("user", """Analyze this method call for test double usage:

Method: {method_name}
Arguments: {args}
Keyword Arguments: {kwargs}
Current Double Type: {double_type}

Provide analysis as JSON with:
- recommended_type: Best double type for this method
- reasoning: Why this type is recommended
- complexity_score: 0.0-1.0 score for implementation complexity
- usage_patterns: List of observed usage patterns

Double types: dummy (parameter filler), fake (simplified working), spy (record calls), mock (verify expectations), stub (fixed responses)""")
        ])

Explanation: On first call, ask LLM to analyze method and recommend best double type.

        chain = prompt | self.llm
        response = chain.invoke({
            "method_name": method_name,
            "args": str(args),
            "kwargs": str(kwargs),
            "double_type": self.double_type.value
        })

        try:
            analysis_data = json.loads(response.content)
            state["analysis"] = DoubleAnalysis(
                recommended_type=DoubleType(analysis_data.get("recommended_type", self.double_type.value)),
                reasoning=analysis_data.get("reasoning", "Default reasoning"),
                complexity_score=analysis_data.get("complexity_score", 0.5),
                usage_patterns=analysis_data.get("usage_patterns", [])
            )
        except:
            state["analysis"] = DoubleAnalysis(
                recommended_type=self.double_type,
                reasoning="Using configured double type",
                complexity_score=0.5,
                usage_patterns=[]
            )

Explanation: Invoke LLM and parse JSON response into analysis object. Fallback to current type if parsing fails.

    return state

Lines 218-230: Configure Behavior Node

def _configure_behavior_node(self, state: TestDoubleState) -> TestDoubleState:
    """Configure double behavior."""
    method_name = state["method_name"]

    # Store method configuration if not exists
    if method_name not in self.methods:
        self.methods[method_name] = {
            "return_value": None,
            "configured": False
        }

    return state

Explanation: Ensure method has configuration entry. Simple node that prepares for execution.

Lines 232-291: Execute Call Node

def _execute_call_node(self, state: TestDoubleState) -> TestDoubleState:
    """Execute the test double call."""
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]
    double_type = state["double_type"]

    result = None

Explanation: Extract parameters and initialize result.

    if double_type == DoubleType.DUMMY:
        # Dummy: Do nothing, return None
        result = None

Explanation: DUMMY – simplest double, does nothing. Used when object is required but never actually used.

    elif double_type == DoubleType.STUB:
        # Stub: Return configured value
        method_config = self.methods.get(method_name, {})
        result = method_config.get("return_value")

Explanation: STUB – returns pre-configured value. No logic, just returns what you told it to.

    elif double_type == DoubleType.FAKE:
        # Fake: Use simplified implementation
        if method_name in self.fake_implementations:
            try:
                result = self.fake_implementations[method_name](*args, **kwargs)
            except Exception as e:
                result = {"error": str(e)}
        else:
            # Default fake behavior
            result = {"status": "fake_success", "method": method_name, "data": "fake_data"}

Explanation: FAKE – has working implementation, but simplified (e.g., in-memory instead of real database). Actually executes logic.

    elif double_type == DoubleType.SPY:
        # Spy: Record call and execute real implementation if available
        if method_name in self.fake_implementations:
            try:
                result = self.fake_implementations[method_name](*args, **kwargs)
            except Exception as e:
                result = {"error": str(e)}
        else:
            result = {"status": "spy_recorded", "method": method_name}

Explanation: SPY – like fake but emphasizes call recording. Executes real code while tracking calls.

    elif double_type == DoubleType.MOCK:
        # Mock: Return expected value if configured
        expectations = self.expectations.get(method_name, [])
        if expectations:
            # Use first expectation's return value
            result = expectations[0].return_value
        else:
            result = {"status": "mock_success", "method": method_name}

Explanation: MOCK – returns value from expectations. Unlike stub, mock will be verified later.

    state["result"] = result

    # Record call
    call_record = CallRecord(
        method=method_name,
        args=args,
        kwargs=kwargs,
        result=result,
        double_type=double_type
    )
    self.call_records.append(call_record)
    self.stats["total_calls"] += 1

    return state

Explanation: Store result, record call in history, update statistics.

Lines 293-346: Verify Expectations Node

def _verify_expectations_node(self, state: TestDoubleState) -> TestDoubleState:
    """Verify expectations for mocks."""
    method_name = state["method_name"]
    args = state["args"]
    kwargs = state["kwargs"]

    expectations_met = True

Explanation: Initialize verification – only meaningful for mocks.

    if state["double_type"] == DoubleType.MOCK:
        expectations = self.expectations.get(method_name, [])

        for expectation in expectations:
            # Check call count
            actual_calls = self.call_counts[method_name]
            if actual_calls != expectation.expected_calls:
                expectations_met = False
                state["insights"] = state.get("insights", [])
                state["insights"].append(
                    f"Expected {expectation.expected_calls} calls to {method_name}, got {actual_calls}"
                )

Explanation: Verify mock was called expected number of times. This is the key difference between mocks and stubs – mocks verify behavior.

            # Check arguments if specified
            if expectation.expected_args is not None and args != expectation.expected_args:
                expectations_met = False
                state["insights"] = state.get("insights", [])
                state["insights"].append(
                    f"Expected args {expectation.expected_args}, got {args}"
                )

            if expectation.expected_kwargs is not None and kwargs != expectation.expected_kwargs:
                expectations_met = False
                state["insights"] = state.get("insights", [])
                state["insights"].append(
                    f"Expected kwargs {expectation.expected_kwargs}, got {kwargs}"
                )

Explanation: Verify arguments match expectations if specified.

    state["expectations_met"] = expectations_met
    state["verification_status"] = VerificationStatus.PASSED if expectations_met else VerificationStatus.FAILED

    if expectations_met:
        self.stats["verifications_passed"] += 1
    else:
        self.stats["verifications_failed"] += 1

    return state

Explanation: Set verification status and update statistics.

Lines 348-361: Generate Report Node

def _generate_report_node(self, state: TestDoubleState) -> TestDoubleState:
    """Generate insights report."""
    if not state.get("insights"):
        state["insights"] = []

    # Add summary insight
    state["insights"].append(
        f"Test double '{self.name}' of type {state['double_type'].value} executed {state['method_name']}"
    )

    return state

Explanation: Add summary insight to state. Simple reporting node.

Lines 363-403: Public Configuration Methods

def configure_method(
    self,
    method_name: str,
    return_value: Any = None,
    implementation: Optional[Callable] = None
):
    """Configure method behavior."""
    self.methods[method_name] = {
        "return_value": return_value,
        "configured": True
    }

    if implementation:
        self.fake_implementations[method_name] = implementation

Explanation: Configure what method should return (for stubs) or how it should behave (for fakes – actual function).

def expect_call(
    self,
    method_name: str,
    times: int = 1,
    with_args: Optional[tuple] = None,
    with_kwargs: Optional[Dict[str, Any]] = None,
    returns: Any = None
):
    """Set expectation for mock."""
    expectation = Expectation(
        method=method_name,
        expected_calls=times,
        expected_args=with_args,
        expected_kwargs=with_kwargs,
        return_value=returns
    )
    self.expectations[method_name].append(expectation)

Explanation: Set expectations for mocks. Used in test setup phase. Example: mock.expect_call("save", times=1, with_args=("data",)).

def call_method(self, method_name: str, *args, **kwargs) -> Any:
    """Execute method call through workflow."""
    initial_state: TestDoubleState = {
        "method_name": method_name,
        "args": args,
        "kwargs": kwargs,
        "double_type": self.double_type,
        "analysis": None,
        "result": None,
        "verification_status": VerificationStatus.NOT_VERIFIED,
        "expectations_met": True,
        "insights": [],
        "call_count": 0
    }

    final_state = self.workflow.invoke(initial_state)
    return final_state["result"]

Explanation: Main entry point – execute workflow and return result.

Lines 428-475: TestDoublePattern Coordinator

class TestDoublePattern:
    """
    Comprehensive Test Double Pattern coordinator with LangChain/LangGraph.
    """

    def __init__(self, model_name: str = "gpt-4"):
        self.doubles: Dict[str, IntelligentTestDouble] = {}
        self.model_name = model_name

Explanation: Coordinator managing multiple test doubles.

    def create_dummy(self, name: str) -> IntelligentTestDouble:
        """Create a dummy double."""
        double = IntelligentTestDouble(name, DoubleType.DUMMY, model_name=self.model_name)
        self.doubles[name] = double
        return double

Explanation: Factory methods for each double type. Clear, explicit API.

Similar factory methods for stub, fake, spy, and mock types follow the same pattern.

Lines 477-675: Demo Code

Scenario 1: Dummy Double

dummy = pattern.create_dummy("DummyLogger")
result1 = pattern.call("DummyLogger", "log", message="Test message")

Explanation: Dummy does nothing, returns None. Used when parameter is required but not used (e.g., logger in test).

Scenario 2: Stub Double

stub = pattern.create_stub("DatabaseStub")
stub.configure_method("get_user", return_value={"id": 1, "name": "Test User", "email": "test@example.com"})

Explanation: Stub returns fixed value. Good for tests that just need data, don’t care about interactions.

Scenario 3: Fake Double

fake = pattern.create_fake("FakeEmailService")

def fake_send_email(to: str, subject: str, body: str = "") -> Dict[str, Any]:
    return {
        "status": "sent",
        "message_id": f"FAKE_{hash(to + subject)}",
        "to": to,
        "subject": subject
    }

fake.configure_method("send_email", implementation=fake_send_email)

Explanation: Fake has working implementation (simplified). Actually processes inputs, but doesn’t hit real email server.

Scenario 4: Spy Double

spy = pattern.create_spy("MetricsSpy")

def track_metric(metric_name: str, value: float) -> Dict[str, Any]:
    return {
        "tracked": True,
        "metric": metric_name,
        "value": value,
        "timestamp": datetime.now().isoformat()
    }

spy.configure_method("track", implementation=track_metric)

# Later, verify calls were made
calls = spy.get_calls("track")
for i, call in enumerate(calls, 1):
    print(f"  Call {i}: args={call.args}, kwargs={call.kwargs}")

Explanation: Spy executes real code AND records calls. Used to verify side effects.

Scenario 5: Mock Double

mock = pattern.create_mock("PaymentMock")

# Set expectations
mock.expect_call(
    "process_payment",
    times=2,
    returns={"status": "success", "transaction_id": "TXN123"}
)

# Call twice
result8 = pattern.call("PaymentMock", "process_payment", amount=100.0)
result9 = pattern.call("PaymentMock", "process_payment", amount=50.0)

# Verify expectations
print(f"\nVerifying 'process_payment' called 2 times: {mock.verify('process_payment', times=2)}")

Explanation: Mock has expectations that must be met. Test fails if method not called expected times. Key difference from stub – verification is mandatory.

Test Double Comparison

Double Purpose Has Logic? Verifies Behavior? Example
Dummy Fill parameter No No Null logger
Stub Provide data No No Return fixed user
Fake Working impl Yes No In-memory DB
Spy Record calls Yes Optional Track metrics
Mock Verify expectations No Yes Assert called twice

When to Use Each

  • Dummy: Parameter required but not used (null object pattern)
  • Stub: Need data, don’t care about interactions
  • Fake: Need working implementation but real one is too slow/complex
  • Spy: Need to verify side effects while using real code
  • Mock: Need to verify specific interactions occurred

Key Design Patterns

  1. Strategy Pattern: Different double types implement different strategies
  2. Template Method: Common workflow with type-specific execution
  3. Observer Pattern: Call recording observes all invocations
  4. Proxy Pattern: Doubles proxy real objects
  5. Factory Pattern: Coordinator creates appropriate doubles

Performance Considerations

  • Dummy/Stub/Mock: O(1) – instant
  • Fake/Spy: O(implementation) – depends on fake logic
  • Verification: O(calls × expectations) – linear in recorded calls
  • Memory: O(calls) – stores all call records

Extension Points

  1. Add partial mocks (some methods real, some mocked)
  2. Implement argument matchers (any(), contains(), etc.)
  3. Add sequence verification (call A before call B)
  4. Implement call chains for fluent APIs
  5. Add async/await support for async methods

Comments

Leave a Reply

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

More posts