VRS Integration Testing with Python

A new trading partner connection inside Trading Partner Verification & Onboarding is not ready for go-live just because a manual test call returned verified: true once from a browser. Part of the wider DSCSA Compliance Architecture & Standards Mapping discipline, this page builds a repeatable pytest suite that drives a Verification Router Service (VRS) connection through every path it will hit in production: a known serial that verifies, an unknown serial that does not, a request signed with an expired credential, and a partner that answers too slowly to meet your latency SLA. The suite runs against a mock responder in CI and against the partner’s staging endpoint before cutover, so a broken integration surfaces in a pull request instead of during a live dispenser lookup.

Prerequisites #

• Python 3.10+ — the snippets use X | None unions and dataclass-style Pydantic v2 models.
• pytest ≥ 7 and pytest-asyncio — set asyncio_mode = "auto" under [tool.pytest.ini_options] in pyproject.toml so async def test_... functions run without per-test decorators.
• httpx ≥ 0.27 — its AsyncClient gives you real timeout semantics, and httpx.MockTransport lets you swap in a fake responder without touching the client code under test.
• Pydantic v2 — to model the GS1 Lightweight Messaging Standard request and response payloads as typed, validated objects rather than raw dictionaries.
• DSCSA/GS1 data prerequisites — the partner’s verification endpoint path and auth scheme, a pool of serials known to be commissioned in the test environment, at least one serial guaranteed not to exist, and a credential you can deliberately expire or revoke. Every request needs the same product identifier (01) (21) (10) (17) the partner expects on a live lookup.

Step-by-Step Solution #

Step 1 — Model the verification request and response contract #

Typing the GS1 Lightweight Messaging Standard payload as a Pydantic model catches malformed test fixtures before they ever leave the machine, and gives every later assertion a stable, validated shape to check against.

from __future__ import annotations

from datetime import date
from uuid import uuid4

from pydantic import BaseModel, Field, field_validator


class VerificationRequest(BaseModel):
    """A GS1 Lightweight Messaging Standard verification lookup."""

    request_id: str = Field(default_factory=lambda: uuid4().hex)
    gtin: str                        # AI (01)
    serial_number: str               # AI (21)
    lot_number: str | None = None    # AI (10)
    expiration_date: date | None = None  # AI (17)
    requestor_gln: str

    @field_validator("gtin")
    @classmethod
    def _gtin_shape(cls, v: str) -> str:
        if len(v) != 14 or not v.isdigit():
            raise ValueError("GTIN (01) must be 14 numeric digits")
        return v

    @field_validator("serial_number")
    @classmethod
    def _serial_shape(cls, v: str) -> str:
        if not (1 <= len(v) <= 20):
            raise ValueError("serial (21) must be 1-20 characters")
        return v


class VerificationResponse(BaseModel):
    request_id: str
    verified: bool
    reason: str | None = None

DSCSA/GS1 note: the request carries the same four data elements printed in the unit’s 2D DataMatrix — GTIN (01), serial (21), lot (10), expiry (17) — because the VRS resolves authenticity from that exact product identifier set, not from an internal database key.

Step 2 — Build a mock VRS responder fixture #

httpx.MockTransport intercepts outbound requests before they touch a socket, so the suite runs offline and deterministically. The handler below inspects the credential header and the serial number to decide which of the four scenarios to emulate.

import asyncio
import json

import httpx

KNOWN_SERIALS = {"SERIAL0000000001", "SERIAL0000000002"}


async def mock_vrs_handler(request: httpx.Request) -> httpx.Response:
    """Emulate a GS1 Lightweight Messaging Standard VRS endpoint."""
    if request.headers.get("Authorization") == "Bearer expired-token":
        return httpx.Response(401, json={"error": "credential_expired"})

    payload = json.loads(request.content)
    serial = payload["serial_number"]

    if serial == "TIMEOUT_SERIAL":
        await asyncio.sleep(5)  # exceeds the client's configured SLA timeout

    verified = serial in KNOWN_SERIALS
    body = {
        "request_id": payload["request_id"],
        "verified": verified,
        "reason": None if verified else "serial_not_commissioned",
    }
    return httpx.Response(200, json=body)

DSCSA/GS1 note: returning serial_not_commissioned rather than a bare false mirrors how a production VRS should behave — an unverified result must still be actionable evidence for a Suspect Product Investigation Workflow, not a dead end.

Step 3 — Wrap the client with an SLA-shaped timeout #

The whole point of the harness is to prove the connection meets your integration SLA, so the timeout is not an afterthought — it is the value under test. A short connect timeout and a bounded total timeout mean a hung partner fails the test suite instead of hanging CI.

import time

import httpx
import pytest


@pytest.fixture
def mock_vrs_client() -> httpx.AsyncClient:
    transport = httpx.MockTransport(mock_vrs_handler)
    return httpx.AsyncClient(
        transport=transport,
        base_url="https://vrs.example-partner.com",
        timeout=httpx.Timeout(2.0, connect=1.0),
    )


async def verify_unit(
    client: httpx.AsyncClient,
    request: VerificationRequest,
    credential: str,
) -> tuple[VerificationResponse, float]:
    """Send a verification lookup and return (response, elapsed_seconds)."""
    started = time.perf_counter()
    resp = await client.post(
        "/GS1/lightweight/verify",
        json=request.model_dump(mode="json"),
        headers={"Authorization": f"Bearer {credential}"},
    )
    elapsed = time.perf_counter() - started
    resp.raise_for_status()
    return VerificationResponse.model_validate(resp.json()), elapsed

DSCSA/GS1 note: the 2.0-second total timeout is a deliberate stand-in for whatever round-trip SLA you negotiated during trading-partner onboarding; a production router typically pairs this same timeout with the retry-and-fallback logic covered in Implementing Circuit Breakers for VRS API Calls.

Step 4 — Assert the verified and unverified happy/sad paths #

These two tests are the baseline: a commissioned serial must verify, and a serial the partner never received must not — with a reason your quarantine workflow can act on.

from datetime import date

import pytest


async def test_known_serial_is_verified(mock_vrs_client):
    async with mock_vrs_client as client:
        request = VerificationRequest(
            gtin="00312345678905",
            serial_number="SERIAL0000000001",
            lot_number="LOT42",
            expiration_date=date(2027, 12, 31),
            requestor_gln="0312345000015",
        )
        response, elapsed = await verify_unit(client, request, credential="valid-token")

    assert response.verified is True
    assert response.request_id == request.request_id
    assert elapsed < 2.0  # go-live SLA: sub-2s round trip


async def test_unknown_serial_is_not_verified(mock_vrs_client):
    async with mock_vrs_client as client:
        request = VerificationRequest(
            gtin="00312345678905",
            serial_number="UNKNOWN9999999",
            requestor_gln="0312345000015",
        )
        response, _ = await verify_unit(client, request, credential="valid-token")

    assert response.verified is False
    assert response.reason == "serial_not_commissioned"

DSCSA/GS1 note: asserting on request_id round-tripping unchanged confirms the partner’s VRS is echoing the correlation identifier your non-repudiation log depends on to reconcile the request against its response.

Step 5 — Assert the negative paths: expired credential and SLA breach #

An expired credential must fail loudly with a 401, never silently as a false “not verified,” because those two failure modes require completely different remediation — one is a security incident, the other is a data-quality question.

import httpx
import pytest


async def test_expired_credential_is_rejected(mock_vrs_client):
    async with mock_vrs_client as client:
        request = VerificationRequest(
            gtin="00312345678905",
            serial_number="SERIAL0000000001",
            requestor_gln="0312345000015",
        )
        with pytest.raises(httpx.HTTPStatusError) as excinfo:
            await verify_unit(client, request, credential="expired-token")

    assert excinfo.value.response.status_code == 401


async def test_slow_partner_breaches_timeout_sla(mock_vrs_client):
    async with mock_vrs_client as client:
        request = VerificationRequest(
            gtin="00312345678905",
            serial_number="TIMEOUT_SERIAL",
            requestor_gln="0312345000015",
        )
        with pytest.raises(httpx.TimeoutException):
            await verify_unit(client, request, credential="valid-token")

DSCSA/GS1 note: treating a timeout as a distinct, asserted exception — rather than letting it surface as a generic failure — is what lets the production router decide correctly whether to retry, fail open, or trip a circuit breaker instead of blocking a saleable-unit decision indefinitely.

Verification #

Run the suite with pytest -v tests/test_vrs_integration.py and confirm all four scenarios pass before any partner is marked ready for cutover. Because a single passing run does not prove the connection holds up under repeated calls, add a batch check that measures the 95th-percentile latency across a run of requests, using the standard library’s statistics module:

import statistics

import pytest


async def test_p95_latency_meets_go_live_sla(mock_vrs_client):
    latencies: list[float] = []
    async with mock_vrs_client as client:
        for _ in range(30):
            request = VerificationRequest(
                gtin="00312345678905",
                serial_number="SERIAL0000000001",
                requestor_gln="0312345000015",
            )
            _, elapsed = await verify_unit(client, request, credential="valid-token")
            latencies.append(elapsed)

    p95 = statistics.quantiles(latencies, n=20)[18]  # 95th percentile
    assert p95 < 1.5, f"p95 latency {p95:.3f}s exceeds the 1.5s go-live SLA"

Once the mock suite is green, re-point base_url at the partner’s staging VRS endpoint (with staging credentials, never production ones) and rerun the identical test module unchanged — if the assertions still hold against a real network hop, the connection is ready to move to production, and the CI job should be kept in place to re-run on every credential rotation.

Gotchas & Edge Cases #

• Leading-zero GTINs in test fixtures. Never store or compare a test GTIN as an int; a stripped leading zero silently produces a 13-digit value that no longer matches the partner’s (01) and every assertion against it becomes meaningless.
• Idempotency-key collisions on retry. If a test retries a request after a timeout without generating a fresh request_id, some partner VRS implementations deduplicate the second attempt and return a cached response, masking a real reliability problem as a pass.
• Credential clock skew. A short-lived staging token that expires mid-suite produces a flaky 401 that looks identical to the intentional expired-credential test. Mint a token with a validity window longer than the suite’s total runtime, and use a clearly separate, deliberately invalid token for the negative case.
• Mock drift from the real GS1 Lightweight Messaging Standard schema. A hand-written mock responder can quietly diverge from what the partner’s staging environment actually returns. Re-run the same test module against staging on a schedule, not just once at onboarding, so schema or field-naming changes on the partner side are caught immediately.
• Confusing httpx.ConnectError with httpx.TimeoutException. A DNS failure or refused connection and a slow-but-reachable partner require different remediation; assert on the specific exception type rather than a bare except Exception, or your circuit-breaker logic downstream will treat both failure modes identically.

Related #

• Up to the parent topic: Trading Partner Verification & Onboarding
• Negotiating EPCIS Capability with Trading Partners — the onboarding conversation this test suite validates the outcome of
• Verification Router Service Architecture — the routing, rate-limiting, and non-repudiation design this connection must fit into
• Implementing Circuit Breakers for VRS API Calls — the production-side pattern that consumes the same timeout and failure semantics this suite proves out