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

# Override and observe behavior

> Use CEL stages in tests to replace results, inject failures conditionally, and capture samples of real behavior for assertions.

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](/testing/overview): insert → sync → trigger → assert.

## Overrides

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

```python theme={null}
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`](/guides/ai-assisted-testing) 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:

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

```go theme={null}
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](/guides/instrument-a-service#context-functions) 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:

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

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

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

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

### 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:

<CodeGroup>
  ```python Python theme={null}
  import pytest

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

  ```typescript TypeScript theme={null}
  await expect(
    varianz.assertSample('email-sent').within(3_000).exists()
  ).rejects.toThrow();
  ```
</CodeGroup>

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:

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