Skip to main content

testing.assert_contract()

Assert that a node’s contract is proved, and that it holds on data.

Usage

Source

testing.assert_contract(node: Node, data: None = None, expected: None = None, engine: str = 'duckdb') -> None
 
testing.assert_contract(node: Node, data: Mapping[str, Rows], expected: Rows | None = None, engine: str = 'duckdb') -> pa.Table

Every contracted transform the node is built from must have its postconditions proved and every precondition on another node’s result proved from that node’s guarantees. A transform with on_unproved="error" (the default) already raises when the node is built; this catches "warn" and "check", which let an unproved contract through.

With data, the node runs on those rows: they must meet the preconditions (a violation is a mistake in the test data, and is reported with the offending rows), and the output must meet the postconditions on engine. With expected, the output must also be those rows, compared as bags (doubles to 12 significant digits), with the missing and unexpected rows listed when it is not.

Parameters

node: Node

A node: a source or node piped into contracted transforms (source(...).pipe(paid)).

data: Mapping[str, Rows] | None = None

The rows of each table the node reads, by name: its sources, and the stored nodes it reads (Node.write). Default: none, which only checks the proof.

expected: Rows | None = None

The rows the node should give on data: dicts by column name, tuples in column order, or an Arrow table.

engine: str = "duckdb"
Where to run the node: "duckdb" (the default), "polars", "datafusion" or "spark".

Returns

pa.Table | None
The node’s output on data, or None without data.

Raises

AssertionError

The contract is not proved, the data violates a precondition, or the output violates a postcondition or differs from expected.

ValueError

data misses a table the node reads, or has one it does not read.

TypeError
node is not a node, or expected is given without data.

Examples

import sverify as sv
from ibis import _
from sverify.testing import assert_contract, source

@sv.transform(
    requires={"orders": [_.amount >= 0]},
    ensures=[_.doubled >= 0],
)
def doubled(orders):
    return orders.select(orders.id, doubled=orders.amount * 2)

orders = source([], {"id": "int64", "amount": "int64"}, name="orders")
assert_contract(orders.pipe(doubled))
assert_contract(
    orders.pipe(doubled),
    data={"orders": [(1, 30), (2, 0)]},
    expected=[(1, 60), (2, 0)],
)