mirror of
https://git.datalinker.icu/ltdrdata/ComfyUI-Manager
synced 2025-12-10 14:45:20 +08:00
2b778fd42c
Fix two critical issues causing test failures in test_installed_api_enabled_priority.py: 1. API Parameter Mismatch (Primary Issue): - Tests were using outdated parameter names (node_name, install_type) - Server expects: id, version, selected_version - Fixed all 6+ parameter usages in test file - Impact: test_installed_api_shows_only_enabled_when_both_exist now passes 2. Cross-Type Package Matching (manager_core.py:1861-1873): - API incorrectly returned both enabled CNR and disabled Nightly packages - Root cause: Logic only checked same-type matches (CNR→CNR, Nightly→Nightly) - Added cross-type matching: disabled Nightly aux_id ↔ enabled CNR cnr_id - Extract package name from aux_id, compare with cnr_id - Impact: Disabled packages correctly excluded when enabled version exists Infrastructure Improvements: - Added monitor_test.sh for background process monitoring - Updated run_automated_tests.sh to use tee for output forwarding - Added test_installed_api_shows_disabled_when_no_enabled_exists to skip list Test Results: - Before: 60/63 tests passing (95.2%), 7/10 environments - After: 61/63 tests passing (96.8%), 8/10 environments - Improvement: +1.6% pass rate, +14.3% environment success rate Remaining Issues (test-specific, not code bugs): - test_installed_api_cnr_priority_when_both_disabled: Nightly installation issue - test_installed_api_shows_disabled_when_no_enabled_exists: Session fixture interference Documentation: - Complete troubleshooting session documented in .claude/livecontext/ 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1029 lines
34 KiB
Python
1029 lines
34 KiB
Python
"""
|
|
Pytest configuration for glob API tests.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
# Add project root to Python path
|
|
project_root = Path(__file__).parent.parent.parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
|
|
def pytest_configure(config):
|
|
"""Configure pytest with custom markers."""
|
|
config.addinivalue_line("markers", "slow: marks tests as slow (deselect with '-m \"not slow\"')")
|
|
config.addinivalue_line(
|
|
"markers", "requires_server: marks tests that require a running ComfyUI server"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "priority_high: marks tests as high priority for comprehensive test suites"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "priority_medium: marks tests as medium priority"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "priority_low: marks tests as low priority"
|
|
)
|
|
config.addinivalue_line(
|
|
"markers", "complex_scenario: marks tests for complex multi-version scenarios"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def server_url():
|
|
"""Get server URL from environment or use default."""
|
|
# Support both TEST_SERVER_PORT (for parallel tests) and COMFYUI_TEST_URL (for custom URLs)
|
|
port = os.environ.get("TEST_SERVER_PORT", "8188")
|
|
return os.environ.get("COMFYUI_TEST_URL", f"http://127.0.0.1:{port}")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def custom_nodes_path():
|
|
"""Get custom nodes path from environment or use default."""
|
|
default_path = project_root / "tests" / "env" / "ComfyUI" / "custom_nodes"
|
|
return Path(os.environ.get("COMFYUI_CUSTOM_NODES_PATH", str(default_path)))
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def check_server_running(server_url):
|
|
"""Check if ComfyUI server is running before running tests."""
|
|
try:
|
|
response = requests.get(f"{server_url}/system_stats", timeout=5)
|
|
if response.status_code != 200:
|
|
pytest.exit(f"ComfyUI server not responding at {server_url}", returncode=1)
|
|
except requests.exceptions.RequestException as e:
|
|
pytest.exit(
|
|
f"ComfyUI server not running at {server_url}. Start server with: cd tests/env && ./run.sh",
|
|
returncode=1,
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def install_test_package(server_url, custom_nodes_path):
|
|
"""
|
|
Install a test package for API tests that require pre-installed packages.
|
|
|
|
This fixture runs once per test session and installs ComfyUI_SigmoidOffsetScheduler
|
|
so that tests like test_installed_api_* have at least one package to work with.
|
|
"""
|
|
# Wait for check_server_running to complete
|
|
time.sleep(1)
|
|
|
|
# Check if package already installed
|
|
try:
|
|
response = requests.get(f"{server_url}/v2/customnode/installed", timeout=5)
|
|
if response.status_code == 200:
|
|
installed = response.json()
|
|
# If we already have packages, skip installation
|
|
if len(installed) > 0:
|
|
print(f"\n✓ Test packages already installed: {len(installed)} packages found")
|
|
return
|
|
except Exception as e:
|
|
print(f"\n⚠ Could not check installed packages: {e}")
|
|
|
|
# Install test package
|
|
print(f"\n⚙ Installing test package for session: ComfyUI_SigmoidOffsetScheduler")
|
|
|
|
try:
|
|
# Queue installation task
|
|
response = requests.post(
|
|
f"{server_url}/v2/manager/queue/task",
|
|
json={
|
|
"kind": "install",
|
|
"ui_id": "session_test_package",
|
|
"client_id": "test_session",
|
|
"params": {
|
|
"id": "ComfyUI_SigmoidOffsetScheduler",
|
|
"version": "1.0.1",
|
|
"selected_version": "latest",
|
|
},
|
|
},
|
|
timeout=10,
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
# Start queue
|
|
requests.get(f"{server_url}/v2/manager/queue/start", timeout=10)
|
|
|
|
# Wait for installation to complete (poll for completion)
|
|
max_wait = 30
|
|
start_time = time.time()
|
|
installed_package_path = custom_nodes_path / "ComfyUI_SigmoidOffsetScheduler"
|
|
|
|
while time.time() - start_time < max_wait:
|
|
if installed_package_path.exists() and (installed_package_path / ".tracking").exists():
|
|
print(f"✓ Test package installed successfully")
|
|
break
|
|
time.sleep(2)
|
|
else:
|
|
print(f"⚠ Test package installation may not have completed")
|
|
|
|
except Exception as e:
|
|
print(f"⚠ Could not install test package: {e}")
|
|
print(f" Some tests may fail due to missing packages")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def ensure_test_package_exists(server_url, custom_nodes_path, request):
|
|
"""
|
|
Function-scoped fixture that ensures test package exists before each test.
|
|
|
|
This handles the case where other test fixtures may have cleaned up the
|
|
session test package. Only runs for tests that don't manipulate the
|
|
test package themselves.
|
|
"""
|
|
# Skip for tests that manipulate test packages themselves
|
|
skip_tests = [
|
|
# Phase 1: Complex scenarios
|
|
"test_enable_cnr_when_both_disabled",
|
|
"test_enable_nightly_when_both_disabled",
|
|
# Phase 3: Disable complex scenarios
|
|
"test_disable_cnr_when_nightly_disabled",
|
|
# Phase 4: Update complex scenarios
|
|
"test_update_cnr_with_nightly_disabled",
|
|
"test_update_nightly_with_cnr_disabled",
|
|
"test_update_enabled_with_multiple_disabled",
|
|
# Phase 5: Install complex scenarios
|
|
"test_install_new_version_when_both_disabled",
|
|
"test_install_cnr_when_nightly_enabled",
|
|
"test_install_nightly_when_cnr_enabled",
|
|
# Phase 6: Uninstall complex scenarios
|
|
"test_uninstall_removes_all_versions",
|
|
# Phase 7: Complex version switch chains
|
|
"test_cnr_version_upgrade_with_history",
|
|
"test_sequential_version_switch_chain",
|
|
# Queue API tests that use cleanup_package fixture
|
|
"test_install_package_via_queue",
|
|
"test_uninstall_package_via_queue",
|
|
"test_install_uninstall_cycle",
|
|
"test_case_insensitive_operations",
|
|
"test_version_switch_cnr_to_nightly",
|
|
"test_version_switch_between_cnr_versions",
|
|
# Tests that manage their own package setup/cleanup
|
|
"test_installed_api_shows_only_enabled_when_both_exist",
|
|
"test_installed_api_cnr_priority_when_both_disabled",
|
|
"test_installed_api_shows_disabled_when_no_enabled_exists",
|
|
]
|
|
|
|
if request.node.name in skip_tests:
|
|
yield
|
|
return
|
|
|
|
# Check if package exists
|
|
test_package_path = custom_nodes_path / "ComfyUI_SigmoidOffsetScheduler"
|
|
|
|
# If package doesn't exist, reinstall it
|
|
if not test_package_path.exists():
|
|
print(f"\n🔄 [RESTORE] Test package was removed, reinstalling...")
|
|
|
|
try:
|
|
# Queue installation task
|
|
response = requests.post(
|
|
f"{server_url}/v2/manager/queue/task",
|
|
json={
|
|
"kind": "install",
|
|
"ui_id": "restore_test_package",
|
|
"client_id": "test_restore",
|
|
"params": {
|
|
"id": "ComfyUI_SigmoidOffsetScheduler",
|
|
"version": "1.0.1",
|
|
"selected_version": "latest",
|
|
},
|
|
},
|
|
timeout=10,
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
# Start queue
|
|
requests.get(f"{server_url}/v2/manager/queue/start", timeout=10)
|
|
|
|
# Wait for installation to complete
|
|
max_wait = 30
|
|
start_time = time.time()
|
|
|
|
while time.time() - start_time < max_wait:
|
|
if test_package_path.exists() and (test_package_path / ".tracking").exists():
|
|
print(f"✓ Test package restored successfully")
|
|
break
|
|
time.sleep(2)
|
|
else:
|
|
print(f"⚠ Test package restoration may not have completed")
|
|
|
|
except Exception as e:
|
|
print(f"⚠ Could not restore test package: {e}")
|
|
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def api_client(server_url):
|
|
"""Create API client with base URL."""
|
|
|
|
class APIClient:
|
|
def __init__(self, base_url: str):
|
|
self.base_url = base_url
|
|
self.session = requests.Session()
|
|
|
|
def post(self, path: str, **kwargs) -> requests.Response:
|
|
"""Make POST request to API endpoint."""
|
|
url = f"{self.base_url}{path}"
|
|
return self.session.post(url, **kwargs)
|
|
|
|
def get(self, path: str, **kwargs) -> requests.Response:
|
|
"""Make GET request to API endpoint."""
|
|
url = f"{self.base_url}{path}"
|
|
return self.session.get(url, **kwargs)
|
|
|
|
def queue_task(self, kind: str, ui_id: str, params: dict) -> requests.Response:
|
|
"""Queue a task to the manager queue."""
|
|
url = f"{self.base_url}/v2/manager/queue/task"
|
|
payload = {"kind": kind, "ui_id": ui_id, "client_id": "test", "params": params}
|
|
return self.session.post(url, json=payload)
|
|
|
|
def start_queue(self) -> requests.Response:
|
|
"""Start processing the queue."""
|
|
url = f"{self.base_url}/v2/manager/queue/start"
|
|
return self.session.get(url)
|
|
|
|
def get_pending_queue(self) -> requests.Response:
|
|
"""Get pending tasks in queue."""
|
|
url = f"{self.base_url}/v2/manager/queue/pending"
|
|
return self.session.get(url)
|
|
|
|
def get_queue_history(self) -> requests.Response:
|
|
"""Get queue task history."""
|
|
url = f"{self.base_url}/v2/manager/queue/history"
|
|
return self.session.get(url)
|
|
|
|
def get_installed_packages(self) -> requests.Response:
|
|
"""Get list of installed packages."""
|
|
url = f"{self.base_url}/v2/customnode/installed"
|
|
return self.session.get(url)
|
|
|
|
return APIClient(server_url)
|
|
|
|
|
|
@pytest.fixture
|
|
def wait_for_queue():
|
|
"""Helper to wait for queue processing to complete."""
|
|
|
|
def _wait(seconds=3):
|
|
time.sleep(seconds)
|
|
|
|
return _wait
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_queue(api_client):
|
|
"""Clean up pending queue before and after test."""
|
|
# Clear queue before test
|
|
try:
|
|
api_client.start_queue()
|
|
time.sleep(2)
|
|
except Exception:
|
|
pass
|
|
|
|
yield
|
|
|
|
# Clear queue after test
|
|
try:
|
|
api_client.start_queue()
|
|
time.sleep(2)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ========================================
|
|
# Complex Scenario Fixtures
|
|
# ========================================
|
|
|
|
# Test package configuration
|
|
TEST_PACKAGE_ID = "ComfyUI_SigmoidOffsetScheduler"
|
|
TEST_PACKAGE_CNR_ID = "comfyui_sigmoidoffsetscheduler"
|
|
|
|
# Dynamic versions (set by session-level setup)
|
|
TEST_PACKAGE_OLDEST_VERSION = None # Third newest version (for Phase 7 history tests)
|
|
TEST_PACKAGE_OLD_VERSION = None # Second newest version
|
|
TEST_PACKAGE_NEW_VERSION = None # Latest version
|
|
|
|
# Derived version variables (set by initialize_test_versions fixture)
|
|
CNR_VERSION = None # Alias for TEST_PACKAGE_NEW_VERSION
|
|
CNR_VERSION_OLD = None # Alias for TEST_PACKAGE_OLD_VERSION
|
|
TEST_PACKAGE_VERSION = None # Alias for TEST_PACKAGE_NEW_VERSION
|
|
|
|
# Wait times for operations
|
|
WAIT_TIME_SHORT = 3 # seconds for enable/disable
|
|
WAIT_TIME_MEDIUM = 8 # seconds for install
|
|
WAIT_TIME_LONG = 10 # seconds for update/complex operations
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def initialize_test_versions(server_url):
|
|
"""
|
|
Session-level fixture to initialize test package versions dynamically.
|
|
This runs once per test session and sets global version variables.
|
|
"""
|
|
global TEST_PACKAGE_OLDEST_VERSION, TEST_PACKAGE_OLD_VERSION, TEST_PACKAGE_NEW_VERSION
|
|
global CNR_VERSION, CNR_VERSION_OLD, TEST_PACKAGE_VERSION
|
|
|
|
versions = get_available_cnr_versions(server_url, TEST_PACKAGE_ID)
|
|
|
|
if not versions:
|
|
pytest.skip(f"Could not fetch versions for {TEST_PACKAGE_ID}")
|
|
|
|
# Assign versions based on availability
|
|
TEST_PACKAGE_NEW_VERSION = versions[0] if len(versions) >= 1 else None
|
|
TEST_PACKAGE_OLD_VERSION = versions[1] if len(versions) >= 2 else versions[0]
|
|
TEST_PACKAGE_OLDEST_VERSION = versions[2] if len(versions) >= 3 else versions[-1]
|
|
|
|
# Set derived version variables (aliases for backward compatibility)
|
|
CNR_VERSION = TEST_PACKAGE_NEW_VERSION
|
|
CNR_VERSION_OLD = TEST_PACKAGE_OLD_VERSION
|
|
TEST_PACKAGE_VERSION = TEST_PACKAGE_NEW_VERSION
|
|
|
|
if len(versions) < 2:
|
|
pytest.skip(f"Need at least 2 versions for testing, found {len(versions)}")
|
|
|
|
print(f"\n📦 Test versions initialized:")
|
|
print(f" - NEW (latest): {TEST_PACKAGE_NEW_VERSION}")
|
|
print(f" - OLD (2nd): {TEST_PACKAGE_OLD_VERSION}")
|
|
print(f" - OLDEST (3rd): {TEST_PACKAGE_OLDEST_VERSION}")
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_multi_disabled_cnr_and_nightly(api_client, custom_nodes_path):
|
|
"""
|
|
Install both CNR and Nightly in disabled state.
|
|
|
|
Creates:
|
|
.disabled/ComfyUI_SigmoidOffsetScheduler_1.0.2/ (CNR with .tracking)
|
|
.disabled/ComfyUI_SigmoidOffsetScheduler/ (Nightly with .git)
|
|
|
|
Use case: Test 1.1, 1.2 (Multiple Disabled → Enable)
|
|
"""
|
|
import shutil
|
|
|
|
disabled_path = custom_nodes_path / ".disabled"
|
|
disabled_path.mkdir(exist_ok=True)
|
|
|
|
# Cleanup any existing sigmoid packages before starting
|
|
enabled_package = custom_nodes_path / TEST_PACKAGE_ID
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
|
|
# Step 1: Install CNR v1.0.2
|
|
print(f"\n=== Step 1: Installing CNR v{TEST_PACKAGE_NEW_VERSION} ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_multi_cnr",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": TEST_PACKAGE_NEW_VERSION,
|
|
"selected_version": "latest",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Debug: Check state after CNR install
|
|
print(f"Enabled packages: {list(custom_nodes_path.glob('*Sigmoid*'))}")
|
|
print(f"Disabled packages: {[p.name for p in disabled_path.iterdir() if 'sigmoid' in p.name.lower()]}")
|
|
|
|
# Step 2: Disable CNR (move to .disabled/)
|
|
print(f"\n=== Step 2: Disabling CNR ===")
|
|
response = api_client.queue_task(
|
|
kind="disable",
|
|
ui_id="setup_multi_disable_cnr",
|
|
params={"node_name": TEST_PACKAGE_ID},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM) # Use MEDIUM for disable to ensure completion
|
|
|
|
# Debug: Check state after CNR disable
|
|
print(f"Enabled packages: {list(custom_nodes_path.glob('*Sigmoid*'))}")
|
|
print(f"Disabled packages: {[p.name for p in disabled_path.iterdir() if 'sigmoid' in p.name.lower()]}")
|
|
|
|
# Step 3: Install Nightly
|
|
print(f"\n=== Step 3: Installing Nightly ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_multi_nightly",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": "nightly",
|
|
"selected_version": "nightly",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Debug: Check state after Nightly install
|
|
print(f"Enabled packages: {list(custom_nodes_path.glob('*Sigmoid*'))}")
|
|
print(f"Disabled packages: {[p.name for p in disabled_path.iterdir() if 'sigmoid' in p.name.lower()]}")
|
|
|
|
# Step 4: Disable Nightly
|
|
print(f"\n=== Step 4: Disabling Nightly ===")
|
|
response = api_client.queue_task(
|
|
kind="disable",
|
|
ui_id="setup_multi_disable_nightly",
|
|
params={"node_name": TEST_PACKAGE_ID},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM) # Use MEDIUM for disable to ensure completion
|
|
|
|
# Debug: Check state after Nightly disable
|
|
print(f"Enabled packages: {list(custom_nodes_path.glob('*Sigmoid*'))}")
|
|
print(f"Disabled packages: {[p.name for p in disabled_path.iterdir() if 'sigmoid' in p.name.lower()]}")
|
|
|
|
# Verify both disabled
|
|
enabled_package = custom_nodes_path / TEST_PACKAGE_ID
|
|
assert not enabled_package.exists(), "No package should be enabled"
|
|
|
|
# Use case-insensitive search for disabled packages
|
|
disabled_packages = [
|
|
item for item in disabled_path.iterdir()
|
|
if 'sigmoid' in item.name.lower()
|
|
]
|
|
assert len(disabled_packages) == 2, (
|
|
f"Both CNR and Nightly should be disabled, found {len(disabled_packages)}: "
|
|
f"{[p.name for p in disabled_packages]}"
|
|
)
|
|
|
|
yield
|
|
|
|
# Cleanup
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_cnr_enabled_nightly_disabled(api_client, custom_nodes_path):
|
|
"""
|
|
CNR enabled, Nightly disabled state.
|
|
|
|
Creates:
|
|
custom_nodes/ComfyUI_SigmoidOffsetScheduler/ (CNR 1.0.1 with .tracking)
|
|
.disabled/comfyui_sigmoidoffsetscheduler@nightly/ (Nightly with .git)
|
|
|
|
Use case: Test 4.1 (Update CNR with Nightly disabled)
|
|
"""
|
|
import shutil
|
|
|
|
disabled_path = custom_nodes_path / ".disabled"
|
|
disabled_path.mkdir(exist_ok=True)
|
|
enabled_package = custom_nodes_path / TEST_PACKAGE_ID
|
|
|
|
# Cleanup any existing sigmoid packages
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
|
|
# Step 1: Install Nightly first
|
|
print(f"\n=== Step 1: Installing Nightly ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_nightly_first",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": "nightly",
|
|
"selected_version": "nightly",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Step 2: Install CNR (this will automatically disable Nightly)
|
|
print(f"\n=== Step 2: Installing CNR v{TEST_PACKAGE_OLD_VERSION} (will disable Nightly) ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_cnr_enabled",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": TEST_PACKAGE_OLD_VERSION,
|
|
"selected_version": "latest",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Verify state
|
|
assert enabled_package.exists(), "CNR should be enabled"
|
|
assert (enabled_package / ".tracking").exists(), "CNR should have .tracking"
|
|
|
|
disabled_packages = [
|
|
item for item in disabled_path.iterdir()
|
|
if 'sigmoid' in item.name.lower()
|
|
]
|
|
assert len(disabled_packages) == 1, f"Nightly should be disabled, found {len(disabled_packages)}"
|
|
|
|
print(f"✓ Setup complete: CNR v{TEST_PACKAGE_OLD_VERSION} enabled, Nightly disabled")
|
|
|
|
yield
|
|
|
|
# Cleanup
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_nightly_enabled_cnr_disabled(api_client, custom_nodes_path):
|
|
"""
|
|
Nightly enabled, CNR disabled state.
|
|
|
|
Creates:
|
|
custom_nodes/ComfyUI_SigmoidOffsetScheduler/ (Nightly with .git)
|
|
.disabled/comfyui_sigmoidoffsetscheduler@1_0_2/ (CNR v1.0.2 with .tracking)
|
|
|
|
Use case: Test 4.2 (Update Nightly with CNR disabled)
|
|
"""
|
|
import shutil
|
|
|
|
disabled_path = custom_nodes_path / ".disabled"
|
|
disabled_path.mkdir(exist_ok=True)
|
|
enabled_package = custom_nodes_path / TEST_PACKAGE_ID
|
|
|
|
# Cleanup any existing sigmoid packages
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
|
|
# Step 1: Install CNR first
|
|
print(f"\n=== Step 1: Installing CNR v{TEST_PACKAGE_NEW_VERSION} ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_cnr_first",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": TEST_PACKAGE_NEW_VERSION,
|
|
"selected_version": "latest",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Step 2: Disable CNR
|
|
print(f"\n=== Step 2: Disabling CNR ===")
|
|
response = api_client.queue_task(
|
|
kind="disable",
|
|
ui_id="setup_disable_cnr",
|
|
params={"node_name": TEST_PACKAGE_ID},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Step 3: Install Nightly (enabled)
|
|
print(f"\n=== Step 3: Installing Nightly ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_nightly_enabled",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": "nightly",
|
|
"selected_version": "nightly",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Verify state
|
|
assert enabled_package.exists(), "Nightly should be enabled"
|
|
assert (enabled_package / ".git").exists(), "Nightly should have .git"
|
|
|
|
disabled_packages = [
|
|
item for item in disabled_path.iterdir()
|
|
if 'sigmoid' in item.name.lower()
|
|
]
|
|
assert len(disabled_packages) == 1, f"CNR should be disabled, found {len(disabled_packages)}"
|
|
|
|
print(f"✓ Setup complete: Nightly enabled, CNR v{TEST_PACKAGE_NEW_VERSION} disabled")
|
|
|
|
yield
|
|
|
|
# Cleanup
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_cnr_enabled_multiple_disabled(api_client, custom_nodes_path):
|
|
"""
|
|
Old CNR enabled, multiple versions disabled.
|
|
|
|
Creates:
|
|
custom_nodes/ComfyUI_SigmoidOffsetScheduler/ (CNR 1.0.1 with .tracking)
|
|
.disabled/comfyui_sigmoidoffsetscheduler@1_0_0/ (CNR v1.0.0 - simulated)
|
|
.disabled/comfyui_sigmoidoffsetscheduler@nightly/ (Nightly with .git)
|
|
|
|
Use case: Test 4.3 (Update enabled while multiple disabled exist)
|
|
|
|
Note: We'll simulate v1.0.0 by installing v1.0.2 and renaming it.
|
|
"""
|
|
import shutil
|
|
|
|
disabled_path = custom_nodes_path / ".disabled"
|
|
disabled_path.mkdir(exist_ok=True)
|
|
enabled_package = custom_nodes_path / TEST_PACKAGE_ID
|
|
|
|
# Cleanup any existing sigmoid packages
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
|
|
# Step 1: Install Nightly first
|
|
print(f"\n=== Step 1: Installing Nightly ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_multi_nightly",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": "nightly",
|
|
"selected_version": "nightly",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Step 2: Disable Nightly
|
|
print(f"\n=== Step 2: Disabling Nightly ===")
|
|
response = api_client.queue_task(
|
|
kind="disable",
|
|
ui_id="setup_multi_disable_nightly",
|
|
params={"node_name": TEST_PACKAGE_ID},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Step 3: Install CNR v1.0.1 (enabled)
|
|
print(f"\n=== Step 3: Installing CNR v{TEST_PACKAGE_OLD_VERSION} (enabled) ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_multi_current_cnr",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": TEST_PACKAGE_OLD_VERSION,
|
|
"selected_version": "latest",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Step 4: Manually create a simulated old CNR version in .disabled/
|
|
# Copy the current enabled CNR and rename it to simulate v1.0.0
|
|
print(f"\n=== Step 4: Creating simulated old CNR version ===")
|
|
simulated_old_cnr = disabled_path / "comfyui_sigmoidoffsetscheduler@1_0_0"
|
|
if enabled_package.exists():
|
|
import shutil
|
|
shutil.copytree(enabled_package, simulated_old_cnr)
|
|
print(f"Created simulated old CNR at {simulated_old_cnr.name}")
|
|
|
|
# Verify state
|
|
assert enabled_package.exists(), "CNR v1.0.1 should be enabled"
|
|
assert (enabled_package / ".tracking").exists(), "CNR should have .tracking"
|
|
|
|
disabled_packages = [
|
|
item for item in disabled_path.iterdir()
|
|
if 'sigmoid' in item.name.lower()
|
|
]
|
|
assert len(disabled_packages) == 2, (
|
|
f"Should have 2 disabled packages (old CNR + Nightly), found {len(disabled_packages)}: "
|
|
f"{[p.name for p in disabled_packages]}"
|
|
)
|
|
|
|
print(f"✓ Setup complete: CNR v{TEST_PACKAGE_OLD_VERSION} enabled, 2 versions disabled")
|
|
|
|
yield
|
|
|
|
# Cleanup
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
|
|
@pytest.fixture
|
|
def setup_nightly_enabled_only(api_client, custom_nodes_path):
|
|
"""
|
|
Install Nightly version only (enabled state).
|
|
|
|
Creates:
|
|
custom_nodes/ComfyUI_SigmoidOffsetScheduler/ (Nightly with .git)
|
|
|
|
Use case: Test 5.1 (Nightly enabled → Install CNR)
|
|
"""
|
|
import shutil
|
|
|
|
disabled_path = custom_nodes_path / ".disabled"
|
|
disabled_path.mkdir(exist_ok=True)
|
|
enabled_package = custom_nodes_path / TEST_PACKAGE_ID
|
|
|
|
# Cleanup any existing sigmoid packages
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
|
|
# Install Nightly
|
|
print(f"\n=== Installing Nightly (enabled) ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_nightly_only",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": "nightly",
|
|
"selected_version": "nightly",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Verify state
|
|
assert enabled_package.exists(), "Nightly should be enabled"
|
|
assert (enabled_package / ".git").exists(), "Nightly should have .git directory"
|
|
|
|
disabled_packages = [
|
|
item for item in disabled_path.iterdir()
|
|
if 'sigmoid' in item.name.lower()
|
|
]
|
|
assert len(disabled_packages) == 0, (
|
|
f"No packages should be disabled, found {len(disabled_packages)}: "
|
|
f"{[p.name for p in disabled_packages]}"
|
|
)
|
|
|
|
print(f"✓ Setup complete: Nightly enabled only")
|
|
|
|
yield
|
|
|
|
# Cleanup
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
|
|
|
|
@pytest.fixture
|
|
def setup_cnr_enabled_only(api_client, custom_nodes_path):
|
|
"""
|
|
Install CNR version only (enabled state).
|
|
|
|
Creates:
|
|
custom_nodes/ComfyUI_SigmoidOffsetScheduler/ (CNR v1.0.2 with .tracking)
|
|
|
|
Use case: Test 5.2 (CNR enabled → Install Nightly)
|
|
"""
|
|
import shutil
|
|
|
|
disabled_path = custom_nodes_path / ".disabled"
|
|
disabled_path.mkdir(exist_ok=True)
|
|
enabled_package = custom_nodes_path / TEST_PACKAGE_ID
|
|
|
|
# Cleanup any existing sigmoid packages
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
|
|
# Install CNR v1.0.2
|
|
print(f"\n=== Installing CNR v{TEST_PACKAGE_NEW_VERSION} (enabled) ===")
|
|
response = api_client.queue_task(
|
|
kind="install",
|
|
ui_id="setup_cnr_only",
|
|
params={
|
|
"id": TEST_PACKAGE_ID,
|
|
"version": TEST_PACKAGE_NEW_VERSION,
|
|
"selected_version": "latest",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
api_client.start_queue()
|
|
time.sleep(WAIT_TIME_MEDIUM)
|
|
|
|
# Verify state
|
|
assert enabled_package.exists(), "CNR should be enabled"
|
|
assert (enabled_package / ".tracking").exists(), "CNR should have .tracking file"
|
|
|
|
disabled_packages = [
|
|
item for item in disabled_path.iterdir()
|
|
if 'sigmoid' in item.name.lower()
|
|
]
|
|
assert len(disabled_packages) == 0, (
|
|
f"No packages should be disabled, found {len(disabled_packages)}: "
|
|
f"{[p.name for p in disabled_packages]}"
|
|
)
|
|
|
|
print(f"✓ Setup complete: CNR v{TEST_PACKAGE_NEW_VERSION} enabled only")
|
|
|
|
yield
|
|
|
|
# Cleanup
|
|
for item in disabled_path.iterdir():
|
|
if 'sigmoid' in item.name.lower() and item.is_dir():
|
|
shutil.rmtree(item)
|
|
if enabled_package.exists():
|
|
shutil.rmtree(enabled_package)
|
|
|
|
|
|
|
|
# ============================================================================
|
|
# Dynamic Version Management Helpers
|
|
# ============================================================================
|
|
|
|
|
|
def get_installed_version(package_path) -> str | None:
|
|
"""
|
|
Get currently installed version from pyproject.toml.
|
|
|
|
Args:
|
|
package_path: Path to the package directory (str or Path object)
|
|
|
|
Returns:
|
|
Version string (e.g., "1.0.2") or None if not found
|
|
|
|
Example:
|
|
>>> version = get_installed_version(custom_nodes_path / "ComfyUI_SigmoidOffsetScheduler")
|
|
>>> print(version) # "1.0.2"
|
|
"""
|
|
import re
|
|
from pathlib import Path
|
|
|
|
# Convert to Path if string
|
|
if isinstance(package_path, str):
|
|
package_path = Path(package_path)
|
|
|
|
pyproject = package_path / "pyproject.toml"
|
|
if not pyproject.exists():
|
|
return None
|
|
|
|
content = pyproject.read_text()
|
|
match = re.search(r'version\s*=\s*"([^"]+)"', content)
|
|
return match.group(1) if match else None
|
|
|
|
|
|
def get_available_cnr_versions(server_url: str, package_id: str) -> list[str]:
|
|
"""
|
|
Get list of available CNR versions for a package from ComfyRegistry.
|
|
|
|
This function queries the ComfyRegistry API to get all available versions.
|
|
|
|
Args:
|
|
server_url: ComfyUI server URL (unused, kept for API compatibility)
|
|
package_id: Package identifier (e.g., "ComfyUI_SigmoidOffsetScheduler")
|
|
|
|
Returns:
|
|
List of version strings sorted newest first (e.g., ["1.0.2", "1.0.1"])
|
|
|
|
Example:
|
|
>>> versions = get_available_cnr_versions(server_url, "ComfyUI_SigmoidOffsetScheduler")
|
|
>>> print(versions) # ["1.0.2", "1.0.1", "1.0.0"]
|
|
"""
|
|
from packaging import version as pkg_version
|
|
|
|
try:
|
|
# Import CNR utils
|
|
from comfyui_manager.common import cnr_utils
|
|
|
|
# Query ComfyRegistry for all versions
|
|
version_data = cnr_utils.all_versions_of_node(package_id)
|
|
|
|
if version_data and isinstance(version_data, list):
|
|
# Extract version strings from response
|
|
# Response format: [{"version": "1.0.2", ...}, {"version": "1.0.1", ...}]
|
|
versions = [item.get('version') for item in version_data if 'version' in item]
|
|
|
|
# Sort by semantic version (newest first)
|
|
return sorted(versions, key=lambda v: pkg_version.parse(v), reverse=True)
|
|
|
|
except Exception as e:
|
|
print(f"Info: ComfyRegistry query failed for {package_id}: {e}")
|
|
|
|
# Fallback: Known versions for test package
|
|
if package_id == "ComfyUI_SigmoidOffsetScheduler":
|
|
print(f"Info: Using known versions for {package_id}")
|
|
return ["1.0.2", "1.0.1", "1.0.0"]
|
|
|
|
print(f"Warning: Could not fetch versions for {package_id}")
|
|
return []
|
|
|
|
|
|
def compare_versions(v1: str, v2: str) -> int:
|
|
"""
|
|
Compare two semantic versions.
|
|
|
|
Args:
|
|
v1: First version string (e.g., "1.0.1")
|
|
v2: Second version string (e.g., "1.0.2")
|
|
|
|
Returns:
|
|
-1 if v1 < v2
|
|
0 if v1 == v2
|
|
1 if v1 > v2
|
|
|
|
Example:
|
|
>>> compare_versions("1.0.1", "1.0.2")
|
|
-1
|
|
>>> compare_versions("1.0.2", "1.0.1")
|
|
1
|
|
>>> compare_versions("1.0.1", "1.0.1")
|
|
0
|
|
"""
|
|
from packaging import version
|
|
|
|
v1_obj = version.parse(v1)
|
|
v2_obj = version.parse(v2)
|
|
|
|
if v1_obj < v2_obj:
|
|
return -1
|
|
elif v1_obj > v2_obj:
|
|
return 1
|
|
else:
|
|
return 0
|
|
|
|
|
|
def assert_version_increased(version_before: str, version_after: str, context: str = ""):
|
|
"""
|
|
Assert that version increased after an operation.
|
|
|
|
Args:
|
|
version_before: Version before operation
|
|
version_after: Version after operation
|
|
context: Additional context for error message
|
|
|
|
Raises:
|
|
AssertionError: If version did not increase
|
|
|
|
Example:
|
|
>>> assert_version_increased("1.0.1", "1.0.2", "after upgrade")
|
|
"""
|
|
assert version_after is not None, f"Version after operation is None {context}"
|
|
assert version_before is not None, f"Version before operation is None {context}"
|
|
assert version_after != version_before, (
|
|
f"Version did not change {context}: {version_before} → {version_after}"
|
|
)
|
|
assert compare_versions(version_after, version_before) > 0, (
|
|
f"Version did not increase {context}: {version_before} → {version_after}"
|
|
)
|
|
|
|
|
|
def assert_version_decreased(version_before: str, version_after: str, context: str = ""):
|
|
"""
|
|
Assert that version decreased after an operation (downgrade).
|
|
|
|
Args:
|
|
version_before: Version before operation
|
|
version_after: Version after operation
|
|
context: Additional context for error message
|
|
|
|
Raises:
|
|
AssertionError: If version did not decrease
|
|
|
|
Example:
|
|
>>> assert_version_decreased("1.0.2", "1.0.1", "after downgrade")
|
|
"""
|
|
assert version_after is not None, f"Version after operation is None {context}"
|
|
assert version_before is not None, f"Version before operation is None {context}"
|
|
assert version_after != version_before, (
|
|
f"Version did not change {context}: {version_before} → {version_after}"
|
|
)
|
|
assert compare_versions(version_after, version_before) < 0, (
|
|
f"Version did not decrease {context}: {version_before} → {version_after}"
|
|
)
|