> ## Documentation Index
> Fetch the complete documentation index at: https://docs.varianz.io/llms.txt
> Use this file to discover all available pages before exploring further.

# pytest

> Test with the varianz-pytest plugin: the varianz fixture, stage insertion, sync, sample assertions, and HTTP/gRPC trigger patterns.

Install `varianz-pytest` ([instructions](/installation#python)) and the plugin auto-registers via entry points — no `conftest.py` changes needed. Run pytest with the [standard environment](/reference/configuration): `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:

```python theme={null}
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)
```

<Warning>
  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=`.
</Warning>

## 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](/testing/overview#zero-subscribers) |
| `assert_sample(ident)`                | `SampleAssert` | Start a fluent assertion chain                                                                                                     |
| `get_samples()`                       | `list[dict]`   | All buffered samples                                                                                                               |

## Sample assertions

```python theme={null}
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:

```python theme={null}
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](/reference/naming-and-routing).