"""Lightweight post-generation constraint validation.
Runs a small, fixed set of structural checks over an already-generated
dataset - primary-key uniqueness, non-null columns populated, and
foreign-key values referencing generated parent ids - without
re-implementing a general-purpose schema validator. Deliberately scoped
tight: these are the checks that are cheap to verify and meaningful
regardless of entity type, and they are exactly the ones a hand-rolled
generator is most likely to get wrong (duplicate ids, missing required
fields, foreign keys pointing at nothing).
"""
import inspect
from dataclasses import dataclass, field
from typing import Any, Dict, List, Set, Tuple
from test_data_workbench.adaptation.generator_factory import GeneratorFactory
from test_data_workbench.core.models import SchemaInfo, Table
[docs]
@dataclass
class ConstraintCheckResult:
"""Outcome of running :func:`check_constraints` over a generated
dataset. Each check is one (table, column) or (table, relationship)
assertion - not one check per record - so the counts stay meaningful
for a "N checks passed" style summary."""
checks_passed: int = 0
checks_failed: int = 0
violations: List[str] = field(default_factory=list)
@property
def total_checks(self) -> int:
return self.checks_passed + self.checks_failed
@property
def all_passed(self) -> bool:
return self.checks_failed == 0
[docs]
def summary_line(self) -> str:
"""One-line summary suitable for CLI output."""
if self.checks_failed:
return (
f"Constraint check: {self.checks_passed} checks passed, "
f"{self.checks_failed} failed"
)
return f"Constraint check: {self.checks_passed} checks passed"
[docs]
def check_constraints(
schema: SchemaInfo, dataset: Dict[str, List[Dict[str, Any]]]
) -> ConstraintCheckResult:
"""Validate a generated dataset against the schema's discoverable
constraints.
Args:
schema: The analyzed schema (tables, columns, relationships).
dataset: table name -> list of generated record dicts, e.g. the
output of calling a generated ``*Generator.generate(n)`` for
each table.
Checks performed:
- Primary-key uniqueness: no duplicate non-null primary key
values within a table.
- Not-null: every non-nullable column is populated (not None)
in every record.
- Foreign-key range: for every declared relationship, every
non-null foreign key value in the child table is found among
the referenced column's generated values in the parent table.
A table present in ``schema`` but absent from ``dataset`` is simply
skipped - there is nothing to check. Likewise, a relationship is
skipped when either side of it has no data in ``dataset``, since
there is no parent id set to check the foreign key against.
"""
passed = 0
failed = 0
violations: List[str] = []
# (table_name, column_name) -> set of generated primary key values,
# used below to check foreign keys land inside a real parent id set.
pk_values: Dict[Tuple[str, str], Set[Any]] = {}
for table in schema.tables:
records = dataset.get(table.name)
if not records:
continue
for pk_col in table.primary_keys:
ok, message = _check_pk_uniqueness(table.name, pk_col.name, records)
passed, failed = _tally(ok, passed, failed, violations, message)
pk_values[(table.name, pk_col.name)] = {
r.get(pk_col.name) for r in records if r.get(pk_col.name) is not None
}
for col in table.columns:
if col.nullable:
continue
ok, message = _check_not_null(table.name, col.name, records)
passed, failed = _tally(ok, passed, failed, violations, message)
for rel in schema.relationships:
from_records = dataset.get(rel.from_table)
to_values = pk_values.get((rel.to_table, rel.to_column))
if not from_records or to_values is None:
continue
ok, message = _check_fk_in_range(rel, from_records, to_values)
passed, failed = _tally(ok, passed, failed, violations, message)
return ConstraintCheckResult(checks_passed=passed, checks_failed=failed, violations=violations)
def _tally(ok: bool, passed: int, failed: int, violations: List[str], message: str) -> Tuple[int, int]:
if ok:
return passed + 1, failed
violations.append(message)
return passed, failed + 1
def _check_pk_uniqueness(table_name: str, column_name: str, records: List[Dict[str, Any]]):
values = [r.get(column_name) for r in records if r.get(column_name) is not None]
duplicates = len(values) - len(set(values))
if duplicates == 0:
return True, f"{table_name}.{column_name}: primary key is unique"
return False, f"{table_name}.{column_name}: {duplicates} duplicate primary key value(s)"
def _check_not_null(table_name: str, column_name: str, records: List[Dict[str, Any]]):
missing = sum(1 for r in records if r.get(column_name) is None)
if missing == 0:
return True, f"{table_name}.{column_name}: not-null constraint satisfied"
return False, f"{table_name}.{column_name}: {missing} record(s) missing a required value"
def _check_fk_in_range(rel, from_records: List[Dict[str, Any]], to_values: Set[Any]):
invalid = sum(
1 for r in from_records
if r.get(rel.from_column) is not None and r.get(rel.from_column) not in to_values
)
if invalid == 0:
return True, (
f"{rel.from_table}.{rel.from_column}: all foreign keys reference "
f"{rel.to_table}.{rel.to_column}"
)
return False, (
f"{rel.from_table}.{rel.from_column}: {invalid} value(s) do not reference "
f"an existing {rel.to_table}.{rel.to_column}"
)
[docs]
def generate_sample_dataset(
schema: SchemaInfo,
generated_code: Dict[str, str],
sample_size: int = 20,
) -> Dict[str, List[Dict[str, Any]]]:
"""Execute generated generator source, in dependency order, to build
a small in-memory sample dataset - purely so :func:`check_constraints`
has something real to check. Nothing here is written to disk.
Parent tables are generated first so their primary key values can be
threaded through as ``reference_data`` to child generators that
accept it (the order/review-style templates), matching how
:meth:`GeneratorFactory._create_order_generator` and friends expect
to be wired up.
Best effort by design: a table whose generator source fails to exec,
defines no ``*Generator`` class, or raises while generating, is
simply skipped rather than aborting the whole check - a single bad
generator should not take down deploy's constraint reporting.
"""
factory = GeneratorFactory()
ordered_tables = factory._sort_by_dependencies(schema.tables, schema.relationships)
dataset: Dict[str, List[Dict[str, Any]]] = {}
reference_data: Dict[str, List[Any]] = {}
for table in ordered_tables:
code = generated_code.get(table.name)
if not code:
continue
try:
namespace: Dict[str, Any] = {}
exec(compile(code, f"<{table.name}>", "exec"), namespace)
generator_classes = [
value for value in namespace.values()
if isinstance(value, type) and value.__name__.endswith("Generator")
]
if not generator_classes:
continue
generator_cls = generator_classes[0]
init_params = inspect.signature(generator_cls.__init__).parameters
instance = (
generator_cls(reference_data=reference_data)
if "reference_data" in init_params
else generator_cls()
)
records = instance.generate(sample_size)
except Exception:
continue
dataset[table.name] = records
_register_reference_data(table, records, schema, factory, reference_data)
return dataset
def _register_reference_data(
table: Table,
records: List[Dict[str, Any]],
schema: SchemaInfo,
factory: GeneratorFactory,
reference_data: Dict[str, List[Any]],
) -> None:
"""Make this table's primary-key values available to any
not-yet-generated child table's foreign key lookup.
Generated child code calls ``self._get_reference_id(guessed_name)``
where ``guessed_name`` comes from
:meth:`GeneratorFactory._infer_reference_table` applied to the
child's own FK column name - which does not always match the real
parent table name (e.g. irregular plurals). Registering the values
under both the real table name and every guess a real child
relationship would produce keeps the reference data usable without
having to fix that pluralization heuristic here.
"""
for pk_col in table.primary_keys:
pk_values = [r.get(pk_col.name) for r in records if r.get(pk_col.name) is not None]
if not pk_values:
continue
reference_data.setdefault(table.name, pk_values)
for rel in schema.relationships:
if rel.to_table == table.name and rel.to_column == pk_col.name:
guess = factory._infer_reference_table(rel.from_column)
reference_data.setdefault(guess, pk_values)