pytest

Install varianz-pytest (instructions) and the plugin auto-registers via entry points — no conftest.py changes needed. Run pytest with the standard environment: VARIANZ_ENABLED=true, plus VARIANZ_INSECURE_ALLOW_PLAINTEXT=true for a plaintext local coordinator. Endpoint: --varianz-endpoint <url> → VARIANZ_COORDINATOR_ENDPOINT → endpoint in varianz.toml → http://localhost:50051. The deprecated VARIANZ_COORDINATOR_ADDR is still read when the canonical name is unset, with a DeprecationWarning; it is removed before 1.0.

The `varianz` fixture

A fresh VarianzSession per test — created, activated for the test body, and torn down (stages removed) automatically:

from datetime import timedelta
from varianz import cel

def test_payment_decline(varianz):
    # Override payment to decline
    varianz.insert(cel("payment/charge",
        'ChargeResult { txnId: "", declined: true }'))

    # Probe the email service
    varianz.insert(cel("email/send-confirmation",
        'sample("email-sent", invoke())'))

    varianz.await_sync_or_fail(timeout_ms=5000)

    # Trigger via gRPC (session ID in metadata)...
    metadata = [("x-varianz-id", varianz.session_id)]
    response = stub.PlaceOrder(order_request, metadata=metadata)

    # ...or via HTTP — use a Session so headers survive redirects
    http = requests.Session()
    http.headers["x-varianz-id"] = varianz.session_id
    response = http.post("http://frontend:8080/cart/checkout", json=payload)

Python’s requests strips custom headers when following redirects. Always use a requests.Session() with x-varianz-id set on the session object, not per-call headers=.

pytest hides Varianz’s validation warnings by default. They are written to stderr at session teardown, which pytest captures and discards for a passing test — so a green run on a laptop shows nothing at all. Run pytest -rA (or -s / --capture=no) to see them, and put -rA in your CI invocation. A stage that silently never applied is otherwise indistinguishable from one that worked.

Session API

Method / property	Returns	Description
`session_id`	`str`	32-char hex session ID
`insert(stage)`	`str`	Insert a stage, returns its ID
`await_sync(timeout_ms=5000)`	`SyncResult`	Wait for subscriber confirmation
`await_sync_or_fail(timeout_ms=5000)`	`None`	Wait; raise `AssertionError` on timeout only — validation issues are advisory and do not fail the barrier (why); see zero-subscriber semantics
`assert_sample(ident)`	`SampleAssert`	Start a fluent assertion chain
`get_samples()`	`list[dict]`	All buffered samples

Sample assertions

varianz.assert_sample("charge-result") \
    .within(timedelta(seconds=10)) \
    .has_value("txnId", "expected-id") \
    .has_value("nested.field", 42)

# Negative: the sample must NOT arrive
import pytest
with pytest.raises(AssertionError):
    varianz.assert_sample("email-sent") \
        .within(timedelta(seconds=3)) \
        .exists()

In-process calls

The fixture activates its session for the duration of the test body, so code running in the same process as pytest matches your stages with no extra wiring — just call the instrumented function:

def test_direct_call(varianz):
    varianz.insert(cel("pricing/calculate", "42.0"))
    varianz.await_sync_or_fail(timeout_ms=5000)

    assert calculate_price(100.0, 0.1) == 42.0   # session already active

(The session rides a contextvar, so it follows asyncio tasks started from the test as well.)

If the VPoint you’re testing runs in the pytest process and you don’t need session isolation, skip the fixture entirely: vz.attach_stage(...) takes a plain Python callback and needs no coordinator, no session, and no sync barrier. See Local stages.

For creating stages, cel(target, expression) is the only constructor you need — targets and routing filters are described in Naming and routing.

Getting started

Core concepts

Guides

SDK reference

Testing

Reference

The `varianz` fixture

Session API

Sample assertions

In-process calls

​The varianz fixture

​Session API

​Sample assertions

​In-process calls

The `varianz` fixture

Session API

Sample assertions

In-process calls