Merge f7df03f4e0 into 01799f8cac

comfyui_manager/common/manager_util.py

@@ -28,6 +28,30 @@ use_uv = False
 use_unified_resolver = False
 bypass_ssl = False
 
+# When False, ComfyUI-Manager will not attempt to install, upgrade, or
+# uninstall Python packages on its own (prestartup auto-healing, the unified
+# dependency resolver, and per-node `pip install -r requirements.txt`). This
+# is intended for distro-packaged or otherwise externally-managed Python
+# environments (e.g. Nix, Guix, system package managers, locked-down
+# corporate images) where the interpreter is read-only.
+#
+# Read-only pip operations (`list`, `freeze`, `show`) still run so the UI,
+# inventory, and security checks keep working.
+dependency_management_enabled = True
+
+
+_OFF_VALUES = frozenset({'off', 'false', '0', 'no', 'disabled'})
+
+
+def is_off_value(value):
+    """Return True if a config or env-var string spells "off".
+
+    Anything else — including the empty string and unknown words — is treated
+    as "on", so a typo doesn't silently disable dependency management.
+    """
+    return value.strip().lower() in _OFF_VALUES
+
+
 def is_manager_pip_package():
     return not os.path.exists(os.path.join(comfyui_manager_path, '..', 'custom_nodes'))
 
@@ -412,6 +436,10 @@ class PIPFixer:
         self.manager_files_path = manager_files_path
 
     def fix_broken(self):
+        if not dependency_management_enabled:
+            logging.info("[ComfyUI-Manager] dependency management disabled; skipping PIPFixer.fix_broken()")
+            return
+
         new_pip_versions = get_installed_packages(True)
 
         # remove `comfy` python package

comfyui_manager/prestartup_script.py

@@ -93,6 +93,24 @@ def read_unified_resolver_mode():
     if 'use_unified_resolver' in default_conf:
         manager_util.use_unified_resolver = default_conf['use_unified_resolver'].lower() == 'true'
 
+
+def read_dependency_management_mode():
+    # Config: [default] dependency_management = on|off (default on).
+    # Env override: COMFYUI_MANAGER_DEPENDENCY_MANAGEMENT=on|off wins over config.
+    # `off` opts out of every Python-package mutation Manager performs at
+    # startup (PIPFixer auto-heal, unified resolver, per-node pip install).
+    # Intended for distro-packaged / externally-managed Python environments
+    # where the interpreter is read-only (Nix, Guix, system packages, etc.).
+    env_value = os.environ.get('COMFYUI_MANAGER_DEPENDENCY_MANAGEMENT')
+    if env_value is not None:
+        manager_util.dependency_management_enabled = not manager_util.is_off_value(env_value)
+    elif 'dependency_management' in default_conf:
+        manager_util.dependency_management_enabled = not manager_util.is_off_value(default_conf['dependency_management'])
+
+    if not manager_util.dependency_management_enabled:
+        logging.info("[ComfyUI-Manager] dependency management disabled; auto-install/upgrade/uninstall paths will be skipped.")
+
+
 def check_file_logging():
     global enable_file_logging
     if 'file_logging' in default_conf and default_conf['file_logging'].lower() == 'false':
@@ -102,6 +120,7 @@ def check_file_logging():
 read_config()
 read_uv_mode()
 read_unified_resolver_mode()
+read_dependency_management_mode()
 security_check.security_check()
 check_file_logging()
 
@@ -591,7 +610,11 @@ def execute_lazy_install_script(repo_path, executable):
     install_script_path = os.path.join(repo_path, "install.py")
     requirements_path = os.path.join(repo_path, "requirements.txt")
 
-    if os.path.exists(requirements_path) and not _unified_resolver_succeeded:
+    if (
+        os.path.exists(requirements_path)
+        and not _unified_resolver_succeeded
+        and manager_util.dependency_management_enabled
+    ):
         # Per-node pip install: only runs if unified resolver is disabled or failed
         print(f"Install: pip packages for '{repo_path}'")
 
@@ -764,6 +787,10 @@ def execute_startup_script():
 
 # --- Unified dependency resolver: batch resolution at startup ---
 # Runs unconditionally when enabled, independent of install-scripts.txt existence.
+if manager_util.use_unified_resolver and not manager_util.dependency_management_enabled:
+    logging.info("[ComfyUI-Manager] dependency management disabled; skipping unified dependency resolver")
+    manager_util.use_unified_resolver = False
+
 if manager_util.use_unified_resolver:
     try:
         from .common.unified_dep_resolver import (

tests/common/test_dependency_management_flag.py

@@ -0,0 +1,109 @@
+"""
+Tests for the `dependency_management` opt-out switch.
+
+Covers the module-level flag, the value parser shared with prestartup, and
+the early-return in `PIPFixer.fix_broken()`. The prestartup gates (unified
+resolver, lazy install) are exercised end-to-end; the unit tests here lock
+in the read paths and the short-circuit semantics.
+"""
+
+import subprocess
+import sys
+import types
+import unittest
+from unittest import mock
+
+
+def _import_manager_util():
+    # `comfyui_manager/__init__.py` imports `comfy.cli_args` at module load
+    # time; stub the host-side modules so the unit under test can be imported
+    # without a full ComfyUI runtime.
+    for name in ("comfy", "comfy.cli_args"):
+        sys.modules.setdefault(name, types.ModuleType(name))
+    sys.modules["comfy.cli_args"].args = types.SimpleNamespace(base_directory=None)
+
+    from comfyui_manager.common import manager_util  # noqa: WPS433
+    return manager_util
+
+
+class DependencyManagementFlagTest(unittest.TestCase):
+    def setUp(self):
+        self.manager_util = _import_manager_util()
+        self._prev = self.manager_util.dependency_management_enabled
+        self.addCleanup(setattr, self.manager_util, "dependency_management_enabled", self._prev)
+
+    def test_default_is_enabled(self):
+        # Default must preserve today's behavior.
+        self.assertTrue(self._prev)
+
+    def test_pip_fixer_short_circuits_when_disabled(self):
+        self.manager_util.dependency_management_enabled = False
+
+        class _Stub(self.manager_util.PIPFixer):
+            def __init__(self):  # bypass parent init
+                pass
+
+        with mock.patch.object(subprocess, "check_output") as patched:
+            _Stub().fix_broken()
+
+        patched.assert_not_called()
+
+    def test_pip_fixer_runs_when_enabled(self):
+        # Enabled path should at least call subprocess (via get_installed_packages).
+        self.manager_util.dependency_management_enabled = True
+
+        class _Stub(self.manager_util.PIPFixer):
+            def __init__(self):
+                self.prev_pip_versions = {}
+                self.comfyui_path = "/nonexistent"
+                self.manager_files_path = "/nonexistent"
+
+        with mock.patch.object(subprocess, "check_output", return_value="Package Version\n------- -------\n") as patched:
+            try:
+                _Stub().fix_broken()
+            except Exception:
+                # Other internal paths (opencv, frontend) may raise on a stub
+                # tree — we only care that subprocess was reached at least once.
+                pass
+
+        self.assertGreaterEqual(patched.call_count, 1)
+
+    def test_make_pip_cmd_not_gated(self):
+        # make_pip_cmd is consulted for both read and write ops (pip list,
+        # pip install, …). Gating it would break read paths the UI relies on.
+        # Assert the helper still produces a runnable command when the flag
+        # is off — gating happens at call sites that perform writes.
+        self.manager_util.dependency_management_enabled = False
+
+        cmd = self.manager_util.make_pip_cmd(["list"])
+        self.assertIsInstance(cmd, list)
+        self.assertGreaterEqual(len(cmd), 3)
+        self.assertIn("list", cmd)
+
+        cmd = self.manager_util.make_pip_cmd(["install", "some-pkg"])
+        self.assertIn("install", cmd)
+        self.assertIn("some-pkg", cmd)
+
+
+class IsOffValueTest(unittest.TestCase):
+    """`is_off_value` is the single source of truth for the config/env vocabulary."""
+
+    def setUp(self):
+        self.manager_util = _import_manager_util()
+
+    def test_off_synonyms(self):
+        for value in ("off", "OFF", " false ", "0", "no", "Disabled"):
+            with self.subTest(value=value):
+                self.assertTrue(self.manager_util.is_off_value(value))
+
+    def test_on_values(self):
+        # Anything not in the off vocabulary — including the empty string and
+        # typos — keeps dependency management enabled. Better to be noisy than
+        # to silently swallow installs.
+        for value in ("on", "true", "1", "yes", "enabled", "auto", "", "  ", "off-ish"):
+            with self.subTest(value=value):
+                self.assertFalse(self.manager_util.is_off_value(value))
+
+
+if __name__ == "__main__":
+    unittest.main()