109 lines
3.2 KiB
Python
109 lines
3.2 KiB
Python
"""Phase 2 component test: dense Wan2.2-TI2V-5B-Turbo pipeline + LoRA stacking.
|
|
|
|
Verifies:
|
|
- ``Wan22Pipeline`` loads successfully (exercises the real LightX2V
|
|
set_config -> init_runner flow).
|
|
- ``load_loras`` / ``unload_loras`` survive with any user LoRAs at
|
|
``/cache/loras/*.safetensors`` (target='both', dense single DIT).
|
|
|
|
Supports any GGUF quant published in hum-ma/Wan2.2-TI2V-5B-Turbo-GGUF.
|
|
Set ``DIT_QUANT`` to switch (default: ``gguf-Q8_0``).
|
|
|
|
DIT_QUANT=gguf-Q4_K_M docker compose exec voice-chat \
|
|
python -m tests.component.test_02_wan22_loras
|
|
|
|
Requires GPU and a first-run download of the base repo + GGUF DIT.
|
|
If LightX2V isn't installed the test is skipped.
|
|
|
|
Run:
|
|
docker compose exec voice-chat python -m tests.component.test_02_wan22_loras
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import glob
|
|
import os
|
|
import sys
|
|
|
|
from tests.component._common import get_logger
|
|
|
|
log = get_logger("test_02")
|
|
|
|
DIT_QUANT = os.environ.get("DIT_QUANT", "gguf-Q8_0")
|
|
CONFIG_JSON = "/app/configs/lightx2v/wan22_i2v_gguf_5b_turbo.json"
|
|
DIT_REPO = "hum-ma/Wan2.2-TI2V-5B-Turbo-GGUF"
|
|
|
|
|
|
def run():
|
|
try:
|
|
from server.video_models.wan22 import Wan22Pipeline
|
|
except ImportError as e:
|
|
log.error("Wan22Pipeline import failed: %s", e)
|
|
log.warning("SKIP: phase 2 deps not installed")
|
|
sys.exit(0)
|
|
|
|
from server.video import LoRASpec
|
|
|
|
log.info("[case 1] Instantiate Wan22Pipeline "
|
|
"(quant=%s, dit_repo=%s)...", DIT_QUANT, DIT_REPO)
|
|
try:
|
|
pipe = Wan22Pipeline(
|
|
base_repo="Wan-AI/Wan2.2-TI2V-5B",
|
|
dit_repo=DIT_REPO,
|
|
config_json=CONFIG_JSON,
|
|
model_cls="wan2.2",
|
|
resolution=480,
|
|
fps=16,
|
|
dit_quant_scheme=DIT_QUANT,
|
|
t5_quantized=True,
|
|
)
|
|
except Exception as e:
|
|
log.error("FAIL: Wan22Pipeline construction raised: %s", e)
|
|
log.error("Check: LightX2V install, HF cache at /cache/huggingface, "
|
|
"VRAM headroom, and that %s exists inside the container.",
|
|
CONFIG_JSON)
|
|
sys.exit(2)
|
|
log.info(" PASS: pipeline constructed")
|
|
|
|
# --- LoRAs ---
|
|
log.info("[case 2] load_loras with empty list -> no-op")
|
|
pipe.load_loras([])
|
|
log.info(" PASS")
|
|
|
|
lora_files = sorted(glob.glob("/cache/loras/*.safetensors"))
|
|
if not lora_files:
|
|
log.warning("SKIP: no LoRA files found in /cache/loras/")
|
|
log.info("ALL PASSED (partial — LoRA cases skipped)")
|
|
return
|
|
|
|
lora_path = lora_files[0]
|
|
log.info("[case 3] load_loras with one 5B-compatible LoRA (%s)", lora_path)
|
|
specs = [
|
|
LoRASpec(
|
|
path=lora_path,
|
|
weight=1.0,
|
|
target="both",
|
|
name=os.path.basename(lora_path),
|
|
),
|
|
]
|
|
try:
|
|
pipe.load_loras(specs)
|
|
except Exception as e:
|
|
log.error("FAIL: load_loras raised: %s", e)
|
|
log.error("Check: LoRA checkpoint shape matches dense 5B DIT.")
|
|
sys.exit(3)
|
|
log.info(" PASS: LoRAs applied")
|
|
|
|
log.info("[case 4] unload_loras")
|
|
try:
|
|
pipe.unload_loras()
|
|
except Exception as e:
|
|
log.error("FAIL: unload_loras raised: %s", e)
|
|
sys.exit(4)
|
|
log.info(" PASS")
|
|
|
|
log.info("ALL PASSED")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|