From 3629246841ee8a5fd97af3dd2c581c177ed36e18 Mon Sep 17 00:00:00 2001 From: CalamitousFelicitousness Date: Tue, 18 Aug 2026 00:47:48 +0100 Subject: [PATCH] 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 | 315 ++++++++++++++++++++++++++++++++++++ 1 file changed, 315 insertions(+) create mode 100644 test/test-video-registry.py diff --git a/test/test-video-registry.py b/test/test-video-registry.py new file mode 100644 index 000000000..b2ca6f1f8 --- /dev/null +++ b/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)