Learning Hub

How to Use Differential Testing to Safely Replace Legacy Code in Python

Learn how to catch behavior-changing bugs during a legacy code migration by running the old and new implementations side by side and comparing what they actually do.

September 17, 2026 21 Min Read
20

If you have ever rewritten a piece of business logic and had it “pass all the tests” while still quietly changing what customers get charged, this tutorial is for you. You are going to learn differential testing: a technique for finding out whether a new implementation actually behaves the same as the old one, by running both of them with the same input and comparing what they do.

Table Of Content

By the end of this tutorial you will have personally built and run a differential test suite in Python that catches two real, reproducible bugs in a rewritten service: a one-cent rounding error that only shows up on some inputs, and a silent fallback that turns an invalid request into a wrong-but-plausible-looking answer instead of an error. You will also build a small “shadow traffic” simulation that runs the new code alongside real requests without letting it affect the response a customer sees, and you will personally reproduce a second, distinct bug: a naive shadow implementation that pollutes production metrics just by existing.

What Is Differential Testing, and Why Isn’t a Normal Test Suite Enough?

Most tests work by comparing a system’s output to a value a human wrote down in advance: assert calculate_total(order) == 42.50. That works well when you already know the right answer. During a migration, you usually don’t, not for every input. You have an old system that has been running in production for years, quietly encoding business rules nobody documented, and a new system that its own authors believe is equivalent.

Differential testing sidesteps the “what’s the right answer” problem entirely. Instead of hand-writing expected values, you run the same input through both implementations and compare their outputs to each other:

same input
    |
    +---> legacy implementation -----> result A
    |
    +---> new implementation --------> result B

                  compare A and B

Every mismatch is a lead. Some are real bugs. Some are intentional improvements. Some are harmless differences in formatting. The job of a differential test suite is to run this comparison at a scale no human reviewer could manage by hand, and to sort the results into those categories so a person only has to look at the ones that matter.

This is a different technique from characterization testing, which locks in what one system already does before you touch it. Differential testing compares two systems against each other. In a real migration you typically use both: characterization tests protect you while you extract a testable boundary (see seams and adapters for that step), and differential tests verify the replacement you eventually swap in behaves the same way.

Prerequisites

  • Python 3.10 or later, comfortable with functions, dictionaries, and exceptions.
  • Basic familiarity with pytest (you’ll install it, no prior experience required).
  • Comfortable running commands in a terminal.
  • No prior testing-framework or migration experience assumed. Every concept is explained as it comes up.

Everything in this tutorial runs locally with nothing but the Python standard library and pytest, no database, no network calls, no cloud account.

Step 1: Set Up an Isolated Project

Create a fresh folder and a virtual environment, so this tutorial’s dependencies stay separate from anything else on your machine:

mkdir difftest-tutorial
cd difftest-tutorial
python -m venv venv
venv\Scripts\activate        (Windows)
source venv/bin/activate      (macOS/Linux)
pip install pytest

You should see pytest install successfully. Confirm it with:

pytest --version
pytest 9.1.1

Your version number may differ slightly. That’s fine as long as it’s pytest 7 or newer.

Step 2: Write the Legacy Implementation

Imagine an e-commerce platform’s shipping quote service. It has been running for years. Nobody loves it, but it works, and a lot of downstream code (billing, a dashboard, a nightly report) depends on its exact behavior. Save this as legacy_shipping.py:

"""The old, currently-running shipping quote calculator.

This module is deliberately written the way a lot of real legacy code looks:
a couple of module-level lookup tables, one function that does everything,
and a side effect (METRICS) that other parts of the system rely on.
"""

import time

ZONE_BASE_RATE = {
    "domestic": 2.50,
    "regional": 4.75,
    "international": 9.90,
}

ZONE_STANDARD_DAYS = {
    "domestic": 2,
    "regional": 4,
    "international": 9,
}

EXPRESS_MULTIPLIER = 1.8
FUEL_SURCHARGE_RATE = 0.12
MIN_CHARGE = 5.00

# A side effect: every call updates a shared counter, and other code
# (a dashboard, a billing job) reads METRICS later.
METRICS = {"domestic": 0, "regional": 0, "international": 0}


def calculate_shipping(weight_kg, zone, express=False, record_metrics=True):
    if weight_kg <= 0:
        raise ValueError(f"weight_kg must be positive, got {weight_kg}")
    if zone not in ZONE_BASE_RATE:
        raise ValueError(f"unknown zone: {zone!r}")

    base_rate = ZONE_BASE_RATE[zone]
    cost = weight_kg * base_rate
    if express:
        cost *= EXPRESS_MULTIPLIER

    fuel_surcharge = cost * FUEL_SURCHARGE_RATE
    cost += fuel_surcharge
    cost = round(cost, 2)
    cost = max(cost, MIN_CHARGE)

    estimated_days = ZONE_STANDARD_DAYS[zone]
    if express:
        estimated_days = max(1, estimated_days // 2)

    carrier = "GlobalShip" if zone == "international" else "RegionalFreight"

    if record_metrics:
        METRICS[zone] += 1

    quote_id = f"LEGACY-{int(time.time() * 1_000_000)}"

    return {
        "quote_id": quote_id,
        "cost": cost,
        "currency": "USD",
        "estimated_days": estimated_days,
        "carrier": carrier,
    }

Nothing here is exotic. It looks up a per-kilogram rate for a shipping zone, applies an express multiplier if requested, adds a 12% fuel surcharge, rounds to the cent, and enforces a $5.00 minimum charge. It raises ValueError for a negative weight or an unrecognized zone. It also updates a module-level METRICS counter every time it’s called, a side effect that a real dashboard or billing job would read later. Note the record_metrics parameter: you’ll use it in Step 9 to run this function for comparison purposes without polluting that counter.

Step 3: Write the New Implementation

Now imagine the team rewrote this service from scratch, with a cleaner, nested response shape that a newer part of the codebase expects. It has its own unit tests, and they pass. Save this as new_shipping.py:

"""The new shipping quote service, mid-migration.

The team rewrote this from scratch with a cleaner, nested response shape.
It passed its own unit tests. Whether it behaves the same as legacy_shipping
for every input is exactly what this tutorial's differential tests exist to
check.
"""

import time

ZONE_BASE_RATE = {
    "domestic": 2.50,
    "regional": 4.75,
    "international": 9.90,
}

ZONE_STANDARD_DAYS = {
    "domestic": 2,
    "regional": 4,
    "international": 9,
}

EXPRESS_MULTIPLIER = 1.8
FUEL_SURCHARGE_RATE = 0.12
MIN_CHARGE = 5.00

METRICS = {"domestic": 0, "regional": 0, "international": 0}


def calculate_shipping(weight_kg, zone, express=False, record_metrics=True):
    if weight_kg <= 0:
        raise ValueError(f"weight_kg must be positive, got {weight_kg}")

    if zone not in ZONE_BASE_RATE:
        # "Nobody ships anywhere else yet, domestic is a safe default
        # until we add real zone validation" - a real PR comment that
        # shipped. This silently masks bad input instead of rejecting it.
        zone = "domestic"

    base_rate = ZONE_BASE_RATE[zone]
    cost = weight_kg * base_rate
    if express:
        cost *= EXPRESS_MULTIPLIER

    # Rounded here, before the fuel surcharge is added, instead of once
    # at the very end like the legacy implementation does.
    cost = round(cost, 2)
    fuel_surcharge = round(cost * FUEL_SURCHARGE_RATE, 2)
    cost = cost + fuel_surcharge
    cost = max(cost, MIN_CHARGE)

    estimated_days = ZONE_STANDARD_DAYS[zone]
    if express:
        estimated_days = max(1, estimated_days // 2)

    carrier = "GlobalShip" if zone == "international" else "RegionalFreight"

    if record_metrics:
        METRICS[zone] += 1

    quote_id = f"NEW-{int(time.time() * 1_000_000)}"

    return {
        "quote_id": quote_id,
        "pricing": {
            "amount": cost,
            "currency": "USD",
        },
        "delivery": {
            "days": estimated_days,
            "carrier": carrier,
        },
    }

Read it once before continuing. It looks reasonable. If you only tested it by hand with a couple of round numbers, you would probably conclude it works. It does not: it has two real bugs, and you are about to find both of them the same way you would in a real migration, by comparing behavior, not by being told where to look.

Step 4: Try the Naive Comparison First, and See Why It Fails

The obvious first move is to call both functions with the same input and compare the results directly. Run this in a Python shell in the same folder:

import legacy_shipping, new_shipping

order = {"weight_kg": 3.2, "zone": "domestic", "express": False}
legacy = legacy_shipping.calculate_shipping(**order)
new = new_shipping.calculate_shipping(**order)
print("legacy:", legacy)
print("new:   ", new)
print()
print("naive equality check:", legacy == new)
legacy: {'quote_id': 'LEGACY-1789674175348170', 'cost': 8.96, 'currency': 'USD', 'estimated_days': 2, 'carrier': 'RegionalFreight'}
new:    {'quote_id': 'NEW-1789674175348177', 'pricing': {'amount': 8.96, 'currency': 'USD'}, 'delivery': {'days': 2, 'carrier': 'RegionalFreight'}}

naive equality check: False

The cost, delivery estimate, and carrier all agree. The equality check still says False, for two reasons that have nothing to do with correctness:

  1. The quote IDs are nondeterministic. Each one embeds the current time in microseconds, so no two calls will ever produce an identical ID, even calling the same function twice in a row.
  2. The response shapes are different on purpose. The new service nests cost under pricing.amount and estimated_days/carrier under delivery. That’s a legitimate API redesign, not a bug, but == on the raw dictionaries can’t tell the difference between “restructured” and “wrong.”

This is the first real lesson of differential testing: comparing raw output blindly tells you almost nothing. You need to compare business meaning, not JSON shape.

Step 5: Extract Business Meaning Before Comparing

The fix is to write a small extractor function per implementation that pulls out the fields that actually matter, in a common shape, ignoring both the nondeterministic ID and the structural differences:

def extract_legacy(result):
    return {
        "cost": result["cost"],
        "estimated_days": result["estimated_days"],
        "carrier": result["carrier"],
    }

def extract_new(result):
    return {
        "cost": result["pricing"]["amount"],
        "estimated_days": result["delivery"]["days"],
        "carrier": result["delivery"]["carrier"],
    }

Now extract_legacy(legacy) == extract_new(new) is the comparison that actually matters. You’ll build this into a reusable harness in Step 7, but first you need more than one hand-picked example to run it against.

Step 6: Generate a Systematic Set of Test Cases

One matching example proves very little. A real migration would export a sample of real historical requests from production logs to use as test cases. This tutorial has no production traffic to export, so instead you’ll generate a dense, systematic sweep across the input space, which is exactly the kind of repetitive coverage a human manually testing “a few examples” never has the patience for. Save this as generate_cases.py:

"""Test case generation for the shipping-quote differential suite.

A real migration would export a sample of real historical requests from
production logs. This sandbox has no production traffic, so
systematic_sweep() stands in for that: it walks every zone and express
combination across a realistic weight range (0.1kg to 49.9kg) in 0.1kg
steps, which is exactly the kind of dense, repetitive input space that a
human reviewer manually testing "a few examples" would never fully cover.
boundary_cases() adds the specific edge cases the business rules call out
by name: the minimum-charge floor and invalid input.
"""


def systematic_sweep():
    cases = []
    for zone in ("domestic", "regional", "international"):
        for express in (False, True):
            for tenths in range(1, 500):
                cases.append({"weight_kg": round(tenths / 10, 1), "zone": zone, "express": express})
    return cases


def boundary_cases():
    return [
        {"weight_kg": 0.1, "zone": "domestic", "express": False},
        {"weight_kg": 0.1, "zone": "domestic", "express": True},
        {"weight_kg": 100.0, "zone": "international", "express": True},
        {"weight_kg": -1.0, "zone": "domestic", "express": False},
        {"weight_kg": 2.0, "zone": "moon-base", "express": False},
    ]

systematic_sweep() walks all 3 zones, both express settings, and 499 weight values from 0.1kg to 49.9kg in 0.1kg steps, for 2,994 combinations. boundary_cases() adds 5 specific edge cases the business rules call out by name: the smallest legal weight, the smallest legal weight with express, a large international express package, and two cases that should be rejected outright (negative weight, an unrecognized zone).

Step 7: Build a Reusable Differential Testing Harness

Now write the comparison logic once, so every case runs through the same rules. Save this as diff_harness.py:

"""A small, reusable differential-testing harness.

The harness never looks at raw JSON. It extracts the business-meaningful
fields from each implementation's own shape, tolerates floating-point noise
that doesn't change what a customer is charged, and still flags every real
divergence: a wrong price, a wrong delivery estimate, or a case where one
implementation raises and the other doesn't.
"""

import math
from dataclasses import dataclass
from enum import Enum


class Verdict(Enum):
    MATCH = "match"
    COSMETIC = "cosmetic"
    BREAKING = "breaking"
    ERROR_MISMATCH = "error_mismatch"


@dataclass
class CaseResult:
    case: dict
    verdict: Verdict
    detail: str
    legacy_value: object = None
    new_value: object = None


def extract_legacy(result):
    """Pull the business-meaningful fields out of legacy_shipping's response shape."""
    return {
        "cost": result["cost"],
        "estimated_days": result["estimated_days"],
        "carrier": result["carrier"],
    }


def extract_new(result):
    """Pull the same business-meaningful fields out of new_shipping's nested shape."""
    return {
        "cost": result["pricing"]["amount"],
        "estimated_days": result["delivery"]["days"],
        "carrier": result["delivery"]["carrier"],
    }


def compare_case(case, legacy_fn, new_fn, cost_tolerance=0.005):
    """Run one input through both implementations and classify the result.

    cost_tolerance=0.005 means "less than half a cent apart" is treated as
    floating-point representation noise, not a real pricing difference.
    """
    legacy_error = new_error = None
    legacy_semantic = new_semantic = None

    try:
        legacy_raw = legacy_fn(**case, record_metrics=False)
        legacy_semantic = extract_legacy(legacy_raw)
    except ValueError as exc:
        legacy_error = exc

    try:
        new_raw = new_fn(**case, record_metrics=False)
        new_semantic = extract_new(new_raw)
    except ValueError as exc:
        new_error = exc

    if (legacy_error is None) != (new_error is None):
        return CaseResult(
            case=case,
            verdict=Verdict.ERROR_MISMATCH,
            detail=(
                f"legacy {'raised: ' + str(legacy_error) if legacy_error else 'succeeded'}; "
                f"new {'raised: ' + str(new_error) if new_error else 'succeeded'}"
            ),
            legacy_value=legacy_semantic,
            new_value=new_semantic,
        )

    if legacy_error is not None and new_error is not None:
        return CaseResult(case, Verdict.MATCH, "both raised ValueError")

    if legacy_semantic["estimated_days"] != new_semantic["estimated_days"]:
        return CaseResult(
            case, Verdict.BREAKING, "estimated_days differs", legacy_semantic, new_semantic
        )
    if legacy_semantic["carrier"] != new_semantic["carrier"]:
        return CaseResult(
            case, Verdict.BREAKING, "carrier differs", legacy_semantic, new_semantic
        )

    legacy_cost = legacy_semantic["cost"]
    new_cost = new_semantic["cost"]
    cost_diff = abs(legacy_cost - new_cost)

    if legacy_cost == new_cost:
        return CaseResult(case, Verdict.MATCH, "identical", legacy_semantic, new_semantic)
    if math.isclose(legacy_cost, new_cost, abs_tol=cost_tolerance):
        return CaseResult(
            case,
            Verdict.COSMETIC,
            f"cost differs by {cost_diff!r} (float noise, within tolerance)",
            legacy_semantic,
            new_semantic,
        )
    return CaseResult(
        case, Verdict.BREAKING, f"cost differs by ${cost_diff:.2f}", legacy_semantic, new_semantic
    )


def run_batch(cases, legacy_fn, new_fn):
    """Run every case and return (results, summary_counts_by_verdict)."""
    results = [compare_case(c, legacy_fn, new_fn) for c in cases]
    summary = {v: 0 for v in Verdict}
    for r in results:
        summary[r.verdict] += 1
    return results, summary

Three things are worth understanding here before you run it:

Why math.isclose Instead of ==

Floating-point numbers cannot exactly represent most decimal fractions, so two calculations that are mathematically identical can land on slightly different bit patterns, like 6.44 versus 6.4399999999999995. Comparing those with == would flag harmless representation noise as a real difference. Python’s standard library provides math.isclose(a, b, rel_tol=1e-09, abs_tol=0.0) for exactly this problem. Per the official documentation, the comparison is abs(a-b) <= max(rel_tol * max(abs(a), abs(b)), abs_tol), and because these costs are small dollar amounts (not values near zero, where rel_tol alone breaks down), the harness passes abs_tol=0.005: half a cent. Anything closer than that is treated as noise. Anything farther apart is a real difference in what a customer would be charged.

Four Possible Verdicts, Not Two

A binary pass/fail throws away information you need. The harness classifies every case into one of four buckets: MATCH (identical, or both raised the same kind of error), COSMETIC (differs only by floating-point noise below the tolerance), BREAKING (a real difference in cost, delivery estimate, or carrier), or ERROR_MISMATCH (one implementation raised an exception and the other didn’t). That last category exists because errors are part of the contract too: silently succeeding on input the old system correctly rejected is its own kind of bug, arguably a more dangerous one.

Comparing Errors as Part of the Contract

Notice that compare_case catches ValueError from both implementations before doing anything else. If exactly one of them raised, that’s flagged immediately as ERROR_MISMATCH, without even looking at cost. If both raised, that counts as a match: they agree that the input was invalid, even though neither produced a comparable result.

Step 8: Run the Suite and Read What It Found

Tie it together. Save this as run_differential.py:

from diff_harness import Verdict, run_batch
from generate_cases import boundary_cases, systematic_sweep
import legacy_shipping
import new_shipping


def main():
    cases = systematic_sweep() + boundary_cases()
    results, summary = run_batch(cases, legacy_shipping.calculate_shipping, new_shipping.calculate_shipping)

    print(f"total cases: {len(cases)}")
    for verdict in Verdict:
        print(f"  {verdict.value:15s} {summary[verdict]}")

    print()
    print("sample BREAKING cases (real business-meaning differences):")
    breaking = [r for r in results if r.verdict == Verdict.BREAKING]
    for r in breaking[:5]:
        print(f"  case={r.case} -> {r.detail}")
        print(f"    legacy={r.legacy_value}  new={r.new_value}")

    print()
    print("sample ERROR_MISMATCH cases:")
    error_mismatches = [r for r in results if r.verdict == Verdict.ERROR_MISMATCH]
    for r in error_mismatches:
        print(f"  case={r.case} -> {r.detail}")

    print()
    print("sample COSMETIC cases (float noise, tolerated):")
    cosmetic = [r for r in results if r.verdict == Verdict.COSMETIC]
    for r in cosmetic[:3]:
        print(f"  case={r.case} -> {r.detail}")


if __name__ == "__main__":
    main()

Run it:

python run_differential.py
total cases: 2999
  match           2092
  cosmetic        480
  breaking        426
  error_mismatch  1

sample BREAKING cases (real business-meaning differences):
  case={'weight_kg': 1.1, 'zone': 'regional', 'express': False} -> cost differs by $0.01
    legacy={'cost': 5.85, 'estimated_days': 4, 'carrier': 'RegionalFreight'}  new={'cost': 5.86, 'estimated_days': 4, 'carrier': 'RegionalFreight'}
  case={'weight_kg': 1.3, 'zone': 'regional', 'express': False} -> cost differs by $0.01
    legacy={'cost': 6.92, 'estimated_days': 4, 'carrier': 'RegionalFreight'}  new={'cost': 6.91, 'estimated_days': 4, 'carrier': 'RegionalFreight'}
  case={'weight_kg': 1.5, 'zone': 'regional', 'express': False} -> cost differs by $0.01
    legacy={'cost': 7.98, 'estimated_days': 4, 'carrier': 'RegionalFreight'}  new={'cost': 7.97, 'estimated_days': 4, 'carrier': 'RegionalFreight'}
  case={'weight_kg': 2.3, 'zone': 'regional', 'express': False} -> cost differs by $0.01
    legacy={'cost': 12.24, 'estimated_days': 4, 'carrier': 'RegionalFreight'}  new={'cost': 12.23, 'estimated_days': 4, 'carrier': 'RegionalFreight'}
  case={'weight_kg': 2.5, 'zone': 'regional', 'express': False} -> cost differs by $0.01
    legacy={'cost': 13.3, 'estimated_days': 4, 'carrier': 'RegionalFreight'}  new={'cost': 13.31, 'estimated_days': 4, 'carrier': 'RegionalFreight'}

sample ERROR_MISMATCH cases:
  case={'weight_kg': 2.0, 'zone': 'moon-base', 'express': False} -> legacy raised: unknown zone: 'moon-base'; new succeeded

sample COSMETIC cases (float noise, tolerated):
  case={'weight_kg': 2.3, 'zone': 'domestic', 'express': False} -> cost differs by 8.881784197001252e-16 (float noise, within tolerance)
  case={'weight_kg': 2.7, 'zone': 'domestic', 'express': False} -> cost differs by 8.881784197001252e-16 (float noise, within tolerance)
  case={'weight_kg': 4.6, 'zone': 'domestic', 'express': False} -> cost differs by 1.7763568394002505e-15 (float noise, within tolerance)

That is real output from real code, and it tells a story: out of 2,999 cases, 2,092 matched cleanly and 480 differed only by floating-point noise smaller than half a cent, both fine. But 426 cases, about 14% of the systematic sweep, are genuinely different prices, and 1 case shows the new implementation accepting input the old one correctly rejects. If you had tested this by hand with 5 or 10 example orders, there’s a real chance every single one would have landed in the “match” or “cosmetic” bucket, and you would have shipped both bugs.

Bug 1: A Rounding-Order Regression

Look at the first breaking case: a 1.1kg regional package costs $5.85 in the legacy system and $5.86 in the new one. Both implementations compute the same fuel surcharge on the same base cost. The difference is when they round. The legacy code adds the fuel surcharge first, then rounds once at the very end. The new code rounds the base cost to two decimals immediately, adds a separately-rounded surcharge to that already-rounded number, and only then applies the minimum charge. Rounding twice at different points in the calculation, instead of once at the end, changes the actual cent value in a real and predictable fraction of cases. This is exactly the kind of change that “looks fine” in code review, because nobody manually re-derives the arithmetic order.

Bug 2: A Silent, Dangerous Fallback

The ERROR_MISMATCH case is worse. Passing an unrecognized zone like "moon-base" makes the legacy system correctly raise ValueError: unknown zone: 'moon-base'. The new system does not raise anything. It silently treats the request as "domestic" and returns a confident-looking $5.60 quote. Somewhere in that codebase’s history, a comment like “nobody ships anywhere else yet, domestic is a safe default” turned a validation bug into a feature. A caller that relied on the exception to reject bad input now gets a wrong answer that looks completely normal. This is precisely why the harness treats error behavior as part of the contract instead of ignoring exceptions.

Step 9: Fix the New Implementation and Confirm It

Both fixes are small: round once, at the end, in the same order as the legacy calculation, and raise instead of silently substituting a zone. Save the corrected version as new_shipping_fixed.py (the response shape stays nested; differential testing checks behavior, not formatting, so fixing these bugs doesn’t force you back to the old shape):

"""The new shipping quote service, after differential testing caught two bugs.

Two changes from new_shipping.py:
1. The fuel surcharge is computed and added before the single, final
   round(), matching legacy_shipping's order of operations.
2. An unknown zone raises ValueError instead of silently falling back
   to "domestic", matching legacy_shipping's error contract.

Nothing about the response *shape* changed: it's still the new, nested
pricing/delivery structure. Differential testing checks behavior, not
formatting, so it doesn't force you to keep the old shape.
"""

import time

ZONE_BASE_RATE = {
    "domestic": 2.50,
    "regional": 4.75,
    "international": 9.90,
}

ZONE_STANDARD_DAYS = {
    "domestic": 2,
    "regional": 4,
    "international": 9,
}

EXPRESS_MULTIPLIER = 1.8
FUEL_SURCHARGE_RATE = 0.12
MIN_CHARGE = 5.00

METRICS = {"domestic": 0, "regional": 0, "international": 0}


def calculate_shipping(weight_kg, zone, express=False, record_metrics=True):
    if weight_kg <= 0:
        raise ValueError(f"weight_kg must be positive, got {weight_kg}")

    if zone not in ZONE_BASE_RATE:
        raise ValueError(f"unknown zone: {zone!r}")

    base_rate = ZONE_BASE_RATE[zone]
    cost = weight_kg * base_rate
    if express:
        cost *= EXPRESS_MULTIPLIER

    fuel_surcharge = cost * FUEL_SURCHARGE_RATE
    cost += fuel_surcharge
    cost = round(cost, 2)
    cost = max(cost, MIN_CHARGE)

    estimated_days = ZONE_STANDARD_DAYS[zone]
    if express:
        estimated_days = max(1, estimated_days // 2)

    carrier = "GlobalShip" if zone == "international" else "RegionalFreight"

    if record_metrics:
        METRICS[zone] += 1

    quote_id = f"NEW-{int(time.time() * 1_000_000)}"

    return {
        "quote_id": quote_id,
        "pricing": {
            "amount": cost,
            "currency": "USD",
        },
        "delivery": {
            "days": estimated_days,
            "carrier": carrier,
        },
    }

Re-run the exact same suite against the fixed version:

from diff_harness import Verdict, run_batch
from generate_cases import boundary_cases, systematic_sweep
import legacy_shipping
import new_shipping_fixed

cases = systematic_sweep() + boundary_cases()
results, summary = run_batch(cases, legacy_shipping.calculate_shipping, new_shipping_fixed.calculate_shipping)
print(f"total cases: {len(cases)}")
for v in Verdict:
    print(f"  {v.value:15s} {summary[v]}")
total cases: 2999
  match           2999
  cosmetic        0
  breaking        0
  error_mismatch  0

Every one of the 2,999 cases now matches exactly, including the 480 that were previously only “close enough.” That second part isn’t a coincidence: once both implementations perform the arithmetic in the same order, they produce bit-for-bit identical floating-point results, not just results within tolerance. That’s a useful thing to notice, but don’t rely on it: keep the tolerance in your harness anyway, because you generally can’t guarantee two independently written implementations will always compute things in the same order, and you don’t want a future, harmless refactor to trip a false alarm.

Step 10: Lock the Result In With a Pytest Suite

A one-time console run is useful for investigation, but you want this check to run automatically forever. Save this as test_differential.py:

import pytest

import legacy_shipping
import new_shipping
import new_shipping_fixed
from diff_harness import Verdict, compare_case, run_batch
from generate_cases import boundary_cases, systematic_sweep


def test_full_sweep_has_zero_breaking_differences_against_fixed_implementation():
    cases = systematic_sweep() + boundary_cases()
    _, summary = run_batch(cases, legacy_shipping.calculate_shipping, new_shipping_fixed.calculate_shipping)
    assert summary[Verdict.BREAKING] == 0
    assert summary[Verdict.ERROR_MISMATCH] == 0


def test_regression_1_1kg_regional_no_longer_off_by_a_cent():
    """This exact input was the first divergence differential testing found
    against the buggy implementation (legacy 5.85 vs buggy new 5.86)."""
    case = {"weight_kg": 1.1, "zone": "regional", "express": False}
    result = compare_case(case, legacy_shipping.calculate_shipping, new_shipping_fixed.calculate_shipping)
    assert result.verdict == Verdict.MATCH


def test_unknown_zone_now_raises_like_legacy_instead_of_falling_back():
    with pytest.raises(ValueError):
        new_shipping_fixed.calculate_shipping(2.0, "moon-base", False)


def test_buggy_implementation_is_still_correctly_flagged_by_the_harness():
    """Guards against the harness itself silently stopping to catch things.
    If this ever starts failing, the comparison logic broke, not the app."""
    cases = systematic_sweep() + boundary_cases()
    _, summary = run_batch(cases, legacy_shipping.calculate_shipping, new_shipping.calculate_shipping)
    assert summary[Verdict.BREAKING] == 426
    assert summary[Verdict.ERROR_MISMATCH] == 1


def test_comparison_never_records_metrics_as_a_side_effect():
    for m in (legacy_shipping, new_shipping, new_shipping_fixed):
        for zone in m.METRICS:
            m.METRICS[zone] = 0

    compare_case(
        {"weight_kg": 3.0, "zone": "domestic", "express": False},
        legacy_shipping.calculate_shipping,
        new_shipping_fixed.calculate_shipping,
    )

    assert legacy_shipping.METRICS == {"domestic": 0, "regional": 0, "international": 0}
    assert new_shipping_fixed.METRICS == {"domestic": 0, "regional": 0, "international": 0}

Run it:

pytest test_differential.py -v
collecting ... collected 5 items

test_differential.py::test_full_sweep_has_zero_breaking_differences_against_fixed_implementation PASSED [ 20%]
test_differential.py::test_regression_1_1kg_regional_no_longer_off_by_a_cent PASSED [ 40%]
test_differential.py::test_unknown_zone_now_raises_like_legacy_instead_of_falling_back PASSED [ 60%]
test_differential.py::test_buggy_implementation_is_still_correctly_flagged_by_the_harness PASSED [ 80%]
test_differential.py::test_comparison_never_records_metrics_as_a_side_effect PASSED [100%]

============================== 5 passed in 0.03s ==============================

Notice test_buggy_implementation_is_still_correctly_flagged_by_the_harness. It deliberately runs the full sweep against the original, buggy new_shipping module and asserts it still finds exactly 426 breaking cases and 1 error mismatch. That test isn’t there to test your application. It’s there to test your test harness: if someone “simplifies” the comparison logic later and it stops catching these two known bugs, this test fails loudly instead of the harness quietly going blind.

Step 11: Use Shadow Traffic to Gain Confidence at Production Scale

A systematic sweep is thorough, but it’s still synthetic. At some point you want to run the new implementation against real, live requests, without letting it affect what any real customer sees. This is usually called shadow traffic or a dark launch. GitHub’s open-source scientist library, built for exactly this purpose during real production refactors, calls the trusted, currently-serving code path the control and the new code path being evaluated the candidate: you always return the control’s result to the caller, and only use the candidate’s result for comparison.

There’s a sharp, easy-to-miss failure mode here: if your shadow call to the candidate has side effects of its own (metrics, logging, cache writes), and you don’t suppress them, your shadow traffic will silently pollute production state for a system that hasn’t actually launched yet. Save this as shadow_traffic.py to see it happen for real, and then fix it:

"""Shadow traffic: run the new implementation alongside real production
requests, purely to compare it against legacy, without ever using its
result for a real response.

This module demonstrates two things:
1. A naive shadow call that forgets to suppress side effects, and the
   real damage that does to shared state like METRICS.
2. The fixed version, plus using the harness to flag real divergences
   found on randomly generated "production-like" traffic instead of the
   full systematic sweep.
"""

import random

import legacy_shipping
import new_shipping
from diff_harness import Verdict, compare_case


def simulate_requests(n=20, seed=7):
    random.seed(seed)
    zones = ["domestic", "regional", "international"]
    requests = []
    for _ in range(n):
        requests.append(
            {
                "weight_kg": round(random.uniform(0.2, 15.0), 1),
                "zone": random.choice(zones),
                "express": random.random() < 0.3,
            }
        )
    return requests


def handle_request(case):
    """What actually serves the customer: the trusted legacy implementation."""
    return legacy_shipping.calculate_shipping(**case)


def naive_shadow_call(case):
    """BROKEN: calls the new implementation with its default side effects."""
    return new_shipping.calculate_shipping(**case)


def safe_shadow_call(case):
    """Correct: the shadow call exists only to compare, so it must not
    record metrics as if it served a real customer."""
    return new_shipping.calculate_shipping(**case, record_metrics=False)


def reset_metrics():
    for m in (legacy_shipping, new_shipping):
        for zone in m.METRICS:
            m.METRICS[zone] = 0


def demo_naive_vs_safe():
    requests = simulate_requests()

    reset_metrics()
    for case in requests:
        handle_request(case)
        naive_shadow_call(case)
    print("after naive shadow calls:")
    print("  legacy_shipping.METRICS:", legacy_shipping.METRICS)
    print("  new_shipping.METRICS:   ", new_shipping.METRICS, "<- polluted by shadow traffic")

    reset_metrics()
    for case in requests:
        handle_request(case)
        safe_shadow_call(case)
    print()
    print("after safe shadow calls:")
    print("  legacy_shipping.METRICS:", legacy_shipping.METRICS)
    print("  new_shipping.METRICS:   ", new_shipping.METRICS, "<- untouched, as it should be")


def demo_shadow_divergence_detection():
    requests = simulate_requests(n=40, seed=11)
    divergences = []
    for case in requests:
        handle_request(case)
        result = compare_case(case, legacy_shipping.calculate_shipping, new_shipping.calculate_shipping)
        if result.verdict not in (Verdict.MATCH, Verdict.COSMETIC):
            divergences.append(result)

    print()
    print(f"shadow-traffic sample size: {len(requests)}")
    print(f"divergences flagged: {len(divergences)}")
    for d in divergences[:5]:
        print(f"  {d.case} -> {d.detail}")


if __name__ == "__main__":
    demo_naive_vs_safe()
    demo_shadow_divergence_detection()

Run it:

python shadow_traffic.py
after naive shadow calls:
  legacy_shipping.METRICS: {'domestic': 8, 'regional': 7, 'international': 5}
  new_shipping.METRICS:    {'domestic': 8, 'regional': 7, 'international': 5} <- polluted by shadow traffic

after safe shadow calls:
  legacy_shipping.METRICS: {'domestic': 8, 'regional': 7, 'international': 5}
  new_shipping.METRICS:    {'domestic': 0, 'regional': 0, 'international': 0} <- untouched, as it should be

shadow-traffic sample size: 40
divergences flagged: 4
  {'weight_kg': 14.7, 'zone': 'regional', 'express': False} -> cost differs by $0.01
  {'weight_kg': 2.8, 'zone': 'international', 'express': True} -> cost differs by $0.01
  {'weight_kg': 3.3, 'zone': 'international', 'express': True} -> cost differs by $0.01
  {'weight_kg': 9.1, 'zone': 'regional', 'express': False} -> cost differs by $0.01

Look closely at the first block. After 20 requests processed by naive_shadow_call, new_shipping.METRICS shows real counts, {'domestic': 8, 'regional': 7, 'international': 5}, identical to legacy’s, even though the new service has never served a single real customer. Any dashboard reading that counter would report new-service volume that doesn’t exist. The fix is exactly the record_metrics=False parameter you added back in Step 2: after resetting the counters and switching to safe_shadow_call, which passes that flag, new_shipping.METRICS stays at zero while legacy_shipping.METRICS still increments correctly from the real, control-path calls.

The second block runs 40 randomly generated, production-shaped requests through the harness itself (which already forces record_metrics=False on both sides internally, since compare_case exists purely to compare, never to serve) and finds 4 more instances of the same rounding bug, on inputs the systematic sweep’s fixed 0.1kg steps never happened to generate. This is the case for running both techniques: a systematic sweep gives you dense, repeatable coverage over a known input space, and shadow traffic on live requests gives you cases you didn’t think to construct.

Common Mistakes and Gotchas

  • Comparing raw output with ==. You saw this fail in Step 4 for reasons that had nothing to do with correctness. Always extract and compare business meaning.
  • Using exact float equality, or a tolerance so large it hides real bugs. A tolerance of $0.005 caught a $0.01 pricing bug in this tutorial. A tolerance of $0.05, chosen carelessly to “stop the noisy failures,” would have hidden it completely.
  • Ignoring exceptions entirely. If your harness only compares return values, an input that should be rejected but silently succeeds will never show up as a problem.
  • Testing a handful of round numbers and calling it done. The rounding bug in this tutorial affected roughly 1 in 7 systematically generated inputs. A programmer manually trying 3.2kg, 5kg, and 10kg domestic packages would very plausibly never hit it.
  • Forgetting that a shadow call has side effects too. You reproduced this directly in Step 11: a shadow call to the new implementation quietly incremented a metrics counter meant to represent real traffic.
  • Treating “my new tests pass” as equivalent to “it matches the old system.” new_shipping.py‘s hypothetical unit tests passing is exactly what makes this technique necessary. Its own tests can’t tell you whether it agrees with something it was never tested against.

How to Know You’re Ready to Cut Over

Before routing real traffic to a replacement, work through this checklist, adapted from what this tutorial actually did:

  1. Have the important input classes been compared? Not just happy-path values. Boundaries, invalid input, and every branch your business rules mention by name.
  2. Is every difference classified, not just counted? “907 differences out of 2,999 cases” is not actionable on its own. “480 float noise, 426 real pricing bugs, 1 dangerous silent fallback” is.
  3. Are the critical differences resolved, and re-verified? Fixing a bug and assuming it’s fixed are different things. Re-run the full suite, don’t just spot-check the one case you fixed.
  4. Do side effects match, not just return values? This tutorial’s METRICS counter is a stand-in for anything else downstream code depends on: a database write, an email, an audit log entry.
  5. Has it been run against real traffic in shadow mode, not just synthetic cases? Synthetic sweeps cover the input space you thought of. Shadow traffic covers the input space your users actually produce.
  6. Can you roll back? Everything here assumed the legacy implementation stays available and authoritative right up until you’re confident. Don’t remove it as part of the same change that introduces the replacement.

Confirm It All Works End to End

To verify everything in this tutorial for yourself, from a clean checkout: create legacy_shipping.py, new_shipping.py, diff_harness.py, generate_cases.py, run_differential.py, new_shipping_fixed.py, test_differential.py, and shadow_traffic.py exactly as shown, in one folder, then run:

pip install pytest
python run_differential.py       # expect 426 breaking, 1 error_mismatch against the buggy version
pytest test_differential.py -v   # expect 5 passed
python shadow_traffic.py         # expect the METRICS pollution, then the fix, then 4 flagged divergences

If your output matches what’s shown above (your quote IDs and exact microsecond timestamps will differ, that’s expected and part of why Step 5 exists), you’ve reproduced the same two real bugs this tutorial found, fixed them the same way, and verified the fix the same way: by running the comparison again, not by trusting that the fix was correct.

Next Steps

  • If your legacy system isn’t cleanly callable as a function yet, start with extracting seams and adapters so you have two comparable implementations to run through a harness like this one in the first place.
  • Before you refactor anything, characterization tests lock in what the current system does, so you have a safety net even before a second implementation exists to compare against.
  • If you’re migrating a codebase you didn’t write and don’t fully understand yet, using AI to investigate it first can help you find the business rules a differential test suite needs to cover, before you start generating test cases.
  • Try adding a third dimension to generate_cases.py: instead of a fixed 0.1kg step, generate random weights with a fixed seed, and see if a larger, randomized sample surfaces anything the systematic sweep’s regular grid missed.

Tags:

Differential TestingLegacy CodepytestPythonTesting

Share

No Comment! Be the first one.

Leave a Reply

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

Related Posts