wangbo/ComfyUI-Manager
1
0
Fork 0
mirror of https://github.com/Comfy-Org/ComfyUI-Manager.git synced 2026-07-26 18:17:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(cli): expand --uv-compile to all node management commands with conflict attribution (#2682)

Browse Source 
* feat(cli): expand --uv-compile to all node management commands with conflict attribution

Add --uv-compile flag to reinstall, update, fix, restore-snapshot,
restore-dependencies, and install-deps commands. Each skips per-node
pip installs and runs batch uv pip compile after all operations.

Change CollectedDeps.sources type to dict[str, list[tuple[str, str]]]
to store (pack_path, pkg_spec) per requester. On resolution failure,
_run_unified_resolve() cross-references conflict packages with sources
using word-boundary regex and displays which node packs requested each
conflicting package.

Update EN/KO user docs and DESIGN/PRD developer docs to cover the
expanded commands and conflict attribution output. Strengthen unit
tests for sources tuple format and compile failure attribution.
Bump version to 4.1b3.

* refactor(cli): extract _finalize_resolve helper, add CNR nightly fallback and pydantic guard

- Extract `_finalize_resolve()` to eliminate 7x duplicated uv-compile
  error handling blocks in cm_cli (~-85 lines)
- Move conflict attribution regex to `attribute_conflicts()` in
  unified_dep_resolver.py for direct testability
- Update 4 attribution tests to call production function instead of
  re-implementing regex
- Add CNR nightly fallback: when node is absent from nightly manifest,
  fall back to cnr_map repository URL (glob + legacy)
- Add pydantic Union guard: use getattr for is_unknown in uninstall
  and disable handlers to prevent Union type mismatch
- Add E2E test suites for endpoint install/uninstall and uv-compile
  CLI commands (conflict + success cases)
- Add nightly CNR fallback regression tests
This commit is contained in:
Dr.Lt.Data
2026-03-14 07:58:29 +09:00
committed by GitHub
parent f042d73b72
commit d7a2277017
14 changed files with 1316 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/e2e/test_e2e_endpoint.py
+306
View File
@@ -0,0 +1,306 @@
"""E2E tests for ComfyUI Manager HTTP API endpoints (install/uninstall).
Starts a real ComfyUI instance, exercises the task-queue-based install
and uninstall endpoints, then verifies the results via the installed-list
endpoint and filesystem checks.
Requires a pre-built E2E environment (from setup_e2e_env.sh).
Set E2E_ROOT env var to point at it, or the tests will be skipped.
Install test methodology follows the main comfyui-manager test suite
(tests/glob/test_queue_task_api.py):
- Uses a CNR-registered package with proper version-based install
- Verifies .tracking file for CNR installs
- Checks installed-list API with cnr_id matching
- Cleans up .disabled/ directory entries
Usage:
E2E_ROOT=/tmp/e2e_full_test pytest tests/e2e/test_e2e_endpoint.py -v
"""
from __future__ import annotations
import os
import shutil
import subprocess
import time
import pytest
import requests
E2E_ROOT = os.environ.get("E2E_ROOT", "")
COMFYUI_PATH = os.path.join(E2E_ROOT, "comfyui") if E2E_ROOT else ""
CUSTOM_NODES = os.path.join(COMFYUI_PATH, "custom_nodes") if COMFYUI_PATH else ""
SCRIPTS_DIR = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "scripts"
)
PORT = 8199
BASE_URL = f"http://127.0.0.1:{PORT}"
# CNR-registered package with multiple versions, no heavy dependencies.
# Same test package used by the main comfyui-manager test suite.
PACK_ID = "ComfyUI_SigmoidOffsetScheduler"
PACK_DIR_NAME = "ComfyUI_SigmoidOffsetScheduler"
PACK_CNR_ID = "comfyui_sigmoidoffsetscheduler"
PACK_VERSION = "1.0.1"
# Polling configuration for async task completion
POLL_TIMEOUT = 30 # max seconds to wait for an operation
POLL_INTERVAL = 0.5 # seconds between polls
pytestmark = pytest.mark.skipif(
not E2E_ROOT
or not os.path.isfile(os.path.join(E2E_ROOT, ".e2e_setup_complete")),
reason="E2E_ROOT not set or E2E environment not ready (run setup_e2e_env.sh first)",
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _start_comfyui() -> int:
"""Start ComfyUI and return its PID."""
env = {**os.environ, "E2E_ROOT": E2E_ROOT, "PORT": str(PORT)}
r = subprocess.run(
["bash", os.path.join(SCRIPTS_DIR, "start_comfyui.sh")],
capture_output=True,
text=True,
timeout=180,
env=env,
)
if r.returncode != 0:
raise RuntimeError(f"Failed to start ComfyUI:\n{r.stderr}")
for part in r.stdout.strip().split():
if part.startswith("COMFYUI_PID="):
return int(part.split("=")[1])
raise RuntimeError(f"Could not parse PID from start_comfyui output:\n{r.stdout}")
def _stop_comfyui():
"""Stop ComfyUI."""
env = {**os.environ, "E2E_ROOT": E2E_ROOT, "PORT": str(PORT)}
subprocess.run(
["bash", os.path.join(SCRIPTS_DIR, "stop_comfyui.sh")],
capture_output=True,
text=True,
timeout=30,
env=env,
)
def _queue_task(task: dict) -> None:
"""Queue a task and start the worker."""
resp = requests.post(
f"{BASE_URL}/v2/manager/queue/task",
json=task,
timeout=10,
)
resp.raise_for_status()
requests.get(f"{BASE_URL}/v2/manager/queue/start", timeout=10)
def _remove_pack(name: str) -> None:
"""Remove a node pack directory and any .disabled/ entries."""
# Active directory
path = os.path.join(CUSTOM_NODES, name)
if os.path.islink(path):
os.unlink(path)
elif os.path.isdir(path):
shutil.rmtree(path, ignore_errors=True)
# .disabled/ entries (CNR versioned + nightly)
disabled_dir = os.path.join(CUSTOM_NODES, ".disabled")
if os.path.isdir(disabled_dir):
cnr_lower = name.lower().replace("_", "").replace("-", "")
for entry in os.listdir(disabled_dir):
entry_lower = entry.lower().replace("_", "").replace("-", "")
if entry_lower.startswith(cnr_lower):
entry_path = os.path.join(disabled_dir, entry)
if os.path.isdir(entry_path):
shutil.rmtree(entry_path, ignore_errors=True)
def _pack_exists(name: str) -> bool:
return os.path.isdir(os.path.join(CUSTOM_NODES, name))
def _has_tracking(name: str) -> bool:
"""Check if the pack has a .tracking file (CNR install marker)."""
return os.path.isfile(os.path.join(CUSTOM_NODES, name, ".tracking"))
def _wait_for(predicate, timeout=POLL_TIMEOUT, interval=POLL_INTERVAL):
"""Poll *predicate* until it returns True or *timeout* seconds elapse.
Returns True if predicate was satisfied, False on timeout.
"""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if predicate():
return True
time.sleep(interval)
return False
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def comfyui():
"""Start ComfyUI once for the module, stop after all tests."""
_remove_pack(PACK_DIR_NAME)
pid = _start_comfyui()
yield pid
_stop_comfyui()
_remove_pack(PACK_DIR_NAME)
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestEndpointInstallUninstall:
"""Install and uninstall via HTTP endpoints on a running ComfyUI.
Follows the same methodology as tests/glob/test_queue_task_api.py in
the main comfyui-manager repo: CNR version-based install, .tracking
verification, installed-list API check.
"""
def test_install_via_endpoint(self, comfyui):
"""POST /v2/manager/queue/task (install) -> pack appears on disk with .tracking."""
_remove_pack(PACK_DIR_NAME)
_queue_task({
"ui_id": "e2e-install",
"client_id": "e2e-install",
"kind": "install",
"params": {
"id": PACK_ID,
"version": PACK_VERSION,
"selected_version": "latest",
"mode": "remote",
"channel": "default",
},
})
assert _wait_for(
lambda: _pack_exists(PACK_DIR_NAME),
), f"{PACK_DIR_NAME} not found in custom_nodes within {POLL_TIMEOUT}s"
assert _has_tracking(PACK_DIR_NAME), f"{PACK_DIR_NAME} missing .tracking (not a CNR install?)"
def test_installed_list_shows_pack(self, comfyui):
"""GET /v2/customnode/installed includes the installed pack."""
if not _pack_exists(PACK_DIR_NAME):
pytest.skip("Pack not installed (previous test may have failed)")
resp = requests.get(f"{BASE_URL}/v2/customnode/installed", timeout=10)
resp.raise_for_status()
installed = resp.json()
# Match by cnr_id (case-insensitive) following main repo pattern
package_found = any(
pkg.get("cnr_id", "").lower() == PACK_CNR_ID.lower()
for pkg in installed.values()
if isinstance(pkg, dict) and pkg.get("cnr_id")
)
assert package_found, (
f"{PACK_CNR_ID} not found in installed list: {list(installed.keys())}"
)
def test_uninstall_via_endpoint(self, comfyui):
"""POST /v2/manager/queue/task (uninstall) -> pack removed from disk."""
if not _pack_exists(PACK_DIR_NAME):
pytest.skip("Pack not installed (previous test may have failed)")
_queue_task({
"ui_id": "e2e-uninstall",
"client_id": "e2e-uninstall",
"kind": "uninstall",
"params": {
"node_name": PACK_CNR_ID,
},
})
assert _wait_for(
lambda: not _pack_exists(PACK_DIR_NAME),
), f"{PACK_DIR_NAME} still exists after uninstall ({POLL_TIMEOUT}s timeout)"
def test_installed_list_after_uninstall(self, comfyui):
"""After uninstall, pack no longer appears in installed list."""
if _pack_exists(PACK_DIR_NAME):
pytest.skip("Pack still exists (previous test may have failed)")
resp = requests.get(f"{BASE_URL}/v2/customnode/installed", timeout=10)
resp.raise_for_status()
installed = resp.json()
package_found = any(
pkg.get("cnr_id", "").lower() == PACK_CNR_ID.lower()
for pkg in installed.values()
if isinstance(pkg, dict) and pkg.get("cnr_id")
)
assert not package_found, f"{PACK_CNR_ID} still in installed list after uninstall"
def test_install_uninstall_cycle(self, comfyui):
"""Complete install/uninstall cycle in a single test."""
_remove_pack(PACK_DIR_NAME)
# Install
_queue_task({
"ui_id": "e2e-cycle-install",
"client_id": "e2e-cycle",
"kind": "install",
"params": {
"id": PACK_ID,
"version": PACK_VERSION,
"selected_version": "latest",
"mode": "remote",
"channel": "default",
},
})
assert _wait_for(
lambda: _pack_exists(PACK_DIR_NAME),
), f"Pack not installed within {POLL_TIMEOUT}s"
assert _has_tracking(PACK_DIR_NAME), "Pack missing .tracking"
# Verify in installed list
resp = requests.get(f"{BASE_URL}/v2/customnode/installed", timeout=10)
resp.raise_for_status()
installed = resp.json()
package_found = any(
pkg.get("cnr_id", "").lower() == PACK_CNR_ID.lower()
for pkg in installed.values()
if isinstance(pkg, dict) and pkg.get("cnr_id")
)
assert package_found, f"{PACK_CNR_ID} not in installed list"
# Uninstall
_queue_task({
"ui_id": "e2e-cycle-uninstall",
"client_id": "e2e-cycle",
"kind": "uninstall",
"params": {
"node_name": PACK_CNR_ID,
},
})
assert _wait_for(
lambda: not _pack_exists(PACK_DIR_NAME),
), f"Pack not uninstalled within {POLL_TIMEOUT}s"
class TestEndpointStartup:
"""Verify ComfyUI startup with unified resolver."""
def test_comfyui_started(self, comfyui):
"""ComfyUI is running and responds to health check."""
resp = requests.get(f"{BASE_URL}/system_stats", timeout=10)
assert resp.status_code == 200
def test_startup_resolver_ran(self, comfyui):
"""Startup log contains unified resolver output."""
log_path = os.path.join(E2E_ROOT, "logs", "comfyui.log")
with open(log_path) as f:
log = f.read()
assert "[UnifiedDepResolver]" in log
assert "startup batch resolution succeeded" in log
tests/e2e/test_e2e_uv_compile.py
+254
View File
@@ -0,0 +1,254 @@
"""E2E tests for cm-cli --uv-compile across all supported commands.
Requires a pre-built E2E environment (from setup_e2e_env.sh).
Set E2E_ROOT env var to point at it, or the tests will be skipped.
Supply-chain safety policy:
To prevent supply-chain attacks, E2E tests MUST only install node packs
from verified, controllable authors (ltdrdata, comfyanonymous, etc.).
Currently this suite uses only ltdrdata's dedicated test packs
(nodepack-test1-do-not-install, nodepack-test2-do-not-install) which
are intentionally designed for conflict testing and contain no
executable code. Adding packs from unverified sources is prohibited.
Usage:
E2E_ROOT=/tmp/e2e_full_test pytest tests/e2e/test_e2e_uv_compile.py -v
"""
from __future__ import annotations
import os
import shutil
import subprocess
import pytest
E2E_ROOT = os.environ.get("E2E_ROOT", "")
COMFYUI_PATH = os.path.join(E2E_ROOT, "comfyui") if E2E_ROOT else ""
CM_CLI = os.path.join(E2E_ROOT, "venv", "bin", "cm-cli") if E2E_ROOT else ""
CUSTOM_NODES = os.path.join(COMFYUI_PATH, "custom_nodes") if COMFYUI_PATH else ""
REPO_TEST1 = "https://github.com/ltdrdata/nodepack-test1-do-not-install"
REPO_TEST2 = "https://github.com/ltdrdata/nodepack-test2-do-not-install"
PACK_TEST1 = "nodepack-test1-do-not-install"
PACK_TEST2 = "nodepack-test2-do-not-install"
pytestmark = pytest.mark.skipif(
not E2E_ROOT or not os.path.isfile(os.path.join(E2E_ROOT, ".e2e_setup_complete")),
reason="E2E_ROOT not set or E2E environment not ready (run setup_e2e_env.sh first)",
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _run_cm_cli(*args: str, timeout: int = 180) -> subprocess.CompletedProcess:
"""Run cm-cli in the E2E environment."""
env = {**os.environ, "COMFYUI_PATH": COMFYUI_PATH}
return subprocess.run(
[CM_CLI, *args],
capture_output=True,
text=True,
timeout=timeout,
env=env,
)
def _remove_pack(name: str) -> None:
"""Remove a node pack from custom_nodes (if it exists)."""
path = os.path.join(CUSTOM_NODES, name)
if os.path.islink(path):
os.unlink(path)
elif os.path.isdir(path):
shutil.rmtree(path, ignore_errors=True)
def _pack_exists(name: str) -> bool:
return os.path.isdir(os.path.join(CUSTOM_NODES, name))
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _clean_test_packs():
"""Ensure test node packs are removed before and after each test."""
_remove_pack(PACK_TEST1)
_remove_pack(PACK_TEST2)
yield
_remove_pack(PACK_TEST1)
_remove_pack(PACK_TEST2)
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestInstall:
"""cm-cli install --uv-compile"""
def test_install_single_pack_resolves(self):
"""Install one test pack with --uv-compile → resolve succeeds."""
r = _run_cm_cli("install", "--uv-compile", REPO_TEST1)
combined = r.stdout + r.stderr
assert _pack_exists(PACK_TEST1)
assert "Installation was successful" in combined
assert "Resolved" in combined
def test_install_conflicting_packs_shows_attribution(self):
"""Install two conflicting packs → conflict attribution output."""
# Install first (no conflict yet)
r1 = _run_cm_cli("install", "--uv-compile", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
assert "Resolved" in r1.stdout + r1.stderr
# Install second → conflict
r2 = _run_cm_cli("install", "--uv-compile", REPO_TEST2)
combined = r2.stdout + r2.stderr
assert _pack_exists(PACK_TEST2)
assert "Installation was successful" in combined
assert "Resolution failed" in combined
assert "Conflicting packages (by node pack):" in combined
assert PACK_TEST1 in combined
assert PACK_TEST2 in combined
assert "ansible" in combined.lower()
class TestReinstall:
"""cm-cli reinstall --uv-compile"""
def test_reinstall_with_uv_compile(self):
"""Reinstall an existing pack with --uv-compile."""
# Install first
_run_cm_cli("install", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
# Reinstall with --uv-compile
r = _run_cm_cli("reinstall", "--uv-compile", REPO_TEST1)
combined = r.stdout + r.stderr
# uv-compile should run (resolve output present)
assert "Resolving dependencies" in combined
class TestUpdate:
"""cm-cli update --uv-compile"""
def test_update_single_with_uv_compile(self):
"""Update an installed pack with --uv-compile."""
_run_cm_cli("install", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
r = _run_cm_cli("update", "--uv-compile", REPO_TEST1)
combined = r.stdout + r.stderr
assert "Resolving dependencies" in combined
def test_update_all_with_uv_compile(self):
"""update all --uv-compile runs uv-compile after updating."""
_run_cm_cli("install", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
r = _run_cm_cli("update", "--uv-compile", "all")
combined = r.stdout + r.stderr
assert "Resolving dependencies" in combined
class TestFix:
"""cm-cli fix --uv-compile"""
def test_fix_single_with_uv_compile(self):
"""Fix an installed pack with --uv-compile."""
_run_cm_cli("install", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
r = _run_cm_cli("fix", "--uv-compile", REPO_TEST1)
combined = r.stdout + r.stderr
assert "Resolving dependencies" in combined
def test_fix_all_with_uv_compile(self):
"""fix all --uv-compile runs uv-compile after fixing."""
_run_cm_cli("install", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
r = _run_cm_cli("fix", "--uv-compile", "all")
combined = r.stdout + r.stderr
assert "Resolving dependencies" in combined
class TestUvCompileStandalone:
"""cm-cli uv-compile (standalone command)"""
def test_uv_compile_no_packs(self):
"""uv-compile with no node packs → 'No custom node packs found'."""
r = _run_cm_cli("uv-compile")
combined = r.stdout + r.stderr
# Only ComfyUI-Manager exists (no requirements.txt in it normally)
# so either "No custom node packs found" or resolves 0
assert r.returncode == 0 or "No custom node packs" in combined
def test_uv_compile_with_packs(self):
"""uv-compile after installing test pack → resolves."""
_run_cm_cli("install", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
r = _run_cm_cli("uv-compile")
combined = r.stdout + r.stderr
assert "Resolving dependencies" in combined
assert "Resolved" in combined
def test_uv_compile_conflict_attribution(self):
"""uv-compile with conflicting packs → shows attribution."""
_run_cm_cli("install", REPO_TEST1)
_run_cm_cli("install", REPO_TEST2)
r = _run_cm_cli("uv-compile")
combined = r.stdout + r.stderr
assert r.returncode != 0
assert "Conflicting packages (by node pack):" in combined
assert PACK_TEST1 in combined
assert PACK_TEST2 in combined
class TestRestoreDependencies:
"""cm-cli restore-dependencies --uv-compile"""
def test_restore_dependencies_with_uv_compile(self):
"""restore-dependencies --uv-compile runs resolver after restore."""
_run_cm_cli("install", REPO_TEST1)
assert _pack_exists(PACK_TEST1)
r = _run_cm_cli("restore-dependencies", "--uv-compile")
combined = r.stdout + r.stderr
assert "Resolving dependencies" in combined
class TestConflictAttributionDetail:
"""Verify conflict attribution output details."""
def test_both_packs_and_specs_shown(self):
"""Conflict output shows pack names AND version specs."""
_run_cm_cli("install", REPO_TEST1)
_run_cm_cli("install", REPO_TEST2)
r = _run_cm_cli("uv-compile")
combined = r.stdout + r.stderr
# Processed attribution must show exact version specs (not raw uv error)
assert "Conflicting packages (by node pack):" in combined
assert "ansible==9.13.0" in combined
assert "ansible-core==2.14.0" in combined
# Both pack names present in attribution block
assert PACK_TEST1 in combined
assert PACK_TEST2 in combined
tests/test_nightly_cnr_fallback.py
+325
View File
@@ -0,0 +1,325 @@
"""Unit tests for CNR fallback in install_by_id nightly path and getattr guard.
Tests two targeted bug fixes:
1. install_by_id nightly: falls back to cnr_map when custom_nodes lookup fails
2. do_uninstall/do_disable: getattr guard prevents AttributeError on Union mismatch
"""
from __future__ import annotations
import asyncio
import types
# ---------------------------------------------------------------------------
# Minimal stubs — avoid importing the full ComfyUI runtime
# ---------------------------------------------------------------------------
class _ManagedResult:
"""Minimal ManagedResult stub matching glob/manager_core.py."""
def __init__(self, action):
self.action = action
self.result = True
self.msg = None
self.target = None
def fail(self, msg):
self.result = False
self.msg = msg
return self
def with_target(self, target):
self.target = target
return self
class _NormalizedKeyDict:
"""Minimal NormalizedKeyDict stub matching glob/manager_core.py."""
def __init__(self):
self._store = {}
self._key_map = {}
def _normalize_key(self, key):
return key.strip().lower() if isinstance(key, str) else key
def __setitem__(self, key, value):
norm = self._normalize_key(key)
self._key_map[norm] = key
self._store[key] = value
def __getitem__(self, key):
norm = self._normalize_key(key)
return self._store[self._key_map[norm]]
def __contains__(self, key):
return self._normalize_key(key) in self._key_map
def get(self, key, default=None):
return self[key] if key in self else default
# ===================================================================
# Test 1: CNR fallback in install_by_id nightly path
# ===================================================================
class TestNightlyCnrFallback:
"""install_by_id with version_spec='nightly' should fall back to cnr_map
when custom_nodes lookup returns None for the node_id."""
def _make_manager(self, cnr_map_entries=None, custom_nodes_entries=None):
"""Create a minimal UnifiedManager-like object with the install_by_id
nightly fallback logic extracted for unit testing."""
mgr = types.SimpleNamespace()
mgr.cnr_map = _NormalizedKeyDict()
if cnr_map_entries:
for k, v in cnr_map_entries.items():
mgr.cnr_map[k] = v
# Mock get_custom_nodes to return a NormalizedKeyDict
custom_nodes = _NormalizedKeyDict()
if custom_nodes_entries:
for k, v in custom_nodes_entries.items():
custom_nodes[k] = v
async def get_custom_nodes(channel=None, mode=None):
return custom_nodes
mgr.get_custom_nodes = get_custom_nodes
# Stubs for is_enabled/is_disabled that always return False (not installed)
mgr.is_enabled = lambda *a, **kw: False
mgr.is_disabled = lambda *a, **kw: False
return mgr
@staticmethod
async def _run_nightly_lookup(mgr, node_id, channel='default', mode='remote'):
"""Execute the nightly lookup logic from install_by_id.
Reproduces lines ~1407-1431 of glob/manager_core.py to test the
CNR fallback path in isolation.
"""
version_spec = 'nightly'
repo_url = None
custom_nodes = await mgr.get_custom_nodes(channel, mode)
the_node = custom_nodes.get(node_id)
if the_node is not None:
repo_url = the_node['repository']
else:
# Fallback for nightly only: use repository URL from CNR map
if version_spec == 'nightly':
cnr_fallback = mgr.cnr_map.get(node_id)
if cnr_fallback is not None and cnr_fallback.get('repository'):
repo_url = cnr_fallback['repository']
else:
result = _ManagedResult('install')
return result.fail(
f"Node '{node_id}@{version_spec}' not found in [{channel}, {mode}]"
)
return repo_url
def test_fallback_to_cnr_map_when_custom_nodes_missing(self):
"""Node absent from custom_nodes but present in cnr_map -> uses cnr_map repo URL."""
mgr = self._make_manager(
cnr_map_entries={
'my-test-pack': {
'id': 'my-test-pack',
'repository': 'https://github.com/test/my-test-pack',
'publisher': 'testuser',
},
},
custom_nodes_entries={}, # empty — node not in nightly manifest
)
result = asyncio.run(
self._run_nightly_lookup(mgr, 'my-test-pack')
)
assert result == 'https://github.com/test/my-test-pack'
def test_fallback_fails_when_cnr_map_also_missing(self):
"""Node absent from both custom_nodes and cnr_map -> ManagedResult.fail."""
mgr = self._make_manager(
cnr_map_entries={},
custom_nodes_entries={},
)
result = asyncio.run(
self._run_nightly_lookup(mgr, 'nonexistent-pack')
)
assert isinstance(result, _ManagedResult)
assert result.result is False
assert 'nonexistent-pack@nightly' in result.msg
def test_fallback_fails_when_cnr_entry_has_no_repository(self):
"""Node in cnr_map but repository is None/empty -> ManagedResult.fail."""
mgr = self._make_manager(
cnr_map_entries={
'no-repo-pack': {
'id': 'no-repo-pack',
'repository': None,
'publisher': 'testuser',
},
},
custom_nodes_entries={},
)
result = asyncio.run(
self._run_nightly_lookup(mgr, 'no-repo-pack')
)
assert isinstance(result, _ManagedResult)
assert result.result is False
def test_fallback_fails_when_cnr_entry_has_empty_repository(self):
"""Node in cnr_map but repository is '' -> ManagedResult.fail (truthy check)."""
mgr = self._make_manager(
cnr_map_entries={
'empty-repo-pack': {
'id': 'empty-repo-pack',
'repository': '',
'publisher': 'testuser',
},
},
custom_nodes_entries={},
)
result = asyncio.run(
self._run_nightly_lookup(mgr, 'empty-repo-pack')
)
assert isinstance(result, _ManagedResult)
assert result.result is False
def test_direct_custom_nodes_hit_skips_cnr_fallback(self):
"""Node present in custom_nodes -> uses custom_nodes directly, no fallback needed."""
mgr = self._make_manager(
cnr_map_entries={
'found-pack': {
'id': 'found-pack',
'repository': 'https://github.com/test/found-cnr',
},
},
custom_nodes_entries={
'found-pack': {
'repository': 'https://github.com/test/found-custom',
'files': ['https://github.com/test/found-custom'],
},
},
)
result = asyncio.run(
self._run_nightly_lookup(mgr, 'found-pack')
)
# Should use custom_nodes repo URL, NOT cnr_map
assert result == 'https://github.com/test/found-custom'
def test_unknown_version_spec_does_not_use_cnr_fallback(self):
"""version_spec='unknown' path should NOT use cnr_map fallback."""
mgr = self._make_manager(
cnr_map_entries={
'unknown-pack': {
'id': 'unknown-pack',
'repository': 'https://github.com/test/unknown-pack',
},
},
custom_nodes_entries={},
)
async def _run_unknown_lookup():
version_spec = 'unknown'
custom_nodes = await mgr.get_custom_nodes()
the_node = custom_nodes.get('unknown-pack')
if the_node is not None:
return the_node['files'][0]
else:
if version_spec == 'nightly':
# This branch should NOT be taken for 'unknown'
cnr_fallback = mgr.cnr_map.get('unknown-pack')
if cnr_fallback is not None and cnr_fallback.get('repository'):
return cnr_fallback['repository']
# Fall through to error for 'unknown'
result = _ManagedResult('install')
return result.fail(
f"Node 'unknown-pack@{version_spec}' not found"
)
result = asyncio.run(_run_unknown_lookup())
assert isinstance(result, _ManagedResult)
assert result.result is False
assert 'unknown' in result.msg
def test_case_insensitive_cnr_map_lookup(self):
"""CNR map uses NormalizedKeyDict — lookup should be case-insensitive."""
mgr = self._make_manager(
cnr_map_entries={
'My-Test-Pack': {
'id': 'my-test-pack',
'repository': 'https://github.com/test/my-test-pack',
},
},
custom_nodes_entries={},
)
result = asyncio.run(
self._run_nightly_lookup(mgr, 'my-test-pack')
)
assert result == 'https://github.com/test/my-test-pack'
# ===================================================================
# Test 2: getattr guard in do_uninstall / do_disable
# ===================================================================
class TestGetAttrGuard:
"""do_uninstall and do_disable use getattr(params, 'is_unknown', False)
to guard against pydantic Union matching UpdatePackParams (which lacks
is_unknown field) instead of UninstallPackParams/DisablePackParams."""
def test_getattr_on_object_with_is_unknown(self):
"""Normal case: params has is_unknown -> returns its value."""
params = types.SimpleNamespace(node_name='test-pack', is_unknown=True)
assert getattr(params, 'is_unknown', False) is True
def test_getattr_on_object_without_is_unknown(self):
"""Bug case: params is UpdatePackParams-like (no is_unknown) -> returns False."""
params = types.SimpleNamespace(node_name='test-pack', node_ver='1.0.0')
# Without getattr guard, this would be: params.is_unknown -> AttributeError
assert getattr(params, 'is_unknown', False) is False
def test_getattr_default_false_on_missing_attribute(self):
"""Minimal case: bare object with only node_name."""
params = types.SimpleNamespace(node_name='test-pack')
assert getattr(params, 'is_unknown', False) is False
def test_pydantic_union_matching_demonstrates_bug(self):
"""Demonstrate why getattr is needed: pydantic Union without discriminator
can match UpdatePackParams for uninstall/disable payloads."""
from pydantic import BaseModel, Field
from typing import Optional, Union
class UpdateLike(BaseModel):
node_name: str
node_ver: Optional[str] = None
class UninstallLike(BaseModel):
node_name: str
is_unknown: Optional[bool] = Field(False)
# When Union tries to match {"node_name": "foo"}, UpdateLike matches first
# because it has fewer required fields and node_name satisfies it
class TaskItem(BaseModel):
params: Union[UpdateLike, UninstallLike]
item = TaskItem(params={"node_name": "foo"})
# The matched type may be UpdateLike (no is_unknown attribute)
# This is the exact scenario the getattr guard protects against
is_unknown = getattr(item.params, 'is_unknown', False)
# Regardless of which Union member matched, getattr safely returns a value
assert isinstance(is_unknown, bool)
tests/test_unified_dep_resolver.py
+120
View File
@@ -244,6 +244,26 @@ class TestCollectRequirements:
deps = r.collect_requirements()
assert len(deps.sources["numpy"]) == 2
def test_sources_stores_pack_path_and_spec_tuple(self, tmp_path):
"""sources entries must be (pack_path, pkg_spec) tuples."""
p = _make_node_pack(str(tmp_path), "pack_a", "numpy>=1.20\n")
r = _resolver([p])
deps = r.collect_requirements()
entries = deps.sources["numpy"]
assert len(entries) == 1
pack_path, pkg_spec = entries[0]
assert pack_path == p
assert pkg_spec == "numpy>=1.20"
def test_sources_captures_spec_per_requester(self, tmp_path):
"""Each requester's exact spec is preserved independently."""
p1 = _make_node_pack(str(tmp_path), "pack_a", "torch>=2.1\n")
p2 = _make_node_pack(str(tmp_path), "pack_b", "torch<2.0\n")
r = _resolver([p1, p2])
deps = r.collect_requirements()
specs = {pkg_spec for _, pkg_spec in deps.sources["torch"]}
assert specs == {"torch>=2.1", "torch<2.0"}
# ===========================================================================
# Input sanitization
@@ -883,6 +903,106 @@ class TestResolveAndInstall:
assert not result.success
assert "compile failed" in result.error
def test_compile_failure_result_includes_collected(self, tmp_path):
"""result.collected must be populated on compile failure for conflict attribution."""
p = _make_node_pack(str(tmp_path), "pack_a", "torch>=2.1\n")
r = _resolver([p])
with mock.patch.object(r, "_get_uv_cmd", return_value=["uv"]):
with mock.patch("subprocess.run", return_value=subprocess.CompletedProcess(
[], 1, stdout="",
stderr="error: Because torch>=2.1 conflicts with torch<2.0, no solution found.",
)):
result = r.resolve_and_install()
assert not result.success
assert result.collected is not None
assert result.lockfile is not None
assert result.lockfile.conflicts # conflict lines present for attribution
def test_conflict_attribution_sources_filter(self, tmp_path):
"""Packages named in conflict lines can be looked up from sources."""
from comfyui_manager.common.unified_dep_resolver import attribute_conflicts
p1 = _make_node_pack(str(tmp_path), "pack_a", "torch>=2.1\n")
p2 = _make_node_pack(str(tmp_path), "pack_b", "torch<2.0\n")
r = _resolver([p1, p2])
conflict_text = "error: torch>=2.1 conflicts with torch<2.0"
with mock.patch.object(r, "_get_uv_cmd", return_value=["uv"]):
with mock.patch("subprocess.run", return_value=subprocess.CompletedProcess(
[], 1, stdout="", stderr=conflict_text,
)):
result = r.resolve_and_install()
assert not result.success
assert result.collected is not None
attributed = attribute_conflicts(result.collected.sources, result.lockfile.conflicts)
assert "torch" in attributed
specs = {spec for _, spec in attributed["torch"]}
assert specs == {"torch>=2.1", "torch<2.0"}
def test_conflict_attribution_no_false_positive_on_underscore_prefix(self, tmp_path):
"""'torch' must NOT match 'torch_audio' in conflict text (underscore boundary)."""
from comfyui_manager.common.unified_dep_resolver import attribute_conflicts
p = _make_node_pack(str(tmp_path), "pack_a", "torch>=2.1\n")
r = _resolver([p])
conflict_text = "error: torch_audio>=2.1 conflicts with torch_audio<2.0"
with mock.patch.object(r, "_get_uv_cmd", return_value=["uv"]):
with mock.patch("subprocess.run", return_value=subprocess.CompletedProcess(
[], 1, stdout="", stderr=conflict_text,
)):
result = r.resolve_and_install()
assert not result.success
assert result.collected is not None
attributed = attribute_conflicts(result.collected.sources, result.lockfile.conflicts)
# 'torch' should NOT match: conflict only mentions 'torch_audio'
assert "torch" not in attributed
def test_conflict_attribution_no_false_positive_on_prefix_match(self, tmp_path):
"""'torch' must NOT match 'torchvision' in conflict text (word boundary)."""
from comfyui_manager.common.unified_dep_resolver import attribute_conflicts
p = _make_node_pack(str(tmp_path), "pack_a", "torch>=2.1\n")
r = _resolver([p])
conflict_text = "error: torchvision>=0.16 conflicts with torchvision<0.15"
with mock.patch.object(r, "_get_uv_cmd", return_value=["uv"]):
with mock.patch("subprocess.run", return_value=subprocess.CompletedProcess(
[], 1, stdout="", stderr=conflict_text,
)):
result = r.resolve_and_install()
assert not result.success
assert result.collected is not None
attributed = attribute_conflicts(result.collected.sources, result.lockfile.conflicts)
# 'torch' should NOT appear: conflict only mentions 'torchvision'
assert "torch" not in attributed
def test_conflict_attribution_hyphen_underscore_normalization(self, tmp_path):
"""Packages stored with hyphens match conflict text using underscores."""
from comfyui_manager.common.unified_dep_resolver import attribute_conflicts
p = _make_node_pack(str(tmp_path), "pack_a", "torch-audio>=2.1\n")
r = _resolver([p])
# uv may print 'torch_audio' (underscore) in conflict output
conflict_text = "error: torch_audio>=2.1 conflicts with torch_audio<2.0"
with mock.patch.object(r, "_get_uv_cmd", return_value=["uv"]):
with mock.patch("subprocess.run", return_value=subprocess.CompletedProcess(
[], 1, stdout="", stderr=conflict_text,
)):
result = r.resolve_and_install()
assert not result.success
assert result.collected is not None
attributed = attribute_conflicts(result.collected.sources, result.lockfile.conflicts)
# _extract_package_name normalizes 'torch-audio' → 'torch_audio'; uv uses underscores too
assert "torch_audio" in attributed
def test_full_success_pipeline(self, tmp_path):
p = _make_node_pack(str(tmp_path), "pack_a", "numpy>=1.20\n")
r = _resolver([p])