test(video): cover the registry invariants and mode derivation

Pins the sentinel contract, row uniqueness, and the equivalence between
dispatch_mode and the ladder it replaced. A row that declares neither a name
marker nor a mapped pipeline class now fails here instead of reaching a runner
that would generate it as text to video.

test/test-video-registry.py

@@ -0,0 +1,315 @@
+#!/usr/bin/env python
+"""
+Offline unit tests for the video model registry in modules.video_models.models_def.
+
+The registry answers two questions about a row: whether it names a loadable model, and how a
+runner should wire its inputs. Both were previously recovered from display-name substrings at
+each call site, which drifted.
+
+Covers:
+
+- the sentinel contract: the None placeholder and the dropdown separators name no model, and
+  every accessor excludes them
+- row uniqueness, so a duplicated entry cannot reach the dropdown twice
+- ``dispatch_mode`` totality: every registered row classifies, so a new row that declares
+  neither a name marker nor a mapped pipeline class fails here rather than generating as t2v
+- ``dispatch_mode`` equivalence against the ladder it replaced, for every row the ladder
+  classified
+- the eight condition rows the ladder did not classify, and the six condition-class rows whose
+  names do declare a mode
+
+No running server required.
+
+Usage:
+    python test/test-video-registry.py
+"""
+
+import os
+import sys
+
+script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, script_dir)
+os.chdir(script_dir)
+
+os.environ['SD_INSTALL_QUIET'] = '1'
+
+# Bootstrap cmd_args before any module that pulls in shared.py.
+import modules.cmd_args  # pylint: disable=wrong-import-position
+import installer  # pylint: disable=wrong-import-position
+orig_argv = sys.argv
+sys.argv = [sys.argv[0]]
+try:
+    modules.cmd_args.parse_args()
+finally:
+    sys.argv = orig_argv
+installer.add_args(modules.cmd_args.parser)
+modules.cmd_args.parsed, _ = modules.cmd_args.parser.parse_known_args([])
+
+from modules.errors import log                       # pylint: disable=wrong-import-position
+from modules.video_models import models_def          # pylint: disable=wrong-import-position
+
+
+results: dict[str, dict] = {}
+
+
+def category(name: str):
+    if name not in results:
+        results[name] = {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []}
+    return name
+
+
+def record(cat: str, passed: bool, name: str, detail: str = ''):
+    status = 'PASS' if passed else 'FAIL'
+    results[cat]['passed' if passed else 'failed'] += 1
+    results[cat]['tests'].append((status, name))
+    msg = f'  {status}: {name}'
+    if detail:
+        msg += f' ({detail})'
+    if passed:
+        log.info(msg)
+    else:
+        log.error(msg)
+
+
+def skip(cat: str, name: str, reason: str):
+    results[cat]['skipped'] += 1
+    results[cat]['tests'].append(('SKIP', name))
+    log.warning(f'  SKIP: {name} ({reason})')
+
+
+def run_test(cat: str, fn):
+    name = fn.__name__
+    try:
+        ok = fn()
+        if ok is False:
+            record(cat, False, name)
+        elif isinstance(ok, str):
+            skip(cat, name, ok)
+        else:
+            record(cat, True, name)
+    except AssertionError as e:
+        record(cat, False, name, str(e))
+    except Exception as e: # pylint: disable=broad-except
+        record(cat, False, name, f'exception: {type(e).__name__}: {e}')
+
+
+def loadable_rows():
+    for engine, rows in models_def.models.items():
+        for row in rows:
+            if models_def.is_model(row):
+                yield engine, row
+
+
+def sentinel_rows():
+    for engine, rows in models_def.models.items():
+        for row in rows:
+            if not models_def.is_model(row):
+                yield engine, row
+
+
+def old_ladder(row):
+    """The name-marker ladder dispatch_mode replaced, as the oracle for equivalence.
+
+    'unknown' stands for the branch in run() that warned and wired nothing; the api ladder
+    reported those same rows as t2v.
+    """
+    if row.workflow is not None:
+        return 'workflow'
+    if 'T2V' in row.name:
+        return 't2v'
+    if 'I2V' in row.name:
+        return 'i2v'
+    if 'FLF2V' in row.name:
+        return 'flf2v'
+    if 'VACE' in row.name:
+        return 'vace'
+    if 'Animate' in row.name:
+        return 'animate'
+    return 'unknown'
+
+
+# ============================================================
+# Sentinel contract and row identity
+# ============================================================
+
+def test_registry_is_populated():
+    assert len(models_def.models) > 0, 'registry failed to build'
+    assert sum(1 for _ in loadable_rows()) > 50, 'registry lost most of its rows'
+
+
+def test_sentinels_name_no_model():
+    for engine, row in sentinel_rows():
+        assert row.name == 'None' or row.name.startswith('─'), f'[{engine}] unexpected sentinel {row.name}'
+        assert row.repo is None, f'[{engine}] sentinel "{row.name}" carries a repo'
+
+
+def test_accessors_exclude_sentinels():
+    for engine in models_def.models:
+        names = models_def.model_names(engine)
+        for name in names:
+            assert name != 'None' and not name.startswith('─'), f'[{engine}] sentinel "{name}" listed as a model'
+    for engine in models_def.engines():
+        assert models_def.model_names(engine), f'[{engine}] listed as an engine with no models'
+
+
+def test_find_rejects_sentinels():
+    for engine, row in sentinel_rows():
+        assert models_def.find(engine, row.name) is None, f'[{engine}] find resolved sentinel "{row.name}"'
+
+
+def test_find_is_case_insensitive():
+    engine, row = next(iter(loadable_rows()))
+    assert models_def.find(engine, row.name) is row
+    assert models_def.find(engine.lower(), row.name.lower()) is row
+    assert models_def.find(engine.upper(), row.name.upper()) is row
+
+
+def test_find_rejects_unknown_names():
+    engine, _row = next(iter(loadable_rows()))
+    assert models_def.find(engine, 'no such model') is None
+    assert models_def.find('no such engine', 'no such model') is None
+    assert models_def.find(engine, None) is None
+
+
+def test_rows_are_unique_within_an_engine():
+    seen = {}
+    for engine, row in loadable_rows():
+        key = (engine, row.name.lower())
+        assert key not in seen, f'[{engine}] duplicate row "{row.name}"'
+        seen[key] = row
+
+
+# ============================================================
+# Mode derivation
+# ============================================================
+
+def test_every_row_classifies():
+    """The extensibility gate: a row declaring neither a marker nor a mapped class fails here."""
+    unknown = [f'[{engine}] {row.name}' for engine, row in loadable_rows() if models_def.dispatch_mode(row) == 'unknown']
+    assert not unknown, f'rows with no mode: {unknown}'
+
+
+def test_mode_matches_the_ladder_it_replaced():
+    """Every row the old ladder classified keeps its answer; only its blind spot changes."""
+    changed = []
+    for engine, row in loadable_rows():
+        old = old_ladder(row)
+        new = models_def.dispatch_mode(row)
+        if old != 'unknown' and old != new:
+            changed.append(f'[{engine}] {row.name}: {old} -> {new}')
+    assert not changed, f'mode changed on rows the ladder already handled: {changed}'
+
+
+def test_the_ladder_blind_spot_is_the_condition_rows():
+    unclassified = [row for _engine, row in loadable_rows() if old_ladder(row) == 'unknown']
+    assert len(unclassified) == 8, f'expected 8 rows the ladder missed, found {len(unclassified)}'
+    for row in unclassified:
+        assert models_def.dispatch_mode(row) == 'condition', f'"{row.name}" resolved as {models_def.dispatch_mode(row)}'
+
+
+def test_named_modes_win_over_the_pipeline_class():
+    """Six LTXConditionPipeline rows are named T2V or I2V and generate as such."""
+    checked = 0
+    for _engine, row in loadable_rows():
+        cls = row.repo_cls if isinstance(row.repo_cls, str) else getattr(row.repo_cls, '__name__', None)
+        if cls not in ('LTXConditionPipeline', 'LTX2ConditionPipeline'):
+            continue
+        if 'T2V' in row.name:
+            assert models_def.dispatch_mode(row) == 't2v', f'"{row.name}" lost its declared mode'
+            checked += 1
+        elif 'I2V' in row.name:
+            assert models_def.dispatch_mode(row) == 'i2v', f'"{row.name}" lost its declared mode'
+            checked += 1
+    assert checked == 6, f'expected 6 condition-class rows declaring a mode, found {checked}'
+
+
+def test_workflow_rows_report_workflow():
+    rows = [row for _engine, row in loadable_rows() if row.workflow is not None]
+    assert rows, 'registry carries no workflow rows'
+    for row in rows:
+        assert models_def.dispatch_mode(row) == 'workflow', f'"{row.name}" resolved as {models_def.dispatch_mode(row)}'
+
+
+def test_flf2v_row_reports_flf2v():
+    rows = [row for _engine, row in loadable_rows() if 'FLF2V' in row.name]
+    assert rows, 'registry carries no flf2v row'
+    for row in rows:
+        assert models_def.dispatch_mode(row) == 'flf2v', f'"{row.name}" resolved as {models_def.dispatch_mode(row)}'
+
+
+def test_mode_resolves_from_a_class_object():
+    """resolve_model synthesizes rows whose repo_cls is a class, not the registry's string."""
+    class WanPipeline: # pylint: disable=too-few-public-methods
+        pass
+    row = models_def.Model(name='local folder with no markers', repo_cls=WanPipeline)
+    assert models_def.dispatch_mode(row) == 't2v'
+
+
+def test_mode_resolves_a_custom_pipeline():
+    row = models_def.Model(name='local folder with no markers', custom='GoogleVeoVideoPipeline')
+    assert models_def.dispatch_mode(row) == 't2v'
+
+
+def test_missing_row_is_unknown():
+    assert models_def.dispatch_mode(None) == 'unknown'
+    assert models_def.dispatch_mode(models_def.Model(name='unregistered')) == 'unknown'
+
+
+def test_class_table_has_no_stale_entries():
+    """Every mapped class is one the registry actually uses, so the table cannot rot unnoticed."""
+    registered = models_def.pipeline_classes()
+    stale = [cls for cls in models_def.CLASS_MODES if cls not in registered]
+    assert not stale, f'class table names pipelines the registry does not carry: {stale}'
+
+
+def run_all():
+    log.warning('=== sentinels and identity ===')
+    cat = category('registry')
+    for fn in [
+        test_registry_is_populated,
+        test_sentinels_name_no_model,
+        test_accessors_exclude_sentinels,
+        test_find_rejects_sentinels,
+        test_find_is_case_insensitive,
+        test_find_rejects_unknown_names,
+        test_rows_are_unique_within_an_engine,
+    ]:
+        run_test(cat, fn)
+
+    log.warning('=== mode derivation ===')
+    cat = category('mode')
+    for fn in [
+        test_every_row_classifies,
+        test_mode_matches_the_ladder_it_replaced,
+        test_the_ladder_blind_spot_is_the_condition_rows,
+        test_named_modes_win_over_the_pipeline_class,
+        test_workflow_rows_report_workflow,
+        test_flf2v_row_reports_flf2v,
+        test_mode_resolves_from_a_class_object,
+        test_mode_resolves_a_custom_pipeline,
+        test_missing_row_is_unknown,
+        test_class_table_has_no_stale_entries,
+    ]:
+        run_test(cat, fn)
+
+    log.warning('=== Results ===')
+    total_passed = 0
+    total_failed = 0
+    total_skipped = 0
+    for cat_name, info in results.items():
+        ok = info['failed'] == 0
+        status = 'PASS' if ok else 'FAIL'
+        log.info(f"  {cat_name}: {info['passed']} passed, {info['failed']} failed, {info['skipped']} skipped [{status}]")
+        total_passed += info['passed']
+        total_failed += info['failed']
+        total_skipped += info['skipped']
+    log.warning(f'Total: {total_passed} passed, {total_failed} failed, {total_skipped} skipped')
+    return total_failed == 0
+
+
+if __name__ == '__main__':
+    import time
+    t0 = time.time()
+    success = run_all()
+    log.warning(f'Total time: {time.time() - t0:.2f}s')
+    sys.exit(0 if success else 1)