Testing Strategies

Testing microservices is fundamentally harder than testing a monolith. Think of it like testing a single factory versus testing a supply chain — in a monolith, you can verify the whole assembly line in one place. With microservices, you need to verify that dozens of independent factories coordinate correctly, handle delayed shipments, and recover when one factory goes offline. This chapter covers comprehensive testing strategies that give you confidence without drowning in slow, brittle tests.

Learning Objectives:

Implement unit testing with mocks
Write integration tests for services
Use contract testing with Pact
Design end-to-end test strategies
Set up test environments

Testing Pyramid for Microservices

Before we dive into a single line of test code, we need a mental model for where to invest testing effort. The testing pyramid is a resource allocation framework disguised as a diagram. The fundamental insight is that tests closer to the bottom (unit tests) are cheap to write, fast to run, and catch localized bugs. Tests closer to the top (E2E) are expensive, slow, and catch integration bugs — but they are also flaky and break for environmental reasons that have nothing to do with your code. If you invert the pyramid and invest heavily in E2E tests, your CI pipeline becomes a coin flip: developers learn to ignore red builds because “the tests are flaky,” and real bugs slip through. The pyramid shape exists because it represents the optimal trade-off between coverage, speed, and signal quality. In microservices, we add two middle layers — contract and component tests — because service boundaries are the highest-risk part of the system and deserve dedicated test types.

┌─────────────────────────────────────────────────────────────────────────────┐
│                    MICROSERVICES TESTING PYRAMID                             │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│                              ▲                                               │
│                             ╱ ╲                                              │
│                            ╱   ╲                                             │
│                           ╱ E2E ╲        Few, slow, expensive               │
│                          ╱───────╲       Test critical user journeys         │
│                         ╱         ╲                                          │
│                        ╱ Component ╲     Service in isolation                │
│                       ╱─────────────╲    with real DB, mocked deps          │
│                      ╱               ╲                                       │
│                     ╱    Contract     ╲  Verify service compatibility       │
│                    ╱───────────────────╲ Consumer-driven contracts          │
│                   ╱                     ╲                                    │
│                  ╱     Integration       ╲  Test with real dependencies     │
│                 ╱─────────────────────────╲ Database, message queue         │
│                ╱                           ╲                                 │
│               ╱           Unit              ╲  Fast, isolated               │
│              ╱───────────────────────────────╲ Many, quick, cheap           │
│                                                                              │
│                                                                              │
│  ┌─────────────┬────────────┬────────────┬───────────────────────────────┐  │
│  │   Level     │   Count    │   Speed    │   What to Test                │  │
│  ├─────────────┼────────────┼────────────┼───────────────────────────────┤  │
│  │ Unit        │ Thousands  │ < 1ms      │ Business logic, utilities     │  │
│  │ Integration │ Hundreds   │ < 1s       │ DB queries, external calls    │  │
│  │ Contract    │ Tens       │ < 5s       │ API compatibility             │  │
│  │ Component   │ Tens       │ < 30s      │ Service endpoints             │  │
│  │ E2E         │ Few        │ < 5min     │ Critical paths                │  │
│  └─────────────┴────────────┴────────────┴───────────────────────────────┘  │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

Testing Strategy Decision Framework

Choosing the right test type for a given scenario is where most teams get it wrong. They either over-invest in E2E tests (slow, flaky, expensive) or under-invest in contract tests (the single most impactful test type for microservices that almost nobody writes). Use this decision framework: “What am I verifying?”

Question	Best Test Type	Why
Does my business logic produce correct results?	Unit test	Fast, isolated, easy to debug
Does my SQL/ORM actually work against a real DB?	Integration test	Catches schema mismatches, query bugs
Can Service A still talk to Service B after B’s latest deploy?	Contract test	Catches breaking API changes before production
Does my service handle the full request lifecycle correctly?	Component test	Catches wiring issues, middleware bugs
Does the critical user journey (checkout, signup) still work?	E2E test	Last line of defense for revenue-critical paths

“How many should I write?”

Test Type	Typical Count Per Service	Run Time Budget	CI Frequency
Unit	200-500+	less than 30 seconds total	Every commit
Integration	30-100	less than 2 minutes	Every commit
Contract	10-30 per consumer	less than 1 minute	Every commit
Component	10-20	less than 5 minutes	Every PR
E2E	5-10 (entire system)	less than 15 minutes	Pre-deploy only

Edge case: when contract tests are not enough. Contract tests verify the shape of responses, not the semantics. If Service B returns {"status": "COMPLETED"} but the meaning of “COMPLETED” has subtly changed (say, it now means “payment authorized” instead of “payment captured”), contract tests pass while your system is broken. For these semantic edge cases, you need integration tests that validate actual behavior, not just schema compliance.

Unit Testing

Testing Business Logic

The key insight with unit testing in microservices: you test your service’s logic in isolation by mocking its dependencies. This is no different from unit testing in a monolith, but the boundaries are cleaner because service APIs are explicit contracts rather than internal method calls. Why mock dependencies? Because the alternative — spinning up real databases and network calls for every test — turns a 10-millisecond unit test into a 500-millisecond integration test. Across thousands of tests, that difference is minutes vs hours of CI time. The trade-off: mocks can drift from reality. You might mock inventoryClient.checkAvailability to return { available: true }, but in production the real API returns a different field name. That is why we complement unit tests with contract tests (which verify the mock matches reality) and integration tests (which exercise real dependencies). Think of unit tests as verifying “my logic is correct assuming my dependencies behave” and other test layers as verifying “my dependencies actually behave the way I assumed.” The dependency injection pattern below is what makes the code testable. If OrderService instantiated its own repository and clients internally, we would have no way to substitute mocks — we would be stuck running the real code against real infrastructure. Constructor injection externalizes these dependencies and flips control: the test decides what OrderService talks to. This applies equally to Node.js and Python; the syntax differs, but the discipline is identical.

Node.js
Python

// services/OrderService.js
// Notice the constructor injection pattern -- this is what makes the class testable.
// Each dependency is an interface we can replace with a mock in tests.
class OrderService {
  constructor(orderRepository, pricingService, inventoryClient) {
    this.orderRepository = orderRepository;
    this.pricingService = pricingService;
    this.inventoryClient = inventoryClient;
  }

  async createOrder(customerId, items) {
    // Validate items
    if (!items || items.length === 0) {
      throw new ValidationError('Order must have at least one item');
    }

    if (items.length > 50) {
      throw new ValidationError('Order cannot have more than 50 items');
    }

    // Check inventory
    const availability = await this.inventoryClient.checkAvailability(items);
    const unavailable = availability.filter(item => !item.available);
    
    if (unavailable.length > 0) {
      throw new InsufficientInventoryError(unavailable);
    }

    // Calculate pricing
    const pricing = await this.pricingService.calculate(items, customerId);

    // Create order
    const order = {
      id: generateId(),
      customerId,
      items,
      subtotal: pricing.subtotal,
      tax: pricing.tax,
      total: pricing.total,
      status: 'PENDING',
      createdAt: new Date()
    };

    await this.orderRepository.save(order);
    
    return order;
  }

  calculateDiscount(order, discountCode) {
    const discounts = {
      'SAVE10': { type: 'percentage', value: 10 },
      'FLAT50': { type: 'fixed', value: 50 },
      'FREESHIP': { type: 'shipping', value: 0 }
    };

    const discount = discounts[discountCode];
    if (!discount) {
      throw new InvalidDiscountError(discountCode);
    }

    switch (discount.type) {
      case 'percentage':
        return order.subtotal * (discount.value / 100);
      case 'fixed':
        return Math.min(discount.value, order.subtotal);
      case 'shipping':
        return order.shippingCost;
      default:
        return 0;
    }
  }
}

# services/order_service.py
# Same constructor injection pattern in Python -- dependencies passed via __init__.
# Python's duck typing means we do not need explicit interfaces; the mock just needs
# to implement the methods we call.
from datetime import datetime
from uuid import uuid4
from typing import List, Dict, Any

from .errors import ValidationError, InsufficientInventoryError, InvalidDiscountError


class OrderService:
    def __init__(self, order_repository, pricing_service, inventory_client):
        self.order_repository = order_repository
        self.pricing_service = pricing_service
        self.inventory_client = inventory_client

    async def create_order(self, customer_id: str, items: List[Dict[str, Any]]) -> Dict[str, Any]:
        # Validate items
        if not items:
            raise ValidationError('Order must have at least one item')

        if len(items) > 50:
            raise ValidationError('Order cannot have more than 50 items')

        # Check inventory
        availability = await self.inventory_client.check_availability(items)
        unavailable = [item for item in availability if not item['available']]

        if unavailable:
            raise InsufficientInventoryError(unavailable)

        # Calculate pricing
        pricing = await self.pricing_service.calculate(items, customer_id)

        # Create order
        order = {
            'id': str(uuid4()),
            'customer_id': customer_id,
            'items': items,
            'subtotal': pricing['subtotal'],
            'tax': pricing['tax'],
            'total': pricing['total'],
            'status': 'PENDING',
            'created_at': datetime.utcnow(),
        }

        await self.order_repository.save(order)

        return order

    def calculate_discount(self, order: Dict[str, Any], discount_code: str) -> float:
        discounts = {
            'SAVE10': {'type': 'percentage', 'value': 10},
            'FLAT50': {'type': 'fixed', 'value': 50},
            'FREESHIP': {'type': 'shipping', 'value': 0},
        }

        discount = discounts.get(discount_code)
        if not discount:
            raise InvalidDiscountError(discount_code)

        if discount['type'] == 'percentage':
            return order['subtotal'] * (discount['value'] / 100)
        elif discount['type'] == 'fixed':
            return min(discount['value'], order['subtotal'])
        elif discount['type'] == 'shipping':
            return order['shipping_cost']
        else:
            return 0

Unit Tests with Jest and pytest

Notice how each mock returns sensible defaults in beforeEach. This is a production pitfall to watch: if your mocks are too tightly coupled to specific test scenarios, a single change in the service code breaks dozens of tests. Default to the “happy path” in setup, then override per test. Why is the happy-path-in-setup pattern so important? Because tests should read like specifications: “Given a valid order, when I call createOrder, then it returns a pending order.” The setup answers “given a valid X” once, at the top, and each individual test focuses only on what makes it different. If every test has to re-configure every mock, you end up with test files where the setup dwarfs the actual assertions, and changing a dependency signature requires touching hundreds of lines. The same principle applies whether you use Jest’s beforeEach or pytest’s fixtures — fixtures compose better (pytest can inject only the fixtures a test needs), but the core idea of “shared happy-path setup, per-test overrides” is the same. One more nuance: verify behavior, not implementation. Tests that assert “mock was called with these exact arguments” are brittle — they fail every time you refactor. Prefer assertions on observable outcomes (what the function returned, what state changed). Only verify interactions when the interaction itself is the behavior you care about (e.g., “did we charge the customer?”).

Caveats & Common Pitfalls: Test Doubles That Do Not Match Production

Hand-written mocks that lie. You hand-code mockInventoryClient.checkAvailability to return { available: true }. In production the real API returns { in_stock: true }. Every unit test passes; every production request 500s. Symptom: 100% green CI, broken in staging.
Mocks returning sync values for async APIs. Your mock returns a plain object; the real client returns a Promise. Tests pass because JavaScript tolerates the difference in many paths. The first .then() in real code returns undefined and the bug surfaces at runtime.
Mocks that never fail. Every mock call succeeds, so error paths are never exercised. Your service has beautiful unit tests and zero resilience; the first network blip in prod takes it down.
Mocks that ignore time. Tests call Date.now() without freezing it, so tests pass at 10am and fail at 11:59pm when a midnight boundary hits.

Solutions & Patterns

Generate mocks from the real API schema. If the producer publishes an OpenAPI spec, generate TypeScript types and mock factories from it. A schema change breaks compilation — impossible to have a silently wrong mock.
Complement unit tests with contract tests. The contract test verifies the mock matches production. If checkAvailability actually returns in_stock, the contract test fails immediately.
Test error paths explicitly. For every happy-path test, write at least one failure case: timeout, 500, malformed response, invalid schema.
Freeze time with jest.useFakeTimers() or freezegun (Python). Any test that touches time must pin it; relying on wall-clock time is how you get flaky nightly builds.
Snapshot real responses. Record a real production response once (in staging), save it as a fixture, and use it as the baseline mock. Refresh when the schema changes.

Node.js
Python

// tests/unit/OrderService.test.js
const { OrderService } = require('../../services/OrderService');
const { ValidationError, InsufficientInventoryError } = require('../../errors');

describe('OrderService', () => {
  let orderService;
  let mockOrderRepository;
  let mockPricingService;
  let mockInventoryClient;

  beforeEach(() => {
    // Create mocks
    mockOrderRepository = {
      save: jest.fn().mockResolvedValue(undefined),
      findById: jest.fn()
    };

    mockPricingService = {
      calculate: jest.fn().mockResolvedValue({
        subtotal: 100,
        tax: 10,
        total: 110
      })
    };

    mockInventoryClient = {
      checkAvailability: jest.fn().mockResolvedValue([
        { productId: 'p1', available: true },
        { productId: 'p2', available: true }
      ])
    };

    orderService = new OrderService(
      mockOrderRepository,
      mockPricingService,
      mockInventoryClient
    );
  });

  describe('createOrder', () => {
    const customerId = 'customer-123';
    const validItems = [
      { productId: 'p1', quantity: 2 },
      { productId: 'p2', quantity: 1 }
    ];

    it('should create order with valid items', async () => {
      const order = await orderService.createOrder(customerId, validItems);

      expect(order).toMatchObject({
        customerId,
        items: validItems,
        subtotal: 100,
        tax: 10,
        total: 110,
        status: 'PENDING'
      });
      expect(order.id).toBeDefined();
      expect(mockOrderRepository.save).toHaveBeenCalledWith(
        expect.objectContaining({ customerId })
      );
    });

    it('should throw ValidationError for empty items', async () => {
      await expect(
        orderService.createOrder(customerId, [])
      ).rejects.toThrow(ValidationError);

      await expect(
        orderService.createOrder(customerId, [])
      ).rejects.toThrow('Order must have at least one item');
    });

    it('should throw ValidationError for too many items', async () => {
      const tooManyItems = Array(51).fill({ productId: 'p1', quantity: 1 });

      await expect(
        orderService.createOrder(customerId, tooManyItems)
      ).rejects.toThrow(ValidationError);
    });

    it('should throw InsufficientInventoryError when items unavailable', async () => {
      mockInventoryClient.checkAvailability.mockResolvedValue([
        { productId: 'p1', available: true },
        { productId: 'p2', available: false }
      ]);

      await expect(
        orderService.createOrder(customerId, validItems)
      ).rejects.toThrow(InsufficientInventoryError);
    });

    it('should check inventory before calculating pricing', async () => {
      await orderService.createOrder(customerId, validItems);

      expect(mockInventoryClient.checkAvailability).toHaveBeenCalledBefore(
        mockPricingService.calculate
      );
    });
  });

  describe('calculateDiscount', () => {
    const order = { subtotal: 200, shippingCost: 15 };

    it('should apply percentage discount correctly', () => {
      const discount = orderService.calculateDiscount(order, 'SAVE10');
      expect(discount).toBe(20); // 10% of 200
    });

    it('should apply fixed discount correctly', () => {
      const discount = orderService.calculateDiscount(order, 'FLAT50');
      expect(discount).toBe(50);
    });

    it('should not exceed subtotal for fixed discount', () => {
      const smallOrder = { subtotal: 30 };
      const discount = orderService.calculateDiscount(smallOrder, 'FLAT50');
      expect(discount).toBe(30);
    });

    it('should apply shipping discount correctly', () => {
      const discount = orderService.calculateDiscount(order, 'FREESHIP');
      expect(discount).toBe(15);
    });

    it('should throw for invalid discount code', () => {
      expect(() => {
        orderService.calculateDiscount(order, 'INVALID');
      }).toThrow('Invalid discount code');
    });
  });
});

# tests/unit/test_order_service.py
# pytest fixtures replace Jest's beforeEach. AsyncMock is the async cousin of MagicMock.
# pytest-asyncio provides the @pytest.mark.asyncio decorator for async test functions.
import pytest
from unittest.mock import AsyncMock, MagicMock

from services.order_service import OrderService
from services.errors import ValidationError, InsufficientInventoryError


@pytest.fixture
def mock_order_repository():
    repo = MagicMock()
    repo.save = AsyncMock(return_value=None)
    repo.find_by_id = AsyncMock()
    return repo


@pytest.fixture
def mock_pricing_service():
    service = MagicMock()
    service.calculate = AsyncMock(return_value={
        'subtotal': 100,
        'tax': 10,
        'total': 110,
    })
    return service


@pytest.fixture
def mock_inventory_client():
    client = MagicMock()
    client.check_availability = AsyncMock(return_value=[
        {'productId': 'p1', 'available': True},
        {'productId': 'p2', 'available': True},
    ])
    return client


@pytest.fixture
def order_service(mock_order_repository, mock_pricing_service, mock_inventory_client):
    return OrderService(mock_order_repository, mock_pricing_service, mock_inventory_client)


class TestCreateOrder:
    customer_id = 'customer-123'
    valid_items = [
        {'productId': 'p1', 'quantity': 2},
        {'productId': 'p2', 'quantity': 1},
    ]

    @pytest.mark.asyncio
    async def test_creates_order_with_valid_items(self, order_service, mock_order_repository):
        order = await order_service.create_order(self.customer_id, self.valid_items)

        assert order['customer_id'] == self.customer_id
        assert order['items'] == self.valid_items
        assert order['subtotal'] == 100
        assert order['status'] == 'PENDING'
        assert order['id'] is not None
        mock_order_repository.save.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_raises_for_empty_items(self, order_service):
        with pytest.raises(ValidationError, match='at least one item'):
            await order_service.create_order(self.customer_id, [])

    @pytest.mark.asyncio
    async def test_raises_for_too_many_items(self, order_service):
        too_many = [{'productId': 'p1', 'quantity': 1}] * 51
        with pytest.raises(ValidationError):
            await order_service.create_order(self.customer_id, too_many)

    @pytest.mark.asyncio
    async def test_raises_when_items_unavailable(self, order_service, mock_inventory_client):
        mock_inventory_client.check_availability.return_value = [
            {'productId': 'p1', 'available': True},
            {'productId': 'p2', 'available': False},
        ]

        with pytest.raises(InsufficientInventoryError):
            await order_service.create_order(self.customer_id, self.valid_items)

    @pytest.mark.asyncio
    async def test_checks_inventory_before_pricing(
        self, order_service, mock_inventory_client, mock_pricing_service
    ):
        call_order = []
        mock_inventory_client.check_availability.side_effect = lambda *a, **k: (
            call_order.append('inventory') or [
                {'productId': 'p1', 'available': True},
                {'productId': 'p2', 'available': True},
            ]
        )
        mock_pricing_service.calculate.side_effect = lambda *a, **k: (
            call_order.append('pricing') or {'subtotal': 100, 'tax': 10, 'total': 110}
        )

        await order_service.create_order(self.customer_id, self.valid_items)
        assert call_order == ['inventory', 'pricing']


class TestCalculateDiscount:
    order = {'subtotal': 200, 'shipping_cost': 15}

    def test_percentage_discount(self, order_service):
        assert order_service.calculate_discount(self.order, 'SAVE10') == 20

    def test_fixed_discount(self, order_service):
        assert order_service.calculate_discount(self.order, 'FLAT50') == 50

    def test_fixed_discount_capped_at_subtotal(self, order_service):
        small = {'subtotal': 30}
        assert order_service.calculate_discount(small, 'FLAT50') == 30

    def test_shipping_discount(self, order_service):
        assert order_service.calculate_discount(self.order, 'FREESHIP') == 15

    def test_invalid_code_raises(self, order_service):
        with pytest.raises(Exception, match='INVALID'):
            order_service.calculate_discount(self.order, 'INVALID')

Integration Testing

Database Integration Tests

Unit tests with mocked repositories can be 100% green while your SQL is completely broken. The mock does not know your column is customer_id and you typed customerid. The mock does not know your foreign key constraint will reject orphaned rows. The mock does not know your JSONB column needs a ::jsonb cast. Every one of these bugs only surfaces when code meets a real database. That is the purpose of integration tests: to verify that your persistence layer actually works against the real storage engine, with real schemas, real constraints, and real SQL dialects. The architectural insight is that integration tests are where your “database per service” principle gets tested. In a microservices world, each service owns its schema. Integration tests verify that your service correctly manages its own schema without reaching into shared tables or assuming knowledge of other services’ data. They also catch performance cliffs — a query that runs in 2 ms on an empty table might take 2 seconds at 10 million rows. Seeding realistic data volumes in integration tests surfaces these issues early. The trade-off: integration tests are slower than unit tests (seconds instead of milliseconds) and require infrastructure (a running Postgres, a migration system). That is fine — you write fewer of them, and you target the code paths that unit tests cannot cover: the SQL, the ORM mappings, the transactional boundaries. If you skip integration tests, your first integration “test” runs in production, with real customer data, at 3 AM.

Node.js
Python

// tests/integration/OrderRepository.test.js
const { OrderRepository } = require('../../repositories/OrderRepository');
const { createTestDatabase, closeTestDatabase } = require('../helpers/database');

describe('OrderRepository Integration', () => {
  let orderRepository;
  let db;

  beforeAll(async () => {
    db = await createTestDatabase();
    orderRepository = new OrderRepository(db);
  });

  afterAll(async () => {
    await closeTestDatabase(db);
  });

  beforeEach(async () => {
    await db.query('DELETE FROM orders');
    await db.query('DELETE FROM order_items');
  });

  describe('save', () => {
    it('should persist order to database', async () => {
      const order = {
        id: 'order-123',
        customerId: 'customer-456',
        items: [{ productId: 'p1', quantity: 2, price: 50 }],
        total: 100,
        status: 'PENDING'
      };

      await orderRepository.save(order);

      const result = await db.query(
        'SELECT * FROM orders WHERE id = $1',
        [order.id]
      );

      expect(result.rows[0]).toMatchObject({
        id: order.id,
        customer_id: order.customerId,
        total: '100.00',
        status: 'PENDING'
      });
    });

    it('should save order items', async () => {
      const order = {
        id: 'order-123',
        customerId: 'customer-456',
        items: [
          { productId: 'p1', quantity: 2, price: 50 },
          { productId: 'p2', quantity: 1, price: 30 }
        ],
        total: 130,
        status: 'PENDING'
      };

      await orderRepository.save(order);

      const result = await db.query(
        'SELECT * FROM order_items WHERE order_id = $1',
        [order.id]
      );

      expect(result.rows).toHaveLength(2);
    });
  });

  describe('findByCustomerId', () => {
    beforeEach(async () => {
      // Seed test data
      await orderRepository.save({
        id: 'order-1',
        customerId: 'customer-A',
        items: [],
        total: 100,
        status: 'COMPLETED'
      });
      await orderRepository.save({
        id: 'order-2',
        customerId: 'customer-A',
        items: [],
        total: 200,
        status: 'PENDING'
      });
      await orderRepository.save({
        id: 'order-3',
        customerId: 'customer-B',
        items: [],
        total: 150,
        status: 'COMPLETED'
      });
    });

    it('should return orders for specific customer', async () => {
      const orders = await orderRepository.findByCustomerId('customer-A');

      expect(orders).toHaveLength(2);
      expect(orders.every(o => o.customerId === 'customer-A')).toBe(true);
    });

    it('should return empty array for customer with no orders', async () => {
      const orders = await orderRepository.findByCustomerId('customer-C');
      expect(orders).toEqual([]);
    });
  });
});

# tests/integration/test_order_repository.py
# Uses pytest fixtures with session scope for DB lifecycle, function scope for cleanup.
# Testcontainers-python provides real PostgreSQL instances for genuinely reliable tests;
# an in-memory SQLite will lie to you about dialect-specific features (JSONB, arrays, etc).
import pytest
import pytest_asyncio
from testcontainers.postgres import PostgresContainer
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from repositories.order_repository import OrderRepository
from db.migrations import run_migrations


@pytest.fixture(scope='session')
def postgres_container():
    with PostgresContainer('postgres:15-alpine') as pg:
        yield pg


@pytest_asyncio.fixture(scope='session')
async def db_engine(postgres_container):
    url = postgres_container.get_connection_url().replace('postgresql://', 'postgresql+asyncpg://')
    engine = create_async_engine(url)
    await run_migrations(engine)
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def db_session(db_engine):
    async_session = sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        # Clean slate before each test
        await session.execute('DELETE FROM order_items')
        await session.execute('DELETE FROM orders')
        await session.commit()
        yield session


@pytest_asyncio.fixture
async def order_repository(db_session):
    return OrderRepository(db_session)


class TestSave:
    @pytest.mark.asyncio
    async def test_persists_order(self, order_repository, db_session):
        order = {
            'id': 'order-123',
            'customer_id': 'customer-456',
            'items': [{'productId': 'p1', 'quantity': 2, 'price': 50}],
            'total': 100,
            'status': 'PENDING',
        }

        await order_repository.save(order)

        result = await db_session.execute(
            'SELECT id, customer_id, total, status FROM orders WHERE id = :id',
            {'id': order['id']}
        )
        row = result.first()
        assert row.id == order['id']
        assert row.customer_id == order['customer_id']
        assert float(row.total) == 100.00
        assert row.status == 'PENDING'

    @pytest.mark.asyncio
    async def test_saves_order_items(self, order_repository, db_session):
        order = {
            'id': 'order-123',
            'customer_id': 'customer-456',
            'items': [
                {'productId': 'p1', 'quantity': 2, 'price': 50},
                {'productId': 'p2', 'quantity': 1, 'price': 30},
            ],
            'total': 130,
            'status': 'PENDING',
        }

        await order_repository.save(order)

        result = await db_session.execute(
            'SELECT * FROM order_items WHERE order_id = :id',
            {'id': order['id']}
        )
        assert len(result.all()) == 2


class TestFindByCustomerId:
    @pytest_asyncio.fixture(autouse=True)
    async def seed_data(self, order_repository):
        await order_repository.save({
            'id': 'order-1', 'customer_id': 'customer-A', 'items': [],
            'total': 100, 'status': 'COMPLETED',
        })
        await order_repository.save({
            'id': 'order-2', 'customer_id': 'customer-A', 'items': [],
            'total': 200, 'status': 'PENDING',
        })
        await order_repository.save({
            'id': 'order-3', 'customer_id': 'customer-B', 'items': [],
            'total': 150, 'status': 'COMPLETED',
        })

    @pytest.mark.asyncio
    async def test_returns_orders_for_customer(self, order_repository):
        orders = await order_repository.find_by_customer_id('customer-A')
        assert len(orders) == 2
        assert all(o['customer_id'] == 'customer-A' for o in orders)

    @pytest.mark.asyncio
    async def test_returns_empty_for_unknown_customer(self, order_repository):
        orders = await order_repository.find_by_customer_id('customer-C')
        assert orders == []

API Integration Tests

Integration tests at the API level verify the entire request-response lifecycle: routing, middleware, authentication, request parsing, business logic, database writes, and response serialization. This is the layer where most “works on my machine” bugs live — a middleware that strips authentication headers in production but not in dev, a request body parser that silently truncates large payloads, a response serializer that exposes internal fields. The supertest library in Node.js and httpx.AsyncClient in Python both test the application without opening a real network socket. They invoke the application’s request handler directly with a synthetic request object. This is faster than spinning up an HTTP server, and it gives you the same level of realism for middleware and routing. You are still testing the real app — just without the network layer that would add latency and potential flakiness. A common mistake in API integration tests is relying on side effects from previous tests (a test that creates an order, followed by a test that assumes the order exists). This creates order-dependent tests that pass locally but fail in CI when tests run in parallel or a different order. Always reset state between tests. The beforeEach delete in the example below is not optional paranoia — it is the difference between a test suite that works and one that develops mysterious flakiness over time.

Caveats & Common Pitfalls: Integration Tests Hitting Shared Staging DB

Shared staging database across developers and CI. Two CI runs collide; one deletes the other’s test data mid-test; both runs fail intermittently. Engineers learn to “just rerun the pipeline” and stop trusting CI entirely.
No cleanup, accumulating test data. Your test_orders table grows to 10M rows over months. Suddenly the find_by_customer_id test takes 40 seconds because the index is bloated and the seed phase churns through locks.
Schema drift between environments. Local Postgres 14, staging Postgres 15, prod Aurora Postgres 13. A JSONB behavior difference passes locally and fails in prod — or worse, passes in staging and fails in prod.
Tests depending on test data that does not exist in fresh DBs. A test assumes customer_id cust-1 exists because someone seeded it manually two years ago. A fresh-spin environment has no such row; the test fails with a cryptic foreign key error.

Solutions & Patterns

Testcontainers per test suite. Every CI run spins its own ephemeral Postgres container. Zero shared state, zero cross-run contamination, identical version and config as production.
Transactional rollback per test. Start a transaction in beforeEach, roll back in afterEach. The test runs against real SQL but leaves zero trace. Dramatically faster than TRUNCATE between tests.
Pin the Postgres version to match production. If prod is aws:aurora-postgres:14.7, your Testcontainers image is postgres:14.7-alpine. Drift between test and prod DB versions is a silent bug generator.
Ephemeral schema per test run. Create a uniquely-named schema (test_<run-id>) at start, drop it at end. Multiple CI runs can share a single container without collision.
Factories, not fixtures. Every test creates the data it needs via factories (FactoryBot, factory_boy). No test depends on data created by another test or a seed script.

Node.js
Python

// tests/integration/OrderAPI.test.js
const request = require('supertest');
const { createApp } = require('../../app');
const { createTestDatabase, closeTestDatabase } = require('../helpers/database');

describe('Order API Integration', () => {
  let app;
  let db;
  let authToken;

  beforeAll(async () => {
    db = await createTestDatabase();
    app = createApp({ db });
    
    // Get auth token
    authToken = await getTestAuthToken();
  });

  afterAll(async () => {
    await closeTestDatabase(db);
  });

  beforeEach(async () => {
    await db.query('DELETE FROM orders');
  });

  describe('POST /orders', () => {
    it('should create order and return 201', async () => {
      const orderData = {
        items: [
          { productId: 'product-1', quantity: 2 }
        ]
      };

      const response = await request(app)
        .post('/orders')
        .set('Authorization', `Bearer ${authToken}`)
        .send(orderData)
        .expect(201);

      expect(response.body).toMatchObject({
        id: expect.any(String),
        status: 'PENDING',
        items: expect.arrayContaining([
          expect.objectContaining({ productId: 'product-1' })
        ])
      });
    });

    it('should return 400 for invalid order data', async () => {
      const response = await request(app)
        .post('/orders')
        .set('Authorization', `Bearer ${authToken}`)
        .send({ items: [] })
        .expect(400);

      expect(response.body.error).toContain('at least one item');
    });

    it('should return 401 without auth token', async () => {
      await request(app)
        .post('/orders')
        .send({ items: [{ productId: 'p1', quantity: 1 }] })
        .expect(401);
    });
  });

  describe('GET /orders/:id', () => {
    it('should return order by id', async () => {
      // Create order first
      const createResponse = await request(app)
        .post('/orders')
        .set('Authorization', `Bearer ${authToken}`)
        .send({ items: [{ productId: 'p1', quantity: 1 }] });

      const orderId = createResponse.body.id;

      const response = await request(app)
        .get(`/orders/${orderId}`)
        .set('Authorization', `Bearer ${authToken}`)
        .expect(200);

      expect(response.body.id).toBe(orderId);
    });

    it('should return 404 for non-existent order', async () => {
      await request(app)
        .get('/orders/non-existent-id')
        .set('Authorization', `Bearer ${authToken}`)
        .expect(404);
    });
  });
});

# tests/integration/test_order_api.py
# httpx.AsyncClient with ASGITransport calls the FastAPI app directly -- no network,
# no running server, but full middleware and routing coverage.
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport

from app import create_app
from tests.helpers.auth import get_test_auth_token


@pytest_asyncio.fixture(scope='module')
async def app(db_engine):
    return create_app(db_engine=db_engine)


@pytest_asyncio.fixture
async def client(app):
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url='http://test') as c:
        yield c


@pytest.fixture(scope='module')
def auth_token():
    return get_test_auth_token()


@pytest_asyncio.fixture(autouse=True)
async def clean_orders(db_session):
    await db_session.execute('DELETE FROM orders')
    await db_session.commit()


class TestCreateOrderEndpoint:
    @pytest.mark.asyncio
    async def test_creates_order_returns_201(self, client, auth_token):
        response = await client.post(
            '/orders',
            headers={'Authorization': f'Bearer {auth_token}'},
            json={'items': [{'productId': 'product-1', 'quantity': 2}]},
        )

        assert response.status_code == 201
        body = response.json()
        assert 'id' in body
        assert body['status'] == 'PENDING'
        assert any(item['productId'] == 'product-1' for item in body['items'])

    @pytest.mark.asyncio
    async def test_returns_400_for_invalid_data(self, client, auth_token):
        response = await client.post(
            '/orders',
            headers={'Authorization': f'Bearer {auth_token}'},
            json={'items': []},
        )
        assert response.status_code == 400
        assert 'at least one item' in response.json()['error']

    @pytest.mark.asyncio
    async def test_returns_401_without_token(self, client):
        response = await client.post(
            '/orders', json={'items': [{'productId': 'p1', 'quantity': 1}]}
        )
        assert response.status_code == 401


class TestGetOrderEndpoint:
    @pytest.mark.asyncio
    async def test_returns_order_by_id(self, client, auth_token):
        create_response = await client.post(
            '/orders',
            headers={'Authorization': f'Bearer {auth_token}'},
            json={'items': [{'productId': 'p1', 'quantity': 1}]},
        )
        order_id = create_response.json()['id']

        response = await client.get(
            f'/orders/{order_id}',
            headers={'Authorization': f'Bearer {auth_token}'},
        )

        assert response.status_code == 200
        assert response.json()['id'] == order_id

    @pytest.mark.asyncio
    async def test_returns_404_for_missing_order(self, client, auth_token):
        response = await client.get(
            '/orders/non-existent-id',
            headers={'Authorization': f'Bearer {auth_token}'},
        )
        assert response.status_code == 404

Contract Testing with Pact

Contract testing is the unsung hero of microservices testing. Here is the core problem it solves: Service A depends on Service B’s API. Service B deploys a change that breaks Service A. Without contract tests, you only discover this in staging (or worse, production). With contract tests, Service B’s CI pipeline fails before the breaking change ever merges. Think of it like a legal contract between two businesses — the consumer writes down what they expect, and the provider agrees to deliver it. If the provider changes the deal, the contract breaks visibly. Why not just use integration tests across services? Because integration tests require running both services together, which means you need an integration environment, shared test data, and coordinated deployments. Contract tests invert the relationship: the consumer runs its test against a Pact mock server (fast, isolated), which records what the consumer expects. The provider later replays those expectations against its real implementation (also fast, also isolated). No shared environment is required. Each team tests independently, on their own CI, and incompatibilities are detected at PR time rather than at integration time. This is the testing strategy that makes independent deployability actually work. The trade-off: contract tests verify syntax (field names, types, HTTP status codes), not semantics. A provider can rename a field and fail the contract; it cannot change the meaning of a field and still pass the contract. For semantic verification, you need integration tests or explicit versioning. Contract tests are necessary but not sufficient — they are the first line of defense, not the only line.

Consumer Contract Test

Node.js
Python

// tests/contract/OrderService.consumer.test.js
const { Pact } = require('@pact-foundation/pact');
const { PaymentClient } = require('../../clients/PaymentClient');
const path = require('path');

describe('Order Service - Payment Service Contract', () => {
  const provider = new Pact({
    consumer: 'OrderService',
    provider: 'PaymentService',
    port: 8081,
    log: path.resolve(process.cwd(), 'logs', 'pact.log'),
    dir: path.resolve(process.cwd(), 'pacts'),
    logLevel: 'warn'
  });

  beforeAll(() => provider.setup());
  afterAll(() => provider.finalize());
  afterEach(() => provider.verify());

  describe('processPayment', () => {
    it('should process payment successfully', async () => {
      // Define expected interaction
      await provider.addInteraction({
        state: 'customer has valid payment method',
        uponReceiving: 'a request to process payment',
        withRequest: {
          method: 'POST',
          path: '/payments',
          headers: {
            'Content-Type': 'application/json'
          },
          body: {
            orderId: '12345',
            amount: 100.00,
            currency: 'USD',
            customerId: 'customer-123'
          }
        },
        willRespondWith: {
          status: 200,
          headers: {
            'Content-Type': 'application/json'
          },
          body: {
            id: like('pay_abc123'),
            orderId: '12345',
            status: 'COMPLETED',
            amount: 100.00,
            processedAt: like('2024-01-15T10:30:00Z')
          }
        }
      });

      // Execute the actual client
      const client = new PaymentClient(`http://localhost:8081`);
      const result = await client.processPayment({
        orderId: '12345',
        amount: 100.00,
        currency: 'USD',
        customerId: 'customer-123'
      });

      expect(result.status).toBe('COMPLETED');
      expect(result.orderId).toBe('12345');
    });

    it('should handle payment failure', async () => {
      await provider.addInteraction({
        state: 'customer has insufficient funds',
        uponReceiving: 'a request to process payment',
        withRequest: {
          method: 'POST',
          path: '/payments',
          headers: {
            'Content-Type': 'application/json'
          },
          body: {
            orderId: '12345',
            amount: 10000.00,
            currency: 'USD',
            customerId: 'customer-456'
          }
        },
        willRespondWith: {
          status: 402,
          headers: {
            'Content-Type': 'application/json'
          },
          body: {
            error: 'INSUFFICIENT_FUNDS',
            message: like('Payment declined due to insufficient funds')
          }
        }
      });

      const client = new PaymentClient(`http://localhost:8081`);
      
      await expect(
        client.processPayment({
          orderId: '12345',
          amount: 10000.00,
          currency: 'USD',
          customerId: 'customer-456'
        })
      ).rejects.toThrow('INSUFFICIENT_FUNDS');
    });
  });
});

// Pact matchers
const { like, eachLike, term } = require('@pact-foundation/pact').Matchers;

# tests/contract/test_order_service_consumer.py
# pact-python provides a Consumer/Provider API that mirrors the JS version.
# The mock service runs on a local port while the test executes; the Pact file is
# published at teardown and later verified by the provider's pipeline.
import atexit
import pytest
from pact import Consumer, Provider, Like, Term

from clients.payment_client import PaymentClient


pact = Consumer('OrderService').has_pact_with(
    Provider('PaymentService'),
    host_name='localhost',
    port=8081,
    pact_dir='./pacts',
)
pact.start_service()
atexit.register(pact.stop_service)


class TestProcessPayment:
    def test_payment_success(self):
        expected = {
            'id': Like('pay_abc123'),
            'orderId': '12345',
            'status': 'COMPLETED',
            'amount': 100.00,
            'processedAt': Like('2024-01-15T10:30:00Z'),
        }

        (pact
            .given('customer has valid payment method')
            .upon_receiving('a request to process payment')
            .with_request(
                method='POST',
                path='/payments',
                headers={'Content-Type': 'application/json'},
                body={
                    'orderId': '12345',
                    'amount': 100.00,
                    'currency': 'USD',
                    'customerId': 'customer-123',
                },
            )
            .will_respond_with(
                status=200,
                headers={'Content-Type': 'application/json'},
                body=expected,
            ))

        with pact:
            client = PaymentClient('http://localhost:8081')
            result = client.process_payment({
                'orderId': '12345',
                'amount': 100.00,
                'currency': 'USD',
                'customerId': 'customer-123',
            })

            assert result['status'] == 'COMPLETED'
            assert result['orderId'] == '12345'

    def test_payment_insufficient_funds(self):
        (pact
            .given('customer has insufficient funds')
            .upon_receiving('a request to process payment')
            .with_request(
                method='POST',
                path='/payments',
                headers={'Content-Type': 'application/json'},
                body={
                    'orderId': '12345',
                    'amount': 10000.00,
                    'currency': 'USD',
                    'customerId': 'customer-456',
                },
            )
            .will_respond_with(
                status=402,
                headers={'Content-Type': 'application/json'},
                body={
                    'error': 'INSUFFICIENT_FUNDS',
                    'message': Like('Payment declined due to insufficient funds'),
                },
            ))

        with pact:
            client = PaymentClient('http://localhost:8081')
            with pytest.raises(Exception, match='INSUFFICIENT_FUNDS'):
                client.process_payment({
                    'orderId': '12345',
                    'amount': 10000.00,
                    'currency': 'USD',
                    'customerId': 'customer-456',
                })

Provider Contract Test

The provider side of contract testing is where the consumer’s expectations meet the provider’s reality. The provider downloads pact files (usually from a Pact Broker), then replays each interaction against its real, running service. Critically, the provider must restore its database state before each interaction so that the “given” preconditions hold — this is what stateHandlers are for. If you skip provider verification, contract tests give you false confidence. The consumer test only verifies that the consumer code works against its own assumptions. Until the provider runs verification, you have no proof that the real provider actually behaves that way. Teams that “have Pact tests” but never run provider verification are in the worst of both worlds — they have the maintenance burden of Pact without the benefit.

Caveats & Common Pitfalls: Contract Tests Not Run in CI

Consumer writes contracts, provider never verifies them. Pact files accumulate in a broker, nobody enforces them, and the provider ships a breaking change that only surfaces in staging. The contracts became documentation, not gates.
Provider verification runs only nightly, not on every PR. A breaking merge happens at 10am; the verification failure lands at 2am; by morning it is blocking 20 other PRs. Verification must be a PR gate, not a nightly job.
Contracts without can-i-deploy checks. Even with verification passing on both sides, you deploy a provider version the consumer has never seen. Pact Broker’s can-i-deploy answers “is this deploy safe for all current consumers?” — skip it and you ship risk blindly.
Contracts covering only the happy path. The consumer records 200 OK responses but never the 404 or 429 error shapes. Provider changes the error envelope; consumer parses it wrong; contract passes because nobody wrote an error-case contract.

Solutions & Patterns

Wire provider verification into the PR pipeline. Every provider PR runs pact-verifier against the latest consumer contracts before merging. If verification fails, the PR is blocked.
Use Pact Broker’s webhooks + can-i-deploy. Consumer CI asks the broker: “based on the verified pacts for provider version X, can I deploy consumer version Y?” If no, halt the deploy.
Write contracts for error cases. Document 404, 400, 401, 429, 5xx response shapes as explicit interactions. These are where breakages most often hide.
Version the broker state. Tag pacts with git SHA and environment (dev, staging, prod). A production deploy only proceeds if the prod tag is compatible, not the floating latest.
Make the consumer publish on every CI run, not just merges. This surfaces the consumer-provider mismatch at PR review time, long before merge.

Node.js
Python

// tests/contract/PaymentService.provider.test.js
const { Verifier } = require('@pact-foundation/pact');
const { createApp } = require('../../app');
const path = require('path');

describe('Payment Service - Provider Verification', () => {
  let server;

  beforeAll(async () => {
    const app = createApp();
    server = app.listen(8082);
  });

  afterAll(() => {
    server.close();
  });

  it('should validate the expectations of Order Service', async () => {
    const opts = {
      provider: 'PaymentService',
      providerBaseUrl: 'http://localhost:8082',
      pactUrls: [
        path.resolve(process.cwd(), 'pacts', 'orderservice-paymentservice.json')
      ],
      // Or use Pact Broker
      // pactBrokerUrl: 'https://your-broker.pactflow.io',
      // pactBrokerToken: process.env.PACT_BROKER_TOKEN,
      stateHandlers: {
        'customer has valid payment method': async () => {
          // Set up test data for valid payment method
          await seedTestData({
            customerId: 'customer-123',
            paymentMethods: [{ type: 'card', valid: true }]
          });
        },
        'customer has insufficient funds': async () => {
          // Set up test data for insufficient funds
          await seedTestData({
            customerId: 'customer-456',
            balance: 0
          });
        }
      },
      publishVerificationResult: process.env.CI === 'true',
      providerVersion: process.env.GIT_COMMIT || '1.0.0'
    };

    await new Verifier(opts).verifyProvider();
  });
});

# tests/contract/test_payment_service_provider.py
# pact-python exposes a CLI verifier; we invoke it via the Verifier class.
# State handlers are exposed as HTTP endpoints on a setup URL -- Pact POSTs to that
# URL with the state name before each interaction.
import os
import subprocess
import pytest
from multiprocessing import Process

import uvicorn
from pact import Verifier

from app import create_app
from tests.helpers.seed import seed_test_data


PROVIDER_PORT = 8082


def run_server():
    app = create_app()

    # Expose a state setup endpoint required by Pact
    @app.post('/_pact/provider_states')
    async def provider_states(payload: dict):
        state = payload['state']
        if state == 'customer has valid payment method':
            await seed_test_data({
                'customer_id': 'customer-123',
                'payment_methods': [{'type': 'card', 'valid': True}],
            })
        elif state == 'customer has insufficient funds':
            await seed_test_data({'customer_id': 'customer-456', 'balance': 0})
        return {'result': state}

    uvicorn.run(app, host='0.0.0.0', port=PROVIDER_PORT, log_level='warning')


@pytest.fixture(scope='module')
def provider_server():
    p = Process(target=run_server, daemon=True)
    p.start()
    yield
    p.terminate()
    p.join()


def test_provider_satisfies_consumer_contract(provider_server):
    verifier = Verifier(
        provider='PaymentService',
        provider_base_url=f'http://localhost:{PROVIDER_PORT}',
    )

    exit_code, _ = verifier.verify_pacts(
        './pacts/orderservice-paymentservice.json',
        provider_states_setup_url=f'http://localhost:{PROVIDER_PORT}/_pact/provider_states',
        publish_version=os.environ.get('GIT_COMMIT', '1.0.0'),
        publish_verification_results=os.environ.get('CI') == 'true',
    )

    assert exit_code == 0

Component Testing

Component tests sit between integration and E2E tests. The idea: spin up your service with real infrastructure (databases, message brokers) but mock other services. This catches the bugs that unit tests miss (ORM misconfigurations, SQL syntax errors, message serialization issues) without the fragility of full E2E tests. Testcontainers makes this practical by spinning up real Docker containers that are destroyed after each test suite. Why not just use more integration tests? Because integration tests typically scope to one module (a repository, a handler). Component tests scope to the whole service: HTTP in, HTTP out, with real infrastructure underneath. They are the test you trust when you ask “if I deploy this service to prod alone, does it work?” They catch wiring bugs (did I register that middleware?), configuration bugs (did I wire the correct environment variable?), and startup ordering bugs (did I await the database connection before accepting requests?). No other test type catches these. Production pitfall: The 60-second timeout on beforeAll is not optional. Container startup is unpredictable — if your CI runner is under load, Postgres might take 30 seconds to initialize. Flaky tests here almost always trace back to insufficient timeouts. A related trap: reusing containers across test runs for speed. It works until a previous run’s state leaks into the current one. Prefer fresh containers per suite and invest in parallelism to keep total time reasonable.

Node.js
Python

// tests/component/OrderService.component.test.js
const { GenericContainer, Wait } = require('testcontainers');
const request = require('supertest');
const { createApp } = require('../../app');

describe('Order Service Component Test', () => {
  let postgresContainer;
  let kafkaContainer;
  let app;

  beforeAll(async () => {
    // Start PostgreSQL container
    postgresContainer = await new GenericContainer('postgres:15-alpine')
      .withEnvironment({
        POSTGRES_DB: 'orders_test',
        POSTGRES_USER: 'test',
        POSTGRES_PASSWORD: 'test'
      })
      .withExposedPorts(5432)
      .withWaitStrategy(Wait.forLogMessage('database system is ready'))
      .start();

    // Start Kafka container (optional)
    kafkaContainer = await new GenericContainer('confluentinc/cp-kafka:7.5.0')
      .withExposedPorts(9092)
      .start();

    // Create app with real containers
    const dbUrl = `postgresql://test:test@${postgresContainer.getHost()}:${postgresContainer.getMappedPort(5432)}/orders_test`;
    
    app = createApp({
      databaseUrl: dbUrl,
      kafkaBrokers: `${kafkaContainer.getHost()}:${kafkaContainer.getMappedPort(9092)}`
    });

    // Run migrations
    await runMigrations(dbUrl);
  }, 60000);

  afterAll(async () => {
    await postgresContainer?.stop();
    await kafkaContainer?.stop();
  });

  describe('Order Lifecycle', () => {
    it('should complete full order flow', async () => {
      // 1. Create order
      const createResponse = await request(app)
        .post('/orders')
        .set('Authorization', 'Bearer test-token')
        .send({
          customerId: 'customer-123',
          items: [
            { productId: 'product-1', quantity: 2 }
          ]
        })
        .expect(201);

      const orderId = createResponse.body.id;
      expect(createResponse.body.status).toBe('PENDING');

      // 2. Confirm order
      await request(app)
        .post(`/orders/${orderId}/confirm`)
        .set('Authorization', 'Bearer test-token')
        .expect(200);

      // 3. Check order status
      const statusResponse = await request(app)
        .get(`/orders/${orderId}`)
        .set('Authorization', 'Bearer test-token')
        .expect(200);

      expect(statusResponse.body.status).toBe('CONFIRMED');
    });
  });
});

# tests/component/test_order_service_component.py
# testcontainers-python provides the same Testcontainers experience as Node.
# The session-scoped fixture keeps the containers alive for the whole suite to
# amortize the startup cost across many tests.
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from testcontainers.postgres import PostgresContainer
from testcontainers.kafka import KafkaContainer

from app import create_app
from db.migrations import run_migrations


@pytest.fixture(scope='session')
def infra():
    with PostgresContainer('postgres:15-alpine') as pg, KafkaContainer() as kafka:
        yield {
            'db_url': pg.get_connection_url().replace('postgresql://', 'postgresql+asyncpg://'),
            'kafka_brokers': kafka.get_bootstrap_server(),
        }


@pytest_asyncio.fixture(scope='session')
async def app(infra):
    await run_migrations(infra['db_url'])
    return create_app(
        database_url=infra['db_url'],
        kafka_brokers=infra['kafka_brokers'],
    )


@pytest_asyncio.fixture
async def client(app):
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url='http://test') as c:
        yield c


class TestOrderLifecycle:
    @pytest.mark.asyncio
    async def test_complete_order_flow(self, client):
        # 1. Create order
        create_response = await client.post(
            '/orders',
            headers={'Authorization': 'Bearer test-token'},
            json={
                'customerId': 'customer-123',
                'items': [{'productId': 'product-1', 'quantity': 2}],
            },
        )
        assert create_response.status_code == 201
        body = create_response.json()
        assert body['status'] == 'PENDING'
        order_id = body['id']

        # 2. Confirm order
        confirm_response = await client.post(
            f'/orders/{order_id}/confirm',
            headers={'Authorization': 'Bearer test-token'},
        )
        assert confirm_response.status_code == 200

        # 3. Check status
        status_response = await client.get(
            f'/orders/{order_id}',
            headers={'Authorization': 'Bearer test-token'},
        )
        assert status_response.status_code == 200
        assert status_response.json()['status'] == 'CONFIRMED'

End-to-End Testing

E2E tests verify complete user journeys across multiple services. They are the most realistic but also the most expensive and fragile. The rule of thumb: have fewer than 10 E2E tests covering only your critical revenue paths. If your checkout flow breaks, you lose money. If a profile avatar upload is slightly slow, nobody notices. Why limit E2E tests? Because every additional service in the test adds a multiplier to failure modes. A test that crosses 5 services has 5 networks to fail, 5 databases to desynchronize, 5 version mismatches to worry about, and 5 containers to start. The math is brutal: even if each service has 99% reliability during tests, the combined suite has 0.99^5 = 95% reliability — meaning 1 in 20 runs fails for environmental reasons alone. This is how “just rerun CI” becomes a team’s default debugging technique, which in turn makes real regressions invisible. Keep E2E tests scarce and scope them to journeys where failure costs real money. Production pitfall: The setTimeout in the test below (waiting for async processing) is a code smell but sometimes unavoidable. In production E2E suites, replace it with a polling mechanism that checks for the expected state with a timeout, rather than a fixed delay. Fixed delays are the single biggest source of E2E flakiness — they either wait too little (test fails intermittently) or too much (test suite takes 30 minutes when it should take 3).

Caveats & Common Pitfalls: E2E Tests So Slow Nobody Runs Them

Full E2E suite takes 45 minutes. Engineers stop running it locally, run it only in a scheduled nightly job, and merge PRs without E2E coverage. Bugs land on main and are caught 12 hours later.
E2E suite covers happy paths for every feature. Fifty tests for CRUD flows. The ratio of value to runtime is terrible — most of these could be integration tests.
E2E against shared staging environment. Developer A’s manual test data mutates what developer B’s E2E assumes. Flakes fly. People stop trusting failures. Real regressions merge.
Retries as an anti-flakiness solution. “Retry failed E2E tests 3 times” hides genuine bugs. A test that fails once in five runs and passes on retry is telling you something — probably about production — and you are silencing it.

Solutions & Patterns

Cap E2E at 5-10 tests, covering only revenue-critical journeys (checkout, signup, payment). Every other behavior belongs in integration or contract tests.
Run E2E pre-deploy, not pre-merge. E2E gates the promotion to production, not the merge to main. PR CI runs unit + integration + contract; a separate pipeline runs E2E against a freshly-built release candidate.
Ephemeral E2E environments per test run. Spin up the full stack via docker-compose or a dedicated k8s namespace, run the suite, tear it down. Zero shared state, zero cross-test contamination.
Polling with timeouts, never fixed sleeps. await poll_until(condition, timeout=10) is reliable; await sleep(5) is either slow or flaky. Almost always both.
When a test flakes, quarantine or fix it same day. Flaky tests are technical debt that compounds. Each “flaky” test teaches the team to ignore failures.

Node.js
Python

// tests/e2e/checkout.e2e.test.js
const axios = require('axios');

describe('Checkout E2E Flow', () => {
  const baseUrl = process.env.E2E_BASE_URL || 'http://localhost:8080';
  let authToken;
  let userId;

  beforeAll(async () => {
    // Login and get auth token
    const loginResponse = await axios.post(`${baseUrl}/auth/login`, {
      email: 'e2e-test@example.com',
      password: 'test-password'
    });
    
    authToken = loginResponse.data.accessToken;
    userId = loginResponse.data.userId;
  });

  const api = () => axios.create({
    baseURL: baseUrl,
    headers: { Authorization: `Bearer ${authToken}` }
  });

  describe('Complete Checkout Journey', () => {
    let orderId;

    it('should add items to cart', async () => {
      const response = await api().post('/cart/items', {
        productId: 'product-123',
        quantity: 2
      });

      expect(response.status).toBe(200);
      expect(response.data.items).toHaveLength(1);
    });

    it('should create order from cart', async () => {
      const response = await api().post('/orders', {
        shippingAddress: {
          street: '123 Test St',
          city: 'Test City',
          country: 'US',
          postalCode: '12345'
        }
      });

      expect(response.status).toBe(201);
      orderId = response.data.id;
      expect(response.data.status).toBe('PENDING');
    });

    it('should process payment', async () => {
      const response = await api().post(`/orders/${orderId}/pay`, {
        paymentMethod: {
          type: 'card',
          token: 'tok_visa'
        }
      });

      expect(response.status).toBe(200);
      expect(response.data.paymentStatus).toBe('COMPLETED');
    });

    it('should show order as confirmed', async () => {
      // Wait for async processing
      await new Promise(resolve => setTimeout(resolve, 2000));

      const response = await api().get(`/orders/${orderId}`);

      expect(response.status).toBe(200);
      expect(response.data.status).toBe('CONFIRMED');
      expect(response.data.paymentId).toBeDefined();
    });

    it('should update inventory', async () => {
      const response = await api().get('/inventory/product-123');

      expect(response.status).toBe(200);
      // Verify inventory was reduced
    });
  });
});

# tests/e2e/test_checkout_flow.py
# E2E tests use httpx against the actual running system (docker-compose, staging, etc).
# The polling helper replaces the brittle "sleep 2 seconds" anti-pattern.
import asyncio
import os
import pytest
import pytest_asyncio
import httpx


BASE_URL = os.environ.get('E2E_BASE_URL', 'http://localhost:8080')


async def poll_until(fn, *, timeout_s=10, interval_s=0.2):
    """Poll fn() until it returns truthy or timeout expires."""
    deadline = asyncio.get_event_loop().time() + timeout_s
    while asyncio.get_event_loop().time() < deadline:
        result = await fn()
        if result:
            return result
        await asyncio.sleep(interval_s)
    raise TimeoutError(f'Condition not met within {timeout_s}s')


@pytest_asyncio.fixture(scope='module')
async def auth_context():
    async with httpx.AsyncClient(base_url=BASE_URL) as c:
        response = await c.post('/auth/login', json={
            'email': 'e2e-test@example.com',
            'password': 'test-password',
        })
        return response.json()


@pytest_asyncio.fixture
async def api(auth_context):
    async with httpx.AsyncClient(
        base_url=BASE_URL,
        headers={'Authorization': f"Bearer {auth_context['accessToken']}"},
    ) as c:
        yield c


class TestCompleteCheckoutJourney:
    # Shared state across ordered tests. In pytest we use a class attribute + pytest-ordering.
    order_id = None

    @pytest.mark.asyncio
    @pytest.mark.order(1)
    async def test_add_items_to_cart(self, api):
        response = await api.post('/cart/items', json={
            'productId': 'product-123',
            'quantity': 2,
        })
        assert response.status_code == 200
        assert len(response.json()['items']) == 1

    @pytest.mark.asyncio
    @pytest.mark.order(2)
    async def test_create_order_from_cart(self, api):
        response = await api.post('/orders', json={
            'shippingAddress': {
                'street': '123 Test St',
                'city': 'Test City',
                'country': 'US',
                'postalCode': '12345',
            },
        })
        assert response.status_code == 201
        body = response.json()
        assert body['status'] == 'PENDING'
        TestCompleteCheckoutJourney.order_id = body['id']

    @pytest.mark.asyncio
    @pytest.mark.order(3)
    async def test_process_payment(self, api):
        response = await api.post(
            f'/orders/{self.order_id}/pay',
            json={'paymentMethod': {'type': 'card', 'token': 'tok_visa'}},
        )
        assert response.status_code == 200
        assert response.json()['paymentStatus'] == 'COMPLETED'

    @pytest.mark.asyncio
    @pytest.mark.order(4)
    async def test_order_confirmed_after_async_processing(self, api):
        async def is_confirmed():
            r = await api.get(f'/orders/{self.order_id}')
            return r.json().get('status') == 'CONFIRMED'

        await poll_until(is_confirmed, timeout_s=10)

        response = await api.get(f'/orders/{self.order_id}')
        body = response.json()
        assert body['status'] == 'CONFIRMED'
        assert body.get('paymentId') is not None

    @pytest.mark.asyncio
    @pytest.mark.order(5)
    async def test_inventory_updated(self, api):
        response = await api.get('/inventory/product-123')
        assert response.status_code == 200
        # Verify inventory was reduced

Test Environment Setup

A dedicated test environment via Docker Compose gives every developer the same reproducible test infrastructure. Notice the tmpfs mount for Postgres below — this stores the database entirely in memory, making integration tests 3-5x faster since there are no disk writes. This single trick can cut your CI pipeline time significantly. Why dedicate infrastructure for tests? Because sharing dev infrastructure creates order-dependent flakiness. Developer A’s lunch-break test run leaves stale data; developer B’s CI run fails mysteriously. A test-only compose file ensures every run starts from a clean, known state. The memory-backed database trick works because tests rarely need durability — a disk crash during CI is no worse than any other CI failure. Trading durability for speed is exactly the kind of test-environment-specific optimization that is wrong in production but right here.

# docker-compose.test.yml
version: '3.8'

services:
  test-db:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: test
      POSTGRES_USER: test
      POSTGRES_PASSWORD: test
    ports:
      - "5433:5432"
    tmpfs:
      - /var/lib/postgresql/data  # Speed up tests

  test-redis:
    image: redis:7-alpine
    ports:
      - "6380:6379"

  test-kafka:
    image: confluentinc/cp-kafka:7.5.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9093
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@test-kafka:9093
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: 'test-cluster-id'
    ports:
      - "9093:9092"

Test Runner Configuration

Splitting tests into projects (unit/integration/e2e) lets developers run fast tests often and slow tests rarely. The coverageThreshold gate prevents silent coverage regressions — if a PR drops coverage below 80%, CI fails. This is more useful than it sounds: in practice, it forces engineers to at least think about testing before merging, which is a surprisingly effective cultural lever.

Node.js
Python

// jest.config.js
module.exports = {
  projects: [
    {
      displayName: 'unit',
      testMatch: ['<rootDir>/tests/unit/**/*.test.js'],
      testEnvironment: 'node',
      setupFilesAfterEach: ['<rootDir>/tests/setup/unit.js']
    },
    {
      displayName: 'integration',
      testMatch: ['<rootDir>/tests/integration/**/*.test.js'],
      testEnvironment: 'node',
      setupFilesAfterEach: ['<rootDir>/tests/setup/integration.js'],
      globalSetup: '<rootDir>/tests/setup/integration-global-setup.js',
      globalTeardown: '<rootDir>/tests/setup/integration-global-teardown.js'
    },
    {
      displayName: 'e2e',
      testMatch: ['<rootDir>/tests/e2e/**/*.test.js'],
      testEnvironment: 'node',
      testTimeout: 30000
    }
  ],
  collectCoverageFrom: [
    'src/**/*.js',
    '!src/**/*.test.js'
  ],
  coverageThreshold: {
    global: {
      branches: 80,
      functions: 80,
      lines: 80,
      statements: 80
    }
  }
};

# pyproject.toml (or pytest.ini)
# pytest uses markers to separate unit/integration/e2e. Combined with pytest-cov
# for coverage thresholds and pytest-xdist for parallel execution.

# [tool.pytest.ini_options]
# asyncio_mode = "auto"
# markers = [
#     "unit: fast tests with mocked dependencies",
#     "integration: tests against real databases/services",
#     "e2e: full system tests, run pre-deploy only",
#     "contract: Pact consumer/provider verification",
# ]
# addopts = [
#     "--strict-markers",
#     "--cov=src",
#     "--cov-report=term-missing",
#     "--cov-fail-under=80",
# ]

# Run commands:
#   pytest -m unit                    # fast, every commit
#   pytest -m integration             # slower, every commit
#   pytest -m e2e --timeout=30        # pre-deploy
#   pytest -m "not e2e" -n auto       # everything except e2e, parallel

# Example conftest.py for shared fixtures:
import pytest


def pytest_collection_modifyitems(config, items):
    """Auto-apply markers based on test file location."""
    for item in items:
        path = str(item.fspath)
        if '/unit/' in path:
            item.add_marker(pytest.mark.unit)
        elif '/integration/' in path:
            item.add_marker(pytest.mark.integration)
        elif '/e2e/' in path:
            item.add_marker(pytest.mark.e2e)
        elif '/contract/' in path:
            item.add_marker(pytest.mark.contract)

Interview Questions

Q1: How do you test microservices interactions?

Answer:Approaches:

Contract Testing (Pact)
- Consumer defines expected interactions
- Provider verifies it can meet expectations
- Catches breaking changes early
Integration Testing
- Test with real dependencies (test containers)
- Slower but more realistic
- Good for database interactions
Service Virtualization
- Mock external services
- Consistent test data
- Faster than real services

Best Practice:

Contract tests for API compatibility
Integration tests for data integrity
Mocks for unit tests

Q2: What is contract testing and why is it important?

Answer:Contract Testing:

Consumer defines expectations (contract)
Provider verifies it meets contract
Changes detected before deployment

Why important:

Prevents integration failures
Enables independent deployment
Faster feedback than E2E tests
Documents service interactions

Consumer-Driven Contracts:

Consumer writes contract tests
Generates contract file
Provider runs verification
Breaks build if contract broken

Tools: Pact, Spring Cloud Contract

Q3: How do you handle test data in microservices?

Answer:Strategies:

Test Containers
- Spin up real databases
- Isolated per test suite
- Clean state each run
Test Data Builders / Factories

Node.js
Python

OrderBuilder.create()
  .withCustomer('123')
  .withItems([...])
  .build();

# factory_boy gives you composable, reusable factories
import factory
from factory import Faker

class OrderFactory(factory.Factory):
    class Meta:
        model = dict
    id = factory.LazyFunction(lambda: str(uuid4()))
    customer_id = Faker('uuid4')
    status = 'PENDING'
    total = Faker('pyfloat', positive=True, max_value=1000)

order = OrderFactory(customer_id='123', items=[...])

Database Seeding
- Known state before tests
- Fixtures or factories
Data Cleanup
- Truncate after tests
- Transaction rollback
- Isolated test databases

Best Practices:

Each test manages own data
No dependencies between tests
Use factories for complex objects

Testing Anti-Patterns and Edge Cases

The “Ice Cream Cone” Anti-Pattern

Many teams invert the testing pyramid — heavy on E2E and manual tests, light on unit and contract tests. This creates slow CI pipelines (45+ minute builds), flaky test suites that everyone ignores, and a false sense of security because the tests that pass are testing the wrong things. How to detect it: If your team says “just run it in staging and check” more than twice a week, you have an ice cream cone.

Edge Case: Testing Async Event Flows

Testing event-driven interactions (Kafka, RabbitMQ) is where microservices testing gets genuinely hard. The challenge: you publish an event, another service consumes it, and you need to verify the side effect. You cannot just assert immediately because the consumer processes asynchronously. Approaches ranked by reliability:

Approach	Reliability	Speed	Complexity
Poll for expected state with timeout	High	Slow (seconds)	Low
Use an in-memory event bus in tests	High	Fast (milliseconds)	Medium
Testcontainers with real Kafka	Highest	Slow (startup overhead)	High
Mock the message broker entirely	Low (misses serialization bugs)	Fastest	Low

A senior engineer would say: “For unit tests, mock the broker. For integration tests, use Testcontainers with real Kafka. For contract tests, verify the event schema independently of the transport.”

Edge Case: Testing Database Migrations

Migrations that work on an empty database often fail on a production database with millions of rows. Your test suite should include:

Forward migration test — apply migration to a seeded database with realistic data volumes
Rollback test — verify the down migration restores the previous schema
Data integrity test — verify existing data survives the migration without corruption
Performance test — ensure the migration completes within your maintenance window (an ALTER TABLE on a 500M row table can take hours)

Summary

Key Takeaways

Follow the testing pyramid
Unit tests for business logic
Contract tests for API compatibility
Integration tests with test containers
E2E for critical paths only

Next Steps

In the next chapter, we’ll cover Interview Preparation - common questions and system design exercises.

Interview Deep-Dive

'Your team has 15 microservices. Integration tests take 45 minutes to run and break frequently due to test environment instability. How do you fix this?'

Strong Answer:45-minute integration tests that break frequently are a symptom of two problems: the test suite is doing too much, and the test environment is not isolated.The first fix is pushing tests down the pyramid. Most of those 45 minutes are probably spent on end-to-end tests that spin up multiple services. I would audit the test suite and reclassify: if a test verifies business logic, it should be a unit test with mocked dependencies (runs in milliseconds). If it verifies database queries, it should be an integration test against a local database using Testcontainers (runs in seconds). If it verifies the API contract between two services, it should be a contract test with Pact (runs in seconds). Only critical user journeys (place order, process payment, receive confirmation) should be E2E tests.The second fix is test isolation. Each test should spin up its own dependencies and clean up after itself. Testcontainers creates disposable Docker containers for PostgreSQL, Redis, and Kafka per test suite. No shared staging database that accumulates test data and creates flaky failures when two CI runs interfere with each other.The third fix is parallelization. If I have 15 services, their test suites should run in parallel in CI, not sequentially. Each service’s pipeline runs independently — Service A’s tests should never wait for Service B’s tests. The only shared gate is contract tests, which verify compatibility between services.After these changes, the typical result is: unit tests (30 seconds), integration tests with Testcontainers (2-3 minutes), contract tests (1-2 minutes), and E2E tests for 3-5 critical paths (5-10 minutes). Total: under 15 minutes, running mostly in parallel.Follow-up: “How do contract tests work in practice, and who is responsible for writing them — the consumer or the provider?”The consumer writes the contract. This is called Consumer-Driven Contract Testing (Pact). The Order Service (consumer) writes a test that says “when I call GET /users/123 on User Service, I expect a response with {id, name, email}.” This contract is published to a Pact Broker. The User Service (provider) CI pipeline downloads all consumer contracts and verifies that its API satisfies them. If User Service renames “email” to “emailAddress,” the contract test fails before the change reaches production. The beauty is that neither team runs the other’s service — the contract is the intermediary. This catches breaking changes without requiring a full integration environment.

'How do you test a saga that spans 4 services (Order, Payment, Inventory, Notification) without running all 4 services?'

Strong Answer:You test the saga at three levels, and only the last level requires multiple services.Level one: unit test the saga orchestrator logic. The orchestrator is a state machine — given state X and event Y, it transitions to state Z and sends command W. I test every state transition and every compensation path with unit tests. The downstream services are mocked interfaces. These tests verify the workflow logic: “if payment fails after inventory is reserved, the orchestrator sends a release-inventory command and then a cancel-order command.” This catches logic errors in the saga definition itself.Level two: integration test each service’s saga participant independently. Test that the Payment Service correctly handles a “charge-payment” command and publishes a “payment-charged” event. Test that it handles a “refund-payment” compensation command. These tests use Testcontainers for the service’s own database and a local Kafka/RabbitMQ instance. Each service is tested in isolation with real infrastructure but without other services.Level three: a focused E2E test for the happy path and the most critical failure paths. I run all 4 services (using docker-compose in CI) and test: (1) successful order placement end-to-end, (2) payment failure triggers compensation across all services, and (3) partial failure mid-saga (kill inventory service during processing) to verify the orchestrator retries and eventually completes. These tests are slow (30-60 seconds each) so I keep them to 3-5 scenarios.The key insight: level one catches 80% of saga bugs (wrong state transitions, missing compensation steps) and runs in milliseconds. Level two catches integration bugs (serialization mismatches, database constraint violations) and runs in seconds. Level three is a confidence check, not a primary testing layer.Follow-up: “How do you test idempotency — the guarantee that processing the same message twice produces the same result?”I write explicit idempotency tests at level two. For each saga participant, I send the same command twice (with the same idempotency key) and verify that the second invocation is a no-op: no duplicate database records, no duplicate side effects (no double charge), and the response is the same as the first invocation. I also test the edge case where the service crashes after processing but before acknowledging the message — on restart, it receives the message again and must handle it idempotently. Testcontainers makes this testable by letting me kill and restart the service mid-test.

'What is the testing diamond and how does it differ from the testing pyramid for microservices?'

Strong Answer:The traditional testing pyramid (many unit tests, fewer integration tests, even fewer E2E tests) was designed for monoliths where the boundaries are all in-process function calls. In microservices, the most valuable bugs are at the integration boundaries — API contract mismatches, serialization errors, incorrect event schemas, timeout behavior under load. Unit tests do not catch these because they mock the boundaries away.The testing diamond (or testing honeycomb, as Spotify calls it) places the most emphasis on integration tests and contract tests — the middle layer. It has fewer unit tests (only for complex business logic), more integration tests (service + real database, service + real message broker), more contract tests (verify API compatibility between services), and fewer E2E tests (only for critical user journeys).The rationale: in a microservice with a typical CRUD API, a unit test that mocks the database verifies that your mock returns what you told it to return — it does not verify that the SQL query actually works, that the database constraints are correct, or that the ORM mapping is accurate. An integration test with Testcontainers running a real PostgreSQL instance catches all of these.However, I do not completely abandon unit tests. Complex business logic — pricing calculations, saga state machine transitions, validation rules — should absolutely be unit tested because the logic is intricate and the tests are fast. The diamond shape means: thin unit layer (only for pure logic), thick integration layer (service + real dependencies), thick contract layer (API compatibility), thin E2E layer (critical paths only).The pragmatic approach I use: if a test can catch the bug with mocked dependencies, make it a unit test. If it can only catch the bug with real dependencies, make it an integration test. If it can only catch the bug with multiple services, make it a contract test or E2E test. This naturally produces a diamond shape for most microservices.Follow-up: “How do you keep integration tests fast enough that developers actually run them before pushing?”Testcontainers with reusable containers. Instead of starting a fresh PostgreSQL container for every test file, I start one container per test suite run and reset the database between tests (truncate tables or use transactional test isolation). A Testcontainers PostgreSQL instance starts in 2-3 seconds and stays alive for the entire test suite. With this approach, 100 integration tests run in under 30 seconds. The key is test isolation through transactions (start transaction before test, rollback after) rather than through container recreation.

Scenario-Driven Interview Questions

You are building a new microservice that calls 4 other services (payments, inventory, user, notifications). Design the testing strategy from unit to production.

Strong Answer Framework

Start with the risk map. Which integrations are financial (payments), which are eventual-consistency safe (notifications), which affect user data? The testing investment follows risk: payments gets contract + integration + an explicit E2E; notifications gets contract + a single smoke test.
Unit tests for internal logic. Inject each of the 4 dependencies as interfaces. Test business rules in isolation with mocks that return controlled responses: happy path, timeout, 4xx, 5xx, malformed body. Target 200-500 unit tests running in under 30 seconds total.
Integration tests for your own persistence. Spin up real Postgres via Testcontainers. Exercise every repository method, every migration, every complex query. No mocks for your own DB — ever. Target 30-100 tests in under 2 minutes.
Contract tests for every downstream. Write consumer-driven contracts for each of the 4 services using Pact. The contracts live in your repo; publish them to a broker on every merge. Each downstream’s CI verifies the contracts before merging their own changes. This is the single most valuable test tier for multi-service architectures.
Component tests for the service boundary. Start your service with Testcontainers Postgres, real Kafka, and WireMock stubs for the 4 downstreams. Make HTTP requests against the service. This catches wiring bugs (middleware order, config injection, startup sequence) that unit tests cannot.
A handful of E2E tests for critical paths. One test per revenue-critical flow (e.g., full order placement end-to-end). Run these in a dedicated staging pipeline, not per-commit. Keep total count under 10.
Production verification. Synthetic transactions every 60 seconds from a third-party prober (or Kubernetes CronJob). Alerts fire if the user-facing journey breaks. This is your continuous E2E.

Real-World ExampleMonzo Bank’s engineering team wrote publicly in 2020 about how they test ~1,500 microservices. The core insight: they rely overwhelmingly on contract tests (they built their own fork of Pact called Coach) and integration tests with ephemeral environments. E2E tests are kept under 20 across the entire bank, each covering a critical customer journey (money transfer, card payment, account opening). Unit and contract tests run per commit; E2E runs on release candidates only. Their deploy frequency reached ~100/day by 2020 with this model.Senior Follow-up Questions

Q: What about chaos tests and performance tests — where do they fit?A: Orthogonal to the functional pyramid. Chaos tests run in staging on a schedule (weekly), injecting failures (kill a downstream, slow a DB). Performance tests run pre-release against a production-shaped environment. Both produce alerts rather than gating deploys — you catch regressions over time, not per-commit.

Q: How do you test failure modes across 4 downstreams (one down, two slow, etc.)?A: Use WireMock or a similar programmable stub. Write component tests that configure each downstream to return errors, timeouts, or slow responses, then assert that your service degrades gracefully (circuit breaker opens, fallback activates, correct error returned to caller). These tests live at the component tier so you exercise real middleware behavior.

Q: How do you measure success of this testing strategy?A: Four signals: (1) escape rate — bugs that reach prod per 1000 commits, trending down; (2) CI time — under 15 minutes for the full pre-merge pipeline; (3) flakiness — test failure rate when nothing changed, under 1%; (4) confidence — engineers merge on green without “let me test this in staging first” qualifiers.

Common Wrong Answers

“Lots of E2E tests covering every feature.” Inverts the pyramid. Slow, flaky, and catches bugs too late. Discussed at length as the ice cream cone anti-pattern.
“Mock all 4 services in unit tests, skip contracts.” Mocks drift from reality. The first time you find out the real service’s schema changed is in staging, if you are lucky, or production if not.
“Run everything against a shared staging environment.” Cross-test contamination guarantees flakiness. Shared state is the enemy of reliable tests.

Further Reading

Martin Fowler’s “Microservice Testing” series: martinfowler.com/articles/microservice-testing — the canonical reference.
“Testing Microservices” by Sam Newman (chapters from “Building Microservices”, 2nd Edition).
Monzo Engineering Blog: “How we test at Monzo” — real-world example at scale.

Your team's integration test suite takes 45 minutes and breaks 3 times a week due to flakiness. What is your diagnostic process and what fixes do you apply?

Strong Answer Framework

Instrument before you guess. Turn on per-test timing and failure counts. The 45-minute runtime is usually 20% of tests consuming 80% of time; the flakiness is usually 5% of tests producing 90% of failures. Data first, fixes second.
Sort tests by flakiness rate. For each test, compute failures / total runs over the last 30 days. Any test above 2% is either genuinely racy, depends on shared state, or asserts something non-deterministic.
Categorize the flaky tests. Three buckets: (a) timing — setTimeout-based waits, insufficient polling; (b) shared state — order-dependent, shared DB/environment; (c) external dependency — hitting a real API, random data from the network.
Fix, do not retry. Never wrap flaky tests in retry logic — that hides real bugs. Instead: replace fixed sleeps with polling, replace shared state with ephemeral state, replace external APIs with Testcontainers or stubs.
Attack the slowest tests. For tests above 5 seconds each, ask: can this be a unit test? Can I seed fixtures once instead of per-test? Can I run this test in parallel? Can I use transactional rollback instead of TRUNCATE?
Parallelize the suite. With proper isolation (each test owns its data), integration tests parallelize trivially. pytest -n auto and Jest’s --maxWorkers cut wall-clock time by the worker count.
Quarantine, fix, reinstate. A test that is flaky and cannot be immediately fixed goes to a quarantine suite — still runs, still reports, but does not block merges. File a ticket, fix within a sprint, or delete.

Real-World ExampleSpotify’s famous “Flaky Test Management” initiative (2018-2019) discovered that 1% of their tests generated 99% of CI failures. They built tooling that auto-quarantined tests with failure rate above 2%, notified owners, and blocked deploys after 30 days of unfixed quarantine. Within a quarter, CI reliability went from 80% to 97%. The counter-intuitive lesson: aggressively removing tests made the suite more valuable because the remaining tests were trusted.Senior Follow-up Questions

Q: How do you prevent the flakiness from creeping back in?A: Three mechanisms. (1) CI reports flakiness rates on every test file — visible in PR reviews. (2) Add a pre-merge check that tests cannot call real external APIs or Date.now() without a mock or frozen clock. (3) Flakiness budgets — each service owns a metric and pages the team if it exceeds threshold.

Q: Is it ever OK to retry a test?A: Only when the flakiness is in the infrastructure, not the test. Example: a Testcontainer failed to start because Docker was slow — retry is legitimate. Example: the test inconsistently asserts on timing — retry is covering a bug. Rule of thumb: if the retry policy lives in your CI config, it is infra; if it lives in the test file, it is a bug.

Q: How do you measure success of the effort?A: Three metrics trend weekly. (1) CI time P95 — should drop from 45 min to under 15. (2) Test flakiness rate (failures when nothing changed) — should drop below 1%. (3) Developer trust, measured indirectly by “retry count after failure” — if the average engineer retries once and then investigates, CI is trusted; if they retry three times before reading the failure, it is not.

Common Wrong Answers

“Add retries to the flaky tests.” Hides bugs, increases CI time, teaches engineers to ignore failures.
“Delete the slow tests.” Loses coverage without fixing the cause. Only valid after confirming the deleted test is redundant.
“Move everything to E2E so we do not need integration tests.” Inverts the pyramid entirely. Ten-minute integration tests become forty-minute E2E tests with worse signal.

Further Reading

Google Testing Blog: “Flaky Tests at Google and How We Mitigate Them” (2016) — the foundational paper.
Spotify Engineering: “Flaky Test Management at Scale” — practical quarantine workflow.
“Working Effectively with Legacy Code” by Michael Feathers — the chapter on fast feedback applies directly to CI speed.

A production incident happened because a breaking API change shipped from the User service and broke the Order service. You had contract tests. Why did they not catch it, and how do you fix the process?

Strong Answer Framework

Reproduce the miss. Pull the exact pact file at the time of the breaking User-service merge. Did it contain the assertion that would have caught the break? If not, the consumer’s contract was incomplete; if yes, the provider’s verification pipeline was not running on this PR.
Check the verification pipeline. Was pact-verifier a mandatory gate on User-service PRs? Most contract-test failures trace to “we had contracts, but verification was a nightly job, not a PR gate.” Fix: wire verifier into the PR required checks.
Check contract completeness. Contract tests verify the shapes and status codes that the consumer actually exercises. If Order never called the specific endpoint that User broke, the contract never covered it. Fix: audit consumer code paths against contract coverage, file tickets for gaps.
Check can-i-deploy in the deploy pipeline. Even with green verification, deploying User-service v2.5 without checking compatibility against the consumer version running in prod is risky. can-i-deploy answers: “are all consumers currently in prod compatible with what I am about to deploy?” If not wired in, wire it now.
Audit tag hygiene. Pact Broker tags like prod, staging, master must move atomically on successful deploys. Drifted tags (stale prod) make can-i-deploy report false positives. Verify tag update is part of the deploy pipeline, not a manual step.
Backfill with a regression test. Write a contract test that would have caught this specific break. Future consumer contracts inherit this shape, so the next team cannot accidentally ship the same mistake.
Postmortem the gap, not the break. The real failure is the process: contracts existed but did not gate. Document the process change (PR gate, can-i-deploy, tag hygiene) and treat this as a lesson for all service teams.

Real-World ExampleSoundCloud wrote about a 2017 incident where the search service shipped a response-envelope change that broke the recommendations service for 4 hours. Both teams had Pact tests. The provider’s verification ran nightly, not on PRs; the breaking merge happened at 10am and was caught at 2am the next day. Their postmortem outcome: verification became a required PR check across all 500+ services, and can-i-deploy was added as a deploy gate. Similar incidents dropped to zero over the following year.Senior Follow-up Questions

Q: What if the breaking change was semantic, not structural (the shape stayed the same but the meaning changed)?A: Contract tests cannot catch semantics by design. Complement with integration tests on the consumer side for behaviorally-critical flows, and use stronger typing with runtime validation (Zod, Pydantic). For truly critical semantics, add explicit version fields and reject unknown versions.

Q: How do you handle the case where the provider has dozens of consumers and cannot coordinate with all of them?A: This is exactly what Pact Broker’s can-i-deploy solves at scale. The provider asks: “given all consumer contracts currently in prod, can I ship this new version?” If even one consumer is incompatible, deploy halts and the provider contacts that specific team. Alternative: backward compatibility as a policy — providers support both old and new shapes for a deprecation window.

Q: How do you measure whether the fixed process is working?A: Track breaking-change-related incidents per quarter. If the count is nonzero, something in the chain is still broken. Also monitor: % of services with required pact gates, % of deploys that triggered can-i-deploy failures (high is good — it means the gate caught real issues), time from provider PR to consumer awareness of break.

Common Wrong Answers

“We need more E2E tests.” E2E tests would have caught this break once, in a shared environment, probably after merge. Contract tests catch it at PR time. E2E is a fallback, not a primary defense against API compatibility.
“The provider should announce breaking changes.” Relies on humans remembering and writing correct announcements. The whole point of contract testing is that the test fails automatically; process changes that rely on human discipline regress over time.

Further Reading

Pact documentation: “Versioning and Branches” and “Can I Deploy” — docs.pact.io.
“Consumer-Driven Contracts: A Service Evolution Pattern” by Ian Robinson — martinfowler.com/articles/consumerDrivenContracts.html.
SoundCloud Engineering: “Scaling Pact Across 500 Services” — real adoption story with pitfalls.

Overview

Testing & Code Quality

Crash Courses

AI Engineering

Math for ML - Understanding Linear Algebra

Probability & Statistics for ML

Math for ML - Understanding Calculus

ML Mastery

Deep Learning Mastery

NestJS Mastery

Microservices Mastery

Low Level Design

OOP Concepts

SOLID Principles

Design Patterns

LLD Case Studies

System Design (HLD)

Senior Level (L5+/Staff)

HLD Case Studies

Engineering Fundamentals

DevOps & Operations

Azure Cloud Engineering

AWS Cloud

AWS Monitoring & Observability

AWS Security Services

AWS Serverless

AWS Operations

AWS Advanced

AWS Case Studies

GCP Cloud Engineering

DevOps Tools

Database Engineering

HIPAA Compliance Mastery

Operating Systems

Linux Internals

Distributed Systems

Networking Mastery

Build Your Own X

Go Lang Mastery

C Programming

Classic Research Papers

Distributed System Tools

Documentation Index

​Testing Strategies

​Testing Pyramid for Microservices

​Testing Strategy Decision Framework

​Unit Testing

​Testing Business Logic

​Unit Tests with Jest and pytest

​Integration Testing

​Database Integration Tests

​API Integration Tests

​Contract Testing with Pact

​Consumer Contract Test

​Provider Contract Test

​Component Testing

​End-to-End Testing

​Test Environment Setup

​Test Runner Configuration

​Interview Questions

​Testing Anti-Patterns and Edge Cases

​The “Ice Cream Cone” Anti-Pattern

​Edge Case: Testing Async Event Flows

​Edge Case: Testing Database Migrations

​Summary

Key Takeaways

Next Steps

​Interview Deep-Dive

​Scenario-Driven Interview Questions

Testing Strategies

Testing Pyramid for Microservices

Testing Strategy Decision Framework

Unit Testing

Testing Business Logic

Unit Tests with Jest and pytest

Integration Testing

Database Integration Tests

API Integration Tests

Contract Testing with Pact

Consumer Contract Test

Provider Contract Test

Component Testing

End-to-End Testing

Test Environment Setup

Test Runner Configuration

Interview Questions

Testing Anti-Patterns and Edge Cases

The “Ice Cream Cone” Anti-Pattern

Edge Case: Testing Async Event Flows

Edge Case: Testing Database Migrations

Summary

Interview Deep-Dive

Scenario-Driven Interview Questions