fastflowtransform.utest

run_unit_specs

run_unit_specs(specs, executor, jenv, only_case=None, *, cache_mode='off', reuse_meta=False)

Execute discovered unit-test specs. Returns the number of failed cases.

Parameters:

Name	Type	Description	Default
cache_mode	str	'off' | 'ro' | 'rw'. Default 'off' for deterministic runs.	'off'
reuse_meta	bool	reserved (no-op).	False

Source code in src/fastflowtransform/utest.py

def run_unit_specs(
    specs: list[UnitSpec],
    executor: Any,
    jenv: Any,
    only_case: str | None = None,
    *,
    cache_mode: str = "off",
    reuse_meta: bool = False,
) -> int:
    """
    Execute discovered unit-test specs. Returns the number of failed cases.

    Args:
        cache_mode: 'off' | 'ro' | 'rw'. Default 'off' for deterministic runs.
        reuse_meta: reserved (no-op).
    """
    cache_mode = _normalize_cache_mode(cache_mode)

    project_dir = _get_project_dir_safe()
    engine_name = _detect_engine_name(executor)
    env_ctx = _make_env_ctx(engine_name)
    cache = _make_cache(project_dir, engine_name)

    ctx = UtestCtx(
        executor=executor,
        jenv=jenv,
        engine_name=engine_name,
        env_ctx=env_ctx,
        cache=cache,
        cache_mode=cache_mode,
    )

    for spec in specs:
        node = REGISTRY.nodes.get(spec.model)
        if not node:
            print(f"⚠️  Model '{spec.model}' not found (in {spec.path})")
            ctx.failures += 1
            continue

        for case in spec.cases:
            if only_case and case.name != only_case:
                continue
            print(f"→ {spec.model} :: {case.name}")

            if not reuse_meta:
                with suppress(Exception):
                    delete_meta_for_node(executor, node.name)

            cand_fp = _fingerprint_case(node, spec, case, ctx)

            # Inputs laden/prüfen (zählt failures selbst)
            ctx.failures += _load_inputs_for_case(executor, spec, case, node)

            # ggf. Skip
            if _maybe_skip_by_cache(node, cand_fp, ctx):
                _read_and_assert(spec, case, ctx)
                continue

            # ausführen + ggf. Cache aktualisieren
            if not _execute_and_update_cache(node, cand_fp, ctx):
                continue

            # Ergebnis prüfen
            _read_and_assert(spec, case, ctx)

    # Cache persistieren (nur rw)
    if ctx.cache and ctx.computed_fps and ctx.cache_mode == "rw":  # pragma: no cover
        ctx.cache.update_many(ctx.computed_fps)
        ctx.cache.save()

    return ctx.failures