fastflowtransform.testing

TestFailure

Bases: Exception

Raised when a data-quality check fails.

Source code in src/fastflowtransform/testing.py

class TestFailure(Exception):
    """Raised when a data-quality check fails."""

    # Prevent pytest from collecting this as a test when imported into a test module.
    __test__ = False
    pass

accepted_values

accepted_values(con, table, column, *, values, where=None)

Checks that all non-NULL values of table.column are in the set 'values'.

Source code in src/fastflowtransform/testing.py

def accepted_values(
    con: Any, table: str, column: str, *, values: list, where: str | None = None
) -> bool:
    """
    Checks that all non-NULL values of table.column are in the set 'values'.
    """
    in_list = _sql_list(values or [])
    sql = f"select count(*) from {table} where {column} is not null and {column} not in ({in_list})"
    if where:
        sql += f" and ({where})"
    n = _scalar(con, sql)
    if int(n or 0) > 0:
        # Beispielwerte zeigen
        sample_sql = (
            f"select distinct {column} "
            f"from {table} "
            f"where {column} is not null and {column} not in ({in_list})"
        )
        if where:
            sample_sql += f" and ({where})"
        sample_sql += " limit 5"
        rows = [r[0] for r in _exec(con, sample_sql).fetchall()]
        raise TestFailure(f"{table}.{column} has {n} value(s) outside accepted set; e.g. {rows}")
    return True

not_null

not_null(con, table, column, where=None)

Fails if any non-filtered row has NULL in column.

Source code in src/fastflowtransform/testing.py

def not_null(con: Any, table: str, column: str, where: str | None = None) -> None:
    """Fails if any non-filtered row has NULL in `column`."""
    sql = f"select count(*) from {table} where {column} is null"
    if where:
        sql += f" and ({where})"
    try:
        c = _scalar(con, sql)
    except Exception as e:
        raise _wrap_db_error("not_null", table, column, sql, e) from e
    c = _scalar(con, sql)
    dprint("not_null:", sql, "=>", c)
    if c and c != 0:
        _fail("not_null", table, column, sql, f"has {c} NULL-values")

unique

unique(con, table, column, where=None)

Fails if any duplicate appears in column within the (optionally) filtered set.

Source code in src/fastflowtransform/testing.py

def unique(con: Any, table: str, column: str, where: str | None = None) -> None:
    """Fails if any duplicate appears in `column` within the (optionally) filtered set."""
    sql = (
        "select count(*) from (select {col} as v, "
        "count(*) as c from {tbl}{w} group by 1 having count(*) > 1) as q"
    )
    w = f" where ({where})" if where else ""
    sql = sql.format(col=column, tbl=table, w=w)
    try:
        c = _scalar(con, sql)
    except Exception as e:
        raise _wrap_db_error("unique", table, column, sql, e) from e
    dprint("unique:", sql, "=>", c)
    if c and c != 0:
        _fail("unique", table, column, sql, f"contains {c} duplicates")

reconcile_equal

reconcile_equal(con, left, right, abs_tolerance=None, rel_tolerance_pct=None)

Assert left == right within absolute and/or relative tolerances.

Both sides are dictionaries: {"table": str, "expr": str, "where": Optional[str]}. If both tolerances are omitted, exact equality is enforced.

Source code in src/fastflowtransform/testing.py

def reconcile_equal(
    con: Any,
    left: dict,
    right: dict,
    abs_tolerance: float | None = None,
    rel_tolerance_pct: float | None = None,
) -> None:
    """Assert left == right within absolute and/or relative tolerances.

    Both sides are dictionaries: {"table": str, "expr": str, "where": Optional[str]}.
    If both tolerances are omitted, exact equality is enforced.
    """
    L = _scalar_where(con, left["table"], left["expr"], left.get("where"))
    R = _scalar_where(con, right["table"], right["expr"], right.get("where"))
    if L is None or R is None:
        raise TestFailure(f"One side is NULL (left={L}, right={R})")
    diff = abs(float(L) - float(R))

    # Absolute tolerance check
    if abs_tolerance is not None and diff <= float(abs_tolerance):
        return

    # Relative tolerance check (percentage)
    if rel_tolerance_pct is not None:
        denom = max(abs(float(R)), 1e-12)
        rel = diff / denom
        if (rel * 100.0) <= float(rel_tolerance_pct):
            return

    # If neither tolerance was provided, enforce strict equality via diff==0.
    if abs_tolerance is None and rel_tolerance_pct is None and diff == 0.0:
        return
    raise TestFailure(
        f"Reconcile equal failed: left={L}, right={R}, diff={diff}, "
        f"rel%={(diff / max(abs(float(R)), 1e-12)) * 100:.6f}"
    )

reconcile_ratio_within

reconcile_ratio_within(con, left, right, min_ratio, max_ratio)

Assert min_ratio <= (left/right) <= max_ratio.

Source code in src/fastflowtransform/testing.py

def reconcile_ratio_within(
    con: Any, left: dict, right: dict, min_ratio: float, max_ratio: float
) -> None:
    """Assert min_ratio <= (left/right) <= max_ratio."""
    L = _scalar_where(con, left["table"], left["expr"], left.get("where"))
    R = _scalar_where(con, right["table"], right["expr"], right.get("where"))
    if L is None or R is None:
        raise TestFailure(f"One side is NULL (left={L}, right={R})")
    eps = 1e-12
    denom = float(R) if abs(float(R)) > eps else eps
    ratio = float(L) / denom
    if not (float(min_ratio) <= ratio <= float(max_ratio)):
        raise TestFailure(
            f"Ratio {ratio:.6f} out of bounds [{min_ratio}, {max_ratio}] (L={L}, R={R})"
        )

reconcile_diff_within

reconcile_diff_within(con, left, right, max_abs_diff)

Assert |left - right| <= max_abs_diff.

Source code in src/fastflowtransform/testing.py

def reconcile_diff_within(con: Any, left: dict, right: dict, max_abs_diff: float) -> None:
    """Assert |left - right| <= max_abs_diff."""
    L = _scalar_where(con, left["table"], left["expr"], left.get("where"))
    R = _scalar_where(con, right["table"], right["expr"], right.get("where"))
    if L is None or R is None:
        raise TestFailure(f"One side is NULL (left={L}, right={R})")
    diff = abs(float(L) - float(R))
    if diff > float(max_abs_diff):
        raise TestFailure(f"Abs diff {diff} > max_abs_diff {max_abs_diff} (L={L}, R={R})")

reconcile_coverage

reconcile_coverage(con, source, target, source_where=None, target_where=None)

Assert that every key from source exists in target (anti-join count == 0).

Source code in src/fastflowtransform/testing.py

def reconcile_coverage(
    con: Any,
    source: dict,
    target: dict,
    source_where: str | None = None,
    target_where: str | None = None,
) -> None:
    """Assert that every key from `source` exists in `target` (anti-join count == 0)."""
    s_tbl, s_key = source["table"], source["key"]
    t_tbl, t_key = target["table"], target["key"]
    s_w = f" where {source_where}" if source_where else ""
    t_w = f" where {target_where}" if target_where else ""
    sql = f"""
      with src as (select {s_key} as k from {s_tbl}{s_w}),
           tgt as (select {t_key} as k from {t_tbl}{t_w})
      select count(*) from src s
      left join tgt t on s.k = t.k
      where t.k is null
    """
    missing = _scalar(con, sql)
    dprint("reconcile_coverage:", sql, "=>", missing)
    if missing and missing != 0:
        raise TestFailure(f"Coverage failed: {missing} source keys missing in target")