darth-veitcher
bb9ede6da9
- Refactored all nodes to accept both manual string input and connected models_json - Removed rigid enum-based dropdowns in favor of flexible STRING inputs - Added optional models_json input to Select, Load, and Unload nodes - Nodes now parse JSON and update cache when models_json is connected - Supports Ollama's auto-pull behavior for models not yet downloaded - Updated category from "Ollama/Enum" to "Ollama" for cleaner UI organization Benefits: - No UI refresh required for manual model entry - Automatic model pulling when typing new model names - Still supports connected workflow pattern (Refresh → Select → Load) - Maximum flexibility - users choose their preferred workflow Test updates: - Fixed all tests for new flexible input signatures - Added comprehensive warning filters for pytest mock internals - All 28 tests passing with 0 warnings Documentation: - Updated README with both usage patterns (connected and manual) - Added example workflows for different use cases - Clarified benefits of each approach
41 lines
998 B
Python
41 lines
998 B
Python
"""Tests for async utilities module."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from comfyui_ollama_model_manager.async_utils import run_async
|
|
|
|
|
|
async def sample_coroutine(value):
|
|
"""A simple coroutine for testing."""
|
|
await asyncio.sleep(0.001)
|
|
return value * 2
|
|
|
|
|
|
def test_run_async_basic():
|
|
"""Test running a simple coroutine."""
|
|
result = run_async(sample_coroutine(5))
|
|
assert result == 10
|
|
|
|
|
|
def test_run_async_with_exception():
|
|
"""Test that exceptions are propagated correctly."""
|
|
|
|
async def failing_coroutine():
|
|
await asyncio.sleep(0.001)
|
|
raise ValueError("Test error")
|
|
|
|
with pytest.raises(ValueError, match="Test error"):
|
|
run_async(failing_coroutine())
|
|
|
|
|
|
def test_run_async_returns_value():
|
|
"""Test that return values are passed through correctly."""
|
|
|
|
async def return_dict():
|
|
return {"status": "success", "data": [1, 2, 3]}
|
|
|
|
result = run_async(return_dict())
|
|
assert result == {"status": "success", "data": [1, 2, 3]}
|