iGavroche
740b347e92
- ROCmSamplerCustomAdvanced: new attention_backend input (auto/pytorch/ comfy_kitchen_int8/flash), installed on a clone of guider.model_options so the shared ModelPatcher stays clean and sparse-attention overrides still chain on top - Removed dead inputs: precision_mode, batch_optimization, flash_attention, test_steps; fix_empty_latent_channels now passes downscale ratios like stock; disable_pbar actually forwarded - ROCmVAEDecode: deleted unreachable VAE re-dtype machinery (first_stage_model .dtype is None for every shipped VAE); decode follows vae.vae_dtype and never mutates shared weights; fp16 accumulation now goes through apply_rocm_backend_settings (rdna3_5 pre-7.14 guard respected), same fix in ROCmGGUFLoader which also stopped forcing allow_tf32=False - Deleted ROCmSamplerCustomAdvancedBenchmark node (ROCmFluxBenchmark remains) - batch_cap field + *_BATCH_CAP constants, select_precision, dead tail of apply_rocm_backend_settings removed - conftest stubs extended (comfy.audio, model_patcher, ldm attention, folder_paths, interrupt hook): 184 passed, 24 skipped, 0 failed - Docs: node docs, README, Benchmarks, ARCHITECTURE refreshed to current stack (ROCm 7.16 / torch 2.15.0a0+rocm10.1.0)
147 lines
4.4 KiB
Python
147 lines
4.4 KiB
Python
"""
|
|
Integration tests for ROCM workflows
|
|
"""
|
|
import pytest
|
|
import torch
|
|
import sys
|
|
import os
|
|
import json
|
|
|
|
# Add the custom nodes directory to path
|
|
sys.path.insert(0, '/home/nino/ComfyUI/custom_nodes/rocm_ninodes')
|
|
|
|
from rocm_nodes.core.vae import ROCmVAEDecode as ROCmVAEDecodeInstrumented
|
|
|
|
class TestWorkflowIntegration:
|
|
"""Integration tests for ROCM workflows"""
|
|
|
|
def test_flux_workflow_simulation(self, sample_vae):
|
|
"""Test simulation of Flux workflow components"""
|
|
node = ROCmVAEDecodeInstrumented()
|
|
|
|
# Simulate Flux workflow latent
|
|
flux_latent = {
|
|
"samples": torch.randn(1, 4, 64, 64) # 512x512 image
|
|
}
|
|
|
|
result = node.decode(
|
|
vae=sample_vae,
|
|
samples=flux_latent,
|
|
tile_size=768,
|
|
overlap=96,
|
|
use_rocm_optimizations=True
|
|
)
|
|
|
|
assert isinstance(result, tuple)
|
|
assert len(result) == 1
|
|
assert isinstance(result[0], torch.Tensor)
|
|
assert len(result[0].shape) == 4
|
|
|
|
@pytest.mark.skip(reason="Video processing needs more complex implementation")
|
|
def test_wan_video_workflow_simulation(self, sample_vae):
|
|
"""Test simulation of WAN video workflow components"""
|
|
node = ROCmVAEDecodeInstrumented()
|
|
|
|
# Simulate WAN video workflow latent
|
|
wan_latent = {
|
|
"samples": torch.randn(1, 4, 16, 32, 32) # 16 frames, 256x256
|
|
}
|
|
|
|
result = node.decode(
|
|
vae=sample_vae,
|
|
samples=wan_latent,
|
|
tile_size=512,
|
|
overlap=64,
|
|
use_rocm_optimizations=True
|
|
)
|
|
|
|
assert isinstance(result, tuple)
|
|
assert len(result) == 1
|
|
assert isinstance(result[0], torch.Tensor)
|
|
|
|
def test_high_resolution_workflow(self, sample_vae):
|
|
"""Test high resolution workflow simulation"""
|
|
node = ROCmVAEDecodeInstrumented()
|
|
|
|
# Test 1280x1280 image
|
|
high_res_latent = {
|
|
"samples": torch.randn(1, 4, 160, 160) # 1280x1280
|
|
}
|
|
|
|
result = node.decode(
|
|
vae=sample_vae,
|
|
samples=high_res_latent,
|
|
tile_size=1024,
|
|
overlap=128,
|
|
use_rocm_optimizations=True
|
|
)
|
|
|
|
assert isinstance(result, tuple)
|
|
assert len(result) == 1
|
|
assert isinstance(result[0], torch.Tensor)
|
|
|
|
def test_batch_workflow_simulation(self, sample_vae):
|
|
"""Test batch processing workflow simulation"""
|
|
node = ROCmVAEDecodeInstrumented()
|
|
|
|
# Test batch processing
|
|
batch_latent = {
|
|
"samples": torch.randn(4, 4, 32, 32) # Batch size 4
|
|
}
|
|
|
|
result = node.decode(
|
|
vae=sample_vae,
|
|
samples=batch_latent,
|
|
tile_size=512,
|
|
overlap=64,
|
|
use_rocm_optimizations=True
|
|
)
|
|
|
|
assert isinstance(result, tuple)
|
|
assert len(result) == 1
|
|
assert isinstance(result[0], torch.Tensor)
|
|
assert result[0].shape[0] == 4 # Batch size preserved
|
|
|
|
def test_memory_constrained_workflow(self, sample_vae):
|
|
"""Test memory-constrained workflow simulation"""
|
|
node = ROCmVAEDecodeInstrumented()
|
|
|
|
# Test with memory optimizations
|
|
memory_latent = {
|
|
"samples": torch.randn(1, 4, 80, 80) # 640x640
|
|
}
|
|
|
|
result = node.decode(
|
|
vae=sample_vae,
|
|
samples=memory_latent,
|
|
tile_size=512,
|
|
overlap=64,
|
|
use_rocm_optimizations=True
|
|
)
|
|
|
|
assert isinstance(result, tuple)
|
|
assert len(result) == 1
|
|
assert isinstance(result[0], torch.Tensor)
|
|
|
|
def test_workflow_consistency(self, sample_vae):
|
|
"""Test workflow consistency across multiple runs"""
|
|
node = ROCmVAEDecodeInstrumented()
|
|
|
|
# Test consistency
|
|
latent = {"samples": torch.randn(1, 4, 32, 32)}
|
|
|
|
results = []
|
|
for _ in range(3):
|
|
result = node.decode(
|
|
vae=sample_vae,
|
|
samples=latent,
|
|
tile_size=512,
|
|
overlap=64,
|
|
use_rocm_optimizations=True
|
|
)
|
|
results.append(result[0])
|
|
|
|
# All results should have same shape
|
|
for result in results:
|
|
assert result.shape == results[0].shape
|