Override and observe behavior

Every Varianz test does one or both of two things to a VPoint: override it (control what it returns) or observe it (capture what actually happened, without changing it). Both are single CEL expressions inserted into your test session. This guide assumes the test lifecycle from How tests work: insert → sync → trigger → assert.

Overrides

An expression without invoke() fully replaces the function’s result. The real code never runs:

from varianz import cel

# Constant scalar
varianz.insert(cel("pricing/calculate", "42.0"))

# Struct result — field names and types validated against the schema
varianz.insert(cel("payment/charge",
    'ChargeResult { txnId: "test-001", declined: false }'))

All required fields of a struct must be provided, names are case-sensitive, and the type name is the schema’s simple name (ChargeResult, not a package-qualified one). If you’re unsure of the exact names, inspect the schema first — get_vpoint_detail shows types, fields, and examples.

Conditional overrides

args.<param> reads the call’s inputs, and invoke() executes the real function — combine them to intercept only the calls you care about:

# Decline only large charges; everything else behaves normally
varianz.insert(cel("payment/charge",
    'args.input.amount > 1000 '
    '? ChargeResult { txnId: "", declined: true } '
    ': invoke()'))

Transforming the real result

invoke() returns the real result, so you can post-process it:

session.Insert(varianztest.Cel("pricing/calculate-price", "invoke() * 2"))
session.Insert(varianztest.Cel("pricing/calculate-price", "ctx.applyTax(invoke(), 0.1)"))

ctx.<fn>(...) calls a context function the service registered.

Observing with samples

sample("name", value) captures a value for your test and returns it unchanged. Wrapping invoke() gives you a probe — the real function runs, and you get a copy of its result:

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

Samples stream back to the test session, where you assert with the fluent API (identical shape in every test SDK):

varianz.assert_sample("email-sent") \
    .within(timedelta(seconds=10)) \
    .exists()

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

You can sample anything, not just the result — sample("input-amount", args.input.amount) captures an argument, and samples compose with overrides:

# Observe the attempted amount AND decline large charges:
varianz.insert(cel("payment/charge",
    'let amt = sample("attempted-amount", args.input.amount); '
    'amt > 1000.0 ? ChargeResult { txnId: "", declined: true } : invoke()'))

The simple composition to remember: sample("x", invoke()) observes; a bare struct literal overrides; condition ? override : invoke() does both selectively.

Negative assertions

To verify something did not happen — the notification that must not fire after a declined payment — probe it and assert the sample never arrives:

import pytest

with pytest.raises(AssertionError):
    varianz.assert_sample("email-sent") \
        .within(timedelta(seconds=3)) \
        .exists()

await expect(
  varianz.assertSample('email-sent').within(3_000).exists()
).rejects.toThrow();

Keep the negative-window timeout short — it bounds how long the test waits to prove absence.

A complete example

Overriding one service while observing another is the bread-and-butter multi-service test:

from datetime import timedelta
from varianz import cel

def test_payment_decline_blocks_email(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 the checkout flow with the session ID
    http = requests.Session()
    http.headers["x-varianz-id"] = varianz.session_id
    resp = http.post("http://localhost:8080/cart/checkout", json=payload)

    assert resp.status_code == 200

    # The decline must prevent the confirmation email
    import pytest
    with pytest.raises(AssertionError):
        varianz.assert_sample("email-sent").within(timedelta(seconds=3)).exists()

The same test drives services written in any supported language — see Test across services.

Rules worth memorizing

Inputs are always args.<param>. Bare names and $. roots are rejected when the stage compiles.
invoke() takes zero arguments (forward the original args) or exactly the VPoint’s arity (substitute different ones).
sample(name, value) takes exactly two arguments and returns value.
Always await_sync_or_fail between insert and trigger — without it, the stage may not have reached the service yet.
Full syntax, optionals, list predicates, and current limitations: CEL reference.

Getting started

Core concepts

Guides

SDK reference

Testing

Reference

Override and observe behavior

Overrides

Conditional overrides

Transforming the real result

Observing with samples

Negative assertions

A complete example

Rules worth memorizing

​Overrides

​Conditional overrides

​Transforming the real result

​Observing with samples

​Negative assertions

​A complete example

​Rules worth memorizing

Overrides

Conditional overrides

Transforming the real result

Observing with samples

Negative assertions

A complete example

Rules worth memorizing