Add support for extra custom node path.

EXTRA_CUSTOM_NODE_PATHS_README.md

@@ -0,0 +1,120 @@
+# Extra Custom Node Paths
+
+This feature allows you to load custom nodes from multiple directories, similar to how extra model paths work in ComfyUI.
+
+## Overview
+
+ComfyUI now supports loading custom nodes from multiple directories through a YAML configuration file called `extra_custom_node_paths.yaml`. This is useful when you want to:
+
+- Organize custom nodes in different locations
+- Share custom nodes between multiple ComfyUI installations
+- Keep custom nodes separate from the main ComfyUI directory
+
+## Configuration File
+
+Create a file called `extra_custom_node_paths.yaml` in your ComfyUI root directory:
+
+```yaml
+# Configuration for custom nodes
+# The custom_nodes directory is always called "custom_nodes" and is located under base_path
+# Each config section can only have one base_path
+
+# Example configuration using base_path
+# This will look for custom_nodes under the specified base_path
+custom_nodes_example:
+    base_path: /path/to/base1
+
+# Another example with a different path
+other_custom_nodes:
+    base_path: /path/to/base
+
+# You can add more custom node path configurations as needed
+# additional_custom_nodes:
+#     base_path: another/base/path
+```
+
+### Configuration Options
+
+- **base_path**: The base directory path where the `custom_nodes` subdirectory is located
+  - Can be relative or absolute paths
+  - Relative paths are resolved relative to the YAML file location
+  - The system automatically looks for a `custom_nodes` subdirectory under each base_path
+```
+
+## How It Works
+
+1. **Automatic Loading**: ComfyUI automatically looks for `extra_custom_node_paths.yaml` in the root directory
+2. **Path Resolution**: All base paths are resolved to absolute paths and validated
+3. **Subdirectory Detection**: The system automatically appends `custom_nodes` to each base path
+4. **Integration**: Custom node paths are added to the existing custom nodes system
+5. **Compatibility**: Works seamlessly with existing custom node functionality
+
+## Example Use Cases
+
+### Shared Custom Nodes
+
+```yaml
+# Share custom nodes between multiple ComfyUI installations
+shared_nodes:
+    base_path: /shared/custom_nodes
+```
+
+### Relative Paths
+
+```yaml
+# Use relative paths for portable configurations
+portable_config:
+    base_path: ../shared_custom_nodes
+```
+
+## File Structure
+
+The custom node directories should follow the standard ComfyUI custom node structure:
+
+```
+base_path/
+└── custom_nodes/
+    ├── node_name_1/
+    │   ├── __init__.py
+    │   └── nodes.py
+    ├── node_name_2/
+    │   ├── __init__.py
+    │   └── nodes.py
+    └── ...
+```
+
+## Troubleshooting
+
+### Path Not Found
+- Ensure the directory exists and is accessible
+- Check file permissions
+- Verify the path is correctly formatted in the YAML file
+- Make sure there's a `custom_nodes` subdirectory under the specified base_path
+
+### Custom Nodes Not Loading
+- Check the ComfyUI console for error messages
+- Verify the YAML syntax is correct
+- Ensure the custom node directories contain valid Python modules
+
+## Technical Details
+
+- Custom node paths are loaded during ComfyUI startup
+- Paths are validated for existence before being added
+- The system maintains the order of paths (default paths first)
+- Duplicate paths are automatically handled
+- All paths are normalized and resolved to absolute paths
+- Each config section can only specify one base_path
+
+## Compatibility
+
+This feature is compatible with:
+- All existing custom node functionality
+- The existing custom node loading system
+- Command line arguments and configuration files
+- All supported operating systems
+
+## See Also
+
+- [Extra Model Paths](../extra_model_paths.yaml) - Similar functionality for model paths
+- [Custom Nodes Documentation](../custom_nodes/) - General custom node information
+- [ComfyUI CLI Arguments](../comfy/cli_args.py) - Command line options

extra_custom_node_paths.yaml.example

@@ -0,0 +1,18 @@
+# Rename this to extra_custom_node_paths.yaml and ComfyUI will load it
+
+# Configuration for custom nodes
+# The custom_nodes directory is always called "custom_nodes" and is located under base_path
+# Each config section can only have one base_path
+
+# Example configuration using base_path
+# This will look for custom_nodes under the specified base_path
+comfyui:
+      base_path: ../
+
+# Another example with a different path
+# other_custom_nodes:
+#     base_path: /path/to/base
+
+# You can add more custom node path configurations as needed
+# additional_custom_nodes:
+#     base_path: another/base/path

folder_paths.py

@@ -427,3 +427,13 @@ def get_input_subfolders() -> list[str]:
         return sorted(folders)
     except FileNotFoundError:
         return []
+
+def add_custom_node_directory(directory: str) -> None:
+    """Add a new custom node directory to the list of custom node directories"""
+    global folder_names_and_paths
+    if "custom_nodes" in folder_names_and_paths:
+        paths, _exts = folder_names_and_paths["custom_nodes"]
+        if directory not in paths:
+            paths.append(directory)
+    else:
+        folder_names_and_paths["custom_nodes"] = ([directory], set())

main.py

@@ -8,7 +8,7 @@ import time
 from comfy.cli_args import args
 from app.logger import setup_logger
 import itertools
-import utils.extra_config
+import utils.extra_config, utils.extra_custom_node_config
 import logging
 import sys
 from comfy_execution.progress import get_progress_state
@@ -32,6 +32,11 @@ def apply_custom_paths():
         for config_path in itertools.chain(*args.extra_model_paths_config):
             utils.extra_config.load_extra_path_config(config_path)
 
+    #extra custom node paths
+    extra_custom_node_paths_config_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "extra_custom_node_paths.yaml")
+    if os.path.isfile(extra_custom_node_paths_config_path):
+        utils.extra_custom_node_config.load_extra_custom_node_path_config(extra_custom_node_paths_config_path)
+    
     # --output-directory, --input-directory, --user-directory
     if args.output_directory:
         output_dir = os.path.abspath(args.output_directory)

tests-unit/folder_paths_test/test_cache_helper.py

@@ -0,0 +1,55 @@
+import pytest
+import time
+from unittest.mock import patch
+from comfy import folder_paths
+
+class TestCacheHelper:
+    def test_cache_helper_initialization(self):
+        """Test that CacheHelper initializes with empty cache and inactive state."""
+        cache = folder_paths.CacheHelper()
+        assert not cache.active
+        assert cache.cache == {}
+
+    def test_cache_operations(self):
+        """Test basic cache operations (get, set, clear)."""
+        cache = folder_paths.CacheHelper()
+        cache.active = True
+        
+        # Test setting and getting a value
+        test_value = (["file1.txt"], {"file1.txt": 123.45}, 1.0)
+        cache.set("test_key", test_value)
+        assert cache.get("test_key") == test_value
+        
+        # Test getting non-existent key
+        assert cache.get("non_existent") is None
+        
+        # Test clearing the cache
+        cache.clear()
+        assert cache.cache == {}
+
+    def test_context_manager(self):
+        """Test that the context manager properly handles activation state."""
+        cache = folder_paths.CacheHelper()
+        
+        with cache:
+            assert cache.active
+            assert cache.get("test") is None  # Shouldn't raise
+            
+        assert not cache.active
+        assert cache.cache == {}  # Should be cleared after context
+
+    def test_cache_inactive(self):
+        """Test that cache doesn't store when inactive."""
+        cache = folder_paths.CacheHelper()
+        cache.set("test", (["file.txt"], {}, 1.0))
+        assert cache.get("test") is None
+
+    @patch('time.time', return_value=100.0)
+    def test_cache_helper_with_time(self, mock_time):
+        """Test cache helper with mocked time."""
+        cache = folder_paths.CacheHelper()
+        cache.active = True
+        
+        test_value = (["file.txt"], {"file.txt": 100.0}, 100.0)
+        cache.set("test", test_value)
+        assert cache.get("test") == test_value

tests-unit/folder_paths_test/test_content_type_filtering.py

@@ -0,0 +1,90 @@
+import pytest
+from unittest.mock import patch, MagicMock
+from comfy import folder_paths
+
+class TestContentTypeFiltering:
+    def test_filter_files_content_types(self):
+        """Test filtering files by content types."""
+        test_files = [
+            "image.jpg",
+            "document.pdf",
+            "model.fbx",
+            "video.mp4",
+            "audio.mp3",
+            "unknown.xyz"
+        ]
+        
+        # Test image filtering
+        result = folder_paths.filter_files_content_types(test_files, ["image"])
+        assert result == ["image.jpg"]
+        
+        # Test model filtering
+        result = folder_paths.filter_files_content_types(test_files, ["model"])
+        assert result == ["model.fbx"]
+        
+        # Test multiple content types
+        result = folder_paths.filter_files_content_types(test_files, ["image", "model"])
+        assert set(result) == {"image.jpg", "model.fbx"}
+        
+        # Test with unknown content type
+        result = folder_paths.filter_files_content_types(test_files, ["unknown"])
+        assert result == []
+        
+        # Test with empty file list
+        assert folder_paths.filter_files_content_types([], ["image"]) == []
+        
+        # Test with empty content types
+        assert folder_paths.filter_files_content_types(test_files, []) == []
+    
+    @patch('mimetypes.guess_type')
+    def test_filter_files_content_types_mock_mime(self, mock_guess):
+        """Test content type filtering with mocked mime types."""
+        # Setup mock to return different mime types
+        def mock_guess_type(filename):
+            if filename == "test.jpg":
+                return ("image/jpeg", None)
+            elif filename == "test.mp4":
+                return ("video/mp4", None)
+            return (None, None)
+            
+        mock_guess.side_effect = mock_guess_type
+        
+        files = ["test.jpg", "test.mp4", "test.unknown"]
+        
+        # Test image filtering
+        result = folder_paths.filter_files_content_types(files, ["image"])
+        assert result == ["test.jpg"]
+        
+        # Test video filtering
+        result = folder_paths.filter_files_content_types(files, ["video"])
+        assert result == ["test.mp4"]
+        
+        # Test unknown type filtering
+        result = folder_paths.filter_files_content_types(files, ["audio"])
+        assert result == []
+    
+    def test_extension_mimetypes_cache(self):
+        """Test that extension to mimetype cache works correctly."""
+        # Test with known extensions in cache
+        assert folder_paths.extension_mimetypes_cache["webp"] == "image"
+        assert folder_paths.extension_mimetypes_cache["fbx"] == "model"
+        
+        # Test with unknown extension (should not be in cache)
+        assert "xyz" not in folder_paths.extension_mimetypes_cache
+        
+    @patch('mimetypes.guess_type')
+    def test_custom_extension_handling(self, mock_guess):
+        """Test handling of custom file extensions."""
+        # Setup mock to return None for unknown types
+        mock_guess.return_value = (None, None)
+        
+        # Add a custom extension to the cache
+        folder_paths.extension_mimetypes_cache["custom"] = "custom_type"
+        
+        files = ["file.custom"]
+        result = folder_paths.filter_files_content_types(files, ["custom_type"])
+        assert result == ["file.custom"]
+        
+        # Test with different case
+        result = folder_paths.filter_files_content_types(["FILE.CUSTOM"], ["custom_type"])
+        assert result == ["FILE.CUSTOM"]

tests-unit/folder_paths_test/test_directory_operations.py

@@ -0,0 +1,104 @@
+import os
+import pytest
+import tempfile
+import shutil
+from unittest.mock import patch, MagicMock, call
+from comfy import folder_paths
+
+class TestDirectoryOperations:
+    @pytest.fixture(autouse=True)
+    def setup_method(self):
+        # Create a temporary directory
+        self.temp_dir = tempfile.mkdtemp()
+        self.original_paths = folder_paths.folder_names_and_paths.copy()
+        yield
+        # Cleanup
+        shutil.rmtree(self.temp_dir)
+        folder_paths.folder_names_and_paths = self.original_paths
+
+    def test_add_model_folder_path(self):
+        """Test adding a new model folder path."""
+        test_path = os.path.join(self.temp_dir, "test_models")
+        os.makedirs(test_path, exist_ok=True)
+        
+        # Add a new folder path
+        folder_paths.add_model_folder_path("test_models", test_path)
+        
+        # Verify it was added correctly
+        paths, extensions = folder_paths.folder_names_and_paths["test_models"]
+        assert test_path in paths
+        assert extensions == set()
+        
+        # Verify it's the last item by default
+        assert paths[-1] == test_path
+
+    def test_add_model_folder_path_as_default(self):
+        """Test adding a folder path as default (should be first in list)."""
+        test_path = os.path.join(self.temp_dir, "test_models")
+        os.makedirs(test_path, exist_ok=True)
+        
+        # Add as default
+        folder_paths.add_model_folder_path("test_models", test_path, is_default=True)
+        
+        # Verify it's first in the list
+        paths, _ = folder_paths.folder_names_and_paths["test_models"]
+        assert paths[0] == test_path
+
+    def test_get_folder_paths(self):
+        """Test retrieving folder paths."""
+        test_path = os.path.join(self.temp_dir, "test_models")
+        os.makedirs(test_path, exist_ok=True)
+        
+        # Add a test path
+        folder_paths.add_model_folder_path("test_models", test_path)
+        
+        # Test getting the paths
+        paths = folder_paths.get_folder_paths("test_models")
+        assert paths == [test_path]
+        
+        # Test getting non-existent folder
+        assert folder_paths.get_folder_paths("non_existent") is None
+
+    def test_get_full_path(self):
+        """Test getting full path for a file in a folder."""
+        # Create a test file
+        test_dir = os.path.join(self.temp_dir, "test_models")
+        os.makedirs(test_dir, exist_ok=True)
+        test_file = os.path.join(test_dir, "model.ckpt")
+        with open(test_file, 'w') as f:
+            f.write("test")
+        
+        # Add the test directory
+        folder_paths.add_model_folder_path("test_models", test_dir)
+        
+        # Test getting full path
+        result = folder_paths.get_full_path("test_models", "model.ckpt")
+        assert result == test_file
+        
+        # Test with non-existent file
+        result = folder_paths.get_full_path("test_models", "nonexistent.ckpt")
+        assert result is None
+        
+        # Test with non-existent folder
+        result = folder_paths.get_full_path("non_existent", "model.ckpt")
+        assert result is None
+
+    @patch('os.walk')
+    def test_recursive_search(self, mock_walk):
+        """Test recursive file search."""
+        # Mock os.walk to return test directory structure
+        mock_walk.return_value = [
+            ('/test', ['subdir'], ['file1.txt']),
+            ('/test/subdir', [], ['file2.txt'])
+        ]
+        
+        # Test recursive search
+        result = folder_paths.recursive_search("/test")
+        assert set(result) == {
+            os.path.join('/test', 'file1.txt'),
+            os.path.join('/test/subdir', 'file2.txt')
+        }
+        
+        # Test with excluded directories
+        result = folder_paths.recursive_search("/test", excluded_dir_names=["subdir"])
+        assert result == [os.path.join('/test', 'file1.txt')]

tests-unit/folder_paths_test/test_file_caching.py

@@ -0,0 +1,91 @@
+import os
+import pytest
+import time
+from unittest.mock import patch, MagicMock, call
+from comfy import folder_paths
+
+class TestFileCaching:
+    @pytest.fixture(autouse=True)
+    def setup_method(self, tmp_path):
+        self.temp_dir = tmp_path
+        self.test_dir = self.temp_dir / "test_models"
+        self.test_dir.mkdir()
+        
+        # Create some test files
+        (self.test_dir / "model1.ckpt").write_text("test1")
+        (self.test_dir / "model2.ckpt").write_text("test2")
+        
+        # Save original state
+        self.original_cache = folder_paths.filename_list_cache.copy()
+        self.original_paths = folder_paths.folder_names_and_paths.copy()
+        
+        # Add test directory to paths
+        folder_paths.add_model_folder_path("test_models", str(self.test_dir))
+        
+        yield
+        
+        # Restore original state
+        folder_paths.filename_list_cache = self.original_cache
+        folder_paths.folder_names_and_paths = self.original_paths
+    
+    def test_get_filename_list_caching(self):
+        """Test that file lists are properly cached."""
+        # First call should populate cache
+        result1 = folder_paths.get_filename_list("test_models")
+        assert set(result1) == {"model1.ckpt", "model2.ckpt"}
+        
+        # Verify cache was populated
+        cache_key = str(self.test_dir)
+        assert cache_key in folder_paths.filename_list_cache
+        
+        # Second call should use cache
+        with patch('os.path.getmtime') as mock_mtime:
+            mock_mtime.return_value = 1000
+            result2 = folder_paths.get_filename_list("test_models")
+            assert result2 == result1
+            # Verify getmtime wasn't called (using cache)
+            mock_mtime.assert_not_called()
+    
+    @patch('os.path.getmtime')
+    def test_cache_invalidation(self, mock_mtime):
+        """Test that cache is invalidated when files change."""
+        # Initial call to populate cache
+        mock_mtime.return_value = 1000
+        folder_paths.get_filename_list("test_models")
+        
+        # Change modification time to trigger cache invalidation
+        mock_mtime.return_value = 2000
+        
+        # This should trigger a cache refresh
+        result = folder_paths.get_filename_list("test_models")
+        assert set(result) == {"model1.ckpt", "model2.ckpt"}
+        
+        # Verify getmtime was called for each file
+        assert mock_mtime.call_count >= 2
+    
+    def test_cached_filename_list_helper(self):
+        """Test the cached filename list helper function."""
+        # Test with empty cache
+        with patch('os.path.getmtime') as mock_mtime:
+            mock_mtime.return_value = 1000
+            result = folder_paths.cached_filename_list_("test_models")
+            assert set(result[0]) == {"model1.ckpt", "model2.ckpt"}
+            assert len(result[1]) == 2  # Should have mtimes for both files
+            assert result[2] == 1000    # Should have the current time
+        
+        # Test with valid cache
+        with patch('os.path.getmtime') as mock_mtime:
+            mock_mtime.return_value = 1000
+            # Call again, should use cache
+            result = folder_paths.cached_filename_list_("test_models")
+            mock_mtime.assert_not_called()  # Shouldn't check mtimes when using cache
+    
+    def test_get_filename_list_nonexistent_dir(self):
+        """Test behavior with non-existent directory."""
+        # Add a non-existent directory to the paths
+        non_existent = self.temp_dir / "nonexistent"
+        folder_paths.add_model_folder_path("test_models", str(non_existent))
+        
+        # Should not raise and should return empty list
+        result = folder_paths.get_filename_list("test_models")
+        assert result == []

tests-unit/folder_paths_test/test_path_manipulation.py

@@ -0,0 +1,70 @@
+import os
+import pytest
+from unittest.mock import patch, MagicMock
+from comfy import folder_paths
+
+def test_map_legacy():
+    """Test the legacy path mapping function."""
+    assert folder_paths.map_legacy("unet") == "diffusion_models"
+    assert folder_paths.map_legacy("clip") == "text_encoders"
+    assert folder_paths.map_legacy("unknown") == "unknown"
+
+def test_annotated_filepath():
+    """Test parsing of annotated file paths."""
+    # Test with no annotation
+    assert folder_paths.annotated_filepath("test.txt") == ("test.txt", None)
+    
+    # Test with annotation
+    assert folder_paths.annotated_filepath("test.txt [output]") == \
+           ("test.txt", folder_paths.get_output_directory())
+    
+    # Test with annotation and spaces
+    assert folder_paths.annotated_filepath("test file.txt [output]") == \
+           ("test file.txt", folder_paths.get_output_directory())
+    
+    # Test with invalid annotation
+    assert folder_paths.annotated_filepath("test.txt [invalid]") == \
+           ("test.txt [invalid]", None)
+
+def test_get_annotated_filepath():
+    """Test getting absolute path from annotated filename."""
+    with patch('os.path.exists', return_value=True):
+        # Test with no annotation
+        result = folder_paths.get_annotated_filepath("test.txt", "/default/dir")
+        assert result == os.path.join("/default/dir", "test.txt")
+        
+        # Test with annotation
+        result = folder_paths.get_annotated_filepath("test.txt [output]")
+        assert result == os.path.join(folder_paths.get_output_directory(), "test.txt")
+
+def test_exists_annotated_filepath():
+    """Test checking if an annotated file path exists."""
+    with patch('os.path.exists') as mock_exists:
+        mock_exists.return_value = True
+        assert folder_paths.exists_annotated_filepath("test.txt [output]")
+        mock_exists.assert_called_once()
+        
+        mock_exists.return_value = False
+        assert not folder_paths.exists_annotated_filepath("nonexistent.txt [output]")
+
+def test_filter_files_extensions():
+    """Test filtering files by extensions."""
+    files = ["test.txt", "image.jpg", "document.pdf", "script.py"]
+    
+    # Test with single extension
+    result = folder_paths.filter_files_extensions(files, [".txt"])
+    assert result == ["test.txt"]
+    
+    # Test with multiple extensions
+    result = folder_paths.filter_files_extensions(files, [".jpg", ".pdf"])
+    assert set(result) == {"image.jpg", "document.pdf"}
+    
+    # Test with no matches
+    result = folder_paths.filter_files_extensions(files, [".png"])
+    assert result == []
+    
+    # Test with empty file list
+    assert folder_paths.filter_files_extensions([], [".txt"]) == []
+    
+    # Test with empty extensions list
+    assert folder_paths.filter_files_extensions(files, []) == []

tests-unit/utils/test_extra_custom_node_config.py

@@ -0,0 +1,142 @@
+import pytest
+import os
+import sys
+from unittest.mock import Mock, patch, mock_open
+import yaml
+
+from utils.extra_custom_node_config import load_extra_custom_node_path_config, get_current_custom_node_paths
+import folder_paths
+
+@pytest.fixture()
+def clear_folder_paths():
+    # Save original state
+    original_paths = folder_paths.folder_names_and_paths.copy()
+    original_custom_nodes = folder_paths.get_custom_nodes_directories()
+    
+    # Clear the custom nodes directories
+    for path in original_custom_nodes:
+        folder_paths.folder_names_and_paths["custom_nodes"][0].remove(path)
+    
+    yield
+    
+    # Restore original state
+    folder_paths.folder_names_and_paths = original_paths
+
+@pytest.fixture
+def mock_yaml_content():
+    return {
+        'custom_nodes_config': {
+            'base_path': '~/custom_nodes_dir',
+        },
+        'another_config': {
+            'base_path': '/absolute/path/to/nodes'
+        }
+    }
+
+@pytest.fixture
+def yaml_config_with_vars():
+    return """
+    custom_nodes_config:
+      base_path: '%APPDATA%/ComfyUI/custom_nodes'
+    """
+
+@patch('builtins.open', new_callable=mock_open, read_data="dummy file content")
+@patch('os.path.expanduser')
+@patch('yaml.safe_load')
+def test_load_extra_custom_node_paths(
+    mock_yaml_load, mock_expanduser, mock_file, mock_yaml_content, clear_folder_paths
+):
+    # Setup mocks
+    mock_yaml_load.return_value = mock_yaml_content
+    mock_expanduser.side_effect = lambda x: x.replace('~/', '/home/user/')
+    
+    # Mock add_custom_node_directory
+    with patch('folder_paths.add_custom_node_directory') as mock_add_dir:
+        load_extra_custom_node_path_config('dummy_path.yaml')
+        
+        # Verify the directories were added
+        expected_paths = [
+            '/home/user/custom_nodes_dir/custom_nodes',
+            '/absolute/path/to/nodes/custom_nodes'
+        ]
+        
+        # Check that add_custom_node_directory was called with the expected paths
+        assert mock_add_dir.call_count == 2
+        actual_paths = [call[0][0] for call in mock_add_dir.call_args_list]
+        assert set(actual_paths) == set(expected_paths)
+
+@patch('builtins.open', new_callable=mock_open, read_data="dummy file content")
+@patch('os.path.expandvars')
+@patch('yaml.safe_load')
+def test_load_extra_custom_node_paths_with_env_vars(
+    mock_yaml_load, mock_expandvars, mock_file, yaml_config_with_vars, clear_folder_paths
+):
+    # Setup mocks
+    mock_yaml_load.return_value = yaml.safe_load(yaml_config_with_vars)
+    
+    def expandvars_side_effect(path):
+        if '%APPDATA%' in path:
+            if sys.platform == 'win32':
+                return path.replace('%APPDATA%', 'C:\\Users\\TestUser\\AppData\\Roaming')
+            else:
+                return path.replace('%APPDATA%', '/Users/TestUser/AppData/Roaming')
+        return path
+    
+    mock_expandvars.side_effect = expandvars_side_effect
+    
+    # Mock add_custom_node_directory
+    with patch('folder_paths.add_custom_node_directory') as mock_add_dir:
+        load_extra_custom_node_path_config('dummy_path.yaml')
+        
+        # Verify the directory was added with expanded path
+        expected_path = os.path.join(
+            expandvars_side_effect('%APPDATA%/ComfyUI/custom_nodes'),
+            'custom_nodes'
+        )
+        mock_add_dir.assert_called_once_with(expected_path)
+
+@patch('builtins.open', new_callable=mock_open, read_data="dummy file content")
+@patch('logging.warning')
+@patch('os.path.exists', return_value=False)
+def test_load_extra_custom_node_paths_nonexistent(
+    mock_exists, mock_warning, mock_file, mock_yaml_content, clear_folder_paths
+):
+    # Setup mocks
+    with patch('yaml.safe_load', return_value=mock_yaml_content):
+        with patch('os.path.expanduser', side_effect=lambda x: x.replace('~/', '/home/user/')):
+            with patch('folder_paths.add_custom_node_directory') as mock_add_dir:
+                load_extra_custom_node_path_config('dummy_path.yaml')
+                
+                # Verify warning was logged for non-existent paths
+                assert mock_warning.call_count == 2
+                for call in mock_warning.call_args_list:
+                    assert "does not exist, skipping" in call[0][0]
+                
+                # Verify no directories were added
+                mock_add_dir.assert_not_called()
+
+def test_get_current_custom_node_paths(clear_folder_paths):
+    # Add some test paths
+    test_paths = ['/path/one', '/path/two']
+    for path in test_paths:
+        folder_paths.add_custom_node_directory(path)
+    
+    # Test getting the paths
+    result = get_current_custom_node_paths()
+    assert set(result) == set(test_paths)
+
+@patch('builtins.open', side_effect=Exception("Test error"))
+@patch('logging.error')
+def test_load_extra_custom_node_paths_error(mock_error, mock_file, clear_folder_paths):
+    # Test error handling when loading YAML fails
+    load_extra_custom_node_path_config('invalid.yaml')
+    mock_error.assert_called_once()
+    assert "Failed to load extra custom node paths config from invalid.yaml" in str(mock_error.call_args[0][0])
+
+@patch('builtins.open', new_callable=mock_open, read_data="dummy file content")
+@patch('yaml.safe_load', return_value={})
+@patch('logging.info')
+def test_load_extra_custom_node_paths_empty_config(mock_info, mock_yaml_load, mock_file, clear_folder_paths):
+    # Test with empty config
+    load_extra_custom_node_path_config('empty.yaml')
+    mock_info.assert_called_with("No custom node paths found in configuration")

utils/extra_custom_node_config.py

@@ -0,0 +1,59 @@
+import os
+import yaml
+import folder_paths
+import logging
+
+def load_extra_custom_node_path_config(yaml_path):
+    """
+    Load extra custom node paths from a YAML configuration file.
+    Similar to load_extra_path_config but specifically for custom nodes.
+    Each config section can have one base_path, and custom_nodes will be looked for under that path.
+    """
+    try:
+        with open(yaml_path, 'r', encoding='utf-8') as stream:
+            config = yaml.safe_load(stream)
+    except Exception as e:
+        logging.error(f"Failed to load extra custom node paths config from {yaml_path}: {e}")
+        return
+    
+    yaml_dir = os.path.dirname(os.path.abspath(yaml_path))
+    
+    # Collect all custom node paths from the YAML configuration
+    all_custom_node_paths = []
+    
+    if not config:
+        return
+
+    for c in config:
+        conf = config[c]
+        if conf is None:
+            continue
+        
+        # Handle base_path (creates custom_nodes subdirectory)
+        if "base_path" in conf:
+            base_path = conf["base_path"]
+            # Process the base path
+            expanded_path = os.path.expandvars(os.path.expanduser(base_path))
+            if not os.path.isabs(expanded_path):
+                expanded_path = os.path.abspath(os.path.join(yaml_dir, expanded_path))
+            
+            # Create the custom_nodes subdirectory path
+            custom_nodes_path = os.path.join(expanded_path, "custom_nodes")
+            all_custom_node_paths.append(custom_nodes_path)
+    
+    # Add all custom node paths to the custom_nodes folder list
+    if all_custom_node_paths:
+        for custom_node_path in all_custom_node_paths:
+            if os.path.exists(custom_node_path):
+                logging.info(f"Adding extra custom node search path: {custom_node_path}")
+                folder_paths.add_custom_node_directory(custom_node_path)
+            else:
+                logging.warning(f"Custom node path does not exist, skipping: {custom_node_path}")
+        
+        logging.info(f"Added {len(all_custom_node_paths)} custom node directories")
+    else:
+        logging.info("No custom node paths found in configuration")
+
+def get_current_custom_node_paths() -> list[str]:
+    """Get the current list of all custom node paths for debugging purposes"""
+    return folder_paths.get_custom_nodes_directories()