Table Validation Implementation Plan¶

import narwhals as nw
import polars as pl
import pytest

from tusk.database import TableSchema
from tusk.exceptions import TuskError, ValidationError
from tusk.validation import (
    CHECKS,
    check_unique_primary_key,
    resolve_checks,
    validate_table,
)


def frame(values):
    """A one-column lazy frame keyed by 'id'."""
    return nw.from_native(pl.LazyFrame({"id": values}))


def schema(primary_key="id"):
    return TableSchema("customers", primary_key, None, {"id": nw.Int64})


def test_duplicate_primary_key_is_reported():
    with pytest.raises(ValidationError) as excinfo:
        check_unique_primary_key(frame([7, 7, 12, 12, 19, 19, 1]), schema())
    message = str(excinfo.value)
    assert "primary_key 'id'" in message
    assert "'customers'" in message
    assert "7 rows" in message
    assert "4 distinct" in message
    assert "7" in message and "12" in message and "19" in message


def test_unique_primary_key_passes():
    check_unique_primary_key(frame([1, 2, 3]), schema())


def test_empty_table_passes():
    check_unique_primary_key(frame([]), schema())


def test_table_without_a_primary_key_is_skipped():
    # add_table already warned about this; failing here would make
    # db.validate() unusable on any database holding a keyless table.
    check_unique_primary_key(frame([7, 7]), schema(primary_key=None))


def test_repeated_nulls_are_not_unique():
    with pytest.raises(ValidationError, match="not unique"):
        check_unique_primary_key(frame([1, None, None]), schema())


def test_a_single_null_passes_the_uniqueness_check():
    # A lone null key is a nullability defect, not a uniqueness one.
    check_unique_primary_key(frame([1, 2, None]), schema())


def test_validation_error_is_a_tusk_error():
    assert issubclass(ValidationError, TuskError)


def test_validate_table_runs_the_registry_by_default():
    with pytest.raises(ValidationError):
        validate_table(frame([7, 7]), schema())


def test_validate_table_runs_nothing_when_false():
    validate_table(frame([7, 7]), schema(), False)


def test_selector_forms_all_select_the_check():
    for selector in (True, "unique_primary_key", ["unique_primary_key"]):
        assert resolve_checks(selector) == (CHECKS["unique_primary_key"],)


def test_empty_selectors_select_nothing():
    assert resolve_checks(False) == ()
    assert resolve_checks([]) == ()


def test_a_repeated_name_runs_the_check_once():
    assert resolve_checks(["unique_primary_key", "unique_primary_key"]) == (
        CHECKS["unique_primary_key"],
    )


def test_checks_run_in_registry_order_not_argument_order(monkeypatch):
    calls = []
    ordered = {
        "first": lambda f, s: calls.append("first"),
        "second": lambda f, s: calls.append("second"),
    }
    monkeypatch.setattr("tusk.validation.CHECKS", ordered)
    validate_table(frame([1]), schema(), ["second", "first"])
    assert calls == ["first", "second"]


def test_an_unknown_check_name_raises_value_error():
    with pytest.raises(ValueError) as excinfo:
        resolve_checks("uniqe_primary_key")
    message = str(excinfo.value)
    assert "uniqe_primary_key" in message
    assert "unique_primary_key" in message


def test_an_unknown_check_name_is_not_a_validation_error():
    # `except ValidationError` must never swallow a typo.
    with pytest.raises(ValueError) as excinfo:
        resolve_checks(["unique_primary_key", "nope"])
    assert not isinstance(excinfo.value, TuskError)

class ValidationError(TuskError):
    """Raised when a validation check finds a defect in a table's data.

    Distinct from :class:`SchemaError`, which reports a malformed database
    before any row is read. A ``ValidationError`` means the declarations are
    well-formed but the data contradicts them.

    An unknown check *name* is a caller mistake rather than a data defect and
    raises :class:`ValueError`, so ``except ValidationError`` never silently
    swallows a typo.
    """

"""Checks that confirm a table's declarations hold against its data.

The rest of the schema layer takes declarations on trust: naming a column as
``primary_key`` asserts that it identifies a row, and nothing confirms it.
These checks spend real queries to find out. Nothing here runs unless the user
asks, through :meth:`tusk.Database.validate` or ``add_table(validate=…)``.

Adding a check is a new function plus a new entry in :data:`CHECKS`; no call
signature changes and both entry points pick it up.
"""

from __future__ import annotations

from collections.abc import Callable, Iterable, Mapping
from typing import TYPE_CHECKING

import narwhals as nw

from tusk.exceptions import ValidationError

if TYPE_CHECKING:  # pragma: no cover - import cycle guard
    from tusk.database import TableSchema

Check = Callable[["nw.LazyFrame", "TableSchema"], None]
"""A check: given a table's frame and schema, raise or return."""

Checks = bool | str | Iterable[str]
"""Selects checks: every check, none, one by name, or several by name."""

_SAMPLE_SIZE = 5
_COUNT = "__tusk_count"


def check_unique_primary_key(frame: nw.LazyFrame, schema: TableSchema) -> None:
    """Confirm the declared primary key holds no repeated value.

    Runs in two phases so the passing case stays cheap: one aggregate compares
    the row count to the distinct-value count, and only a disagreement pays for
    a second query naming offending values.

    A table with no ``primary_key`` is skipped rather than failed --
    ``add_table`` already warned about it, and failing here would make
    :meth:`tusk.Database.validate` unusable on any database that deliberately
    holds a keyless table.

    Nulls count as one distinct value, so repeated nulls fail this check while
    a single null passes it. A lone null key is a nullability defect, not a
    uniqueness one.

    Args:
        frame: The table's lazy frame.
        schema: The table's schema, naming the column to check.

    Raises:
        ValidationError: If the primary key column holds a repeated value.
    """
    key = schema.primary_key
    if key is None:
        return

    counts = frame.select(total=nw.len(), distinct=nw.col(key).n_unique()).collect()
    total, distinct = counts["total"].item(), counts["distinct"].item()
    if total == distinct:
        return

    duplicates = (
        frame.group_by(key)
        .agg(nw.len().alias(_COUNT))
        .filter(nw.col(_COUNT) > 1)
        .sort(key)
        .head(_SAMPLE_SIZE)
        .collect()
    )
    raise ValidationError(
        f"primary_key {key!r} of {schema.name!r} is not unique: "
        f"{total} rows, {distinct} distinct values, "
        f"e.g. {duplicates[key].to_list()}"
    )


CHECKS: Mapping[str, Check] = {
    "unique_primary_key": check_unique_primary_key,
}
"""Every available check, by name. Iteration order is the order checks run."""


def resolve_checks(selector: Checks) -> tuple[Check, ...]:
    """Turn a selector into the checks it names.

    Selected checks are deduplicated and returned in :data:`CHECKS` order, not
    argument order, so which failure a caller sees never depends on how they
    happened to type the list.

    Args:
        selector: ``True`` for every check, ``False`` (or an empty iterable)
            for none, a string for one by name, or an iterable of names.

    Returns:
        The selected checks, in registry order.

    Raises:
        ValueError: If a name is not in :data:`CHECKS`.
    """
    if isinstance(selector, bool):
        return tuple(CHECKS.values()) if selector else ()

    names = {selector} if isinstance(selector, str) else set(selector)
    unknown = sorted(names - set(CHECKS))
    if unknown:
        listed = ", ".join(repr(name) for name in unknown)
        available = ", ".join(repr(name) for name in CHECKS)
        raise ValueError(f"unknown check {listed}; available checks: {available}")
    return tuple(check for name, check in CHECKS.items() if name in names)


def validate_table(
    frame: nw.LazyFrame, schema: TableSchema, checks: Checks = True
) -> None:
    """Run the selected checks against one table.

    Args:
        frame: The table's lazy frame.
        schema: The table's schema.
        checks: Which checks to run. See :func:`resolve_checks`.

    Raises:
        ValidationError: If a check finds a defect. Raised by the first check
            that fails; later checks do not run.
        ValueError: If ``checks`` names a check that does not exist.
    """
    for check in resolve_checks(checks):
        check(frame, schema)

@pytest.fixture
def spy(monkeypatch):
    """Replace the registry with a recorder, so plumbing is observable."""
    calls = []
    monkeypatch.setattr(
        "tusk.validation.CHECKS",
        {"unique_primary_key": lambda f, s: calls.append(s.name)},
    )
    return calls


def dupes():
    return pl.LazyFrame({"id": [7, 7, 12], "v": [1.0, 2.0, 3.0]})


def test_add_table_does_not_validate_by_default(spy):
    db = tusk.Database("x").add_table("t", dupes(), primary_key="id")
    assert spy == []
    assert db.table_names == ("t",)


def test_add_table_validates_when_asked(spy):
    tusk.Database("x").add_table("t", dupes(), primary_key="id", validate=True)
    assert spy == ["t"]


def test_add_table_accepts_every_selector_form():
    for selector in (True, "unique_primary_key", ["unique_primary_key"]):
        with pytest.raises(ValidationError, match="not unique"):
            tusk.Database("x").add_table(
                "t", dupes(), primary_key="id", validate=selector
            )


def test_a_failed_add_table_leaves_the_database_unchanged():
    db = tusk.Database("x")
    with pytest.raises(ValidationError):
        db.add_table("t", dupes(), primary_key="id", validate=True)
    assert db.table_names == ()
    with pytest.raises(tusk.exceptions.SchemaError):
        db.frame("t")


def test_schema_errors_still_precede_validation():
    # A primary_key that is not a column must stay a SchemaError; querying a
    # column that does not exist would report it as something else.
    with pytest.raises(tusk.exceptions.SchemaError, match="nope"):
        tusk.Database("x").add_table("t", dupes(), primary_key="nope", validate=True)


def test_database_validate_runs_every_table_in_insertion_order(spy):
    db = (
        tusk.Database("x")
        .add_table("a", pl.LazyFrame({"id": [1]}), primary_key="id")
        .add_table("b", pl.LazyFrame({"id": [2]}), primary_key="id")
    )
    assert db.validate() is db
    assert spy == ["a", "b"]


def test_database_validate_reports_the_offending_table():
    db = (
        tusk.Database("x")
        .add_table("a", pl.LazyFrame({"id": [1, 2]}), primary_key="id")
        .add_table("b", dupes(), primary_key="id")
    )
    with pytest.raises(ValidationError, match="'b'"):
        db.validate()


def test_database_validate_false_is_a_no_op(spy):
    db = tusk.Database("x").add_table("t", dupes(), primary_key="id")
    assert db.validate(False) is db
    assert spy == []


def test_database_validate_skips_a_keyless_table():
    with pytest.warns(tusk.exceptions.MissingPrimaryKeyWarning):
        db = tusk.Database("x").add_table("t", dupes())
    assert db.validate() is db


def test_unknown_check_names_reach_both_entry_points():
    with pytest.raises(ValueError, match="unknown check"):
        tusk.Database("x").add_table(
            "t", pl.LazyFrame({"id": [1]}), primary_key="id", validate="nope"
        )
    db = tusk.Database("x").add_table("t", pl.LazyFrame({"id": [1]}), primary_key="id")
    with pytest.raises(ValueError, match="unknown check"):
        db.validate("nope")

from tusk.validation import Checks, validate_table

    def add_table(
        self,
        name: str,
        table: Any,
        primary_key: str | None = None,
        row_creation_time: str | None = None,
        *,
        validate: Checks = False,
    ) -> Database:

        self._frames[name] = lazy
        self._schemas[name] = TableSchema(name, primary_key, row_creation_time, dtypes)
        return self

        schema = TableSchema(name, primary_key, row_creation_time, dtypes)
        validate_table(lazy, schema, validate)

        self._frames[name] = lazy
        self._schemas[name] = schema
        return self

    def validate(self, checks: Checks = True) -> Database:
        """Run validation checks against every table in the database.

        Tables are checked in insertion order and the first failure raises, so
        a database with several defects reports the earliest one.

        Unlike the rest of the schema layer, this reads rows: each check runs
        real queries against the data.

        Args:
            checks: Which checks to run. ``True`` (the default) runs every
                check; ``False`` runs none; a name or list of names runs
                those. See :func:`tusk.validation.validate_table`.

        Returns:
            This database, to allow chaining.

        Raises:
            ValidationError: If a check finds a defect in a table's data.
            ValueError: If ``checks`` names a check that does not exist.
        """
        for name, schema in self._schemas.items():
            validate_table(self._frames[name], schema, checks)
        return self

def test_uniqueness_check_runs_on_duckdb(duck_db):
    db, con = duck_db
    assert db.validate() is db

    con.execute("CREATE TABLE dupes AS SELECT * FROM (VALUES (1),(1),(2)) t(id)")
    with pytest.raises(tusk.exceptions.ValidationError, match="not unique"):
        db.add_table("dupes", con.table("dupes"), primary_key="id", validate=True)
    assert "dupes" not in db.table_names