API and Contract Testing Questions

Direct answer

The migration works best as a gradual, verifiable transition rather than a cutover: run the new contract-backed stubs alongside the existing file-based mocks for a period, prove for each service pair that the new stub produces the same or better fidelity, and only remove the old mocks once that's demonstrated, not on a fixed calendar date.

Structured elaboration

Why the old approach is brittle. File-based local mocks are hand-maintained: someone writes down a plausible-looking response once, and nothing keeps that response in sync with what the real dependency actually returns. They drift silently, and by the time someone notices, the tests have been giving false confidence for a while.

Coexistence during migration. Rather than replacing all mocks for a service at once, migrate one dependency relationship at a time. For each one, stand up the contract-backed stub alongside the existing file-based mock, and run both in parallel for a period, ideally comparing their outputs against a real recorded baseline to confirm the new stub is at least as accurate.

Encouraging adoption. A migration that's purely a mandate tends to stall. What actually gets teams to adopt it: making the new approach genuinely easier once it exists, if the contract-backed stub is generated from something the provider team already maintains, it needs less manual upkeep than a hand-written file, and pointing to a concrete case where the old file-based mock had gone stale and hidden a real bug, since that's the argument that makes the migration's value tangible rather than theoretical.

Automating validation. As each dependency migrates, add an automated check that the contract-backed stub's responses still match the provider's actual current contract, this is what prevents the new approach from developing the same silent-drift problem the old one had. A scheduled job that re-verifies stubs against the real (or sandboxed) provider, and fails loudly on a mismatch, is the mechanism that makes the improvement durable rather than a one-time cleanup.

Rollback. Because both approaches coexist during migration, rolling back a single dependency that's causing trouble in its new stubbed form is cheap, revert to the file-based mock for that one dependency while you investigate, without affecting the dependencies that have already migrated cleanly.

Measuring success. Concrete signals worth tracking: the number of dependencies still on file-based mocks (should trend toward zero on a realistic timeline, not a rushed one), the number of test failures caught by the new stubs that the old mocks would have missed (evidence the migration is finding real value, not just churn), and whether any production incidents trace back to a stub that was still stale, which would indicate the validation automation isn't yet solid enough to fully trust.

Trade-offs and pitfalls

The temptation to declare victory once every service has "some" contract-backed stub, without actually removing the old file-based ones or automating their validation, leaves you maintaining two systems indefinitely and getting the benefit of neither. The migration isn't complete until the old mocks are gone and the new stubs are automatically, continuously validated against reality, not just initially accurate.

Direct answer

Generating the signature is a few lines: HMAC (hash-based message authentication code) SHA-256 over the timestamp and raw body, using the shared secret. Verifying it correctly on the receiver side is the part that actually matters, and it has three genuinely separate checks: the signature is valid, the timestamp is fresh, and this exact request hasn't been processed before.

Structured elaboration

Signing (sender side, what the Postman pre-request script does). The vendor's convention here is standard: concatenate the timestamp and the raw request body with a separator, then HMAC-SHA256 that combined string with the shared secret, and send the result as a header alongside the timestamp itself. The receiver has to sign the exact same bytes the same way to check it, so the pre-request script and the receiver's verification logic must agree on the signed-content format down to the separator character.

Verifying (receiver side), three checks, each catching a different failure:

1. Timestamp freshness. Reject anything outside a tolerance window (a few minutes is typical) before doing anything else. This is what limits how long a captured request stays replayable even before the replay check runs, and it's cheap to check first so an obviously stale request doesn't cost a cryptographic comparison.
2. Signature validation. Recompute the expected signature from the secret, the timestamp, and the body, and compare it to the header using a constant-time comparison, never a plain string equality, which can leak timing information about how many leading bytes matched and make the secret guessable byte by byte over many attempts.
3. Replay protection. Even a validly-signed, fresh request should only be accepted once. Track signatures (or a vendor-supplied event ID, if one exists) already processed, and reject a repeat.

Worked example

The pre-request script that generates the signature, using CryptoJS, the crypto library Postman's sandbox exposes as a global:

const secret = 'test-webhook-secret-shared-with-vendor';
const timestamp = Math.floor(Date.now() / 1000).toString();
const body = pm.request.body.raw;
const signedContent = timestamp + '.' + body;
const signature = CryptoJS.HmacSHA256(signedContent, secret).toString(CryptoJS.enc.Hex);
pm.request.headers.upsert({ key: 'X-Webhook-Timestamp', value: timestamp });
pm.request.headers.upsert({ key: 'X-Webhook-Signature', value: signature });

And the receiver-side verification, in Python:

import hmac, hashlib, time

def verify_webhook(secret, timestamp, body, signature_header, seen_signatures, max_age_seconds=300):
    ts = int(timestamp)
    if abs(time.time() - ts) > max_age_seconds:
        return False, "stale timestamp"
    signed_content = timestamp.encode() + b"." + body
    expected = hmac.new(secret.encode(), signed_content, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature_header):
        return False, "signature mismatch"
    if signature_header in seen_signatures:
        return False, "replay detected"
    seen_signatures.add(signature_header)
    return True, "accepted"

Verified end to end with a minimal receiver, not just the isolated function above. Wiring verify_webhook into a small Flask endpoint and driving it with Flask's test client (real HTTP request/response objects, no mocking) confirms the three cases the tests need to cover:

from flask import Flask, request, jsonify

SECRET = "test-webhook-secret-shared-with-vendor"
seen_signatures = set()
app = Flask(__name__)

@app.route("/webhook", methods=["POST"])
def webhook():
    timestamp = request.headers.get("X-Webhook-Timestamp", "")
    signature = request.headers.get("X-Webhook-Signature", "")
    body = request.get_data()
    ok, reason = verify_webhook(SECRET, timestamp, body, signature, seen_signatures)
    if ok:
        return jsonify({"status": reason}), 200
    if reason == "replay detected":
        return jsonify({"error": reason}), 409
    return jsonify({"error": reason}), 400

Driving it with three real round trips through app.test_client():

fresh first-time request -> 200 {'status': 'accepted'}
replayed request          -> 409 {'error': 'replay detected'}
stale (1hr old) request   -> 400 {'error': 'stale timestamp'}

Exactly the three outcomes the acceptance criteria need: a fresh, correctly-signed request accepted, the identical request replayed and rejected as a duplicate, and a correctly-signed but hour-old request rejected as stale.

Trade-offs and pitfalls

A real gotcha found while building this, worth knowing before you hit it live. The obvious "modernization" of the pre-request script, replacing the bare CryptoJS global with const CryptoJS = require('crypto-js'), actually breaks in the current Postman sandbox: CryptoJS is already bound as a global, and redeclaring it throws SyntaxError: Identifier 'CryptoJS' has already been declared. The bare global form is deprecated (Postman's own console warns about it) but is still the one that actually works today; don't "fix" a deprecation warning by introducing a naming collision that breaks the script outright, verify a replacement actually runs before trusting a deprecation notice's suggested fix.

The signed-content format (timestamp, a separator, then the raw body, in that exact order and byte form) has to match the vendor's real convention exactly; a mismatch anywhere (a different separator, a parsed-and-re-serialized body instead of the raw bytes, a different byte encoding) makes every signature fail to verify even though the logic is otherwise correct, so the first thing to check against a real vendor's docs, not assume, is the exact signed-content construction.

Direct answer

Below is a reusable HTTP GET function with retry and exponential backoff for transient server errors and network errors, with configurable attempt count and base delay.

Structured elaboration

The function distinguishes what's worth retrying (a timeout, a connection error, a 5xx that's likely transient) from what isn't (a 4xx, which means the request itself is wrong and retrying an unmodified request will just fail the same way again), and backs off exponentially between attempts so a struggling server isn't hit with an immediate retry storm.

Worked example

import time
import requests


def resilient_get(url, max_attempts=5, initial_backoff=0.5, backoff_factor=2,
                   retry_statuses=(500, 502, 503, 504)):
    last_exception = None
    delay = initial_backoff

    for attempt in range(1, max_attempts + 1):
        try:
            resp = requests.get(url, timeout=5)
            if resp.status_code not in retry_statuses:
                return resp  # success, or a non-retryable error (e.g. 4xx): return as-is
            last_exception = None
        except (requests.ConnectionError, requests.Timeout) as e:
            last_exception = e
            resp = None

        if attempt == max_attempts:
            if last_exception:
                raise last_exception
            return resp  # exhausted retries on a retryable status; return the last response

        time.sleep(delay)
        delay *= backoff_factor

    raise RuntimeError("unreachable")  # defensive; loop always returns or raises above

Thread-safety. As written, resilient_get has no shared mutable state at all: delay, attempt, and last_exception are all local to each call, so many test threads calling it concurrently don't interact with each other in any way. The one thing worth being deliberate about in concurrent use is the underlying requests session: this version uses the module-level requests.get, which creates a new connection per call and is safe under concurrency but doesn't reuse connections. If you switch to a shared requests.Session() for connection pooling (a reasonable optimization under high concurrency), the Session object itself needs to be either one per thread or explicitly documented as thread-safe for your use case, since requests.Session is not guaranteed thread-safe for concurrent use by requests' own documentation.

Verified with a fixture that fails twice with a 503 and then succeeds:

call_count = [0]

def flaky_get(url, timeout):
    call_count[0] += 1
    class FakeResp:
        status_code = 503 if call_count[0] <= 2 else 200
    return FakeResp()

requests.get = flaky_get  # monkeypatched into requests.get for this test

resp = resilient_get("http://fake/x", initial_backoff=0.01)
print(f"attempts made: {call_count[0]}")
print(f"final status_code: {resp.status_code}")

Running resilient_get against this fixture (with initial_backoff=0.01 to keep the test fast) returns a 200 after exactly 3 attempts, confirming the retry loop and the eventual-success path both work:

attempts made: 3
final status_code: 200

Trade-offs and pitfalls

A 4xx status code deliberately does NOT trigger a retry in this implementation: retrying an unmodified request that the server has already rejected as invalid wastes time and, in the worst case, can look like an attempted abuse pattern to the server (repeated requests to an endpoint that keeps rejecting them). If a caller genuinely wants to retry a 429 (rate-limited) specifically, that status needs to be added to retry_statuses deliberately, and ideally the delay should respect a Retry-After header if the server provides one, rather than blindly following the function's own generic backoff schedule.

Direct answer

At organizational scale, a contract testing rollout has three parts working together: a clear authoring and versioning discipline for contracts, a broker as the shared source of truth, and CI gates on both the consumer and provider side that actually block a bad deploy rather than just reporting on it after the fact.

Structured elaboration

Authoring and ownership. Each consumer team owns the contracts that describe what it needs from a provider; each provider team owns making its own CI verify against every contract published against it. This is what keeps contracts from silently drifting out of sync: since a consumer's contract is generated by running its own test, it's grounded in what the code actually depends on, not documentation someone forgot to update.

Storage and versioning. A broker stores every contract, tagged by consumer version and branch, and every verification result, tagged by which provider version verified against which consumer version. This is the piece that scales the approach past a handful of services: instead of everyone needing to know everyone else's state, the broker answers "is it safe for provider version X to deploy, given what consumer versions are actually running in production right now" as a single query, sometimes called a can-I-deploy check.

CI integration. Two hooks matter. On the consumer side, a merge to main publishes the new contract to the broker. On the provider side, the pipeline pulls the latest relevant contracts and runs provider verification, replaying each contract's interactions against the real service, before allowing the build to proceed. Verification timing is usually split: on every pull request against the most recent contracts (fast feedback), and again on a schedule or before release against whatever's currently deployed (catching drift).

Handling backward-incompatible change. When a provider needs to make a breaking change, the discipline is to introduce the new behavior alongside the old one (an additive change, a new field, a new version), get every affected consumer to update and re-verify against the new shape, and only then retire the old behavior once the broker shows no live consumer still depends on it. The can-I-deploy check is what makes this safe to do incrementally rather than as a coordinated big-bang release.

Handling mismatches in CI. When provider verification fails, that failure belongs in the provider's own build, not the consumer's, and the pipeline should block the provider's deploy rather than let it ship and only fail visibly in production. Multiple consumers with conflicting expectations for the same interaction is a real failure mode: it needs to be resolved as a genuine compatibility conversation between teams, not silently overridden by whichever contract happened to verify last.

Trade-offs and pitfalls

The two failure modes worth naming explicitly: rolling this out with no governance, so every team invents its own conventions and the broker becomes noise instead of a source of truth, and rolling it out as pure tooling with no CI enforcement, so contracts exist but nothing actually blocks a bad deploy, which teaches everyone to ignore them. Retrofitting this onto an organization with many existing services and no prior contract-testing practice works better as a staged migration: pick a handful of well-understood, high-change-frequency service pairs first, prove the workflow and the can-I-deploy gate actually catch something real, then expand, rather than mandating it everywhere at once with no working example to point to.

Direct answer

Verifying completeness across pagination means tracking every item you've seen by its unique ID as you walk the pages, then asserting at the end that the set of IDs collected matches the full expected set exactly, no duplicates and nothing missing, rather than just checking that each individual page request succeeded.

Structured elaboration

The two pagination styles need different loop-termination logic but the same completeness check underneath: page-number pagination stops when a page comes back empty (or when you've reached a known total), token-based pagination stops when next_page comes back null. A subtlety worth being deliberate about: the collection step itself must NOT silently deduplicate as it goes, if it does, a real bug where the API returns the same item twice across two pages gets quietly absorbed instead of caught. Collect everything exactly as returned, then check for duplicates and completeness as a separate step afterward.

Worked example

import requests


def fetch_all_items(base_url, use_tokens: bool):
    # No dedup here on purpose: deduping while collecting would hide a
    # duplicate-across-pages bug instead of surfacing it.
    collected_ids = []

    if use_tokens:
        next_page = None
        while True:
            params = {"next_page": next_page} if next_page else {}
            resp = requests.get(f"{base_url}/items", params=params)
            data = resp.json()
            collected_ids.extend(item["id"] for item in data["items"])
            next_page = data.get("next_page")
            if not next_page:
                break
    else:
        page = 1
        while True:
            resp = requests.get(f"{base_url}/items", params={"page": page, "page_size": 50})
            data = resp.json()
            if not data["items"]:
                break
            collected_ids.extend(item["id"] for item in data["items"])
            page += 1

    return collected_ids


def test_pagination_completeness_no_duplicates(base_url, expected_ids: set, use_tokens: bool):
    collected = fetch_all_items(base_url, use_tokens)
    assert len(collected) == len(set(collected)), (
        f"duplicates found across pages: {len(collected)} items but only "
        f"{len(set(collected))} unique ids"
    )
    assert set(collected) == expected_ids, (
        f"missing: {expected_ids - set(collected)}, unexpected: {set(collected) - expected_ids}"
    )

Verification harness. Monkeypatching requests.get with a fixture stands in for the real API without a network call, and lets both the buggy and fixed cases actually run and print their own result:

class FakeResp:
    def __init__(self, data):
        self._data = data
    def json(self):
        return self._data


def build_fixture(duplicate_bug: bool):
    """137 items across 3 pages; optionally leaks item 90 into a second page."""
    all_items = [{"id": i} for i in range(1, 138)]
    p1, p2, p3 = all_items[0:50], all_items[50:100], all_items[100:137]
    if duplicate_bug:
        p3 = [{"id": 90}] + p3  # item 90 duplicated across two consecutive pages
    pages = {
        None: {"items": p1, "next_page": "tok2"},
        "tok2": {"items": p2, "next_page": "tok3"},
        "tok3": {"items": p3, "next_page": None},
    }
    def fake_get(url, params=None):
        return FakeResp(pages[params.get("next_page") if params else None])
    return fake_get


expected_ids = set(range(1, 138))

requests.get = build_fixture(duplicate_bug=True)
try:
    test_pagination_completeness_no_duplicates("http://fake", expected_ids, use_tokens=True)
except AssertionError as e:
    print(f"AssertionError: {e}")

requests.get = build_fixture(duplicate_bug=False)
collected = fetch_all_items("http://fake", use_tokens=True)
test_pagination_completeness_no_duplicates("http://fake", expected_ids, use_tokens=True)
print(f"test passed: {len(set(collected))}/{len(expected_ids)} unique items collected, expected set matched exactly")

Against the buggy fixture (item 90 duplicated across two consecutive pages):

AssertionError: duplicates found across pages: 138 items but only 137 unique ids

And against the same fixture with the duplication bug fixed:

test passed: 137/137 unique items collected, expected set matched exactly

Trade-offs and pitfalls

The subtlety in the collection step above is worth calling out explicitly because it's an easy bug to write into the TEST itself: an earlier version of this exact function deduplicated ids as it collected them (only appending an id the first time it was seen), which meant the "duplicates found" assertion could never fire, no matter how badly the API was actually duplicating items across pages, because the dedup step silently absorbed the duplication before the check ever ran. A completeness test needs to collect the raw, undeduplicated data first and treat deduplication as part of the ASSERTION, not part of the collection.

A completeness test like this also needs a known, fixed dataset to compare against: testing against a live, changing dataset makes "missing" and "unexpected" ambiguous, since an item could legitimately have been added or removed between the start and end of the pagination walk rather than being a real bug. In CI, this argues for seeding a deterministic dataset before the test runs rather than pointing the test at shared, mutable data.