diff --git a/cm_cli/__main__.py b/cm_cli/__main__.py index 114d46eb..a2b5d1f1 100644 --- a/cm_cli/__main__.py +++ b/cm_cli/__main__.py @@ -688,21 +688,7 @@ def install( pip_fixer = manager_util.PIPFixer(manager_util.get_installed_packages(), comfy_path, context.manager_files_path) for_each_nodes(nodes, act=install_node, exit_on_fail=exit_on_fail) - if uv_compile: - try: - _run_unified_resolve() - except ImportError as e: - print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") - raise typer.Exit(1) - except typer.Exit: - raise - except Exception as e: - print(f"[bold red]Batch resolution failed: {e}[/bold red]") - raise typer.Exit(1) - finally: - pip_fixer.fix_broken() - else: - pip_fixer.fix_broken() + _finalize_resolve(pip_fixer, uv_compile) @app.command(help="Reinstall custom nodes") @@ -757,21 +743,7 @@ def reinstall( pip_fixer = manager_util.PIPFixer(manager_util.get_installed_packages(), comfy_path, context.manager_files_path) for_each_nodes(nodes, act=reinstall_node) - if uv_compile: - try: - _run_unified_resolve() - except ImportError as e: - print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") - raise typer.Exit(1) - except typer.Exit: - raise - except Exception as e: - print(f"[bold red]Batch resolution failed: {e}[/bold red]") - raise typer.Exit(1) - finally: - pip_fixer.fix_broken() - else: - pip_fixer.fix_broken() + _finalize_resolve(pip_fixer, uv_compile) @app.command(help="Uninstall custom nodes") @@ -843,21 +815,7 @@ def update( update_parallel(nodes) - if uv_compile: - try: - _run_unified_resolve() - except ImportError as e: - print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") - raise typer.Exit(1) - except typer.Exit: - raise - except Exception as e: - print(f"[bold red]Batch resolution failed: {e}[/bold red]") - raise typer.Exit(1) - finally: - pip_fixer.fix_broken() - else: - pip_fixer.fix_broken() + _finalize_resolve(pip_fixer, uv_compile) @app.command(help="Disable custom nodes") @@ -964,21 +922,7 @@ def fix( pip_fixer = manager_util.PIPFixer(manager_util.get_installed_packages(), comfy_path, context.manager_files_path) for_each_nodes(nodes, fix_node, allow_all=True) - if uv_compile: - try: - _run_unified_resolve() - except ImportError as e: - print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") - raise typer.Exit(1) - except typer.Exit: - raise - except Exception as e: - print(f"[bold red]Batch resolution failed: {e}[/bold red]") - raise typer.Exit(1) - finally: - pip_fixer.fix_broken() - else: - pip_fixer.fix_broken() + _finalize_resolve(pip_fixer, uv_compile) @app.command("show-versions", help="Show all available versions of the node") @@ -1249,21 +1193,7 @@ def restore_snapshot( pip_fixer.fix_broken() raise typer.Exit(code=1) - if uv_compile: - try: - _run_unified_resolve() - except ImportError as e: - print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") - raise typer.Exit(1) - except typer.Exit: - raise - except Exception as e: - print(f"[bold red]Batch resolution failed: {e}[/bold red]") - raise typer.Exit(1) - finally: - pip_fixer.fix_broken() - else: - pip_fixer.fix_broken() + _finalize_resolve(pip_fixer, uv_compile) @app.command( @@ -1306,21 +1236,7 @@ def restore_dependencies( unified_manager.execute_install_script('', x, instant_execution=True, no_deps=bool(uv_compile)) i += 1 - if uv_compile: - try: - _run_unified_resolve() - except ImportError as e: - print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") - raise typer.Exit(1) - except typer.Exit: - raise - except Exception as e: - print(f"[bold red]Batch resolution failed: {e}[/bold red]") - raise typer.Exit(1) - finally: - pip_fixer.fix_broken() - else: - pip_fixer.fix_broken() + _finalize_resolve(pip_fixer, uv_compile) @app.command( @@ -1399,30 +1315,36 @@ def install_deps( else: # disabled core.gitclone_set_active([k], False) - if uv_compile: - try: - _run_unified_resolve() - except ImportError as e: - print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") - raise typer.Exit(1) - except typer.Exit: - raise - except Exception as e: - print(f"[bold red]Batch resolution failed: {e}[/bold red]") - raise typer.Exit(1) - finally: - pip_fixer.fix_broken() - else: - pip_fixer.fix_broken() + _finalize_resolve(pip_fixer, uv_compile) print("Dependency installation and activation complete.") +def _finalize_resolve(pip_fixer, uv_compile) -> None: + """Run batch resolution if --uv-compile is set, then fix broken packages.""" + if uv_compile: + try: + _run_unified_resolve() + except ImportError as e: + print(f"[bold red]Failed to import unified_dep_resolver: {e}[/bold red]") + raise typer.Exit(1) + except typer.Exit: + raise + except Exception as e: + print(f"[bold red]Batch resolution failed: {e}[/bold red]") + raise typer.Exit(1) + finally: + pip_fixer.fix_broken() + else: + pip_fixer.fix_broken() + + def _run_unified_resolve(): """Shared logic for unified batch dependency resolution.""" from comfyui_manager.common.unified_dep_resolver import ( UnifiedDepResolver, UvNotAvailableError, + attribute_conflicts, collect_base_requirements, collect_node_pack_paths, ) @@ -1459,14 +1381,8 @@ def _run_unified_resolve(): print("[bold green]Resolution complete (no deps needed).[/bold green]") else: print(f"[bold red]Resolution failed: {result.error}[/bold red]") - # Show which node packs requested each conflicting package. if result.lockfile and result.lockfile.conflicts and result.collected: - conflict_text = "\n".join(result.lockfile.conflicts).lower().replace("-", "_") - attributed = { - pkg: reqs - for pkg, reqs in result.collected.sources.items() - if re.search(r'(? dict[str, list[tuple[str, str]]]: + """Cross-reference conflict packages with their requesting node packs. + + Uses word-boundary regex to prevent false-positive prefix matches + (e.g. ``torch`` does NOT match ``torchvision`` or ``torch_audio``). + """ + conflict_text = "\n".join(conflicts).lower().replace("-", "_") + return { + pkg: reqs + for pkg, reqs in sources.items() + if re.search( + r'(? str: node_name = params.node_name + is_unknown = getattr(params, 'is_unknown', False) # guard: pydantic Union may match UpdatePackParams logging.debug( "[ComfyUI-Manager] Disabling node: name=%s, is_unknown=%s", node_name, - params.is_unknown, + is_unknown, ) try: - res = core.unified_manager.unified_disable(node_name, params.is_unknown) + res = core.unified_manager.unified_disable(node_name, is_unknown) if res: return OperationResult.success.value diff --git a/comfyui_manager/legacy/manager_core.py b/comfyui_manager/legacy/manager_core.py index 5cf3fff9..065bf3ab 100644 --- a/comfyui_manager/legacy/manager_core.py +++ b/comfyui_manager/legacy/manager_core.py @@ -1411,8 +1411,18 @@ class UnifiedManager: else: # nightly repo_url = the_node['repository'] else: - result = ManagedResult('install') - return result.fail(f"Node '{node_id}@{version_spec}' not found in [{channel}, {mode}]") + # Fallback for nightly only: use repository URL from CNR map + # when node is registered in CNR but absent from nightly manifest + if version_spec == 'nightly': + cnr_fallback = self.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}]") + else: + result = ManagedResult('install') + return result.fail(f"Node '{node_id}@{version_spec}' not found in [{channel}, {mode}]") if self.is_enabled(node_id, version_spec): return ManagedResult('skip').with_target(f"{node_id}@{version_spec}") diff --git a/tests/e2e/test_e2e_endpoint.py b/tests/e2e/test_e2e_endpoint.py new file mode 100644 index 00000000..d9aab575 --- /dev/null +++ b/tests/e2e/test_e2e_endpoint.py @@ -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 diff --git a/tests/e2e/test_e2e_uv_compile.py b/tests/e2e/test_e2e_uv_compile.py new file mode 100644 index 00000000..44bf28f5 --- /dev/null +++ b/tests/e2e/test_e2e_uv_compile.py @@ -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 diff --git a/tests/test_nightly_cnr_fallback.py b/tests/test_nightly_cnr_fallback.py new file mode 100644 index 00000000..c88033c7 --- /dev/null +++ b/tests/test_nightly_cnr_fallback.py @@ -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) diff --git a/tests/test_unified_dep_resolver.py b/tests/test_unified_dep_resolver.py index 7fbb88a4..f1280bb1 100644 --- a/tests/test_unified_dep_resolver.py +++ b/tests/test_unified_dep_resolver.py @@ -922,7 +922,7 @@ class TestResolveAndInstall: def test_conflict_attribution_sources_filter(self, tmp_path): """Packages named in conflict lines can be looked up from sources.""" - import re as _re + 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]) @@ -937,20 +937,14 @@ class TestResolveAndInstall: assert not result.success assert result.collected is not None - sources = result.collected.sources - conflict_lower = "\n".join(result.lockfile.conflicts).lower().replace("-", "_") - # Simulate the attribution filter used in _run_unified_resolve() (word-boundary version) - def _matches(pkg): - normalized = pkg.lower().replace("-", "_") - return bool(_re.search(r'(?=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).""" - import re as _re + 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]) @@ -964,18 +958,13 @@ class TestResolveAndInstall: assert not result.success assert result.collected is not None - sources = result.collected.sources - conflict_lower = "\n".join(result.lockfile.conflicts).lower().replace("-", "_") - def _matches(pkg): - normalized = pkg.lower().replace("-", "_") - return bool(_re.search(r'(?=2.1\n") r = _resolver([p]) @@ -989,18 +978,13 @@ class TestResolveAndInstall: assert not result.success assert result.collected is not None - sources = result.collected.sources - conflict_lower = "\n".join(result.lockfile.conflicts).lower().replace("-", "_") - def _matches(pkg): - normalized = pkg.lower().replace("-", "_") - return bool(_re.search(r'(?=2.1\n") r = _resolver([p]) @@ -1015,12 +999,7 @@ class TestResolveAndInstall: assert not result.success assert result.collected is not None - sources = result.collected.sources - conflict_lower = "\n".join(result.lockfile.conflicts).lower().replace("-", "_") - def _matches(pkg): - normalized = pkg.lower().replace("-", "_") - return bool(_re.search(r'(?