fastflowtransform.testing.discovery

discover_sql_tests

discover_sql_tests(project_dir)

Discover SQL-based DQ tests under tests/*/.ff.sql.

Each file must

• start with a {{ config(type="...", params=[...]) }} block
• contain exactly one SELECT that returns a scalar "violation count".

For each test we

• parse config(type=..., params=[...])
• build a Pydantic params model from params (if provided)
• register the test via register_sql_test(...)

Runtime behaviour

• params from project.yml are validated against that model (unknown keys → error)
• Jinja template is rendered with: table, column, params, where (where = params.get('where'), always present)
• the SQL is executed and interpreted as "violation count".

Source code in src/fastflowtransform/testing/discovery.py

def discover_sql_tests(project_dir: Path) -> None:
    """
    Discover SQL-based DQ tests under tests/**/*.ff.sql.

    Each file must:
      - start with a {{ config(type="...", params=[...]) }} block
      - contain exactly one SELECT that returns a scalar "violation count".

    For each test we:
      - parse config(type=..., params=[...])
      - build a Pydantic params model from `params` (if provided)
      - register the test via `register_sql_test(...)`

    Runtime behaviour:
      - params from project.yml are validated against that model (unknown keys → error)
      - Jinja template is rendered with:
            table, column, params, where
        (where = params.get('where'), always present)
      - the SQL is executed and interpreted as "violation count".
    """
    tests_dir = project_dir / "tests"
    if not tests_dir.exists():
        return

    # Ensure the Jinja env is initialized, even though register_sql_test
    # will call REGISTRY.get_env() again internally.
    REGISTRY.get_env()

    for path in sorted(tests_dir.rglob("*.ff.sql")):
        try:
            text = path.read_text(encoding="utf-8")
        except Exception as exc:
            logger.error("Failed to read SQL test file %s: %s", path, exc)
            continue

        cfg = _parse_test_config(text, path)
        tname = cfg.get("type")
        if not tname or not isinstance(tname, str):
            logger.warning(
                "%s: SQL test file missing config(type='...'); skipping",
                path,
            )
            continue

        try:
            params_model = _build_params_model_from_config(tname, path, cfg)
        except ModelConfigError:
            # Config errors should be fatal, like for models
            raise

        register_sql_test(
            kind=tname,
            path=path,
            params_model=params_model,
        )
        logger.debug("Registered SQL DQ test '%s' from %s", tname, path)

discover_python_tests

discover_python_tests(project_dir)

Discover Python-based DQ tests under tests/*/.ff.py.

The files should define functions decorated with @dq_test("type_name"). Importing the module is enough; the decorator will register the runner.

Source code in src/fastflowtransform/testing/discovery.py

def discover_python_tests(project_dir: Path) -> None:
    """
    Discover Python-based DQ tests under tests/**/*.ff.py.

    The files should define functions decorated with @dq_test("type_name").
    Importing the module is enough; the decorator will register the runner.
    """
    tests_dir = project_dir / "tests"
    if not tests_dir.exists():
        return

    for path in sorted(tests_dir.rglob("*.ff.py")):
        # Reuse the Registry's module loader so we get consistent behaviour.
        try:
            REGISTRY._load_py_module(path)
        except ModuleLoadError:
            # Fail fast: broken DQ test modules should not be silently ignored.
            raise
        except Exception as exc:
            # Other exceptions: surface as a clear message
            raise ModuleLoadError(f"Failed to import DQ test module {path}: {exc}") from exc