Fake Driven Testing

Fake-Driven Testing Architecture for Python

Use this skill when: Writing tests, fixing bugs, adding features, or modifying gateway layers in Python projects.

No prerequisites. This skill is self-contained. It focuses on testing architecture, not language-specific style.

Overview

This skill provides a defense-in-depth testing strategy with five layers for Python applications:

┌─────────────────────────────────────────────────┐
│  Layer 5 "smoke": Business Logic Integration Tests (5%)  │  ← Smoke tests over real system
├─────────────────────────────────────────────────┤
│  Layer 4 "logic": Business Logic Tests (70%)             │  ← Tests over fakes (MOST TESTS)
├─────────────────────────────────────────────────┤
│  Layer 3 "pure": Pure Unit Tests (10%)                   │  ← Zero dependencies, isolated testing
├─────────────────────────────────────────────────┤
│  Layer 2 "real-sanity": Integration Sanity Tests (10%)   │  ← Fast validation with mocking
├─────────────────────────────────────────────────┤
│  Layer 1 "fake-check": Fake Infrastructure Tests (5%)    │  ← Verify test doubles work
└─────────────────────────────────────────────────┘

Philosophy: Test business logic extensively over fast in-memory fakes. Use real implementations sparingly for integration validation.

Terminology note: The "gateway layer" (also called adapters/providers) refers to thin wrappers around heavyweight external APIs (databases, filesystems, HTTP APIs, message queues, etc.). The pattern matters more than the name.

Quick Decision: What Should I Read?

Adding a feature or fixing a bug? → Read quick-reference.md first, then workflows.md#adding-a-new-feature

Need to understand where to put a test? → Read testing-strategy.md

Working with Python-specific patterns? → Read python-specific.md

Adding/changing a gateway interface? → Read gateway-architecture.md, then workflows.md#adding-a-gateway-method

Wondering where the DI boundary is? → Read gateway-architecture.md#the-di-boundary-only-fake-gateways — only gateways get fakes

Need to understand non-ideal states vs exceptions? → Read non-ideal-states.md

Found tests using unittest.mock that should use fakes? → Read mock-to-fake-conversion.md

Need to implement a specific pattern (CliRunner, builders, etc.)? → Read patterns.md

Want to extend the gateway system (e.g., dry-run preview)? → Read advanced-extensions.md

Not sure if I'm doing it right? → Read anti-patterns.md

Just need a quick lookup? → Read quick-reference.md

When to Read Each Reference Document

📖 gateway-architecture.md

Read when:

• Adding or changing gateway/ABC interfaces
• Understanding the ABC/Real/Fake pattern
• Need examples of gateway implementations
• Want to understand what gateways are (and why they're thin)
• Creating a backend (higher-level abstraction that composes gateways)

Contents:

• What are gateway classes? (naming: gateways/adapters/providers)
• The three core implementations (ABC, Real, Fake)
• Code examples for each
• When to add/change gateway methods
• Design principles (keep gateways thin)
• Common gateway types (Database, API, FileSystem, MessageQueue)
• The DI boundary — only gateways get fakes

📖 non-ideal-states.md

Read when:

• Designing return types for gateway operations that can fail
• Deciding between exceptions and discriminated unions
• Implementing error injection in fakes
• Understanding error boundaries (where try/except belongs)

Contents:

• Non-ideal states vs exceptions (the core distinction)
• Decision framework for choosing between them
• How this shapes gateway signatures and fake design
• Error boundaries (try/except only in Real implementations)
• Three test categories per discriminated union operation
• The tracking-on-error decision
• isinstance() for type narrowing (never truthiness)

📖 mock-to-fake-conversion.md

Read when:

• Tests use unittest.mock.patch or @patch decorators
• An agent wrote tests with mocks instead of gateway fakes
• Converting existing mock-based tests to the gateway pattern

Contents:

• Step-by-step conversion workflow (audit, find/create gateway, inject, rewrite)
• "subprocess.run is never the right gateway boundary"
• Monkeypatch decision tree (when it's still OK)
• Common pitfalls (wrong abstraction level, subprocess-level gateways)

📖 testing-strategy.md

Read when:

• Deciding where to put a test
• Understanding the five testing layers
• Need test distribution guidance (5/70/10/10/5 rule)
• Want to know which layer tests what

Contents:

• Layer 1 "fake-check": Unit tests of fakes (verify test infrastructure)
• Layer 2 "real-sanity": Integration sanity tests with mocking (quick validation)
• Layer 3 "pure": Pure unit tests (zero dependencies, isolated testing)
• Layer 4 "logic": Business logic over fakes (majority of tests)
• Layer 5 "smoke": Business logic integration tests (smoke tests over real systems)
• Decision tree: where should my test go?
• Test distribution examples

📖 python-specific.md

Read when:

• Working with pytest fixtures
• Need Python mocking patterns
• Testing Flask/FastAPI/Django applications
• Understanding Python testing tools
• Need Python-specific commands

Contents:

• pytest fixtures and parametrization
• Mocking with unittest.mock and pytest-mock
• Testing web frameworks (Flask, FastAPI, Django)
• Python testing commands
• Type hints in tests
• Python packaging for test utilities

📖 workflows.md

Read when:

• Adding a new feature (step-by-step)
• Fixing a bug (step-by-step)
• Adding a gateway method (complete checklist)
• Changing an interface (what to update)
• Managing dry-run features

Contents:

• Adding a new feature (TDD workflow)
• Fixing a bug (reproduce → fix → regression test)
• Adding a gateway method (8-step checklist with examples)
• Changing an interface (update all layers)
• Managing dry-run features (wrapping pattern)
• Testing with builder patterns

📖 patterns.md

Read when:

• Implementing constructor injection for fakes
• Adding mutation tracking to fakes
• Using CliRunner for CLI tests
• Building complex test scenarios with builders
• Testing dry-run behavior
• Need code examples of specific patterns

Contents:

• Constructor injection (how and why)
• Mutation tracking properties (read-only access)
• Using CliRunner (not subprocess)
• Builder patterns for complex scenarios
• Simulated environment pattern
• Error injection pattern
• Dry-run testing pattern

📖 anti-patterns.md

Read when:

• Unsure if your approach is correct
• Want to avoid common mistakes
• Reviewing code for bad patterns
• Debugging why tests are slow/brittle

Contents:

• ❌ Testing speculative features
• ❌ Hardcoded paths in tests (catastrophic)
• ❌ Not updating all layers
• ❌ Using subprocess in unit tests
• ❌ Complex logic in gateway classes
• ❌ Fakes with I/O operations
• ❌ Testing implementation details
• ❌ Incomplete test coverage for gateways

📖 quick-reference.md

Read when:

• Quick lookup for file locations
• Finding example tests to reference
• Looking up common fixtures
• Need command reference
• Want test distribution guidelines

Contents:

• Decision tree (where to add test)
• File location map (source + tests)
• Common fixtures (tmp_path, CliRunner, etc.)
• Common test patterns (code snippets)
• Example tests to reference
• Useful commands (pytest, ty, etc.)
• Quick checklist for adding gateway methods

Quick Navigation by Task

I'm adding a new feature

1. Quick start: quick-reference.md → Decision tree 2. Step-by-step: workflows.md#adding-a-new-feature 3. Patterns: patterns.md (CliRunner, builders) 4. Avoid: anti-patterns.md (speculative tests, hardcoded paths)

I'm fixing a bug

1. Step-by-step: workflows.md#fixing-a-bug 2. Patterns: patterns.md#constructor-injection-for-fakes 3. Examples: quick-reference.md#example-tests-to-reference

I'm adding/changing a gateway method

1. Understanding: gateway-architecture.md 2. Step-by-step: workflows.md#adding-a-gateway-method 3. Checklist: quick-reference.md#quick-checklist-adding-a-new-gateway-method 4. Avoid: anti-patterns.md#not-updating-all-layers

I don't know where my test should go

1. Decision tree: quick-reference.md#decision-tree 2. Detailed guide: testing-strategy.md 3. Examples: quick-reference.md#example-tests-to-reference

I need to implement a pattern

1. All patterns: patterns.md 2. Examples: quick-reference.md#common-test-patterns

I think I'm doing something wrong

1. Anti-patterns: anti-patterns.md 2. Correct approach: workflows.md

Visual Layer Guide

┌──────────────────────────────────────────────────────────────┐
│ Layer 5 "smoke": Business Logic Integration Tests (5%)       │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Real database, filesystem, APIs, actual subprocess        │ │
│ │ Purpose: Smoke tests, catch integration issues           │ │
│ │ When: Sparingly, for critical workflows                  │ │
│ │ Speed: Seconds per test                                   │ │
│ │ Location: tests/e2e/ or tests/integration/               │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│ Layer 4 "logic": Business Logic Tests (70%) ← MOST TESTS    │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ FakeDatabase, FakeApiClient, FakeFileSystem              │ │
│ │ Purpose: Test features and business logic extensively    │ │
│ │ When: For EVERY feature and bug fix                      │ │
│ │ Speed: Milliseconds per test                              │ │
│ │ Location: tests/unit/, tests/services/, tests/commands/  │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│ Layer 3 "pure": Pure Unit Tests (10%)                        │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Zero dependencies, no fakes, no mocks                    │ │
│ │ Purpose: Test isolated utilities and helpers             │ │
│ │ When: For pure functions, data structures, parsers       │ │
│ │ Speed: Milliseconds per test                              │ │
│ │ Location: tests/unit/                                     │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│ Layer 2 "real-sanity": Integration Sanity Tests (10%)        │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ RealDatabase with mocked connections                     │ │
│ │ Purpose: Quick validation, catch syntax errors           │ │
│ │ When: When adding/changing real implementation           │ │
│ │ Speed: Fast (mocked)                                      │ │
│ │ Location: tests/integration/test_real_*.py               │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│ Layer 1 "fake-check": Fake Infrastructure Tests (5%)         │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Test FakeDatabase itself                                 │ │
│ │ Purpose: Verify test infrastructure is reliable          │ │
│ │ When: When adding/changing fake implementation           │ │
│ │ Speed: Milliseconds per test                              │ │
│ │ Location: tests/unit/fakes/test_fake_*.py               │ │
│ └──────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘

Key Principles

1. Thin gateway layer: Wrap external state, push complexity to business logic 2. Fast tests over fakes: 70% of tests should use in-memory fakes 3. Defense in depth: Fakes → sanity tests → pure unit → business logic → integration 4. Test what you're building: No speculative tests, only active work 5. Update all layers: When changing interfaces, update ABC/Real/Fake 6. The DI boundary: Only gateways get fakes; everything above them is tested with real logic and fake gateways

Layer Selection Guide

Distinguishing Layer 3 "pure" from Layer 4 "logic":

• Layer 3 "pure" (Pure Unit Tests): ZERO dependencies - no fakes, no mocks, no external state
• Testing string utilities: sanitize_branch_name("feat/FOO") → "feat-foo"
• Testing parsers: parse_git_status("## main") → {"branch": "main"}
• Testing data structures: LinkedList.append() without any external dependencies

• Layer 4 "logic" (Business Logic Tests): Uses fakes for external dependencies
• Testing commands: create_worktree(fake_git, name="feature")
• Testing workflows: submit_pr(fake_gh, fake_git, ...)
• Testing business logic that coordinates multiple integrations

*If your test imports a Fake\, it belongs in Layer 4 "logic", not Layer 3 "pure".**

Default Testing Strategy

When in doubt:

• Write test over fakes (Layer 4 "logic") for business logic
• Write pure unit test (Layer 3 "pure") for utilities/helpers with no dependencies
• Use pytest with fixtures
• Use tmp_path fixture (not hardcoded paths)
• Follow examples in quick-reference.md

Summary

For quick tasks: Start with quick-reference.md

For understanding: Start with testing-strategy.md or gateway-architecture.md

For step-by-step guidance: Use workflows.md

For implementation details: Use patterns.md

For error handling design: Check non-ideal-states.md

For converting mocks to fakes: Check mock-to-fake-conversion.md

For validation: Check anti-patterns.md

For Python specifics: Check python-specific.md

Advanced Gateway Extensions

Read this when: You want to extend the gateway system beyond the core ABC/Real/Fake pattern.

Overview

The core gateway pattern (ABC → Real → Fake) handles most testing needs. This document covers optional extensions you can add when your project requires them.

Extension: Dry-Run Wrapper

A DryRun wrapper intercepts write operations and prints what would happen, while delegating read operations to the wrapped implementation. This is useful for CLI tools with a --dry-run flag.

Pattern

class DryRunDatabaseGateway(DatabaseGateway):
    """Wrapper that prints instead of executing writes."""

    def __init__(self, gateway: DatabaseGateway) -> None:
        self._gateway = gateway  # Wrap any implementation

    def query(self, sql: str) -> list[dict]:
        """Read operation: delegate to wrapped."""
        return self._gateway.query(sql)

    def execute(self, sql: str) -> None:
        """Write operation: print instead of executing."""
        print(f"[DRY RUN] Would execute: {sql}")
        # Does NOT call self._gateway.execute()

Key Characteristics

• Decorator pattern: Wraps any gateway implementation (Real or Fake)
• Read operations: Pass through to wrapped implementation
• Write operations: Print [DRY RUN] message, don't execute
• Same interface: Implements the same ABC as Real and Fake

Wiring

def create_context(*, dry_run: bool) -> AppContext:
    database = RealDatabaseGateway(connection_string)
    if dry_run:
        database = DryRunDatabaseGateway(database)
    return AppContext(database=database)

Testing Dry-Run

def test_migration_dry_run(capsys) -> None:
    """Verify --dry-run doesn't modify data."""
    fake_db = FakeDatabaseGateway(
        users=[{"id": 1, "name": "Alice"}]
    )
    # Wrap fake with dry-run (same as production wraps real)
    dry_run_db = DryRunDatabaseGateway(fake_db)

    service = MigrationService(database=dry_run_db)
    service.migrate()

    # Write operations were intercepted
    assert len(fake_db.executed_commands) == 0
    captured = capsys.readouterr()
    assert "[DRY RUN]" in captured.out

When to Add DryRun

Add a DryRun wrapper when:

• Your application has a --dry-run CLI flag
• Users need to preview what operations would do before executing
• Write operations are destructive or expensive

Most gateways do not need DryRun. Only add it for gateways whose mutations are user-facing and benefit from preview.

Extension: Sub-Gateway Composition

When a gateway grows too large, extract related methods into sub-gateways accessed via properties:

class GitGateway(ABC):
    """Main gateway: pure facade with property accessors."""

    @property
    @abstractmethod
    def branch(self) -> GitBranchOps:
        """Access branch operations."""

    @property
    @abstractmethod
    def remote(self) -> GitRemoteOps:
        """Access remote operations."""

class RealGitGateway(GitGateway):
    def __init__(self) -> None:
        self._branch = RealGitBranchOps()
        self._remote = RealGitRemoteOps()

    @property
    def branch(self) -> GitBranchOps:
        return self._branch

    @property
    def remote(self) -> GitRemoteOps:
        return self._remote

When to Use Sub-Gateways

• Gateway has 15+ methods spanning distinct domains
• Methods naturally cluster (branch ops, remote ops, status ops)
• You want to inject only a subset of operations in some tests

The Fake Shares State

class FakeGitGateway(GitGateway):
    def __init__(self, *, branches: list[str] | None = None) -> None:
        # Shared state
        self._branches = branches or []
        # Sub-gateways share state
        self._branch_ops = FakeGitBranchOps(branches=self._branches)
        self._remote_ops = FakeGitRemoteOps(branches=self._branches)

    @property
    def branch(self) -> GitBranchOps:
        return self._branch_ops

    @property
    def remote(self) -> GitRemoteOps:
        return self._remote_ops

Related Documentation

• gateway-architecture.md - Core gateway pattern
• patterns.md - Constructor injection and mutation tracking
• testing-strategy.md - Which layer to test at

Testing Anti-Patterns

Read this when: You're unsure if your approach is correct, or want to avoid common mistakes.

Overview

This document covers common anti-patterns in Python testing and how to avoid them. Each anti-pattern includes examples of what NOT to do and the correct approach.

❌ Testing Speculative Features

NEVER write tests for code that doesn't exist yet (unless doing TDD RIGHT NOW).

Wrong Approach

# ❌ WRONG: Placeholder test for future feature
# def test_feature_we_might_add_later():
#     """TODO: Implement this feature next sprint."""
#     pass

# ❌ WRONG: Test stub for "maybe someday" idea
# def test_hypothetical_feature():
#     """Feature we're considering for Q2."""
#     # Not implemented yet, just a placeholder
#     pass

Correct Approach

# ✅ CORRECT: TDD for feature being implemented NOW
def test_new_feature_im_building_today():
    """Test for feature I'm about to implement."""
    result = process_payment(card="4111111111111111", amount=100.00)
    assert result.status == "success"  # Will implement after this test

# ✅ CORRECT: Test for actively worked bug fix
def test_bug_123_is_fixed():
    """Regression test for bug I'm fixing right now."""
    # Reproducing bug, then will fix it
    ...

Why This Is Wrong

Problems with speculative tests:

• Maintenance burden: Tests need updating when feature changes
• False confidence: Test suite looks comprehensive but validates nothing
• Wasted effort: Planned features often change significantly before implementation
• Stale code: Commented-out tests clutter codebase

Rule: Only write tests for code being actively implemented or fixed in this work session.

TDD Is Explicitly Allowed

TDD workflow is encouraged:

1. Write failing test for feature you're about to implement 2. Implement feature 3. Test passes

This is NOT speculative because you're implementing NOW, not "maybe later."

---

❌ Wrong Test Categorization (Unit vs Integration)

CRITICAL: Tests MUST be categorized correctly to maintain CI performance.

Test Categorization Rules

🔴 A test MUST be categorized as an integration test if:

1. It invokes a subprocess - Any test that calls subprocess.run(), subprocess.Popen(), or similar 2. It uses `time.sleep()` - Tests that rely on actual timing delays (must use mocking or DI instead) 3. It performs extensive real filesystem I/O - Tests that interact with external filesystem locations, create many files, or depend on actual filesystem behavior (limited file I/O with isolated_filesystem() or tmp_path in unit tests is acceptable) 4. It tests subprocess boundaries - Tests validating that abstraction layers correctly wrap external tools

Location Rules

• Unit tests → tests/unit/, tests/commands/, tests/core/
• Use fakes (FakeDatabase, FakeApiClient, etc.)
• Use CliRunner (NOT subprocess)
• No time.sleep() calls
• Fast, in-memory execution

• Integration tests → tests/integration/
• Use real implementations (RealGit, etc.)
• May invoke subprocess calls
• May use tmp_path fixture for real directories
• Slower, tests external tool integration

Wrong Approach

# ❌ WRONG - Unit test location with subprocess call
# Located in tests/commands/test_sync.py
def test_sync_calls_git() -> None:
    result = subprocess.run(["git", "fetch"], capture_output=True)
    # This MUST be moved to tests/integration/

# ❌ WRONG - Unit test with time.sleep()
# Located in tests/unit/test_retry.py
def test_retry_with_backoff() -> None:
    time.sleep(0.5)  # Actual delay
    # This MUST be moved to tests/integration/ OR use mocking

Correct Approach

# ✅ CORRECT - Integration test with subprocess
# Located in tests/integration/test_real_git.py
def test_real_git_fetch(tmp_path: Path) -> None:
    result = subprocess.run(["git", "fetch"], cwd=tmp_path, capture_output=True)
    assert result.returncode == 0

# ✅ CORRECT - Unit test with mocked sleep
# Located in tests/unit/test_retry.py
def test_retry_with_backoff(monkeypatch) -> None:
    mock_sleep = Mock()
    monkeypatch.setattr("time.sleep", mock_sleep)
    # Test logic without actual delay

Why This Matters

• CI performance: Unit tests must remain fast (<2s total) for quick feedback
• Test reliability: Subprocess calls can fail due to environment differences
• Parallel execution: Tests with subprocesses may have race conditions
• Resource usage: Subprocess tests consume more system resources

Rule: If unsure, default to integration test. It's safer to categorize a test as integration than to slow down the unit test suite.

---

❌ Hardcoded Paths in Tests (CATASTROPHIC)

NEVER use hardcoded paths in tests. Always use fixtures.

Wrong Approach

# ❌ WRONG - CATASTROPHICALLY DANGEROUS
def test_something():
    service = FileService(base_path=Path("/test/default/path"))
    service.process_files()

def test_another_thing():
    db = DatabaseAdapter(data_dir=Path("/var/lib/myapp/test"))
    db.initialize()

def test_with_absolute_path():
    config_path = Path("/Users/someone/test/config.yaml")
    # Code may write files to this path!
    config = load_config(config_path)

Correct Approach

# ✅ CORRECT - Use tmp_path fixture
def test_something(tmp_path: Path):
    service = FileService(base_path=tmp_path)
    service.process_files()

# ✅ CORRECT - Use temporary directory
def test_another_thing(tmp_path: Path):
    data_dir = tmp_path / "data"
    data_dir.mkdir()
    db = DatabaseAdapter(data_dir=data_dir)
    db.initialize()

# ✅ CORRECT - Create config in tmp_path
def test_with_config(tmp_path: Path):
    config_path = tmp_path / "config.yaml"
    config_path.write_text("debug: true")
    config = load_config(config_path)

Why This Is Catastrophic

Dangers of hardcoded paths:

1. Global config mutation: Code may write config files at hardcoded paths, polluting real filesystem 2. False isolation: Tests appear isolated but share state through hardcoded paths 3. Security risk: Creating files at system paths can be exploited 4. CI/CD failures: Paths may not exist on CI systems 5. Permission errors: Tests may not have write access to hardcoded paths

Detection: If you see `Path("/` in test code, STOP and use fixtures.

---

❌ Not Updating All Layers When Interface Changes

When changing a gateway interface, you MUST update ALL implementations.

Wrong Approach

# You changed DatabaseAdapter.query() signature:

# 1. DatabaseAdapter (ABC) ✅ Updated
class DatabaseAdapter(ABC):
    @abstractmethod
    def query(self, sql: str, *, timeout: float = 30.0) -> list[dict]:
        ...

# 2. RealDatabaseAdapter ✅ Updated
class RealDatabaseAdapter(DatabaseAdapter):
    def query(self, sql: str, *, timeout: float = 30.0) -> list[dict]:
        # Updated implementation
        ...

# 3. FakeDatabaseAdapter ❌ FORGOT TO UPDATE!
class FakeDatabaseAdapter(DatabaseAdapter):
    def query(self, sql: str) -> list[dict]:
        # Old signature - type error!
        ...

# 4. DryRunDatabaseAdapter ❌ FORGOT TO UPDATE!
class DryRunDatabaseAdapter(DatabaseAdapter):
    def query(self, sql: str) -> list[dict]:
        # Old signature - type error!
        ...

# Result: Type errors, broken tests, runtime failures

Correct Approach

Use this checklist when changing an interface:

• [ ] Update ABC interface (e.g., DatabaseAdapter)
• [ ] Update real implementation (e.g., RealDatabaseAdapter)
• [ ] Update fake implementation (e.g., FakeDatabaseAdapter)
• [ ] Update dry-run wrapper (e.g., DryRunDatabaseAdapter)
• [ ] Update all call sites in business logic
• [ ] Update unit tests of fake
• [ ] Update integration tests of real
• [ ] Update business logic tests that use the method

Tool: Run mypy or ty check to catch signature mismatches.

Why This Is Wrong

Problems:

• Type errors: Implementations don't match interface
• Runtime errors: Tests pass locally but fail in production
• Inconsistent behavior: Different implementations have different behavior
• Broken tests: Tests expect old signature

Rule: When changing interface, update ALL implementation layers (ABC, Real, Fake, DryRun) + tests.

---

❌ Using subprocess in Unit Tests

Use test clients and CliRunner for testing, NOT subprocess.

Wrong Approach

# ❌ WRONG: Slow, harder to debug
def test_cli_command():
    result = subprocess.run(
        ["python", "-m", "myapp", "process", "--file", "data.csv"],
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0
    assert "processed" in result.stdout

# ❌ WRONG: Even worse - shell=True
def test_another_command():
    result = subprocess.run(
        "myapp process --file data.csv",
        shell=True,
        capture_output=True,
    )
    assert result.returncode == 0

Correct Approach

# ✅ CORRECT: Fast, better error messages (for Click CLIs)
from click.testing import CliRunner

def test_cli_command(tmp_path: Path):
    runner = CliRunner()
    data_file = tmp_path / "data.csv"
    data_file.write_text("id,name\n1,Alice")

    result = runner.invoke(process_cmd, ["--file", str(data_file)])

    assert result.exit_code == 0
    assert "processed" in result.output

# ✅ CORRECT: For Flask apps
def test_flask_endpoint(client):
    response = client.post("/process", json={"file": "data.csv"})
    assert response.status_code == 200

# ✅ CORRECT: For FastAPI apps
def test_fastapi_endpoint(client):
    response = client.post("/process", json={"file": "data.csv"})
    assert response.status_code == 200

Why This Is Wrong

Performance:

• Test client/CliRunner: milliseconds per test (~10ms)
• subprocess: seconds per test (~1s)
• ~100x slower with subprocess

Debugging:

• subprocess: Harder to set breakpoints, unclear errors
• Test clients: Direct access to exceptions, clear stack traces

Reliability:

• subprocess: Shell interpretation issues, PATH dependencies
• Test clients: Direct Python invocation, no shell quirks

Rule: Always use appropriate test clients. Only use subprocess for true end-to-end integration tests (Layer 5 "smoke").

---

❌ Complex Logic in Integration class Classes

Integration classes should be THIN wrappers. Push complexity to business logic layer.

Wrong Approach

# ❌ WRONG: Business logic in adapter class
class RealDatabaseAdapter(DatabaseAdapter):
    def get_premium_users_with_expired_subscriptions(self) -> list[dict]:
        """Complex logic to find specific users."""
        users = self.query("SELECT * FROM users WHERE premium = true")

        # 50 lines of complex business logic...
        result = []
        for user in users:
            subscriptions = self.query(
                f"SELECT * FROM subscriptions WHERE user_id = {user['id']}"
            )

            # Complex date calculations
            for sub in subscriptions:
                end_date = datetime.fromisoformat(sub['end_date'])
                grace_period = timedelta(days=7)
                if end_date + grace_period < datetime.now():
                    # More complex logic...
                    if self._should_include_user(user, sub):
                        result.append(user)

        return result

    def _should_include_user(self, user: dict, sub: dict) -> bool:
        # Even more business logic...
        return True

Problems:

• Hard to fake (complex logic in fake too)
• Hard to test (need to mock everything)
• Hard to understand (mixed concerns)
• Hard to change (logic tied to database implementation)

Correct Approach

# ✅ CORRECT: Thin integration class, just wrap database operations
class RealDatabaseAdapter(DatabaseAdapter):
    def query(self, sql: str) -> list[dict[str, Any]]:
        """Just wrap database query - no business logic."""
        conn = psycopg2.connect(self.connection_string)
        cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
        cursor.execute(sql)
        return [dict(row) for row in cursor.fetchall()]

# ✅ CORRECT: Business logic in service layer
class SubscriptionService:
    def __init__(self, database: DatabaseAdapter) -> None:
        self.database = database

    def get_premium_users_with_expired_subscriptions(self) -> list[User]:
        """Complex logic over thin integration class."""
        users = self.database.query("SELECT * FROM users WHERE premium = true")

        # Business logic here - easy to test over fakes!
        result = []
        for user_dict in users:
            user = User.from_dict(user_dict)
            if self._has_expired_subscription(user):
                result.append(user)

        return result

    def _has_expired_subscription(self, user: User) -> bool:
        """Business logic isolated from database."""
        subscriptions = self.database.query(
            f"SELECT * FROM subscriptions WHERE user_id = {user.id}"
        )

        for sub in subscriptions:
            if self._is_expired(sub):
                return True
        return False

Benefits:

• Easy to fake (thin integration class, simple fake)
• Easy to test (business logic tested over fakes)
• Easy to understand (clear separation of concerns)
• Easy to change (business logic independent of database)

Rule

Integration classes should:

• Wrap external system calls
• Parse responses into domain objects
• Validate basic preconditions (file exists, etc.)

Integration classes should NOT:

• Contain business logic
• Make decisions about "what to do"
• Implement algorithms or calculations
• Have complex control flow

Test: If you can't easily fake an integration class, it's too complex. Push logic up.

---

❌ Fakes with I/O Operations

Fakes should be in-memory ONLY (except minimal directory creation).

Wrong Approach

# ❌ WRONG: Fake performs I/O
class FakeDatabaseAdapter(DatabaseAdapter):
    def __init__(self, db_file: Path) -> None:
        self.db_file = db_file

    def query(self, sql: str) -> list[dict]:
        # Reading/writing real files defeats the purpose!
        import sqlite3
        conn = sqlite3.connect(self.db_file)
        cursor = conn.cursor()
        cursor.execute(sql)
        return cursor.fetchall()

class FakeFileService(FileService):
    def process_file(self, path: Path) -> str:
        # Actually reading files defeats the purpose!
        content = path.read_text()
        return content.upper()

Problems:

• Slow (I/O operations)
• Requires real filesystem setup
• Defeats purpose of fakes
• Tests become integration tests

Correct Approach

# ✅ CORRECT: Fake uses in-memory state
class FakeDatabaseAdapter(DatabaseAdapter):
    def __init__(
        self,
        *,
        initial_data: dict[str, list[dict]] | None = None
    ) -> None:
        self._tables = initial_data or {}
        self._executed_queries: list[str] = []

    def query(self, sql: str) -> list[dict]:
        """Return in-memory data."""
        self._executed_queries.append(sql)

        # Simple parsing, return from memory
        if "FROM users" in sql:
            return self._tables.get("users", []).copy()
        return []

class FakeFileService(FileService):
    def __init__(self) -> None:
        self._processed_files: list[str] = []

    def process_file(self, path: Path) -> str:
        """Simulate processing without I/O."""
        self._processed_files.append(str(path))
        return "SIMULATED RESULT"

Benefits:

• Fast (no I/O)
• Simple test setup (configure via constructor)
• True unit testing
• Reliable (no filesystem quirks)

Exception: Directory Creation

Acceptable: Fakes may create real directories when needed for filesystem integration.

# ✅ ACCEPTABLE: Create directory for integration
class FakeFileManager(FileManager):
    def create_project(self, base_path: Path, name: str) -> Path:
        # Create real directory (acceptable for filesystem integration)
        project_path = base_path / name
        project_path.mkdir(parents=True, exist_ok=True)

        # But don't write actual files - keep data in memory
        self._projects[str(project_path)] = {
            "name": name,
            "created": datetime.now()
        }

        return project_path

Rule: Fakes may mkdir(), but should not read/write files.

---

❌ Testing Implementation Details

Test behavior, not implementation.

Wrong Approach

# ❌ WRONG: Testing internal implementation details
def test_service_uses_cache():
    """Test that service uses internal cache."""
    service = UserService(database=fake_db)

    # Checking private implementation details
    assert hasattr(service, "_cache")
    assert isinstance(service._cache, dict)

    service.get_user(1)
    assert 1 in service._cache  # Testing private attribute

def test_service_calls_private_method():
    """Test that service calls private method."""
    service = OrderService(database=fake_db)

    # Mocking private method - fragile!
    service._validate_order = Mock()

    service.process_order(order)
    service._validate_order.assert_called_once()

Correct Approach

# ✅ CORRECT: Testing observable behavior
def test_service_caches_users():
    """Test that service doesn't query database twice for same user."""
    fake_db = FakeDatabaseAdapter()
    service = UserService(database=fake_db)

    # Get same user twice
    user1 = service.get_user(1)
    user2 = service.get_user(1)

    # Assert on observable behavior - only one query
    assert len(fake_db.executed_queries) == 1
    assert user1 == user2

def test_order_validation():
    """Test that invalid orders are rejected."""
    service = OrderService(database=fake_db)

    invalid_order = Order(items=[], total=-50)

    # Test behavior, not how it's implemented
    with pytest.raises(ValueError, match="Invalid order"):
        service.process_order(invalid_order)

Why This Is Wrong

Problems:

• Tests break when refactoring
• Couples tests to implementation
• Doesn't verify user-visible behavior
• Makes code harder to change

Rule: Test what the code does, not how it does it.

---

❌ Incomplete Test Coverage for Integration class Changes

When adding/changing integration class method, you must test ALL implementations.

Wrong Approach

# Added new method to DatabaseAdapter
# ✅ Implemented in RealDatabaseAdapter
# ✅ Implemented in FakeDatabaseAdapter
# ❌ Forgot to test FakeDatabaseAdapter!
# ❌ Forgot to test RealDatabaseAdapter!

# Result: Untested code, potential bugs

Correct Approach

Complete testing checklist:

• [ ] Unit test of fake (tests/unit/fakes/test_fake_database.py)
• [ ] Integration test of real with mocking (tests/integration/test_real_database.py)
• [ ] Business logic test using fake (tests/unit/services/test_my_service.py)
• [ ] (Optional) E2E test with real implementation

See: workflows.md#adding-an-integration class-method for full checklist.

---

❌ Mocking What You Don't Own

Create your own integration classes instead of mocking third-party libraries directly.

Wrong Approach

# ❌ WRONG: Mocking third-party library
@patch("requests.Session")
def test_api_call(mock_session):
    # Fragile - couples to requests internals
    mock_session.return_value.get.return_value.json.return_value = {"data": "test"}

    service = DataService()
    result = service.fetch_data()

@patch("boto3.client")
def test_s3_upload(mock_boto):
    # Fragile - AWS SDK might change
    mock_client = Mock()
    mock_boto.return_value = mock_client
    mock_client.upload_file.return_value = None

Correct Approach

# ✅ CORRECT: Create your own integration class
class StorageAdapter(ABC):
    @abstractmethod
    def upload_file(self, local_path: Path, remote_key: str) -> None:
        """Upload file to storage."""

class S3StorageAdapter(StorageAdapter):
    """Real implementation using boto3."""
    def upload_file(self, local_path: Path, remote_key: str) -> None:
        import boto3
        client = boto3.client("s3")
        client.upload_file(str(local_path), self.bucket, remote_key)

class FakeStorageAdapter(StorageAdapter):
    """Fake for testing."""
    def __init__(self) -> None:
        self.uploaded_files: list[tuple[str, str]] = []

    def upload_file(self, local_path: Path, remote_key: str) -> None:
        self.uploaded_files.append((str(local_path), remote_key))

# Test with your fake
def test_file_upload():
    storage = FakeStorageAdapter()
    service = FileService(storage=storage)

    service.process_and_upload("data.csv")

    assert ("data.csv", "processed/data.csv") in storage.uploaded_files

Benefits:

• Not coupled to third-party library internals
• Easy to test
• Clear interface
• Can switch libraries without changing tests

---

❌ Creating Fake Backends (DI All The Way Down)

NEVER create fake implementations for backends. DI is ONLY at the gateway level.

Understanding the Problem

There's a critical distinction between gateways and backends:

• Gateways = thin wrappers around external systems (Database, ApiClient, FileSystem)
• Need 3 core implementations: ABC, Real, Fake
• Fakes provide in-memory simulation

• Code above gateways = services, backends, managers that COMPOSE gateways
• Only need real implementations
• NO fake implementation needed - inject fake gateways instead

Wrong Approach

# ❌ WRONG: Creating a fake backend
class ManagedPrBackend(ABC):
    @abstractmethod
    def create_managed_pr(self, ...) -> CreateManagedPrResult: ...

class ManagedGitHubPrBackend(ManagedPrBackend):
    def __init__(self, github_issues: GitHubIssues):
        self._github_issues = github_issues

    def create_managed_pr(self, ...) -> CreateManagedPrResult:
        result = self._github_issues.create_issue(...)
        return CreateManagedPrResult(...)

# ❌ WRONG: DON'T DO THIS - fake backend is unnecessary
class FakeManagedPrBackend(ManagedPrBackend):
    def __init__(self, *, managed_prs: dict | None = None):
        self._managed_prs = managed_prs or {}

    def create_managed_pr(self, ...) -> CreateManagedPrResult:
        # Duplicates logic that should be tested via real backend + fake gateway
        ...

Problems:

• Duplicated logic: Fake backend duplicates real backend's business logic
• Untested real code: The actual backend logic goes untested
• Wrong abstraction: DI should stop at the gateway level
• Java-style over-engineering: "DI all the way down" leads to test doubles at every layer

Correct Approach

# ✅ CORRECT: Backend composes gateways, no fake needed
class ManagedPrBackend(ABC):
    @abstractmethod
    def create_managed_pr(self, ...) -> CreateManagedPrResult: ...

class ManagedGitHubPrBackend(ManagedPrBackend):
    def __init__(self, github_issues: GitHubIssues):
        self._github_issues = github_issues  # Gateway injected here

    def create_managed_pr(self, ...) -> CreateManagedPrResult:
        result = self._github_issues.create_issue(...)
        return CreateManagedPrResult(pr_id=str(result.number), url=result.url)

# ✅ CORRECT: Test backend with fake gateway
def test_create_managed_pr():
    fake_issues = FakeGitHubIssues()  # Fake at gateway level
    backend = ManagedGitHubPrBackend(fake_issues)  # Real backend

    result = backend.create_managed_pr(...)

    # Assert on gateway mutations
    assert fake_issues.created_issues[0][0] == "expected title"
    assert result.pr_id == "1"

Why This Is Wrong

1. Gateways are the seam: They're the boundary where we swap real ↔ fake 2. Backends contain business logic: Should be tested with real logic, fake dependencies 3. Avoids duplication: A fake backend just duplicates the real backend's logic 4. DI boundary rule: Only inject dependencies at the gateway level

The DI Boundary Rule

Application entry point → DI container / context
  → OrderService (business logic - REAL in tests)
    → FakePaymentGateway (gateway - FAKE in tests)  ← DI stops here
    → FakeDatabaseAdapter (gateway - FAKE in tests)  ← DI stops here

Rule: DI and fakes apply to gateways only. Business logic is tested with real implementations that receive fake gateways.

---

Summary of Anti-Patterns

Related Documentation

• workflows.md - Step-by-step guides for correct approaches
• patterns.md - Common testing patterns to follow
• testing-strategy.md - Which layer to test at
• gateway-architecture.md - Understanding the gateway layer
• python-specific.md - Python testing best practices

Gateway Layer Architecture

Read this when: You need to understand or modify the gateway layer (the thin wrapper interfaces over external state).

Overview

Naming note: "Gateway" is a common name for this pattern. These classes are also called adapters, providers, or ports in other contexts. The pattern matters more than the name.

What Are Gateway Classes?

Gateway classes are thin wrappers around heavyweight external APIs that:

• Touch external state (filesystem, database, APIs, message queues)
• Could be slow (network calls, disk I/O, subprocess execution)
• Could fail periodically (network issues, rate limits, service outages)
• Are difficult to test directly

The Core Implementations

Every gateway interface has three core implementations (ABC, Real, Fake). A fourth — DryRun — is an optional extension covered in advanced-extensions.md.

1. Abstract Interface (ABC)

Defines the contract all implementations must follow.

Example: DatabaseGateway (src/myapp/gateways/database.py)

from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any

class DatabaseGateway(ABC):
    """Thin wrapper over database operations."""

    @abstractmethod
    def query(self, sql: str, *, timeout: float | None = None) -> list[dict[str, Any]]:
        """Execute a SELECT query."""

    @abstractmethod
    def execute(self, sql: str) -> None:
        """Execute an INSERT, UPDATE, or DELETE."""

    @abstractmethod
    def transaction(self) -> "TransactionContext":
        """Start a database transaction."""

    # ... more methods

Key characteristics:

• Uses ABC (not Protocol)
• All methods are @abstractmethod
• Contains ONLY runtime operations (no test setup methods)
• May have concrete helper methods (all implementations inherit)

2. Real Implementation

Calls actual external systems (database, filesystem, API).

Example: RealDatabaseGateway (src/myapp/gateways/database.py)

import psycopg2
from contextlib import contextmanager

class RealDatabaseGateway(DatabaseGateway):
    """Real database operations via psycopg2."""

    def __init__(self, connection_string: str) -> None:
        self.connection_string = connection_string

    def query(self, sql: str, *, timeout: float | None = None) -> list[dict[str, Any]]:
        """Execute SELECT query against PostgreSQL."""
        conn = psycopg2.connect(
            self.connection_string,
            options=f"-c statement_timeout={int(timeout * 1000)}" if timeout else ""
        )

        try:
            cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
            cursor.execute(sql)
            return [dict(row) for row in cursor.fetchall()]
        finally:
            cursor.close()
            conn.close()

    def execute(self, sql: str) -> None:
        """Execute INSERT/UPDATE/DELETE against PostgreSQL."""
        conn = psycopg2.connect(self.connection_string)

        try:
            cursor = conn.cursor()
            cursor.execute(sql)
            conn.commit()
        finally:
            cursor.close()
            conn.close()

    @contextmanager
    def transaction(self):
        """Transaction context manager."""
        conn = psycopg2.connect(self.connection_string)
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

Key characteristics:

• Uses real libraries (psycopg2, requests, boto3, etc.)
• Handles connection management
• LBYL: checks conditions before operations
• Lets exceptions bubble up (no try/except for control flow)

3. Fake Implementation

In-memory simulation for fast testing.

Example: FakeDatabaseGateway (tests/fakes/database.py)

from typing import Any
from contextlib import contextmanager

class FakeDatabaseGateway(DatabaseGateway):
    """In-memory database simulation for testing."""

    def __init__(
        self,
        *,
        initial_data: dict[str, list[dict]] | None = None,
        should_fail_on: list[str] | None = None,
    ) -> None:
        # Mutable state (private)
        self._tables: dict[str, list[dict]] = initial_data or {}
        self._should_fail_on = should_fail_on or []
        self._in_transaction = False

        # Mutation tracking (private, accessed via properties)
        self._executed_queries: list[str] = []
        self._executed_commands: list[str] = []
        self._transaction_count = 0

    def query(self, sql: str, *, timeout: float | None = None) -> list[dict[str, Any]]:
        """Return in-memory data."""
        # Simulate failure if configured
        if any(pattern in sql for pattern in self._should_fail_on):
            raise RuntimeError(f"Simulated failure for: {sql}")

        # Track operation
        self._executed_queries.append(sql)

        # Parse table name (simplified)
        if "FROM" in sql:
            table = sql.split("FROM")[1].split()[0].strip()
            return self._tables.get(table, []).copy()
        return []

    def execute(self, sql: str) -> None:
        """Update in-memory state."""
        # Track operation
        self._executed_commands.append(sql)

        # Simulate INSERT (simplified parsing)
        if sql.startswith("INSERT INTO"):
            # Extract table and values (simplified)
            parts = sql.split()
            table = parts[2]
            if table not in self._tables:
                self._tables[table] = []
            # Add dummy record
            self._tables[table].append({"id": len(self._tables[table]) + 1})

        # Simulate DELETE (simplified)
        elif sql.startswith("DELETE FROM"):
            parts = sql.split()
            table = parts[2]
            if table in self._tables:
                self._tables[table] = []

    @contextmanager
    def transaction(self):
        """Simulated transaction."""
        self._in_transaction = True
        self._transaction_count += 1
        try:
            yield self
        finally:
            self._in_transaction = False

    @property
    def executed_queries(self) -> list[str]:
        """Read-only access for test assertions."""
        return self._executed_queries.copy()

    @property
    def executed_commands(self) -> list[str]:
        """Read-only access for test assertions."""
        return self._executed_commands.copy()

    @property
    def transaction_count(self) -> int:
        """Read-only access for test assertions."""
        return self._transaction_count

Key characteristics:

• Constructor injection: All initial state via keyword arguments
• In-memory storage: Dictionaries, lists for state
• Mutation tracking: Read-only properties for assertions
• Fast: No I/O, no network calls
• Simulation: May mimic real behavior (e.g., checking constraints)

Mutation tracking pattern:

# In test:
fake_db = FakeDatabaseGateway()
fake_db.execute("INSERT INTO users VALUES (...)")

# Assert operation was called
assert "INSERT INTO users" in fake_db.executed_commands[0]

Common Gateway Types

API Client Gateway

class ApiClient(ABC):
    """Gateway for external API calls."""

    @abstractmethod
    def get(self, endpoint: str, *, params: dict | None = None) -> dict:
        """GET request to API."""

    @abstractmethod
    def post(self, endpoint: str, *, json: dict) -> dict:
        """POST request to API."""

class RealApiClient(ApiClient):
    """Real HTTP client using requests."""

    def __init__(self, base_url: str, api_key: str) -> None:
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def get(self, endpoint: str, *, params: dict | None = None) -> dict:
        import requests
        response = requests.get(
            f"{self.base_url}{endpoint}",
            params=params,
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

class FakeApiClient(ApiClient):
    """Fake API client for testing."""

    def __init__(self, responses: dict[str, Any]) -> None:
        self.responses = responses
        self.requested_endpoints: list[str] = []

    def get(self, endpoint: str, *, params: dict | None = None) -> dict:
        self.requested_endpoints.append(endpoint)
        return self.responses.get(endpoint, {})

File System Gateway

class FileSystemGateway(ABC):
    """Gateway for file system operations."""

    @abstractmethod
    def read_file(self, path: Path) -> str:
        """Read file contents."""

    @abstractmethod
    def write_file(self, path: Path, content: str) -> None:
        """Write file contents."""

    @abstractmethod
    def exists(self, path: Path) -> bool:
        """Check if path exists."""

class RealFileSystemGateway(FileSystemGateway):
    """Real file system operations."""

    def read_file(self, path: Path) -> str:
        if path.exists():
            return path.read_text(encoding="utf-8")
        raise FileNotFoundError(f"File not found: {path}")

    def write_file(self, path: Path, content: str) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")

    def exists(self, path: Path) -> bool:
        return path.exists()

class FakeFileSystemGateway(FileSystemGateway):
    """In-memory file system for testing."""

    def __init__(self) -> None:
        self._files: dict[str, str] = {}

    def read_file(self, path: Path) -> str:
        key = str(path)
        if key in self._files:
            return self._files[key]
        raise FileNotFoundError(f"File not found: {path}")

    def write_file(self, path: Path, content: str) -> None:
        self._files[str(path)] = content

    def exists(self, path: Path) -> bool:
        return str(path) in self._files

Message Queue Gateway

class MessageQueueGateway(ABC):
    """Gateway for message queue operations."""

    @abstractmethod
    def publish(self, topic: str, message: dict) -> None:
        """Publish message to topic."""

    @abstractmethod
    def subscribe(self, topic: str) -> Generator[dict, None, None]:
        """Subscribe to topic messages."""

class FakeMessageQueue(MessageQueueGateway):
    """In-memory message queue for testing."""

    def __init__(self) -> None:
        self._queues: dict[str, list[dict]] = {}
        self._published_messages: list[tuple[str, dict]] = []

    def publish(self, topic: str, message: dict) -> None:
        if topic not in self._queues:
            self._queues[topic] = []
        self._queues[topic].append(message)
        self._published_messages.append((topic, message))

    def subscribe(self, topic: str) -> Generator[dict, None, None]:
        queue = self._queues.get(topic, [])
        while queue:
            yield queue.pop(0)

    @property
    def published_messages(self) -> list[tuple[str, dict]]:
        """For test assertions."""
        return self._published_messages.copy()

Time Gateway

The simplest possible gateway — demonstrates why even stdlib calls should go through gateways for testability.

import time
from abc import ABC, abstractmethod
from datetime import datetime

class Time(ABC):
    """Gateway for time operations."""

    @abstractmethod
    def now(self) -> datetime:
        """Current time (replaces datetime.now())."""

    @abstractmethod
    def sleep(self, seconds: float) -> None:
        """Sleep (replaces time.sleep())."""

class RealTime(Time):
    """Real time operations."""

    def now(self) -> datetime:
        return datetime.now()

    def sleep(self, seconds: float) -> None:
        time.sleep(seconds)

class FakeTime(Time):
    """Deterministic time for testing."""

    def __init__(self, *, current_time: datetime | None = None) -> None:
        self._current_time = current_time or datetime(2024, 1, 15, 14, 30, 0)
        self._sleep_calls: list[float] = []

    def now(self) -> datetime:
        return self._current_time

    def sleep(self, seconds: float) -> None:
        self._sleep_calls.append(seconds)  # Track, don't actually sleep

    @property
    def sleep_calls(self) -> list[float]:
        """Read-only access for test assertions."""
        return list(self._sleep_calls)

Why gateway-ify time? Tests using datetime.now() directly are flaky (timing-dependent) and slow (real time.sleep()). With FakeTime, tests are deterministic and instant.

When to Add/Change Gateway Methods

Adding a Method

If you need to add a method to a gateway interface:

1. Add @abstractmethod to ABC interface 2. Implement in real class with actual I/O 3. Implement in fake class with in-memory state 4. Write unit test of fake implementation 5. Write integration test of real implementation

Changing an Interface

If you need to change an interface:

• Update all implementations (ABC, Real, Fake)
• Update all tests that use the changed method
• Update any business logic that calls the method

Design Principles

Keep Gateways Thin

Gateways should NOT contain business logic. Push complexity to the business layer.

# ❌ WRONG: Business logic in gateway class
class RealDatabaseGateway(DatabaseGateway):
    def get_active_users_with_recent_orders(self) -> list[dict]:
        """Complex logic to find users."""
        users = self.query("SELECT * FROM users WHERE active = true")
        result = []
        for user in users:
            orders = self.query(f"SELECT * FROM orders WHERE user_id = {user['id']}")
            if any(o['created_at'] > datetime.now() - timedelta(days=30) for o in orders):
                result.append(user)
        return result

# ✅ CORRECT: Thin gateway, logic in business layer
class RealDatabaseGateway(DatabaseGateway):
    def query(self, sql: str) -> list[dict[str, Any]]:
        """Just wrap database query."""
        conn = psycopg2.connect(self.connection_string)
        cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
        cursor.execute(sql)
        return [dict(row) for row in cursor.fetchall()]

# Business logic layer:
class UserService:
    def get_active_users_with_recent_orders(self) -> list[User]:
        """Complex logic over thin gateway."""
        users = self.database.query("SELECT * FROM users WHERE active = true")
        result = []
        for user in users:
            orders = self.database.query(f"SELECT * FROM orders WHERE user_id = {user['id']}")
            if any(o['created_at'] > datetime.now() - timedelta(days=30) for o in orders):
                result.append(User.from_dict(user))
        return result

Why: Thin gateways are easier to fake, easier to test, easier to understand.

Fakes Should Be In-Memory

Fakes should avoid I/O operations (except minimal directory creation when testing file operations).

# ❌ WRONG: Fake performs I/O
class FakeFileSystem(FileSystemGateway):
    def read_file(self, path: Path) -> str:
        # Reading real files defeats the purpose of fakes!
        return path.read_text()

# ✅ CORRECT: Fake uses in-memory state
class FakeFileSystem(FileSystemGateway):
    def __init__(self) -> None:
        self._files: dict[str, str] = {}

    def read_file(self, path: Path) -> str:
        key = str(path)
        if key in self._files:
            return self._files[key]
        raise FileNotFoundError(f"File not found: {path}")

Exception: Fakes may create real directories when necessary for integration, but should not read/write actual files.

The DI Boundary: Only Fake Gateways

CRITICAL: DI is ONLY at the gateway level. We do NOT want "DI all the way down" like Java.

Gateways are the thin wrappers around external systems. Everything above them — services, backends, managers, handlers, whatever your project calls them — composes gateways and contains business logic. Only gateways get fakes.

The Distinction

Testing Code Above Gateways

To test business logic, use the real class with fake gateways injected:

# ✅ CORRECT: Real service, fake gateways
def test_process_order():
    fake_db = FakeDatabaseAdapter(users=[{"id": 1, "balance": 100}])
    fake_payment = FakePaymentGateway(approved_cards=["4111111111111111"])
    service = OrderService(database=fake_db, payment=fake_payment)

    result = service.process_order(user_id=1, card="4111111111111111", amount=50)

    assert result.status == "success"
    assert len(fake_payment.processed_transactions) == 1

# ❌ WRONG: Creating a fake service
class FakeOrderService(OrderService):  # DON'T DO THIS
    ...

Why Only Fake Gateways?

1. Gateways are the seam — they're the boundary where we swap real for fake 2. Business logic should be tested, not faked — test with real logic, fake dependencies 3. Avoids duplication — a fake service just duplicates the real service's logic 4. DI stops here — only inject at the gateway level, not deeper

Related Documentation

• non-ideal-states.md - Error handling: non-ideal states vs exceptions
• testing-strategy.md - How to test gateway classes at different layers
• workflows.md - Step-by-step guide for adding gateway methods
• patterns.md - Constructor injection and mutation tracking patterns
• anti-patterns.md - What to avoid in gateway design
• advanced-extensions.md - DryRun wrappers and sub-gateway composition

Mock-to-Fake Conversion

Read this when: You find tests using unittest.mock.patch, @patch decorators, or MagicMock that should use gateway fakes instead.

Why This Happens

AI agents (and developers unfamiliar with the gateway pattern) often reach for unittest.mock when writing tests. This creates tests that are:

• Coupled to import paths — @patch("myapp.services.subprocess.run") breaks when code moves
• Hard to read — nested mock.return_value.json.return_value chains
• Fragile — tests break on refactoring even when behavior is unchanged

The fix is to convert these mocks into gateway fakes.

Step 1: Audit What's Being Mocked

For each patch(...) call, identify the system boundary (tool or service), not the Python function:

Group mocks by test. A single test patching 2-3 things together suggests those things form a unit covered by one gateway.

The Critical Rule

`subprocess.run` is never the right gateway boundary.

The gateway should be named after the _tool_ being called, not the mechanism:

Name the gateway after what it represents, not how it executes.

Step 2: Check for Existing Gateways

Before creating anything new, check if a gateway already exists:

# Search for existing ABCs
grep -r "class.*ABC" src/myapp/gateways/

# Search for existing fakes
grep -r "class Fake" tests/fakes/

Priority when multiple gateways match:

1. A gateway that covers ALL mocked targets in a test 2. A gateway at the highest behavioral level (e.g., PaymentGateway.charge() rather than HttpClient.post()) 3. The lowest-level matching gateway as a last resort

If an existing gateway covers the mocked behavior, skip to Step 4.

Step 3: Create the Gateway (If Needed)

Follow the standard three-file pattern (see gateway-architecture.md):

# src/myapp/gateways/user_api.py

from abc import ABC, abstractmethod

class UserApiClient(ABC):
    @abstractmethod
    def get_users(self) -> list[dict]:
        """Fetch users from API."""

class RealUserApiClient(UserApiClient):
    def __init__(self, base_url: str, api_key: str) -> None:
        self._base_url = base_url
        self._api_key = api_key

    def get_users(self) -> list[dict]:
        import requests
        response = requests.get(
            f"{self._base_url}/users",
            headers={"Authorization": f"Bearer {self._api_key}"}
        )
        response.raise_for_status()
        return response.json()["users"]

# tests/fakes/user_api.py

class FakeUserApiClient(UserApiClient):
    def __init__(self, *, users: list[dict] | None = None) -> None:
        self._users = users or []
        self._get_users_calls: list[None] = []

    def get_users(self) -> list[dict]:
        self._get_users_calls.append(None)
        return list(self._users)

    @property
    def get_users_call_count(self) -> int:
        return len(self._get_users_calls)

Step 4: Make Source Code Injectable

Add the gateway as a constructor parameter:

# Before:
class DataSyncService:
    def sync(self) -> None:
        import requests
        users = requests.get("https://api.example.com/users").json()["users"]
        ...

# After:
class DataSyncService:
    def __init__(self, *, user_api: UserApiClient) -> None:
        self._user_api = user_api

    def sync(self) -> None:
        users = self._user_api.get_users()
        ...

Update production wiring to pass the real implementation:

service = DataSyncService(user_api=RealUserApiClient(base_url=..., api_key=...))

Step 5: Rewrite the Tests

# Before:
@patch("myapp.services.requests.get")
def test_sync_fetches_users(mock_get):
    mock_get.return_value.json.return_value = {"users": [{"id": 1, "name": "Alice"}]}

    service = DataSyncService()
    service.sync()

    mock_get.assert_called_once()

# After:
def test_sync_fetches_users() -> None:
    fake_api = FakeUserApiClient(users=[{"id": 1, "name": "Alice"}])
    service = DataSyncService(user_api=fake_api)

    service.sync()

    assert fake_api.get_users_call_count == 1

Step 6: Verify

Run the affected tests, then lint and type-check:

pytest tests/unit/test_sync_service.py -v
mypy src/myapp/services/

When Monkeypatch Is Still OK

Not everything needs a gateway. Monkeypatch (pytest's built-in) is acceptable for process-level globals:

Is there a gateway for this operation?
├── YES -> Use the fake gateway
│
└── NO -> Should there be a gateway?
    ├── YES -> Create the gateway, then use its fake
    │
    └── NO -> Is it a process-level global?
        ├── YES -> monkeypatch is acceptable
        │         (environment variables, Path.home(), locale)
        │
        └── NO -> Create a gateway

Acceptable monkeypatch uses:

• monkeypatch.setenv("API_KEY", "test_value") — environment variables
• monkeypatch.setattr(Path, "home", lambda: tmp_path) — home directory isolation
• monkeypatch.delenv("DEBUG", raising=False) — removing env vars

Not acceptable (need a gateway instead):

• monkeypatch.setattr(subprocess, "run", ...) — bypasses gateway infrastructure
• monkeypatch.setattr(requests, "get", ...) — should be behind a gateway
• @patch("module.function") — unittest.mock should be replaced entirely

Common Pitfalls

Pitfall 1: Wrong gateway level

If shutil.which("tool") and subprocess.run(["tool", ...]) are both mocked in the same test, the gateway should cover _both_ — something like ToolGateway with an is_available() method and operation methods. Don't create separate gateways for availability checks vs execution.

Pitfall 2: subprocess-level gateways

If you find yourself designing a gateway called ShellRunner, SubprocessGateway, or CommandRunner, stop. That's still mocking at the wrong level. The gateway must be specific to the _tool_ being called.

Pitfall 3: Forgetting production wiring

After adding a constructor parameter, update wherever the class is instantiated in production. Type checkers will catch this — run them.

Pitfall 4: Multiple patches = wrong abstraction

Multiple patch() calls in one test is a red flag. A single fake should replace all of them. If you need 3+ patches, the gateway is probably at the wrong level of abstraction.

Pitfall 5: Keeping unittest.mock "just for this one thing"

Once you have gateways, there's almost never a reason to keep unittest.mock. The exception is monkeypatch for process-level globals (see decision tree above).

Related Documentation

• gateway-architecture.md - How to design gateway interfaces
• non-ideal-states.md - Designing return types for operations that can fail
• patterns.md - Constructor injection and mutation tracking
• anti-patterns.md#mocking-what-you-dont-own - Why mocking third-party libraries is fragile

Non-Ideal States vs Exceptions

Read this when: Designing gateway method return types, deciding how to handle errors in gateway interfaces, or implementing error injection in fakes.

The Core Distinction

Non-ideal states are optional outcomes visible in the type signature. When a gateway method returns UserCreated | UserAlreadyExists, both possible outcomes are self-documenting — any reader (human or AI agent) can see from the signature alone what can happen and write control flow accordingly.

This is the fundamental advantage over exceptions: Python exceptions are not knowable from the type signature. You cannot determine what a function might throw without reading its implementation (and every implementation it calls). This makes exceptions hostile to agent-driven development and harder for humans to reason about.

The distinction:

• Non-ideal states — optional outcomes encoded in the return type. The caller can see them, branch on them, and continue. Self-documenting.
• Exceptions — invisible in the type signature. Used only when all callers terminate identically and no branching logic is needed.

This distinction drives how you design gateway method signatures, which in turn drives how fakes simulate error conditions.

When to Use Exceptions

Use exceptions when all callers terminate identically — the error is just a message, no branching logic, no meaningful field inspection.

class RealFileSystem(FileSystemGateway):
    def read_file(self, path: Path) -> str:
        """Read file contents. Raises if file doesn't exist."""
        if not path.exists():
            raise FileNotFoundError(f"File not found: {path}")
        return path.read_text(encoding="utf-8")

The fake is simple — it raises the same exception when configured to:

class FakeFileSystem(FileSystemGateway):
    def __init__(self, *, files: dict[str, str] | None = None) -> None:
        self._files = files or {}

    def read_file(self, path: Path) -> str:
        key = str(path)
        if key not in self._files:
            raise FileNotFoundError(f"File not found: {path}")
        return self._files[key]

When to Use Non-Ideal States (Discriminated Unions)

Use non-ideal states when callers branch on the error and continue — different handling for different outcomes, field inspection, or type-safe continuation.

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass(frozen=True)
class UserCreated:
    """Success: user was created."""
    user_id: int

@dataclass(frozen=True)
class UserAlreadyExists:
    """Non-ideal state: email is taken."""
    email: str
    message: str

class UserGateway(ABC):
    @abstractmethod
    def create_user(self, email: str, name: str) -> UserCreated | UserAlreadyExists:
        """Create a user. Returns non-ideal state if email is taken."""

The caller branches on the result:

result = gateway.create_user(email, name)
if isinstance(result, UserAlreadyExists):
    # Branch: suggest login instead
    click.echo(f"Account already exists for {result.email}")
    return
# Type narrowing: result is UserCreated
click.echo(f"Created user {result.user_id}")

Decision Framework

Type Narrowing: Always Use isinstance()

Always use `isinstance()` for type narrowing. Never use truthiness checks, .success attributes, or any other mechanism.

# CORRECT: isinstance() enables type narrowing
result = gateway.charge(card, amount)
if isinstance(result, PaymentDeclined):
    # Type checker knows: result is PaymentDeclined
    log(f"Declined: {result.reason}")
    return handle_decline(result)
# Type checker knows: result is ChargeSuccess
send_receipt(result.transaction_id)

# WRONG: truthiness doesn't narrow types
if not result:  # Empty success markers are falsy!
    ...

# WRONG: attribute checks bypass the type system
if result.is_error:  # Type checker can't narrow from this
    ...

Error Boundaries: Where try/except Belongs

In the gateway pattern, only the Real implementation catches exceptions. Fakes never use try/except.

Real: The Exception Boundary

The Real implementation is the only place where external systems raise exceptions. It catches them and converts to discriminated union types:

class RealPaymentGateway(PaymentGateway):
    def charge(self, card: str, amount: float) -> ChargeSuccess | PaymentDeclined:
        response = self._http_client.post("/charges", json={
            "card": card, "amount": amount
        })
        if response.status_code == 402:
            data = response.json()
            return PaymentDeclined(
                reason=data["decline_reason"],
                card_last_four=card[-4:]
            )
        response.raise_for_status()  # Unexpected errors -> exception
        return ChargeSuccess(transaction_id=response.json()["id"])

What gets caught: Expected failure modes from external systems (HTTP 402, command exit codes, file-not-found).

What doesn't get caught: Programming errors (AttributeError, TypeError) — these should crash, not be masked.

Fake: Constructor-Configured (No try/except)

Fakes never catch exceptions because they never call external systems. Error behavior is configured at construction time:

class FakePaymentGateway(PaymentGateway):
    def __init__(
        self,
        *,
        decline_cards: dict[str, str] | None = None,
    ) -> None:
        self._decline_cards = decline_cards or {}
        self._charges: list[tuple[str, float]] = []

    def charge(self, card: str, amount: float) -> ChargeSuccess | PaymentDeclined:
        if card in self._decline_cards:
            return PaymentDeclined(
                reason=self._decline_cards[card],
                card_last_four=card[-4:]
            )
        self._charges.append((card, amount))
        return ChargeSuccess(transaction_id=f"fake-txn-{len(self._charges)}")

Why no try/except? There are no subprocess calls, no network requests, no filesystem operations. Error scenarios are predetermined by test setup, not discovered at runtime.

Three Test Categories Per Operation

Every fake method that returns a discriminated union needs three test categories:

class TestCharge:
    """Tests for FakePaymentGateway.charge()"""

    def test_default_success(self) -> None:
        """No-arg fake returns success."""
        fake = FakePaymentGateway()
        result = fake.charge("4111111111111111", 50.0)
        assert isinstance(result, ChargeSuccess)

    def test_error_injection(self) -> None:
        """Constructor-configured decline is returned."""
        fake = FakePaymentGateway(
            decline_cards={"4000000000000002": "insufficient funds"}
        )
        result = fake.charge("4000000000000002", 50.0)
        assert isinstance(result, PaymentDeclined)
        assert result.reason == "insufficient funds"

    def test_mutation_tracking(self) -> None:
        """Successful charges are tracked."""
        fake = FakePaymentGateway()
        fake.charge("4111111111111111", 50.0)
        assert len(fake.charges) == 1
        assert fake.charges[0] == ("4111111111111111", 50.0)

Organize tests by operation (one test class per method), not by category.

The Tracking-on-Error Decision

The most subtle design choice: should mutation tracking occur when the operation returns an error?

Ask: "If the real operation fails, did a side effect still occur?"

Implementation pattern:

# No side effect on failure -> check error FIRST, skip tracking
def charge(self, card: str, amount: float) -> ChargeSuccess | PaymentDeclined:
    if card in self._decline_cards:
        return PaymentDeclined(...)  # Return before tracking
    self._charges.append((card, amount))  # Track only on success
    return ChargeSuccess(...)

# Side effect on failure -> track FIRST, then check error
def sync_data(self, source: str) -> SyncComplete | SyncPartialFailure:
    self._sync_attempts.append(source)  # Always track the attempt
    if source in self._partial_failure_sources:
        return SyncPartialFailure(...)
    return SyncComplete(...)

Anti-Patterns

Using try/except in fakes

# WRONG — fakes have no exceptions to catch
class FakePaymentGateway(PaymentGateway):
    def charge(self, card, amount):
        try:
            if card in self._decline_cards:
                return PaymentDeclined(...)
            return ChargeSuccess(...)
        except Exception as e:
            return PaymentDeclined(reason=str(e))  # Dead code

Catching exceptions to return None

# WRONG — masks the actual error
class RealApiClient(ApiClient):
    def get_user(self, user_id: int) -> dict | None:
        try:
            return self._http.get(f"/users/{user_id}").json()
        except Exception:
            return None  # Was it 404? 500? Network timeout?

Use discriminated unions to preserve error context:

# CORRECT — caller can branch on the specific error
def get_user(self, user_id: int) -> UserData | UserNotFound | ApiError:
    ...

Testing only the happy path

Every discriminated union method needs both success AND error tests. A fake that only tests success could silently break error injection for all consumers.

Pattern Structure

Success types: Frozen dataclasses with useful fields (IDs, created resources), or empty markers when the operation itself is the result.

Non-ideal state types: Frozen dataclasses with message: str and descriptive fields.

@dataclass(frozen=True)
class PaymentProcessed:
    transaction_id: str
    amount: float

@dataclass(frozen=True)
class InsufficientFunds:
    available: float
    requested: float
    message: str

@dataclass(frozen=True)
class CardDeclined:
    reason: str
    message: str

class PaymentGateway(ABC):
    @abstractmethod
    def charge(
        self, card: str, amount: float
    ) -> PaymentProcessed | InsufficientFunds | CardDeclined:
        """Process payment. Multiple non-ideal states for different failures."""

Related Documentation

• gateway-architecture.md - Gateway interface design
• patterns.md - Constructor injection and error injection patterns
• testing-strategy.md - Layer 1 "fake-check" tests for fakes
• anti-patterns.md - What to avoid

Testing Patterns

Read this when: You need to implement a specific pattern (constructor injection, mutation tracking, CliRunner, builders, etc.).

Overview

This document covers common patterns used throughout Python test suites. Each pattern includes examples and explanations.

Constructor Injection for Fakes

Pattern: Pass all initial state via constructor keyword arguments.

Implementation

from typing import Any
from pathlib import Path

class FakeDatabaseAdapter(DatabaseAdapter):
    def __init__(
        self,
        *,
        initial_data: dict[str, list[dict]] | None = None,
        users: list[dict] | None = None,
        orders: list[dict] | None = None,
        should_fail_on: list[str] | None = None,
    ) -> None:
        # Initialize mutable state from constructor
        self._tables = initial_data or {}
        if users:
            self._tables["users"] = users
        if orders:
            self._tables["orders"] = orders

        self._should_fail_on = should_fail_on or []

        # Initialize mutation tracking
        self._executed_queries: list[str] = []
        self._executed_commands: list[str] = []
        self._transaction_count = 0

Usage in Tests

# ✅ CORRECT: Constructor injection
def test_with_constructor_injection(tmp_path: Path) -> None:
    # Configure fake with initial state
    fake_db = FakeDatabaseAdapter(
        users=[
            {"id": 1, "name": "Alice", "email": "alice@example.com"},
            {"id": 2, "name": "Bob", "email": "bob@example.com"},
        ],
        orders=[
            {"id": 1, "user_id": 1, "total": 100.00},
        ]
    )

    # Fake is fully configured, ready to use
    users = fake_db.query("SELECT * FROM users")
    assert len(users) == 2

Anti-Pattern

# ❌ WRONG: Mutation after construction
def test_with_mutation() -> None:
    fake_db = FakeDatabaseAdapter()

    # Don't mutate private state directly!
    fake_db._tables["users"] = [...]  # Bypasses encapsulation
    fake_db._executed_queries = []  # Fragile, couples to implementation

Why Constructor Injection?

Benefits:

• Declarative: Test setup is explicit and readable
• Encapsulation: Doesn't expose private implementation details
• Maintainable: Changes to fake internals don't break tests
• Clear intent: Constructor signature documents what can be configured

Rule: If tests need to set up state, add a constructor parameter. Don't mutate private fields.

---

Mutation Tracking Properties

Pattern: Track operations in private lists/dicts, expose via read-only properties.

Implementation

class FakeApiClient(ApiClient):
    def __init__(self, responses: dict[str, Any] | None = None) -> None:
        self._responses = responses or {}

        # Private mutation tracking
        self._requested_endpoints: list[str] = []
        self._posted_data: list[tuple[str, dict]] = []
        self._request_count = 0

    def get(self, endpoint: str) -> dict:
        """GET request."""
        # Track mutation
        self._requested_endpoints.append(endpoint)
        self._request_count += 1

        # Return configured response
        return self._responses.get(endpoint, {})

    def post(self, endpoint: str, *, json: dict) -> dict:
        """POST request."""
        # Track mutation
        self._posted_data.append((endpoint, json))
        self._request_count += 1

        # Return configured response
        return self._responses.get(endpoint, {})

    @property
    def requested_endpoints(self) -> list[str]:
        """Read-only access for test assertions."""
        return self._requested_endpoints.copy()  # Return copy to prevent tampering

    @property
    def posted_data(self) -> list[tuple[str, dict]]:
        """Read-only access for test assertions."""
        return self._posted_data.copy()

    @property
    def request_count(self) -> int:
        """Read-only access for test assertions."""
        return self._request_count

Usage in Tests

def test_mutation_tracking() -> None:
    fake_api = FakeApiClient(
        responses={
            "/users": [{"id": 1, "name": "Alice"}],
            "/users/1": {"id": 1, "name": "Alice"},
        }
    )

    # Perform operations
    users = fake_api.get("/users")
    user = fake_api.get("/users/1")
    fake_api.post("/users", json={"name": "Bob"})

    # Assert mutations were tracked
    assert fake_api.requested_endpoints == ["/users", "/users/1"]
    assert len(fake_api.posted_data) == 1
    assert fake_api.posted_data[0] == ("/users", {"name": "Bob"})
    assert fake_api.request_count == 3

Why Track Mutations?

Benefits:

• Verification: Tests can verify operations were called
• Ordering: Lists preserve call order for sequential assertions
• Arguments: Track arguments passed to operations
• Debugging: Easy to see what operations were performed

Rule: For every write operation, track the mutation in a read-only property.

When to Track on Error

The most subtle decision: should mutation tracking occur when the operation returns an error (a non-ideal state)?

Ask: "If the real operation fails, did a side effect still occur?"

# No side effect on failure -> check error FIRST, skip tracking
def charge(self, card: str, amount: float) -> ChargeSuccess | PaymentDeclined:
    if card in self._decline_cards:
        return PaymentDeclined(...)  # Return before tracking
    self._charges.append((card, amount))  # Track only on success
    return ChargeSuccess(...)

# Side effect on failure -> track FIRST, then check error
def sync_data(self, source: str) -> SyncComplete | SyncPartialFailure:
    self._sync_attempts.append(source)  # Always track the attempt
    if source in self._partial_failure_sources:
        return SyncPartialFailure(...)
    return SyncComplete(...)

---

Using CliRunner for CLI Tests

Pattern: Use Click's CliRunner for testing CLI commands, NOT subprocess.

Basic Usage

from click.testing import CliRunner
import click

@click.command()
@click.argument("name")
@click.option("--greeting", default="Hello")
def greet(name: str, greeting: str) -> None:
    """Greet someone."""
    click.echo(f"{greeting}, {name}!")

def test_cli_command() -> None:
    """Test CLI command with CliRunner."""
    runner = CliRunner()

    # Test with argument
    result = runner.invoke(greet, ["Alice"])

    assert result.exit_code == 0
    assert "Hello, Alice!" in result.output

    # Test with option
    result = runner.invoke(greet, ["Bob", "--greeting", "Hi"])
    assert "Hi, Bob!" in result.output

Separating stdout and stderr (Click 8.2+)

Click 8.2+ automatically separates stdout and stderr. Use result.stdout and result.stderr for independent access:

@click.command()
def mixed_output() -> None:
    """Command that writes to both stdout and stderr."""
    click.echo("Normal output")           # Goes to stdout
    click.echo("Error message", err=True) # Goes to stderr

def test_separate_stdout_stderr() -> None:
    """Test stdout and stderr are captured separately."""
    runner = CliRunner()
    result = runner.invoke(mixed_output)

    # result.output contains combined stdout+stderr (for backwards compat)
    # result.stdout contains only stdout
    # result.stderr contains only stderr
    assert "Normal output" in result.stdout
    assert "Error message" in result.stderr
    assert "Normal output" not in result.stderr
    assert "Error message" not in result.stdout

IMPORTANT: Do NOT use CliRunner(mix_stderr=False) - this parameter was removed in Click 8.2. Stdout/stderr separation is now automatic.

With Context Object

class AppContext:
    """Application context passed to commands."""
    def __init__(self, database: DatabaseAdapter, api: ApiClient) -> None:
        self.database = database
        self.api = api

@click.command()
@click.pass_obj
def sync_data(ctx: AppContext) -> None:
    """Sync data from API to database."""
    data = ctx.api.get("/data")
    ctx.database.execute(f"INSERT INTO sync_log VALUES ('{data}')")
    click.echo(f"Synced {len(data)} records")

def test_command_with_context(tmp_path: Path) -> None:
    """Test command that uses context."""
    # Create context with fakes
    fake_db = FakeDatabaseAdapter()
    fake_api = FakeApiClient(responses={"/data": {"records": [1, 2, 3]}})
    ctx = AppContext(database=fake_db, api=fake_api)

    # Invoke command with context
    runner = CliRunner()
    result = runner.invoke(sync_data, obj=ctx)

    # Assert
    assert result.exit_code == 0
    assert "Synced 1 records" in result.output
    assert len(fake_db.executed_commands) == 1

With Isolated Filesystem

@click.command()
@click.argument("project_name")
def init_project(project_name: str) -> None:
    """Initialize a new project."""
    project_dir = Path(project_name)
    project_dir.mkdir()
    (project_dir / "README.md").write_text(f"# {project_name}")
    (project_dir / "config.yaml").write_text("version: 1.0")
    click.echo(f"Created project: {project_name}")

def test_command_creates_files() -> None:
    """Test command that creates files."""
    runner = CliRunner()

    with runner.isolated_filesystem():
        # Command runs in temporary directory
        result = runner.invoke(init_project, ["my_project"])

        assert result.exit_code == 0
        assert Path("my_project").exists()
        assert Path("my_project/README.md").exists()
        assert Path("my_project/config.yaml").exists()

Capturing Exceptions

def test_command_error() -> None:
    """Test command that raises an exception."""
    @click.command()
    def buggy_cmd():
        raise ValueError("Something went wrong!")

    runner = CliRunner()

    # CliRunner catches exceptions and sets exit_code
    result = runner.invoke(buggy_cmd, catch_exceptions=True)

    assert result.exit_code != 0
    assert "ValueError" in result.output

Why CliRunner (NOT subprocess)?

Performance:

• CliRunner: milliseconds per test
• Subprocess: seconds per test
• ~100x faster with CliRunner

Better debugging:

• Direct access to exceptions
• No shell interpretation issues
• Easier to debug with breakpoints

Rule: Always use CliRunner for CLI tests. Only use subprocess for true end-to-end integration tests.

---

Builder Patterns for Complex Scenarios

Pattern: Use builder pattern to construct complex test scenarios declaratively.

Implementation

from dataclasses import dataclass
from typing import Any

@dataclass
class User:
    id: int
    name: str
    email: str
    balance: float = 100.0

@dataclass
class Product:
    id: int
    name: str
    price: float
    stock: int = 100

class TestScenarioBuilder:
    """Builder for complex test scenarios."""

    def __init__(self) -> None:
        self.users: list[dict] = []
        self.products: list[dict] = []
        self.orders: list[dict] = []
        self.api_responses: dict[str, Any] = {}
        self.config: dict[str, Any] = {}

    def with_user(
        self,
        name: str = "Test User",
        email: str | None = None,
        balance: float = 100.0
    ) -> "TestScenarioBuilder":
        """Add a user to the scenario."""
        user_id = len(self.users) + 1
        if email is None:
            email = f"{name.lower().replace(' ', '.')}@example.com"

        self.users.append({
            "id": user_id,
            "name": name,
            "email": email,
            "balance": balance
        })
        return self

    def with_product(
        self,
        name: str = "Test Product",
        price: float = 10.0,
        stock: int = 100
    ) -> "TestScenarioBuilder":
        """Add a product to the scenario."""
        product_id = len(self.products) + 1
        self.products.append({
            "id": product_id,
            "name": name,
            "price": price,
            "stock": stock
        })
        return self

    def with_order(
        self,
        user_id: int,
        product_ids: list[int] | None = None,
        status: str = "pending"
    ) -> "TestScenarioBuilder":
        """Add an order to the scenario."""
        order_id = len(self.orders) + 1
        self.orders.append({
            "id": order_id,
            "user_id": user_id,
            "product_ids": product_ids or [1],
            "status": status
        })
        return self

    def with_api_response(self, endpoint: str, response: Any) -> "TestScenarioBuilder":
        """Configure API response."""
        self.api_responses[endpoint] = response
        return self

    def with_config(self, **kwargs) -> "TestScenarioBuilder":
        """Set configuration values."""
        self.config.update(kwargs)
        return self

    def build(self) -> tuple[FakeDatabaseAdapter, FakeApiClient, dict]:
        """Build configured test environment."""
        fake_db = FakeDatabaseAdapter(
            users=self.users,
            products=self.products,
            orders=self.orders
        )
        fake_api = FakeApiClient(responses=self.api_responses)

        return fake_db, fake_api, self.config

Usage in Tests

def test_complex_e_commerce_scenario() -> None:
    """Test with multiple users, products, and orders."""
    # Fluent, readable test setup
    fake_db, fake_api, config = (
        TestScenarioBuilder()
        .with_user(name="Alice", balance=500)
        .with_user(name="Bob", balance=100)
        .with_product(name="Laptop", price=1000, stock=5)
        .with_product(name="Mouse", price=25, stock=50)
        .with_order(user_id=1, product_ids=[1, 2])
        .with_order(user_id=2, product_ids=[2])
        .with_api_response("/tax", {"rate": 0.08})
        .with_api_response("/shipping", {"cost": 10.00})
        .with_config(enable_discounts=True, discount_rate=0.1)
        .build()
    )

    service = OrderService(database=fake_db, api_client=fake_api, config=config)

    # Test complex business logic
    result = service.calculate_order_total(order_id=1)

    assert result.subtotal == 1025.00
    assert result.tax == 82.00
    assert result.shipping == 10.00
    assert result.discount == 102.50  # 10% discount
    assert result.total == 1014.50

When to Use Builders

Use builders when:

• Setting up complex multi-component scenarios
• Same scenario reused across multiple tests
• Test setup obscures test intent
• Many optional configurations

Don't use builders when:

• Simple single-component setup
• Setup is only used once
• Constructor injection is sufficient

Benefits

Readability: Fluent API makes test intent clear Reusability: Share builder across test suite Maintainability: Changes to setup logic in one place Flexibility: Mix and match components as needed

---

Simulated Environment Pattern

Pattern: Create isolated test environments with proper setup and cleanup.

Implementation

from contextlib import contextmanager
from dataclasses import dataclass

@dataclass
class TestEnvironment:
    """Container for test environment resources."""
    base_path: Path
    config_path: Path
    data_path: Path
    database: FakeDatabaseAdapter
    api_client: FakeApiClient

@contextmanager
def simulated_environment(tmp_path: Path):
    """Create isolated test environment with proper cleanup."""
    # Setup test environment structure
    base_path = tmp_path / "test_env"
    base_path.mkdir()

    config_path = base_path / "config"
    config_path.mkdir()

    data_path = base_path / "data"
    data_path.mkdir()

    # Create default configuration
    (config_path / "app.yaml").write_text("""
    database:
      host: localhost
      port: 5432
    api:
      base_url: https://api.example.com
      timeout: 30
    """)

    # Initialize test doubles
    fake_db = FakeDatabaseAdapter(
        users=[{"id": 1, "name": "Test User"}]
    )
    fake_api = FakeApiClient(
        responses={"/health": {"status": "ok"}}
    )

    env = TestEnvironment(
        base_path=base_path,
        config_path=config_path,
        data_path=data_path,
        database=fake_db,
        api_client=fake_api
    )

    try:
        yield env
    finally:
        # Cleanup happens automatically with tmp_path
        # But we could add explicit cleanup here if needed
        pass

Usage

def test_with_simulated_environment(tmp_path: Path) -> None:
    """Test in isolated environment."""
    with simulated_environment(tmp_path) as env:
        # Use the environment
        service = DataService(
            database=env.database,
            api_client=env.api_client,
            config_dir=env.config_path
        )

        # Perform operations
        service.process_data()

        # Assert using environment's test doubles
        assert len(env.database.executed_queries) > 0
        assert env.api_client.request_count > 0

        # Can also use the filesystem
        output_file = env.data_path / "output.json"
        assert output_file.exists()

Why Simulated Environments?

Benefits:

• Isolation: Each test runs in clean environment
• Safety: No risk of polluting real filesystem
• Cleanup: Automatic cleanup after test
• Realistic: Tests can create real files/directories when needed

Rule: Use simulated environments for integration tests that need filesystem isolation.

---

Error Injection Pattern

Pattern: Configure fakes to raise errors for testing error handling.

Implementation

from typing import Any

class FakePaymentGateway(PaymentGateway):
    def __init__(
        self,
        *,
        approved_cards: list[str] | None = None,
        declined_cards: list[str] | None = None,
        network_error_on: list[str] | None = None,
        rate_limit_after: int | None = None,
    ) -> None:
        self._approved_cards = approved_cards or []
        self._declined_cards = declined_cards or []
        self._network_error_on = network_error_on or []
        self._rate_limit_after = rate_limit_after
        self._request_count = 0
        self._processed_transactions: list[dict] = []

    def charge(self, card_number: str, amount: float) -> str:
        """Process payment with error injection."""
        self._request_count += 1

        # Inject rate limit error
        if self._rate_limit_after and self._request_count > self._rate_limit_after:
            raise RateLimitError("Too many requests")

        # Inject network error
        if card_number in self._network_error_on:
            raise NetworkError("Connection timeout")

        # Simulate declined card
        if card_number in self._declined_cards:
            raise PaymentDeclined(f"Card {card_number[-4:]} declined")

        # Simulate approved card
        if card_number in self._approved_cards:
            transaction_id = f"txn_{self._request_count:04d}"
            self._processed_transactions.append({
                "id": transaction_id,
                "card": card_number,
                "amount": amount,
                "status": "approved"
            })
            return transaction_id

        # Default behavior
        raise ValueError(f"Unknown card: {card_number}")

    @property
    def processed_transactions(self) -> list[dict]:
        """For test assertions."""
        return self._processed_transactions.copy()

Usage in Tests

def test_handles_payment_declined() -> None:
    """Test error handling when payment is declined."""
    # Configure fake to decline specific card
    payment_gateway = FakePaymentGateway(
        approved_cards=["4111111111111111"],
        declined_cards=["4000000000000002"]
    )

    service = PaymentService(payment_gateway=payment_gateway)

    # Test declined card
    result = service.process_payment("4000000000000002", 100.00)

    assert result.status == "failed"
    assert "declined" in result.error_message.lower()
    assert len(payment_gateway.processed_transactions) == 0

def test_handles_network_errors() -> None:
    """Test handling of network errors."""
    payment_gateway = FakePaymentGateway(
        network_error_on=["4242424242424242"]
    )

    service = PaymentService(payment_gateway=payment_gateway)

    # Should retry on network error
    result = service.process_payment_with_retry("4242424242424242", 50.00)

    assert result.status == "failed"
    assert result.retry_count == 3

def test_handles_rate_limiting() -> None:
    """Test rate limit handling."""
    payment_gateway = FakePaymentGateway(
        approved_cards=["4111111111111111"],
        rate_limit_after=5
    )

    service = PaymentService(payment_gateway=payment_gateway)

    # Process 5 successful payments
    for i in range(5):
        result = service.process_payment("4111111111111111", 10.00)
        assert result.status == "success"

    # 6th payment should hit rate limit
    result = service.process_payment("4111111111111111", 10.00)
    assert result.status == "failed"
    assert "rate limit" in result.error_message.lower()

Benefits

Fast: No need for real system to fail Reliable: Errors are deterministic, not flaky Complete: Test all error paths, even rare ones Safe: No risk of corrupting real state

Rule: Add error injection parameters for operations that can fail.

---

Dry-Run Testing Pattern

Pattern: Verify operations are intercepted, not executed.

Implementation

def test_data_migration_dry_run(tmp_path: Path, capsys) -> None:
    """Verify --dry-run doesn't modify data."""
    # Arrange: Set up fake with initial data
    fake_db = FakeDatabaseAdapter(
        users=[
            {"id": 1, "name": "Alice", "old_field": "value1"},
            {"id": 2, "name": "Bob", "old_field": "value2"},
        ]
    )

    service = DataMigrationService(database=fake_db)

    # Act: Run migration with dry-run flag
    service.migrate_schema(dry_run=True)

    # Assert: Operation was NOT executed
    assert len(fake_db.executed_commands) == 0  # No writes
    assert len(fake_db.executed_queries) == 1   # Only read queries

    # Assert: Dry-run messages were printed
    captured = capsys.readouterr()
    assert "[DRY RUN]" in captured.out
    assert "Would migrate 2 users" in captured.out
    assert "Would drop column: old_field" in captured.out

    # Assert: Data unchanged
    users = fake_db.query("SELECT * FROM users")
    assert all("old_field" in user for user in users)

Pattern

1. Arrange: Set up fake with initial state 2. Act: Execute operation with dry_run=True 3. Assert:

• Mutation tracking shows operations NOT executed
• Output contains [DRY RUN] messages
• State unchanged (operations didn't happen)

CLI Command with Dry-Run

@click.command()
@click.option("--dry-run", is_flag=True, help="Show what would be done")
@click.pass_obj
def cleanup_data(ctx: AppContext, dry_run: bool) -> None:
    """Clean up old data."""
    if dry_run:
        # Wrap database with dry-run integration class
        ctx.database = DryRunDatabaseAdapter(ctx.database)

    old_records = ctx.database.query("SELECT * FROM logs WHERE age > 30")
    click.echo(f"Found {len(old_records)} old records")

    for record in old_records:
        ctx.database.execute(f"DELETE FROM logs WHERE id = {record['id']}")

    if not dry_run:
        click.echo("✓ Cleanup complete")

def test_cleanup_dry_run() -> None:
    """Test cleanup command with dry-run."""
    fake_db = FakeDatabaseAdapter(
        logs=[
            {"id": 1, "age": 45, "message": "old"},
            {"id": 2, "age": 10, "message": "new"},
        ]
    )
    ctx = AppContext(database=fake_db, api=FakeApiClient())

    runner = CliRunner()
    result = runner.invoke(cleanup_data, ["--dry-run"], obj=ctx)

    # Verify no deletions
    assert len(fake_db.executed_commands) == 0
    assert "[DRY RUN]" in result.output
    assert "Would execute: DELETE" in result.output

Benefits

Verifies:

• Dry-run wrapper correctly intercepts operations
• Messages accurately describe what would happen
• No side effects occur in dry-run mode

---

Pure Logic Extraction Pattern

Pattern: Separate decision logic from I/O by extracting pure functions that take input dataclasses and return output dataclasses.

Use when: Testing hooks, CLI commands, or any code with many external dependencies that would require heavy mocking.

The Problem

Hooks and CLI commands often have many I/O dependencies:

• Reading stdin/environment
• Calling subprocess (git, etc.)
• Reading/writing files
• Checking file existence

Testing these requires mocking every dependency, leading to brittle tests with 3-5+ patches per test.

The Solution

1. Input Dataclass: Capture all inputs needed for decision logic 2. Pure Function: All decision logic, no I/O 3. Output Dataclass: Decision result including what actions to take 4. I/O Wrappers: Thin functions that gather inputs and execute outputs

Implementation

from dataclasses import dataclass
from enum import Enum

class Action(Enum):
    ALLOW = 0
    BLOCK = 2

@dataclass(frozen=True)
class HookInput:
    """All inputs needed for decision logic."""
    session_id: str | None
    feature_enabled: bool
    marker_exists: bool
    plan_exists: bool

@dataclass(frozen=True)
class HookOutput:
    """Decision result from pure logic."""
    action: Action
    message: str
    delete_marker: bool = False

def determine_action(hook_input: HookInput) -> HookOutput:
    """Pure function - all decision logic, no I/O."""
    if not hook_input.feature_enabled:
        return HookOutput(Action.ALLOW, "")

    if hook_input.session_id is None:
        return HookOutput(Action.ALLOW, "No session")

    if hook_input.marker_exists:
        return HookOutput(Action.ALLOW, "Marker found", delete_marker=True)

    if hook_input.plan_exists:
        return HookOutput(Action.BLOCK, "Plan exists - prompting user")

    return HookOutput(Action.ALLOW, "No plan found")

# I/O layer
def _gather_inputs() -> HookInput:
    """All I/O happens here."""
    return HookInput(
        session_id=_get_session_from_stdin(),
        feature_enabled=_is_feature_enabled(),
        marker_exists=_marker_path().exists() if _marker_path() else False,
        plan_exists=_find_plan() is not None,
    )

def _execute_result(result: HookOutput) -> None:
    """All I/O happens here."""
    if result.delete_marker:
        _marker_path().unlink()
    click.echo(result.message, err=True)
    sys.exit(result.action.value)

# Main entry point
def hook_command() -> None:
    hook_input = _gather_inputs()
    result = determine_action(hook_input)
    _execute_result(result)

Testing Benefits

Before (mocking):

def test_marker_allows_exit(tmp_path):
    with (
        patch("module.is_in_project", return_value=True),
        patch("subprocess.run", return_value=mock_result),
        patch("module.extract_slugs", return_value=["slug"]),
        patch("module._get_branch", return_value="main"),
        patch("pathlib.Path.home", return_value=tmp_path),
    ):
        result = runner.invoke(hook_command, input=stdin_data)
    assert result.exit_code == 0

After (pure logic):

def test_marker_allows_exit():
    result = determine_action(HookInput(
        session_id="abc123",
        feature_enabled=True,
        marker_exists=True,
        plan_exists=True,
    ))
    assert result.action == Action.ALLOW
    assert result.delete_marker is True

When to Use

• Hooks with 3+ external dependencies
• CLI commands with complex conditional logic
• Any code where test setup dominates test assertions

Results

---

Related Documentation

• workflows.md - Step-by-step guides for using these patterns
• testing-strategy.md - Which layer to test at
• gateway-architecture.md - Understanding fakes and gateway layer
• anti-patterns.md - What to avoid
• python-specific.md - Python-specific testing patterns