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 registry. Endpoint: defaults to http://localhost:50051; override with --varianz-endpoint <url> or VARIANZ_REGISTRY_ADDR.

The `varianz` fixture

A fresh VarianzSession per test, with automatic stage cleanup:

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=.

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 or validation failure; 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

When the code under test runs in the same process as pytest (not behind HTTP/gRPC), activate the session on the current context before calling it:

from varianz import Varianz

Varianz.set_session(varianz.session_id)
result = calculate_price(100.0, 0.1)

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