Delete tests-unit

tests-unit/README.md

@@ -1,8 +0,0 @@
-# Pytest Unit Tests
-
-## Install test dependencies
-
-`pip install -r tests-unit/requirements.txt`
-
-## Run tests
-`pytest tests-unit/`

tests-unit/app_test/custom_node_manager_test.py

@@ -1,147 +0,0 @@
-import pytest
-from aiohttp import web
-from unittest.mock import patch
-from app.custom_node_manager import CustomNodeManager
-import json
-
-pytestmark = (
-    pytest.mark.asyncio
-)  # This applies the asyncio mark to all test functions in the module
-
-
-@pytest.fixture
-def custom_node_manager():
-    return CustomNodeManager()
-
-
-@pytest.fixture
-def app(custom_node_manager):
-    app = web.Application()
-    routes = web.RouteTableDef()
-    custom_node_manager.add_routes(
-        routes, app, [("ComfyUI-TestExtension1", "ComfyUI-TestExtension1")]
-    )
-    app.add_routes(routes)
-    return app
-
-
-async def test_get_workflow_templates(aiohttp_client, app, tmp_path):
-    client = await aiohttp_client(app)
-    # Setup temporary custom nodes file structure with 1 workflow file
-    custom_nodes_dir = tmp_path / "custom_nodes"
-    example_workflows_dir = (
-        custom_nodes_dir / "ComfyUI-TestExtension1" / "example_workflows"
-    )
-    example_workflows_dir.mkdir(parents=True)
-    template_file = example_workflows_dir / "workflow1.json"
-    template_file.write_text("")
-
-    with patch(
-        "folder_paths.folder_names_and_paths",
-        {"custom_nodes": ([str(custom_nodes_dir)], None)},
-    ):
-        response = await client.get("/workflow_templates")
-        assert response.status == 200
-        workflows_dict = await response.json()
-        assert isinstance(workflows_dict, dict)
-        assert "ComfyUI-TestExtension1" in workflows_dict
-        assert isinstance(workflows_dict["ComfyUI-TestExtension1"], list)
-        assert workflows_dict["ComfyUI-TestExtension1"][0] == "workflow1"
-
-
-async def test_build_translations_empty_when_no_locales(custom_node_manager, tmp_path):
-    custom_nodes_dir = tmp_path / "custom_nodes"
-    custom_nodes_dir.mkdir(parents=True)
-
-    with patch("folder_paths.get_folder_paths", return_value=[str(custom_nodes_dir)]):
-        translations = custom_node_manager.build_translations()
-        assert translations == {}
-
-
-async def test_build_translations_loads_all_files(custom_node_manager, tmp_path):
-    # Setup test directory structure
-    custom_nodes_dir = tmp_path / "custom_nodes" / "test-extension"
-    locales_dir = custom_nodes_dir / "locales" / "en"
-    locales_dir.mkdir(parents=True)
-
-    # Create test translation files
-    main_content = {"title": "Test Extension"}
-    (locales_dir / "main.json").write_text(json.dumps(main_content))
-
-    node_defs = {"node1": "Node 1"}
-    (locales_dir / "nodeDefs.json").write_text(json.dumps(node_defs))
-
-    commands = {"cmd1": "Command 1"}
-    (locales_dir / "commands.json").write_text(json.dumps(commands))
-
-    settings = {"setting1": "Setting 1"}
-    (locales_dir / "settings.json").write_text(json.dumps(settings))
-
-    with patch(
-        "folder_paths.get_folder_paths", return_value=[tmp_path / "custom_nodes"]
-    ):
-        translations = custom_node_manager.build_translations()
-
-        assert translations == {
-            "en": {
-                "title": "Test Extension",
-                "nodeDefs": {"node1": "Node 1"},
-                "commands": {"cmd1": "Command 1"},
-                "settings": {"setting1": "Setting 1"},
-            }
-        }
-
-
-async def test_build_translations_handles_invalid_json(custom_node_manager, tmp_path):
-    # Setup test directory structure
-    custom_nodes_dir = tmp_path / "custom_nodes" / "test-extension"
-    locales_dir = custom_nodes_dir / "locales" / "en"
-    locales_dir.mkdir(parents=True)
-
-    # Create valid main.json
-    main_content = {"title": "Test Extension"}
-    (locales_dir / "main.json").write_text(json.dumps(main_content))
-
-    # Create invalid JSON file
-    (locales_dir / "nodeDefs.json").write_text("invalid json{")
-
-    with patch(
-        "folder_paths.get_folder_paths", return_value=[tmp_path / "custom_nodes"]
-    ):
-        translations = custom_node_manager.build_translations()
-
-        assert translations == {
-            "en": {
-                "title": "Test Extension",
-            }
-        }
-
-
-async def test_build_translations_merges_multiple_extensions(
-    custom_node_manager, tmp_path
-):
-    # Setup test directory structure for two extensions
-    custom_nodes_dir = tmp_path / "custom_nodes"
-    ext1_dir = custom_nodes_dir / "extension1" / "locales" / "en"
-    ext2_dir = custom_nodes_dir / "extension2" / "locales" / "en"
-    ext1_dir.mkdir(parents=True)
-    ext2_dir.mkdir(parents=True)
-
-    # Create translation files for extension 1
-    ext1_main = {"title": "Extension 1", "shared": "Original"}
-    (ext1_dir / "main.json").write_text(json.dumps(ext1_main))
-
-    # Create translation files for extension 2
-    ext2_main = {"description": "Extension 2", "shared": "Override"}
-    (ext2_dir / "main.json").write_text(json.dumps(ext2_main))
-
-    with patch("folder_paths.get_folder_paths", return_value=[str(custom_nodes_dir)]):
-        translations = custom_node_manager.build_translations()
-
-        assert translations == {
-            "en": {
-                "title": "Extension 1",
-                "description": "Extension 2",
-                "shared": "Override",  # Second extension should override first
-            }
-        }

tests-unit/app_test/frontend_manager_test.py

@@ -1,174 +0,0 @@
-import argparse
-import pytest
-from requests.exceptions import HTTPError
-from unittest.mock import patch
-
-from app.frontend_management import (
-    FrontendManager,
-    FrontEndProvider,
-    Release,
-)
-from comfy.cli_args import DEFAULT_VERSION_STRING
-
-
-@pytest.fixture
-def mock_releases():
-    return [
-        Release(
-            id=1,
-            tag_name="1.0.0",
-            name="Release 1.0.0",
-            prerelease=False,
-            created_at="2022-01-01T00:00:00Z",
-            published_at="2022-01-01T00:00:00Z",
-            body="Release notes for 1.0.0",
-            assets=[{"name": "dist.zip", "url": "https://example.com/dist.zip"}],
-        ),
-        Release(
-            id=2,
-            tag_name="2.0.0",
-            name="Release 2.0.0",
-            prerelease=False,
-            created_at="2022-02-01T00:00:00Z",
-            published_at="2022-02-01T00:00:00Z",
-            body="Release notes for 2.0.0",
-            assets=[{"name": "dist.zip", "url": "https://example.com/dist.zip"}],
-        ),
-    ]
-
-
-@pytest.fixture
-def mock_provider(mock_releases):
-    provider = FrontEndProvider(
-        owner="test-owner",
-        repo="test-repo",
-    )
-    provider.all_releases = mock_releases
-    provider.latest_release = mock_releases[1]
-    FrontendManager.PROVIDERS = [provider]
-    return provider
-
-
-def test_get_release(mock_provider, mock_releases):
-    version = "1.0.0"
-    release = mock_provider.get_release(version)
-    assert release == mock_releases[0]
-
-
-def test_get_release_latest(mock_provider, mock_releases):
-    version = "latest"
-    release = mock_provider.get_release(version)
-    assert release == mock_releases[1]
-
-
-def test_get_release_invalid_version(mock_provider):
-    version = "invalid"
-    with pytest.raises(ValueError):
-        mock_provider.get_release(version)
-
-
-def test_init_frontend_default():
-    version_string = DEFAULT_VERSION_STRING
-    frontend_path = FrontendManager.init_frontend(version_string)
-    assert frontend_path == FrontendManager.default_frontend_path()
-
-
-def test_init_frontend_invalid_version():
-    version_string = "test-owner/test-repo@1.100.99"
-    with pytest.raises(HTTPError):
-        FrontendManager.init_frontend_unsafe(version_string)
-
-
-def test_init_frontend_invalid_provider():
-    version_string = "invalid/invalid@latest"
-    with pytest.raises(HTTPError):
-        FrontendManager.init_frontend_unsafe(version_string)
-
-
-@pytest.fixture
-def mock_os_functions():
-    with (
-        patch("app.frontend_management.os.makedirs") as mock_makedirs,
-        patch("app.frontend_management.os.listdir") as mock_listdir,
-        patch("app.frontend_management.os.rmdir") as mock_rmdir,
-    ):
-        mock_listdir.return_value = []  # Simulate empty directory
-        yield mock_makedirs, mock_listdir, mock_rmdir
-
-
-@pytest.fixture
-def mock_download():
-    with patch("app.frontend_management.download_release_asset_zip") as mock:
-        mock.side_effect = Exception("Download failed")  # Simulate download failure
-        yield mock
-
-
-def test_finally_block(mock_os_functions, mock_download, mock_provider):
-    # Arrange
-    mock_makedirs, mock_listdir, mock_rmdir = mock_os_functions
-    version_string = "test-owner/test-repo@1.0.0"
-
-    # Act & Assert
-    with pytest.raises(Exception):
-        FrontendManager.init_frontend_unsafe(version_string, mock_provider)
-
-    # Assert
-    mock_makedirs.assert_called_once()
-    mock_download.assert_called_once()
-    mock_listdir.assert_called_once()
-    mock_rmdir.assert_called_once()
-
-
-def test_parse_version_string():
-    version_string = "owner/repo@1.0.0"
-    repo_owner, repo_name, version = FrontendManager.parse_version_string(
-        version_string
-    )
-    assert repo_owner == "owner"
-    assert repo_name == "repo"
-    assert version == "1.0.0"
-
-
-def test_parse_version_string_invalid():
-    version_string = "invalid"
-    with pytest.raises(argparse.ArgumentTypeError):
-        FrontendManager.parse_version_string(version_string)
-
-
-def test_init_frontend_default_with_mocks():
-    # Arrange
-    version_string = DEFAULT_VERSION_STRING
-
-    # Act
-    with (
-        patch("app.frontend_management.check_frontend_version") as mock_check,
-        patch.object(
-            FrontendManager, "default_frontend_path", return_value="/mocked/path"
-        ),
-    ):
-        frontend_path = FrontendManager.init_frontend(version_string)
-
-    # Assert
-    assert frontend_path == "/mocked/path"
-    mock_check.assert_called_once()
-
-
-def test_init_frontend_fallback_on_error():
-    # Arrange
-    version_string = "test-owner/test-repo@1.0.0"
-
-    # Act
-    with (
-        patch.object(
-            FrontendManager, "init_frontend_unsafe", side_effect=Exception("Test error")
-        ),
-        patch("app.frontend_management.check_frontend_version") as mock_check,
-        patch.object(
-            FrontendManager, "default_frontend_path", return_value="/default/path"
-        ),
-    ):
-        frontend_path = FrontendManager.init_frontend(version_string)
-
-    # Assert
-    assert frontend_path == "/default/path"
-    mock_check.assert_called_once()

tests-unit/app_test/model_manager_test.py

@@ -1,62 +0,0 @@
-import pytest
-import base64
-import json
-import struct
-from io import BytesIO
-from PIL import Image
-from aiohttp import web
-from unittest.mock import patch
-from app.model_manager import ModelFileManager
-
-pytestmark = (
-    pytest.mark.asyncio
-)  # This applies the asyncio mark to all test functions in the module
-
-@pytest.fixture
-def model_manager():
-    return ModelFileManager()
-
-@pytest.fixture
-def app(model_manager):
-    app = web.Application()
-    routes = web.RouteTableDef()
-    model_manager.add_routes(routes)
-    app.add_routes(routes)
-    return app
-
-async def test_get_model_preview_safetensors(aiohttp_client, app, tmp_path):
-    img = Image.new('RGB', (100, 100), 'white')
-    img_byte_arr = BytesIO()
-    img.save(img_byte_arr, format='PNG')
-    img_byte_arr.seek(0)
-    img_b64 = base64.b64encode(img_byte_arr.getvalue()).decode('utf-8')
-
-    safetensors_file = tmp_path / "test_model.safetensors"
-    header_bytes = json.dumps({
-        "__metadata__": {
-            "ssmd_cover_images": json.dumps([img_b64])
-        }
-    }).encode('utf-8')
-    length_bytes = struct.pack('<Q', len(header_bytes))
-    with open(safetensors_file, 'wb') as f:
-        f.write(length_bytes)
-        f.write(header_bytes)
-
-    with patch('folder_paths.folder_names_and_paths', {
-        'test_folder': ([str(tmp_path)], None)
-    }):
-        client = await aiohttp_client(app)
-        response = await client.get('/experiment/models/preview/test_folder/0/test_model.safetensors')
-
-        # Verify response
-        assert response.status == 200
-        assert response.content_type == 'image/webp'
-
-        # Verify the response contains valid image data
-        img_bytes = BytesIO(await response.read())
-        img = Image.open(img_bytes)
-        assert img.format
-        assert img.format.lower() == 'webp'
-
-        # Clean up
-        img.close()

tests-unit/comfy_api_nodes_test/mapper_utils_test.py

@@ -1,297 +0,0 @@
-from typing import Optional
-from enum import Enum
-
-from pydantic import BaseModel, Field
-
-from comfy.comfy_types.node_typing import IO
-from comfy_api_nodes.mapper_utils import model_field_to_node_input
-
-
-def test_model_field_to_float_input():
-    """Tests mapping a float field with constraints."""
-
-    class ModelWithFloatField(BaseModel):
-        cfg_scale: Optional[float] = Field(
-            default=0.5,
-            description="Flexibility in video generation",
-            ge=0.0,
-            le=1.0,
-            multiple_of=0.001,
-        )
-
-    expected_output = (
-        IO.FLOAT,
-        {
-            "default": 0.5,
-            "tooltip": "Flexibility in video generation",
-            "min": 0.0,
-            "max": 1.0,
-            "step": 0.001,
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.FLOAT, ModelWithFloatField, "cfg_scale"
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_float_input_no_constraints():
-    """Tests mapping a float field with no constraints."""
-
-    class ModelWithFloatField(BaseModel):
-        cfg_scale: Optional[float] = Field(default=0.5)
-
-    expected_output = (
-        IO.FLOAT,
-        {
-            "default": 0.5,
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.FLOAT, ModelWithFloatField, "cfg_scale"
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_int_input():
-    """Tests mapping an int field with constraints."""
-
-    class ModelWithIntField(BaseModel):
-        num_frames: Optional[int] = Field(
-            default=10,
-            description="Number of frames to generate",
-            ge=1,
-            le=100,
-            multiple_of=1,
-        )
-
-    expected_output = (
-        IO.INT,
-        {
-            "default": 10,
-            "tooltip": "Number of frames to generate",
-            "min": 1,
-            "max": 100,
-            "step": 1,
-        },
-    )
-
-    actual_output = model_field_to_node_input(IO.INT, ModelWithIntField, "num_frames")
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_string_input():
-    """Tests mapping a string field."""
-
-    class ModelWithStringField(BaseModel):
-        prompt: Optional[str] = Field(
-            default="A beautiful sunset over a calm ocean",
-            description="A prompt for the video generation",
-        )
-
-    expected_output = (
-        IO.STRING,
-        {
-            "default": "A beautiful sunset over a calm ocean",
-            "tooltip": "A prompt for the video generation",
-        },
-    )
-
-    actual_output = model_field_to_node_input(IO.STRING, ModelWithStringField, "prompt")
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_string_input_multiline():
-    """Tests mapping a string field."""
-
-    class ModelWithStringField(BaseModel):
-        prompt: Optional[str] = Field(
-            default="A beautiful sunset over a calm ocean",
-            description="A prompt for the video generation",
-        )
-
-    expected_output = (
-        IO.STRING,
-        {
-            "default": "A beautiful sunset over a calm ocean",
-            "tooltip": "A prompt for the video generation",
-            "multiline": True,
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.STRING, ModelWithStringField, "prompt", multiline=True
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_combo_input():
-    """Tests mapping a combo field."""
-
-    class MockEnum(str, Enum):
-        option_1 = "option 1"
-        option_2 = "option 2"
-        option_3 = "option 3"
-
-    class ModelWithComboField(BaseModel):
-        model_name: Optional[MockEnum] = Field("option 1", description="Model Name")
-
-    expected_output = (
-        IO.COMBO,
-        {
-            "options": ["option 1", "option 2", "option 3"],
-            "default": "option 1",
-            "tooltip": "Model Name",
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.COMBO, ModelWithComboField, "model_name", enum_type=MockEnum
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_combo_input_no_options():
-    """Tests mapping a combo field with no options."""
-
-    class ModelWithComboField(BaseModel):
-        model_name: Optional[str] = Field(description="Model Name")
-
-    expected_output = (
-        IO.COMBO,
-        {
-            "tooltip": "Model Name",
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.COMBO, ModelWithComboField, "model_name"
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_image_input():
-    """Tests mapping an image field."""
-
-    class ModelWithImageField(BaseModel):
-        image: Optional[str] = Field(
-            default=None,
-            description="An image for the video generation",
-        )
-
-    expected_output = (
-        IO.IMAGE,
-        {
-            "default": None,
-            "tooltip": "An image for the video generation",
-        },
-    )
-
-    actual_output = model_field_to_node_input(IO.IMAGE, ModelWithImageField, "image")
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_node_input_no_description():
-    """Tests mapping a field with no description."""
-
-    class ModelWithNoDescriptionField(BaseModel):
-        field: Optional[str] = Field(default="default value")
-
-    expected_output = (
-        IO.STRING,
-        {
-            "default": "default value",
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.STRING, ModelWithNoDescriptionField, "field"
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_node_input_no_default():
-    """Tests mapping a field with no default."""
-
-    class ModelWithNoDefaultField(BaseModel):
-        field: Optional[str] = Field(description="A field with no default")
-
-    expected_output = (
-        IO.STRING,
-        {
-            "tooltip": "A field with no default",
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.STRING, ModelWithNoDefaultField, "field"
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_node_input_no_metadata():
-    """Tests mapping a field with no metadata or properties defined on the schema."""
-
-    class ModelWithNoMetadataField(BaseModel):
-        field: Optional[str] = Field()
-
-    expected_output = (
-        IO.STRING,
-        {},
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.STRING, ModelWithNoMetadataField, "field"
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]
-
-
-def test_model_field_to_node_input_default_is_none():
-    """
-    Tests mapping a field with a default of `None`.
-    I.e., the default field should be included as the schema explicitly sets it to `None`.
-    """
-
-    class ModelWithNoneDefaultField(BaseModel):
-        field: Optional[str] = Field(
-            default=None, description="A field with a default of None"
-        )
-
-    expected_output = (
-        IO.STRING,
-        {
-            "default": None,
-            "tooltip": "A field with a default of None",
-        },
-    )
-
-    actual_output = model_field_to_node_input(
-        IO.STRING, ModelWithNoneDefaultField, "field"
-    )
-
-    assert actual_output[0] == expected_output[0]
-    assert actual_output[1] == expected_output[1]

tests-unit/comfy_api_test/input_impl_test.py

@@ -1,91 +0,0 @@
-import io
-from comfy_api.input_impl.video_types import (
-    container_to_output_format,
-    get_open_write_kwargs,
-)
-from comfy_api.util import VideoContainer
-
-
-def test_container_to_output_format_empty_string():
-    """Test that an empty string input returns None. `None` arg allows default auto-detection."""
-    assert container_to_output_format("") is None
-
-
-def test_container_to_output_format_none():
-    """Test that None input returns None."""
-    assert container_to_output_format(None) is None
-
-
-def test_container_to_output_format_comma_separated():
-    """Test that a comma-separated list returns a valid singular format from the list."""
-    comma_separated_format = "mp4,mov,m4a"
-    output_format = container_to_output_format(comma_separated_format)
-    assert output_format in comma_separated_format
-
-
-def test_container_to_output_format_single():
-    """Test that a single format string (not comma-separated list) is returned as is."""
-    assert container_to_output_format("mp4") == "mp4"
-
-
-def test_get_open_write_kwargs_filepath_no_format():
-    """Test that 'format' kwarg is NOT set when dest is a file path."""
-    kwargs_auto = get_open_write_kwargs("output.mp4", "mp4", VideoContainer.AUTO)
-    assert "format" not in kwargs_auto, "Format should not be set for file paths (AUTO)"
-
-    kwargs_specific = get_open_write_kwargs("output.avi", "mp4", "avi")
-    fail_msg = "Format should not be set for file paths (Specific)"
-    assert "format" not in kwargs_specific, fail_msg
-
-
-def test_get_open_write_kwargs_base_options_mode():
-    """Test basic kwargs for file path: mode and movflags."""
-    kwargs = get_open_write_kwargs("output.mp4", "mp4", VideoContainer.AUTO)
-    assert kwargs["mode"] == "w", "mode should be set to write"
-
-    fail_msg = "movflags should be set to preserve custom metadata tags"
-    assert "movflags" in kwargs["options"], fail_msg
-    assert kwargs["options"]["movflags"] == "use_metadata_tags", fail_msg
-
-
-def test_get_open_write_kwargs_bytesio_auto_format():
-    """Test kwargs for BytesIO dest with AUTO format."""
-    dest = io.BytesIO()
-    container_fmt = "mov,mp4,m4a"
-    kwargs = get_open_write_kwargs(dest, container_fmt, VideoContainer.AUTO)
-
-    assert kwargs["mode"] == "w"
-    assert kwargs["options"]["movflags"] == "use_metadata_tags"
-
-    fail_msg = (
-        "Format should be a valid format from the container's format list when AUTO"
-    )
-    assert kwargs["format"] in container_fmt, fail_msg
-
-
-def test_get_open_write_kwargs_bytesio_specific_format():
-    """Test kwargs for BytesIO dest with a specific single format."""
-    dest = io.BytesIO()
-    container_fmt = "avi"
-    to_fmt = VideoContainer.MP4
-    kwargs = get_open_write_kwargs(dest, container_fmt, to_fmt)
-
-    assert kwargs["mode"] == "w"
-    assert kwargs["options"]["movflags"] == "use_metadata_tags"
-
-    fail_msg = "Format should be the specified format (lowercased) when output format is not AUTO"
-    assert kwargs["format"] == "mp4", fail_msg
-
-
-def test_get_open_write_kwargs_bytesio_specific_format_list():
-    """Test kwargs for BytesIO dest with a specific comma-separated format."""
-    dest = io.BytesIO()
-    container_fmt = "avi"
-    to_fmt = "mov,mp4,m4a"  # A format string that is a list
-    kwargs = get_open_write_kwargs(dest, container_fmt, to_fmt)
-
-    assert kwargs["mode"] == "w"
-    assert kwargs["options"]["movflags"] == "use_metadata_tags"
-
-    fail_msg = "Format should be a valid format from the specified format list when output format is not AUTO"
-    assert kwargs["format"] in to_fmt, fail_msg

tests-unit/comfy_api_test/video_types_test.py

@@ -1,239 +0,0 @@
-import pytest
-import torch
-import tempfile
-import os
-import av
-import io
-from fractions import Fraction
-from comfy_api.input_impl.video_types import VideoFromFile, VideoFromComponents
-from comfy_api.util.video_types import VideoComponents
-from comfy_api.input.basic_types import AudioInput
-from av.error import InvalidDataError
-
-EPSILON = 0.0001
-
-
-@pytest.fixture
-def sample_images():
-    """3-frame 2x2 RGB video tensor"""
-    return torch.rand(3, 2, 2, 3)
-
-
-@pytest.fixture
-def sample_audio():
-    """Stereo audio with 44.1kHz sample rate"""
-    return AudioInput(
-        {
-            "waveform": torch.rand(1, 2, 1000),
-            "sample_rate": 44100,
-        }
-    )
-
-
-@pytest.fixture
-def video_components(sample_images, sample_audio):
-    """VideoComponents with images, audio, and metadata"""
-    return VideoComponents(
-        images=sample_images,
-        audio=sample_audio,
-        frame_rate=Fraction(30),
-        metadata={"test": "metadata"},
-    )
-
-
-def create_test_video(width=4, height=4, frames=3, fps=30):
-    """Helper to create a temporary video file"""
-    tmp = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
-    with av.open(tmp.name, mode="w") as container:
-        stream = container.add_stream("h264", rate=fps)
-        stream.width = width
-        stream.height = height
-        stream.pix_fmt = "yuv420p"
-
-        for i in range(frames):
-            frame = av.VideoFrame.from_ndarray(
-                torch.ones(height, width, 3, dtype=torch.uint8).numpy() * (i * 85),
-                format="rgb24",
-            )
-            frame = frame.reformat(format="yuv420p")
-            packet = stream.encode(frame)
-            container.mux(packet)
-
-        # Flush
-        packet = stream.encode(None)
-        container.mux(packet)
-
-    return tmp.name
-
-
-@pytest.fixture
-def simple_video_file():
-    """4x4 video with 3 frames at 30fps"""
-    file_path = create_test_video()
-    yield file_path
-    os.unlink(file_path)
-
-
-def test_video_from_components_get_duration(video_components):
-    """Duration calculated correctly from frame count and frame rate"""
-    video = VideoFromComponents(video_components)
-    duration = video.get_duration()
-
-    expected_duration = 3.0 / 30.0
-    assert duration == pytest.approx(expected_duration)
-
-
-def test_video_from_components_get_duration_different_frame_rates(sample_images):
-    """Duration correct for different frame rates including fractional"""
-    # Test with 60 fps
-    components_60fps = VideoComponents(images=sample_images, frame_rate=Fraction(60))
-    video_60fps = VideoFromComponents(components_60fps)
-    assert video_60fps.get_duration() == pytest.approx(3.0 / 60.0)
-
-    # Test with fractional frame rate (23.976fps)
-    components_frac = VideoComponents(
-        images=sample_images, frame_rate=Fraction(24000, 1001)
-    )
-    video_frac = VideoFromComponents(components_frac)
-    expected_frac = 3.0 / (24000.0 / 1001.0)
-    assert video_frac.get_duration() == pytest.approx(expected_frac)
-
-
-def test_video_from_components_get_duration_empty_video():
-    """Duration is zero for empty video"""
-    empty_components = VideoComponents(
-        images=torch.zeros(0, 2, 2, 3), frame_rate=Fraction(30)
-    )
-    video = VideoFromComponents(empty_components)
-    assert video.get_duration() == 0.0
-
-
-def test_video_from_components_get_dimensions(video_components):
-    """Dimensions returned correctly from image tensor shape"""
-    video = VideoFromComponents(video_components)
-    width, height = video.get_dimensions()
-    assert width == 2
-    assert height == 2
-
-
-def test_video_from_file_get_duration(simple_video_file):
-    """Duration extracted from file metadata"""
-    video = VideoFromFile(simple_video_file)
-    duration = video.get_duration()
-    assert duration == pytest.approx(0.1, abs=0.01)
-
-
-def test_video_from_file_get_dimensions(simple_video_file):
-    """Dimensions read from stream without decoding frames"""
-    video = VideoFromFile(simple_video_file)
-    width, height = video.get_dimensions()
-    assert width == 4
-    assert height == 4
-
-
-def test_video_from_file_bytesio_input():
-    """VideoFromFile works with BytesIO input"""
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mp4") as container:
-        stream = container.add_stream("h264", rate=30)
-        stream.width = 2
-        stream.height = 2
-        stream.pix_fmt = "yuv420p"
-
-        frame = av.VideoFrame.from_ndarray(
-            torch.zeros(2, 2, 3, dtype=torch.uint8).numpy(), format="rgb24"
-        )
-        frame = frame.reformat(format="yuv420p")
-        packet = stream.encode(frame)
-        container.mux(packet)
-        packet = stream.encode(None)
-        container.mux(packet)
-
-    buffer.seek(0)
-    video = VideoFromFile(buffer)
-
-    assert video.get_dimensions() == (2, 2)
-    assert video.get_duration() == pytest.approx(1 / 30, abs=0.01)
-
-
-def test_video_from_file_invalid_file_error():
-    """InvalidDataError raised for non-video files"""
-    with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as tmp:
-        tmp.write(b"not a video file")
-        tmp.flush()
-        tmp_name = tmp.name
-
-    try:
-        with pytest.raises(InvalidDataError):
-            video = VideoFromFile(tmp_name)
-            video.get_dimensions()
-    finally:
-        os.unlink(tmp_name)
-
-
-def test_video_from_file_audio_only_error():
-    """ValueError raised for audio-only files"""
-    with tempfile.NamedTemporaryFile(suffix=".m4a", delete=False) as tmp:
-        tmp_name = tmp.name
-
-    try:
-        with av.open(tmp_name, mode="w") as container:
-            stream = container.add_stream("aac", rate=44100)
-            stream.sample_rate = 44100
-            stream.format = "fltp"
-
-            audio_data = torch.zeros(1, 1024).numpy()
-            audio_frame = av.AudioFrame.from_ndarray(
-                audio_data, format="fltp", layout="mono"
-            )
-            audio_frame.sample_rate = 44100
-            audio_frame.pts = 0
-            packet = stream.encode(audio_frame)
-            container.mux(packet)
-
-            for packet in stream.encode(None):
-                container.mux(packet)
-
-        with pytest.raises(ValueError, match="No video stream found"):
-            video = VideoFromFile(tmp_name)
-            video.get_dimensions()
-    finally:
-        os.unlink(tmp_name)
-
-
-def test_single_frame_video():
-    """Single frame video has correct duration"""
-    components = VideoComponents(
-        images=torch.rand(1, 10, 10, 3), frame_rate=Fraction(1)
-    )
-    video = VideoFromComponents(components)
-    assert video.get_duration() == 1.0
-
-
-@pytest.mark.parametrize(
-    "frame_rate,expected_fps",
-    [
-        (Fraction(24000, 1001), 24000 / 1001),
-        (Fraction(30000, 1001), 30000 / 1001),
-        (Fraction(25, 1), 25.0),
-        (Fraction(50, 2), 25.0),
-    ],
-)
-def test_fractional_frame_rates(frame_rate, expected_fps):
-    """Duration calculated correctly for various fractional frame rates"""
-    components = VideoComponents(images=torch.rand(100, 4, 4, 3), frame_rate=frame_rate)
-    video = VideoFromComponents(components)
-    duration = video.get_duration()
-    expected_duration = 100.0 / expected_fps
-    assert duration == pytest.approx(expected_duration)
-
-
-def test_duration_consistency(video_components):
-    """get_duration() consistent with manual calculation from components"""
-    video = VideoFromComponents(video_components)
-
-    duration = video.get_duration()
-    components = video.get_components()
-    manual_duration = float(components.images.shape[0] / components.frame_rate)
-
-    assert duration == pytest.approx(manual_duration)

tests-unit/comfy_extras_test/image_stitch_test.py

@@ -1,243 +0,0 @@
-import torch
-from unittest.mock import patch, MagicMock
-
-# Mock nodes module to prevent CUDA initialization during import
-mock_nodes = MagicMock()
-mock_nodes.MAX_RESOLUTION = 16384
-
-# Mock server module for PromptServer
-mock_server = MagicMock()
-
-with patch.dict('sys.modules', {'nodes': mock_nodes, 'server': mock_server}):
-    from comfy_extras.nodes_images import ImageStitch
-
-
-class TestImageStitch:
-
-    def create_test_image(self, batch_size=1, height=64, width=64, channels=3):
-        """Helper to create test images with specific dimensions"""
-        return torch.rand(batch_size, height, width, channels)
-
-    def test_no_image2_passthrough(self):
-        """Test that when image2 is None, image1 is returned unchanged"""
-        node = ImageStitch()
-        image1 = self.create_test_image()
-
-        result = node.stitch(image1, "right", True, 0, "white", image2=None)
-
-        assert len(result) == 1
-        assert torch.equal(result[0], image1)
-
-    def test_basic_horizontal_stitch_right(self):
-        """Test basic horizontal stitching to the right"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=32, width=24)
-
-        result = node.stitch(image1, "right", False, 0, "white", image2)
-
-        assert result[0].shape == (1, 32, 56, 3)  # 32 + 24 width
-
-    def test_basic_horizontal_stitch_left(self):
-        """Test basic horizontal stitching to the left"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=32, width=24)
-
-        result = node.stitch(image1, "left", False, 0, "white", image2)
-
-        assert result[0].shape == (1, 32, 56, 3)  # 24 + 32 width
-
-    def test_basic_vertical_stitch_down(self):
-        """Test basic vertical stitching downward"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=24, width=32)
-
-        result = node.stitch(image1, "down", False, 0, "white", image2)
-
-        assert result[0].shape == (1, 56, 32, 3)  # 32 + 24 height
-
-    def test_basic_vertical_stitch_up(self):
-        """Test basic vertical stitching upward"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=24, width=32)
-
-        result = node.stitch(image1, "up", False, 0, "white", image2)
-
-        assert result[0].shape == (1, 56, 32, 3)  # 24 + 32 height
-
-    def test_size_matching_horizontal(self):
-        """Test size matching for horizontal concatenation"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=64, width=64)
-        image2 = self.create_test_image(height=32, width=32)  # Different aspect ratio
-
-        result = node.stitch(image1, "right", True, 0, "white", image2)
-
-        # image2 should be resized to match image1's height (64) with preserved aspect ratio
-        expected_width = 64 + 64  # original + resized (32*64/32 = 64)
-        assert result[0].shape == (1, 64, expected_width, 3)
-
-    def test_size_matching_vertical(self):
-        """Test size matching for vertical concatenation"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=64, width=64)
-        image2 = self.create_test_image(height=32, width=32)
-
-        result = node.stitch(image1, "down", True, 0, "white", image2)
-
-        # image2 should be resized to match image1's width (64) with preserved aspect ratio
-        expected_height = 64 + 64  # original + resized (32*64/32 = 64)
-        assert result[0].shape == (1, expected_height, 64, 3)
-
-    def test_padding_for_mismatched_heights_horizontal(self):
-        """Test padding when heights don't match in horizontal concatenation"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=64, width=32)
-        image2 = self.create_test_image(height=48, width=24)  # Shorter height
-
-        result = node.stitch(image1, "right", False, 0, "white", image2)
-
-        # Both images should be padded to height 64
-        assert result[0].shape == (1, 64, 56, 3)  # 32 + 24 width, max(64,48) height
-
-    def test_padding_for_mismatched_widths_vertical(self):
-        """Test padding when widths don't match in vertical concatenation"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=64)
-        image2 = self.create_test_image(height=24, width=48)  # Narrower width
-
-        result = node.stitch(image1, "down", False, 0, "white", image2)
-
-        # Both images should be padded to width 64
-        assert result[0].shape == (1, 56, 64, 3)  # 32 + 24 height, max(64,48) width
-
-    def test_spacing_horizontal(self):
-        """Test spacing addition in horizontal concatenation"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=32, width=24)
-        spacing_width = 16
-
-        result = node.stitch(image1, "right", False, spacing_width, "white", image2)
-
-        # Expected width: 32 + 16 (spacing) + 24 = 72
-        assert result[0].shape == (1, 32, 72, 3)
-
-    def test_spacing_vertical(self):
-        """Test spacing addition in vertical concatenation"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=24, width=32)
-        spacing_width = 16
-
-        result = node.stitch(image1, "down", False, spacing_width, "white", image2)
-
-        # Expected height: 32 + 16 (spacing) + 24 = 72
-        assert result[0].shape == (1, 72, 32, 3)
-
-    def test_spacing_color_values(self):
-        """Test that spacing colors are applied correctly"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=32, width=32)
-
-        # Test white spacing
-        result_white = node.stitch(image1, "right", False, 16, "white", image2)
-        # Check that spacing region contains white values (close to 1.0)
-        spacing_region = result_white[0][:, :, 32:48, :]  # Middle 16 pixels
-        assert torch.all(spacing_region >= 0.9)  # Should be close to white
-
-        # Test black spacing
-        result_black = node.stitch(image1, "right", False, 16, "black", image2)
-        spacing_region = result_black[0][:, :, 32:48, :]
-        assert torch.all(spacing_region <= 0.1)  # Should be close to black
-
-    def test_odd_spacing_width_made_even(self):
-        """Test that odd spacing widths are made even"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=32, width=32)
-
-        # Use odd spacing width
-        result = node.stitch(image1, "right", False, 15, "white", image2)
-
-        # Should be made even (16), so total width = 32 + 16 + 32 = 80
-        assert result[0].shape == (1, 32, 80, 3)
-
-    def test_batch_size_matching(self):
-        """Test that different batch sizes are handled correctly"""
-        node = ImageStitch()
-        image1 = self.create_test_image(batch_size=2, height=32, width=32)
-        image2 = self.create_test_image(batch_size=1, height=32, width=32)
-
-        result = node.stitch(image1, "right", False, 0, "white", image2)
-
-        # Should match larger batch size
-        assert result[0].shape == (2, 32, 64, 3)
-
-    def test_channel_matching_rgb_to_rgba(self):
-        """Test that channel differences are handled (RGB + alpha)"""
-        node = ImageStitch()
-        image1 = self.create_test_image(channels=3)  # RGB
-        image2 = self.create_test_image(channels=4)  # RGBA
-
-        result = node.stitch(image1, "right", False, 0, "white", image2)
-
-        # Should have 4 channels (RGBA)
-        assert result[0].shape[-1] == 4
-
-    def test_channel_matching_rgba_to_rgb(self):
-        """Test that channel differences are handled (RGBA + RGB)"""
-        node = ImageStitch()
-        image1 = self.create_test_image(channels=4)  # RGBA
-        image2 = self.create_test_image(channels=3)  # RGB
-
-        result = node.stitch(image1, "right", False, 0, "white", image2)
-
-        # Should have 4 channels (RGBA)
-        assert result[0].shape[-1] == 4
-
-    def test_all_color_options(self):
-        """Test all available color options"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=32, width=32)
-
-        colors = ["white", "black", "red", "green", "blue"]
-
-        for color in colors:
-            result = node.stitch(image1, "right", False, 16, color, image2)
-            assert result[0].shape == (1, 32, 80, 3)  # Basic shape check
-
-    def test_all_directions(self):
-        """Test all direction options"""
-        node = ImageStitch()
-        image1 = self.create_test_image(height=32, width=32)
-        image2 = self.create_test_image(height=32, width=32)
-
-        directions = ["right", "left", "up", "down"]
-
-        for direction in directions:
-            result = node.stitch(image1, direction, False, 0, "white", image2)
-            assert result[0].shape == (1, 32, 64, 3) if direction in ["right", "left"] else (1, 64, 32, 3)
-
-    def test_batch_size_channel_spacing_integration(self):
-        """Test integration of batch matching, channel matching, size matching, and spacings"""
-        node = ImageStitch()
-        image1 = self.create_test_image(batch_size=2, height=64, width=48, channels=3)
-        image2 = self.create_test_image(batch_size=1, height=32, width=32, channels=4)
-
-        result = node.stitch(image1, "right", True, 8, "red", image2)
-
-        # Should handle: batch matching, size matching, channel matching, spacing
-        assert result[0].shape[0] == 2  # Batch size matched
-        assert result[0].shape[-1] == 4  # Channels matched to max
-        assert result[0].shape[1] == 64  # Height from image1 (size matching)
-        # Width should be: 48 + 8 (spacing) + resized_image2_width
-        expected_image2_width = int(64 * (32/32))  # Resized to height 64
-        expected_total_width = 48 + 8 + expected_image2_width
-        assert result[0].shape[2] == expected_total_width
-

tests-unit/comfy_test/folder_path_test.py

@@ -1,162 +0,0 @@
-### 🗻 This file is created through the spirit of Mount Fuji at its peak
-# TODO(yoland): clean up this after I get back down
-import sys
-import pytest
-import os
-import tempfile
-from unittest.mock import patch
-from importlib import reload
-
-import folder_paths
-import comfy.cli_args
-from comfy.options import enable_args_parsing
-enable_args_parsing()
-
-
-@pytest.fixture()
-def clear_folder_paths():
-    # Reload the module after each test to ensure isolation
-    yield
-    reload(folder_paths)
-
-@pytest.fixture
-def temp_dir():
-    with tempfile.TemporaryDirectory() as tmpdirname:
-        yield tmpdirname
-
-
-@pytest.fixture
-def set_base_dir():
-    def _set_base_dir(base_dir):
-        # Mock CLI args
-        with patch.object(sys, 'argv', ["main.py", "--base-directory", base_dir]):
-            reload(comfy.cli_args)
-            reload(folder_paths)
-    yield _set_base_dir
-    # Reload the modules after each test to ensure isolation
-    with patch.object(sys, 'argv', ["main.py"]):
-        reload(comfy.cli_args)
-        reload(folder_paths)
-
-
-def test_get_directory_by_type(clear_folder_paths):
-    test_dir = "/test/dir"
-    folder_paths.set_output_directory(test_dir)
-    assert folder_paths.get_directory_by_type("output") == test_dir
-    assert folder_paths.get_directory_by_type("invalid") is None
-
-def test_annotated_filepath():
-    assert folder_paths.annotated_filepath("test.txt") == ("test.txt", None)
-    assert folder_paths.annotated_filepath("test.txt [output]") == ("test.txt", folder_paths.get_output_directory())
-    assert folder_paths.annotated_filepath("test.txt [input]") == ("test.txt", folder_paths.get_input_directory())
-    assert folder_paths.annotated_filepath("test.txt [temp]") == ("test.txt", folder_paths.get_temp_directory())
-
-def test_get_annotated_filepath():
-    default_dir = "/default/dir"
-    assert folder_paths.get_annotated_filepath("test.txt", default_dir) == os.path.join(default_dir, "test.txt")
-    assert folder_paths.get_annotated_filepath("test.txt [output]") == os.path.join(folder_paths.get_output_directory(), "test.txt")
-
-def test_add_model_folder_path_append(clear_folder_paths):
-    folder_paths.add_model_folder_path("test_folder", "/default/path", is_default=True)
-    folder_paths.add_model_folder_path("test_folder", "/test/path", is_default=False)
-    assert folder_paths.get_folder_paths("test_folder") == ["/default/path", "/test/path"]
-
-
-def test_add_model_folder_path_insert(clear_folder_paths):
-    folder_paths.add_model_folder_path("test_folder", "/test/path", is_default=False)
-    folder_paths.add_model_folder_path("test_folder", "/default/path", is_default=True)
-    assert folder_paths.get_folder_paths("test_folder") == ["/default/path", "/test/path"]
-
-
-def test_add_model_folder_path_re_add_existing_default(clear_folder_paths):
-    folder_paths.add_model_folder_path("test_folder", "/test/path", is_default=False)
-    folder_paths.add_model_folder_path("test_folder", "/old_default/path", is_default=True)
-    assert folder_paths.get_folder_paths("test_folder") == ["/old_default/path", "/test/path"]
-    folder_paths.add_model_folder_path("test_folder", "/test/path", is_default=True)
-    assert folder_paths.get_folder_paths("test_folder") == ["/test/path", "/old_default/path"]
-
-
-def test_add_model_folder_path_re_add_existing_non_default(clear_folder_paths):
-    folder_paths.add_model_folder_path("test_folder", "/test/path", is_default=False)
-    folder_paths.add_model_folder_path("test_folder", "/default/path", is_default=True)
-    assert folder_paths.get_folder_paths("test_folder") == ["/default/path", "/test/path"]
-    folder_paths.add_model_folder_path("test_folder", "/test/path", is_default=False)
-    assert folder_paths.get_folder_paths("test_folder") == ["/default/path", "/test/path"]
-
-
-def test_recursive_search(temp_dir):
-    os.makedirs(os.path.join(temp_dir, "subdir"))
-    open(os.path.join(temp_dir, "file1.txt"), "w").close()
-    open(os.path.join(temp_dir, "subdir", "file2.txt"), "w").close()
-
-    files, dirs = folder_paths.recursive_search(temp_dir)
-    assert set(files) == {"file1.txt", os.path.join("subdir", "file2.txt")}
-    assert len(dirs) == 2  # temp_dir and subdir
-
-def test_filter_files_extensions():
-    files = ["file1.txt", "file2.jpg", "file3.png", "file4.txt"]
-    assert folder_paths.filter_files_extensions(files, [".txt"]) == ["file1.txt", "file4.txt"]
-    assert folder_paths.filter_files_extensions(files, [".jpg", ".png"]) == ["file2.jpg", "file3.png"]
-    assert folder_paths.filter_files_extensions(files, []) == files
-
-@patch("folder_paths.recursive_search")
-@patch("folder_paths.folder_names_and_paths")
-def test_get_filename_list(mock_folder_names_and_paths, mock_recursive_search):
-    mock_folder_names_and_paths.__getitem__.return_value = (["/test/path"], {".txt"})
-    mock_recursive_search.return_value = (["file1.txt", "file2.jpg"], {})
-    assert folder_paths.get_filename_list("test_folder") == ["file1.txt"]
-
-def test_get_save_image_path(temp_dir):
-    with patch("folder_paths.output_directory", temp_dir):
-        full_output_folder, filename, counter, subfolder, filename_prefix = folder_paths.get_save_image_path("test", temp_dir, 100, 100)
-        assert os.path.samefile(full_output_folder, temp_dir)
-        assert filename == "test"
-        assert counter == 1
-        assert subfolder == ""
-        assert filename_prefix == "test"
-
-
-def test_base_path_changes(set_base_dir):
-    test_dir = os.path.abspath("/test/dir")
-    set_base_dir(test_dir)
-
-    assert folder_paths.base_path == test_dir
-    assert folder_paths.models_dir == os.path.join(test_dir, "models")
-    assert folder_paths.input_directory == os.path.join(test_dir, "input")
-    assert folder_paths.output_directory == os.path.join(test_dir, "output")
-    assert folder_paths.temp_directory == os.path.join(test_dir, "temp")
-    assert folder_paths.user_directory == os.path.join(test_dir, "user")
-
-    assert os.path.join(test_dir, "custom_nodes") in folder_paths.get_folder_paths("custom_nodes")
-
-    for name in ["checkpoints", "loras", "vae", "configs", "embeddings", "controlnet", "classifiers"]:
-        assert folder_paths.get_folder_paths(name)[0] == os.path.join(test_dir, "models", name)
-
-
-def test_base_path_change_clears_old(set_base_dir):
-    test_dir = os.path.abspath("/test/dir")
-    set_base_dir(test_dir)
-
-    assert len(folder_paths.get_folder_paths("custom_nodes")) == 1
-
-    single_model_paths = [
-        "checkpoints",
-        "loras",
-        "vae",
-        "configs",
-        "clip_vision",
-        "style_models",
-        "diffusers",
-        "vae_approx",
-        "gligen",
-        "upscale_models",
-        "embeddings",
-        "hypernetworks",
-        "photomaker",
-        "classifiers",
-    ]
-    for name in single_model_paths:
-        assert len(folder_paths.get_folder_paths(name)) == 1
-
-    for name in ["controlnet", "diffusion_models", "text_encoders"]:
-        assert len(folder_paths.get_folder_paths(name)) == 2

tests-unit/execution_test/validate_node_input_test.py

@@ -1,119 +0,0 @@
-import pytest
-from comfy_execution.validation import validate_node_input
-
-
-def test_exact_match():
-    """Test cases where types match exactly"""
-    assert validate_node_input("STRING", "STRING")
-    assert validate_node_input("STRING,INT", "STRING,INT")
-    assert validate_node_input("INT,STRING", "STRING,INT")  # Order shouldn't matter
-
-
-def test_strict_mode():
-    """Test strict mode validation"""
-    # Should pass - received type is subset of input type
-    assert validate_node_input("STRING", "STRING,INT", strict=True)
-    assert validate_node_input("INT", "STRING,INT", strict=True)
-    assert validate_node_input("STRING,INT", "STRING,INT,BOOLEAN", strict=True)
-
-    # Should fail - received type is not subset of input type
-    assert not validate_node_input("STRING,INT", "STRING", strict=True)
-    assert not validate_node_input("STRING,BOOLEAN", "STRING", strict=True)
-    assert not validate_node_input("INT,BOOLEAN", "STRING,INT", strict=True)
-
-
-def test_non_strict_mode():
-    """Test non-strict mode validation (default behavior)"""
-    # Should pass - types have overlap
-    assert validate_node_input("STRING,BOOLEAN", "STRING,INT")
-    assert validate_node_input("STRING,INT", "INT,BOOLEAN")
-    assert validate_node_input("STRING", "STRING,INT")
-
-    # Should fail - no overlap in types
-    assert not validate_node_input("BOOLEAN", "STRING,INT")
-    assert not validate_node_input("FLOAT", "STRING,INT")
-    assert not validate_node_input("FLOAT,BOOLEAN", "STRING,INT")
-
-
-def test_whitespace_handling():
-    """Test that whitespace is handled correctly"""
-    assert validate_node_input("STRING, INT", "STRING,INT")
-    assert validate_node_input("STRING,INT", "STRING, INT")
-    assert validate_node_input(" STRING , INT ", "STRING,INT")
-    assert validate_node_input("STRING,INT", " STRING , INT ")
-
-
-def test_empty_strings():
-    """Test behavior with empty strings"""
-    assert validate_node_input("", "")
-    assert not validate_node_input("STRING", "")
-    assert not validate_node_input("", "STRING")
-
-
-def test_single_vs_multiple():
-    """Test single type against multiple types"""
-    assert validate_node_input("STRING", "STRING,INT,BOOLEAN")
-    assert validate_node_input("STRING,INT,BOOLEAN", "STRING", strict=False)
-    assert not validate_node_input("STRING,INT,BOOLEAN", "STRING", strict=True)
-
-
-def test_non_string():
-    """Test non-string types"""
-    obj1 = object()
-    obj2 = object()
-    assert validate_node_input(obj1, obj1)
-    assert not validate_node_input(obj1, obj2)
-
-
-class NotEqualsOverrideTest(str):
-    """Test class for ``__ne__`` override."""
-
-    def __ne__(self, value: object) -> bool:
-        if self == "*" or value == "*":
-            return False
-        if self == "LONGER_THAN_2":
-            return not len(value) > 2
-        raise TypeError("This is a class for unit tests only.")
-
-
-def test_ne_override():
-    """Test ``__ne__`` any override"""
-    any = NotEqualsOverrideTest("*")
-    invalid_type = "INVALID_TYPE"
-    obj = object()
-    assert validate_node_input(any, any)
-    assert validate_node_input(any, invalid_type)
-    assert validate_node_input(any, obj)
-    assert validate_node_input(any, {})
-    assert validate_node_input(any, [])
-    assert validate_node_input(any, [1, 2, 3])
-
-
-def test_ne_custom_override():
-    """Test ``__ne__`` custom override"""
-    special = NotEqualsOverrideTest("LONGER_THAN_2")
-
-    assert validate_node_input(special, special)
-    assert validate_node_input(special, "*")
-    assert validate_node_input(special, "INVALID_TYPE")
-    assert validate_node_input(special, [1, 2, 3])
-
-    # Should fail
-    assert not validate_node_input(special, [1, 2])
-    assert not validate_node_input(special, "TY")
-
-
-@pytest.mark.parametrize(
-    "received,input_type,strict,expected",
-    [
-        ("STRING", "STRING", False, True),
-        ("STRING,INT", "STRING,INT", False, True),
-        ("STRING", "STRING,INT", True, True),
-        ("STRING,INT", "STRING", True, False),
-        ("BOOLEAN", "STRING,INT", False, False),
-        ("STRING,BOOLEAN", "STRING,INT", False, True),
-    ],
-)
-def test_parametrized_cases(received, input_type, strict, expected):
-    """Parametrized test cases for various scenarios"""
-    assert validate_node_input(received, input_type, strict) == expected

tests-unit/feature_flags_test.py

@@ -1,98 +0,0 @@
-"""Tests for feature flags functionality."""
-
-from comfy_api.feature_flags import (
-    get_connection_feature,
-    supports_feature,
-    get_server_features,
-    SERVER_FEATURE_FLAGS,
-)
-
-
-class TestFeatureFlags:
-    """Test suite for feature flags functions."""
-
-    def test_get_server_features_returns_copy(self):
-        """Test that get_server_features returns a copy of the server flags."""
-        features = get_server_features()
-        # Verify it's a copy by modifying it
-        features["test_flag"] = True
-        # Original should be unchanged
-        assert "test_flag" not in SERVER_FEATURE_FLAGS
-
-    def test_get_server_features_contains_expected_flags(self):
-        """Test that server features contain expected flags."""
-        features = get_server_features()
-        assert "supports_preview_metadata" in features
-        assert features["supports_preview_metadata"] is True
-        assert "max_upload_size" in features
-        assert isinstance(features["max_upload_size"], (int, float))
-
-    def test_get_connection_feature_with_missing_sid(self):
-        """Test getting feature for non-existent session ID."""
-        sockets_metadata = {}
-        result = get_connection_feature(sockets_metadata, "missing_sid", "some_feature")
-        assert result is False  # Default value
-
-    def test_get_connection_feature_with_custom_default(self):
-        """Test getting feature with custom default value."""
-        sockets_metadata = {}
-        result = get_connection_feature(
-            sockets_metadata, "missing_sid", "some_feature", default="custom_default"
-        )
-        assert result == "custom_default"
-
-    def test_get_connection_feature_with_feature_flags(self):
-        """Test getting feature from connection with feature flags."""
-        sockets_metadata = {
-            "sid1": {
-                "feature_flags": {
-                    "supports_preview_metadata": True,
-                    "custom_feature": "value",
-                },
-            }
-        }
-        result = get_connection_feature(sockets_metadata, "sid1", "supports_preview_metadata")
-        assert result is True
-
-        result = get_connection_feature(sockets_metadata, "sid1", "custom_feature")
-        assert result == "value"
-
-    def test_get_connection_feature_missing_feature(self):
-        """Test getting non-existent feature from connection."""
-        sockets_metadata = {
-            "sid1": {"feature_flags": {"existing_feature": True}}
-        }
-        result = get_connection_feature(sockets_metadata, "sid1", "missing_feature")
-        assert result is False
-
-    def test_supports_feature_returns_boolean(self):
-        """Test that supports_feature always returns boolean."""
-        sockets_metadata = {
-            "sid1": {
-                "feature_flags": {
-                    "bool_feature": True,
-                    "string_feature": "value",
-                    "none_feature": None,
-                },
-            }
-        }
-
-        # True boolean feature
-        assert supports_feature(sockets_metadata, "sid1", "bool_feature") is True
-
-        # Non-boolean values should return False
-        assert supports_feature(sockets_metadata, "sid1", "string_feature") is False
-        assert supports_feature(sockets_metadata, "sid1", "none_feature") is False
-        assert supports_feature(sockets_metadata, "sid1", "missing_feature") is False
-
-    def test_supports_feature_with_missing_connection(self):
-        """Test supports_feature with missing connection."""
-        sockets_metadata = {}
-        assert supports_feature(sockets_metadata, "missing_sid", "any_feature") is False
-
-    def test_empty_feature_flags_dict(self):
-        """Test connection with empty feature flags dictionary."""
-        sockets_metadata = {"sid1": {"feature_flags": {}}}
-        result = get_connection_feature(sockets_metadata, "sid1", "any_feature")
-        assert result is False
-        assert supports_feature(sockets_metadata, "sid1", "any_feature") is False

tests-unit/folder_paths_test/filter_by_content_types_test.py

@@ -1,66 +0,0 @@
-import pytest
-import os
-import tempfile
-from folder_paths import filter_files_content_types, extension_mimetypes_cache
-from unittest.mock import patch
-
-
-@pytest.fixture(scope="module")
-def file_extensions():
-    return {
-        'image': ['gif', 'heif', 'ico', 'jpeg', 'jpg', 'png', 'pnm', 'ppm', 'svg', 'tiff', 'webp', 'xbm', 'xpm'],
-        'audio': ['aif', 'aifc', 'aiff', 'au', 'flac', 'm4a', 'mp2', 'mp3', 'ogg', 'snd', 'wav'],
-        'video': ['avi', 'm2v', 'm4v', 'mkv', 'mov', 'mp4', 'mpeg', 'mpg', 'ogv', 'qt', 'webm', 'wmv'],
-        'model': ['gltf', 'glb', 'obj', 'fbx', 'stl']
-    }
-
-
-@pytest.fixture(scope="module")
-def mock_dir(file_extensions):
-    with tempfile.TemporaryDirectory() as directory:
-        for content_type, extensions in file_extensions.items():
-            for extension in extensions:
-                with open(f"{directory}/sample_{content_type}.{extension}", "w") as f:
-                    f.write(f"Sample {content_type} file in {extension} format")
-        yield directory
-
-
-@pytest.fixture
-def patched_mimetype_cache(file_extensions):
-    # Mock model file extensions since they may not be in the test-runner system's mimetype cache
-    new_cache = extension_mimetypes_cache.copy()
-    for extension in file_extensions["model"]:
-        new_cache[extension] = "model"
-
-    with patch("folder_paths.extension_mimetypes_cache", new_cache):
-        yield
-
-
-def test_categorizes_all_correctly(mock_dir, file_extensions, patched_mimetype_cache):
-    files = os.listdir(mock_dir)
-    for content_type, extensions in file_extensions.items():
-        filtered_files = filter_files_content_types(files, [content_type])
-        for extension in extensions:
-            assert f"sample_{content_type}.{extension}" in filtered_files
-
-
-def test_categorizes_all_uniquely(mock_dir, file_extensions, patched_mimetype_cache):
-    files = os.listdir(mock_dir)
-    for content_type, extensions in file_extensions.items():
-        filtered_files = filter_files_content_types(files, [content_type])
-        assert len(filtered_files) == len(extensions)
-
-
-def test_handles_bad_extensions():
-    files = ["file1.txt", "file2.py", "file3.example", "file4.pdf", "file5.ini", "file6.doc", "file7.md"]
-    assert filter_files_content_types(files, ["image", "audio", "video"]) == []
-
-
-def test_handles_no_extension():
-    files = ["file1", "file2", "file3", "file4", "file5", "file6", "file7"]
-    assert filter_files_content_types(files, ["image", "audio", "video"]) == []
-
-
-def test_handles_no_files():
-    files = []
-    assert filter_files_content_types(files, ["image", "audio", "video"]) == []

tests-unit/folder_paths_test/misc_test.py

@@ -1,51 +0,0 @@
-import pytest
-import os
-import tempfile
-from folder_paths import get_input_subfolders, set_input_directory
-
-@pytest.fixture(scope="module")
-def mock_folder_structure():
-    with tempfile.TemporaryDirectory() as temp_dir:
-        # Create a nested folder structure
-        folders = [
-            "folder1",
-            "folder1/subfolder1",
-            "folder1/subfolder2",
-            "folder2",
-            "folder2/deep",
-            "folder2/deep/nested",
-            "empty_folder"
-        ]
-
-        # Create the folders
-        for folder in folders:
-            os.makedirs(os.path.join(temp_dir, folder))
-
-        # Add some files to test they're not included
-        with open(os.path.join(temp_dir, "root_file.txt"), "w") as f:
-            f.write("test")
-        with open(os.path.join(temp_dir, "folder1", "test.txt"), "w") as f:
-            f.write("test")
-
-        set_input_directory(temp_dir)
-        yield temp_dir
-
-
-def test_gets_all_folders(mock_folder_structure):
-    folders = get_input_subfolders()
-    expected = ["folder1", "folder1/subfolder1", "folder1/subfolder2",
-                "folder2", "folder2/deep", "folder2/deep/nested", "empty_folder"]
-    assert sorted(folders) == sorted(expected)
-
-
-def test_handles_nonexistent_input_directory():
-    with tempfile.TemporaryDirectory() as temp_dir:
-        nonexistent = os.path.join(temp_dir, "nonexistent")
-        set_input_directory(nonexistent)
-        assert get_input_subfolders() == []
-
-
-def test_empty_input_directory():
-    with tempfile.TemporaryDirectory() as temp_dir:
-        set_input_directory(temp_dir)
-        assert get_input_subfolders() == []  # Empty since we don't include root

tests-unit/prompt_server_test/user_manager_test.py

@@ -1,289 +0,0 @@
-import pytest
-import os
-from aiohttp import web
-from app.user_manager import UserManager
-from unittest.mock import patch
-
-pytestmark = (
-    pytest.mark.asyncio
-)  # This applies the asyncio mark to all test functions in the module
-
-
-@pytest.fixture
-def user_manager(tmp_path):
-    um = UserManager()
-    um.get_request_user_filepath = lambda req, file, **kwargs: os.path.join(
-        tmp_path, file
-    ) if file else tmp_path
-    return um
-
-
-@pytest.fixture
-def app(user_manager):
-    app = web.Application()
-    routes = web.RouteTableDef()
-    user_manager.add_routes(routes)
-    app.add_routes(routes)
-    return app
-
-
-async def test_listuserdata_empty_directory(aiohttp_client, app, tmp_path):
-    client = await aiohttp_client(app)
-    resp = await client.get("/userdata?dir=test_dir")
-    assert resp.status == 404
-
-
-async def test_listuserdata_with_files(aiohttp_client, app, tmp_path):
-    os.makedirs(tmp_path / "test_dir")
-    with open(tmp_path / "test_dir" / "file1.txt", "w") as f:
-        f.write("test content")
-
-    client = await aiohttp_client(app)
-    resp = await client.get("/userdata?dir=test_dir")
-    assert resp.status == 200
-    assert await resp.json() == ["file1.txt"]
-
-
-async def test_listuserdata_recursive(aiohttp_client, app, tmp_path):
-    os.makedirs(tmp_path / "test_dir" / "subdir")
-    with open(tmp_path / "test_dir" / "file1.txt", "w") as f:
-        f.write("test content")
-    with open(tmp_path / "test_dir" / "subdir" / "file2.txt", "w") as f:
-        f.write("test content")
-
-    client = await aiohttp_client(app)
-    resp = await client.get("/userdata?dir=test_dir&recurse=true")
-    assert resp.status == 200
-    assert set(await resp.json()) == {"file1.txt", "subdir/file2.txt"}
-
-
-async def test_listuserdata_full_info(aiohttp_client, app, tmp_path):
-    os.makedirs(tmp_path / "test_dir")
-    with open(tmp_path / "test_dir" / "file1.txt", "w") as f:
-        f.write("test content")
-
-    client = await aiohttp_client(app)
-    resp = await client.get("/userdata?dir=test_dir&full_info=true")
-    assert resp.status == 200
-    result = await resp.json()
-    assert len(result) == 1
-    assert result[0]["path"] == "file1.txt"
-    assert "size" in result[0]
-    assert "modified" in result[0]
-
-
-async def test_listuserdata_split_path(aiohttp_client, app, tmp_path):
-    os.makedirs(tmp_path / "test_dir" / "subdir")
-    with open(tmp_path / "test_dir" / "subdir" / "file1.txt", "w") as f:
-        f.write("test content")
-
-    client = await aiohttp_client(app)
-    resp = await client.get("/userdata?dir=test_dir&recurse=true&split=true")
-    assert resp.status == 200
-    assert await resp.json() == [["subdir/file1.txt", "subdir", "file1.txt"]]
-
-
-async def test_listuserdata_invalid_directory(aiohttp_client, app):
-    client = await aiohttp_client(app)
-    resp = await client.get("/userdata?dir=")
-    assert resp.status == 400
-
-
-async def test_listuserdata_normalized_separator(aiohttp_client, app, tmp_path):
-    os_sep = "\\"
-    with patch("os.sep", os_sep):
-        with patch("os.path.sep", os_sep):
-            os.makedirs(tmp_path / "test_dir" / "subdir")
-            with open(tmp_path / "test_dir" / "subdir" / "file1.txt", "w") as f:
-                f.write("test content")
-
-            client = await aiohttp_client(app)
-            resp = await client.get("/userdata?dir=test_dir&recurse=true")
-            assert resp.status == 200
-            result = await resp.json()
-            assert len(result) == 1
-            assert "/" in result[0]  # Ensure forward slash is used
-            assert "\\" not in result[0]  # Ensure backslash is not present
-            assert result[0] == "subdir/file1.txt"
-
-            # Test with full_info
-            resp = await client.get(
-                "/userdata?dir=test_dir&recurse=true&full_info=true"
-            )
-            assert resp.status == 200
-            result = await resp.json()
-            assert len(result) == 1
-            assert "/" in result[0]["path"]  # Ensure forward slash is used
-            assert "\\" not in result[0]["path"]  # Ensure backslash is not present
-            assert result[0]["path"] == "subdir/file1.txt"
-
-
-async def test_post_userdata_new_file(aiohttp_client, app, tmp_path):
-    client = await aiohttp_client(app)
-    content = b"test content"
-    resp = await client.post("/userdata/test.txt", data=content)
-
-    assert resp.status == 200
-    assert await resp.text() == '"test.txt"'
-
-    # Verify file was created with correct content
-    with open(tmp_path / "test.txt", "rb") as f:
-        assert f.read() == content
-
-
-async def test_post_userdata_overwrite_existing(aiohttp_client, app, tmp_path):
-    # Create initial file
-    with open(tmp_path / "test.txt", "w") as f:
-        f.write("initial content")
-
-    client = await aiohttp_client(app)
-    new_content = b"updated content"
-    resp = await client.post("/userdata/test.txt", data=new_content)
-
-    assert resp.status == 200
-    assert await resp.text() == '"test.txt"'
-
-    # Verify file was overwritten
-    with open(tmp_path / "test.txt", "rb") as f:
-        assert f.read() == new_content
-
-
-async def test_post_userdata_no_overwrite(aiohttp_client, app, tmp_path):
-    # Create initial file
-    with open(tmp_path / "test.txt", "w") as f:
-        f.write("initial content")
-
-    client = await aiohttp_client(app)
-    resp = await client.post("/userdata/test.txt?overwrite=false", data=b"new content")
-
-    assert resp.status == 409
-
-    # Verify original content unchanged
-    with open(tmp_path / "test.txt", "r") as f:
-        assert f.read() == "initial content"
-
-
-async def test_post_userdata_full_info(aiohttp_client, app, tmp_path):
-    client = await aiohttp_client(app)
-    content = b"test content"
-    resp = await client.post("/userdata/test.txt?full_info=true", data=content)
-
-    assert resp.status == 200
-    result = await resp.json()
-    assert result["path"] == "test.txt"
-    assert result["size"] == len(content)
-    assert "modified" in result
-
-
-async def test_move_userdata(aiohttp_client, app, tmp_path):
-    # Create initial file
-    with open(tmp_path / "source.txt", "w") as f:
-        f.write("test content")
-
-    client = await aiohttp_client(app)
-    resp = await client.post("/userdata/source.txt/move/dest.txt")
-
-    assert resp.status == 200
-    assert await resp.text() == '"dest.txt"'
-
-    # Verify file was moved
-    assert not os.path.exists(tmp_path / "source.txt")
-    with open(tmp_path / "dest.txt", "r") as f:
-        assert f.read() == "test content"
-
-
-async def test_move_userdata_no_overwrite(aiohttp_client, app, tmp_path):
-    # Create source and destination files
-    with open(tmp_path / "source.txt", "w") as f:
-        f.write("source content")
-    with open(tmp_path / "dest.txt", "w") as f:
-        f.write("destination content")
-
-    client = await aiohttp_client(app)
-    resp = await client.post("/userdata/source.txt/move/dest.txt?overwrite=false")
-
-    assert resp.status == 409
-
-    # Verify files remain unchanged
-    with open(tmp_path / "source.txt", "r") as f:
-        assert f.read() == "source content"
-    with open(tmp_path / "dest.txt", "r") as f:
-        assert f.read() == "destination content"
-
-
-async def test_move_userdata_full_info(aiohttp_client, app, tmp_path):
-    # Create initial file
-    with open(tmp_path / "source.txt", "w") as f:
-        f.write("test content")
-
-    client = await aiohttp_client(app)
-    resp = await client.post("/userdata/source.txt/move/dest.txt?full_info=true")
-
-    assert resp.status == 200
-    result = await resp.json()
-    assert result["path"] == "dest.txt"
-    assert result["size"] == len("test content")
-    assert "modified" in result
-
-    # Verify file was moved
-    assert not os.path.exists(tmp_path / "source.txt")
-    with open(tmp_path / "dest.txt", "r") as f:
-        assert f.read() == "test content"
-
-
-async def test_listuserdata_v2_empty_root(aiohttp_client, app):
-    client = await aiohttp_client(app)
-    resp = await client.get("/v2/userdata")
-    assert resp.status == 200
-    assert await resp.json() == []
-
-
-async def test_listuserdata_v2_nonexistent_subdirectory(aiohttp_client, app):
-    client = await aiohttp_client(app)
-    resp = await client.get("/v2/userdata?path=does_not_exist")
-    assert resp.status == 404
-
-
-async def test_listuserdata_v2_default(aiohttp_client, app, tmp_path):
-    os.makedirs(tmp_path / "test_dir" / "subdir")
-    (tmp_path / "test_dir" / "file1.txt").write_text("content")
-    (tmp_path / "test_dir" / "subdir" / "file2.txt").write_text("content")
-
-    client = await aiohttp_client(app)
-    resp = await client.get("/v2/userdata?path=test_dir")
-    assert resp.status == 200
-    data = await resp.json()
-    file_paths = {item["path"] for item in data if item["type"] == "file"}
-    assert file_paths == {"test_dir/file1.txt", "test_dir/subdir/file2.txt"}
-
-
-async def test_listuserdata_v2_normalized_separators(aiohttp_client, app, tmp_path, monkeypatch):
-    # Force backslash as os separator
-    monkeypatch.setattr(os, 'sep', '\\')
-    monkeypatch.setattr(os.path, 'sep', '\\')
-    os.makedirs(tmp_path / "test_dir" / "subdir")
-    (tmp_path / "test_dir" / "subdir" / "file1.txt").write_text("x")
-
-    client = await aiohttp_client(app)
-    resp = await client.get("/v2/userdata?path=test_dir")
-    assert resp.status == 200
-    data = await resp.json()
-    for item in data:
-        assert "/" in item["path"]
-        assert "\\" not in item["path"]\
-
-async def test_listuserdata_v2_url_encoded_path(aiohttp_client, app, tmp_path):
-    # Create a directory with a space in its name and a file inside
-    os.makedirs(tmp_path / "my dir")
-    (tmp_path / "my dir" / "file.txt").write_text("content")
-
-    client = await aiohttp_client(app)
-    # Use URL-encoded space in path parameter
-    resp = await client.get("/v2/userdata?path=my%20dir&recurse=false")
-    assert resp.status == 200
-    data = await resp.json()
-    assert len(data) == 1
-    entry = data[0]
-    assert entry["name"] == "file.txt"
-    # Ensure the path is correctly decoded and uses forward slash
-    assert entry["path"] == "my dir/file.txt"

tests-unit/requirements.txt

@@ -1,4 +0,0 @@
-pytest>=7.8.0
-pytest-aiohttp
-pytest-asyncio
-websocket-client

tests-unit/server/utils/file_operations_test.py

@@ -1,42 +0,0 @@
-import pytest
-from typing import List
-from api_server.utils.file_operations import FileSystemOperations, FileSystemItem, is_file_info
-
-@pytest.fixture
-def temp_directory(tmp_path):
-    # Create a temporary directory structure
-    dir1 = tmp_path / "dir1"
-    dir2 = tmp_path / "dir2"
-    dir1.mkdir()
-    dir2.mkdir()
-    (dir1 / "file1.txt").write_text("content1")
-    (dir2 / "file2.txt").write_text("content2")
-    (tmp_path / "file3.txt").write_text("content3")
-    return tmp_path
-
-def test_walk_directory(temp_directory):
-    result: List[FileSystemItem] = FileSystemOperations.walk_directory(str(temp_directory))
-
-    assert len(result) == 5  # 2 directories and 3 files
-
-    files = [item for item in result if item['type'] == 'file']
-    dirs = [item for item in result if item['type'] == 'directory']
-
-    assert len(files) == 3
-    assert len(dirs) == 2
-
-    file_names = {file['name'] for file in files}
-    assert file_names == {'file1.txt', 'file2.txt', 'file3.txt'}
-
-    dir_names = {dir['name'] for dir in dirs}
-    assert dir_names == {'dir1', 'dir2'}
-
-def test_walk_directory_empty(tmp_path):
-    result = FileSystemOperations.walk_directory(str(tmp_path))
-    assert len(result) == 0
-
-def test_walk_directory_file_size(temp_directory):
-    result: List[FileSystemItem] = FileSystemOperations.walk_directory(str(temp_directory))
-    files = [item for item in result if is_file_info(item)]
-    for file in files:
-        assert file['size'] > 0  # Assuming all files have some content

tests-unit/utils/extra_config_test.py

@@ -1,303 +0,0 @@
-import pytest
-import yaml
-import os
-import sys
-from unittest.mock import Mock, patch, mock_open
-
-from utils.extra_config import load_extra_path_config
-import folder_paths
-
-
-@pytest.fixture()
-def clear_folder_paths():
-    # Clear the global dictionary before each test to ensure isolation
-    original = folder_paths.folder_names_and_paths.copy()
-    folder_paths.folder_names_and_paths.clear()
-    yield
-    folder_paths.folder_names_and_paths = original
-
-
-@pytest.fixture
-def mock_yaml_content():
-    return {
-        'test_config': {
-            'base_path': '~/App/',
-            'checkpoints': 'subfolder1',
-        }
-    }
-
-
-@pytest.fixture
-def mock_expanded_home():
-    return '/home/user'
-
-
-@pytest.fixture
-def yaml_config_with_appdata():
-    return """
-    test_config:
-      base_path: '%APPDATA%/ComfyUI'
-      checkpoints: 'models/checkpoints'
-    """
-
-
-@pytest.fixture
-def mock_yaml_content_appdata(yaml_config_with_appdata):
-    return yaml.safe_load(yaml_config_with_appdata)
-
-
-@pytest.fixture
-def mock_expandvars_appdata():
-    mock = Mock()
-
-    def expandvars(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.side_effect = expandvars
-    return mock
-
-
-@pytest.fixture
-def mock_add_model_folder_path():
-    return Mock()
-
-
-@pytest.fixture
-def mock_expanduser(mock_expanded_home):
-    def _expanduser(path):
-        if path.startswith('~/'):
-            return os.path.join(mock_expanded_home, path[2:])
-        return path
-    return _expanduser
-
-
-@pytest.fixture
-def mock_yaml_safe_load(mock_yaml_content):
-    return Mock(return_value=mock_yaml_content)
-
-
-@patch('builtins.open', new_callable=mock_open, read_data="dummy file content")
-def test_load_extra_model_paths_expands_userpath(
-    mock_file,
-    monkeypatch,
-    mock_add_model_folder_path,
-    mock_expanduser,
-    mock_yaml_safe_load,
-    mock_expanded_home
-):
-    # Attach mocks used by load_extra_path_config
-    monkeypatch.setattr(folder_paths, 'add_model_folder_path', mock_add_model_folder_path)
-    monkeypatch.setattr(os.path, 'expanduser', mock_expanduser)
-    monkeypatch.setattr(yaml, 'safe_load', mock_yaml_safe_load)
-
-    dummy_yaml_file_name = 'dummy_path.yaml'
-    load_extra_path_config(dummy_yaml_file_name)
-
-    expected_calls = [
-        ('checkpoints', os.path.join(mock_expanded_home, 'App', 'subfolder1'), False),
-    ]
-
-    assert mock_add_model_folder_path.call_count == len(expected_calls)
-
-    # Check if add_model_folder_path was called with the correct arguments
-    for actual_call, expected_call in zip(mock_add_model_folder_path.call_args_list, expected_calls):
-        assert actual_call.args[0] == expected_call[0]
-        assert os.path.normpath(actual_call.args[1]) == os.path.normpath(expected_call[1])  # Normalize and check the path to check on multiple OS.
-        assert actual_call.args[2] == expected_call[2]
-
-    # Check if yaml.safe_load was called
-    mock_yaml_safe_load.assert_called_once()
-
-    # Check if open was called with the correct file path
-    mock_file.assert_called_once_with(dummy_yaml_file_name, 'r', encoding='utf-8')
-
-
-@patch('builtins.open', new_callable=mock_open)
-def test_load_extra_model_paths_expands_appdata(
-    mock_file,
-    monkeypatch,
-    mock_add_model_folder_path,
-    mock_expandvars_appdata,
-    yaml_config_with_appdata,
-    mock_yaml_content_appdata
-):
-    # Set the mock_file to return yaml with appdata as a variable
-    mock_file.return_value.read.return_value = yaml_config_with_appdata
-
-    # Attach mocks
-    monkeypatch.setattr(folder_paths, 'add_model_folder_path', mock_add_model_folder_path)
-    monkeypatch.setattr(os.path, 'expandvars', mock_expandvars_appdata)
-    monkeypatch.setattr(yaml, 'safe_load', Mock(return_value=mock_yaml_content_appdata))
-
-    # Mock expanduser to do nothing (since we're not testing it here)
-    monkeypatch.setattr(os.path, 'expanduser', lambda x: x)
-
-    dummy_yaml_file_name = 'dummy_path.yaml'
-    load_extra_path_config(dummy_yaml_file_name)
-
-    if sys.platform == "win32":
-        expected_base_path = 'C:/Users/TestUser/AppData/Roaming/ComfyUI'
-    else:
-        expected_base_path = '/Users/TestUser/AppData/Roaming/ComfyUI'
-    expected_calls = [
-        ('checkpoints', os.path.normpath(os.path.join(expected_base_path, 'models/checkpoints')), False),
-    ]
-
-    assert mock_add_model_folder_path.call_count == len(expected_calls)
-
-    # Check the base path variable was expanded
-    for actual_call, expected_call in zip(mock_add_model_folder_path.call_args_list, expected_calls):
-        assert actual_call.args == expected_call
-
-    # Verify that expandvars was called
-    assert mock_expandvars_appdata.called
-
-
-@patch("builtins.open", new_callable=mock_open, read_data="dummy yaml content")
-@patch("yaml.safe_load")
-def test_load_extra_path_config_relative_base_path(
-    mock_yaml_load, _mock_file, clear_folder_paths, monkeypatch, tmp_path
-):
-    """
-    Test that when 'base_path' is a relative path in the YAML, it is joined to the YAML file directory, and then
-    the items in the config are correctly converted to absolute paths.
-    """
-    sub_folder = "./my_rel_base"
-    config_data = {
-        "some_model_folder": {
-            "base_path": sub_folder,
-            "is_default": True,
-            "checkpoints": "checkpoints",
-            "some_key": "some_value"
-        }
-    }
-    mock_yaml_load.return_value = config_data
-
-    dummy_yaml_name = "dummy_file.yaml"
-
-    def fake_abspath(path):
-        if path == dummy_yaml_name:
-            # If it's the YAML path, treat it like it lives in tmp_path
-            return os.path.join(str(tmp_path), dummy_yaml_name)
-        return os.path.join(str(tmp_path), path)  # Otherwise, do a normal join relative to tmp_path
-
-    def fake_dirname(path):
-        # We expect path to be the result of fake_abspath(dummy_yaml_name)
-        if path.endswith(dummy_yaml_name):
-            return str(tmp_path)
-        return os.path.dirname(path)
-
-    monkeypatch.setattr(os.path, "abspath", fake_abspath)
-    monkeypatch.setattr(os.path, "dirname", fake_dirname)
-
-    load_extra_path_config(dummy_yaml_name)
-
-    expected_checkpoints = os.path.abspath(os.path.join(str(tmp_path), "my_rel_base", "checkpoints"))
-    expected_some_value = os.path.abspath(os.path.join(str(tmp_path), "my_rel_base", "some_value"))
-
-    actual_paths = folder_paths.folder_names_and_paths["checkpoints"][0]
-    assert len(actual_paths) == 1, "Should have one path added for 'checkpoints'."
-    assert actual_paths[0] == expected_checkpoints
-
-    actual_paths = folder_paths.folder_names_and_paths["some_key"][0]
-    assert len(actual_paths) == 1, "Should have one path added for 'some_key'."
-    assert actual_paths[0] == expected_some_value
-
-
-@patch("builtins.open", new_callable=mock_open, read_data="dummy yaml content")
-@patch("yaml.safe_load")
-def test_load_extra_path_config_absolute_base_path(
-    mock_yaml_load, _mock_file, clear_folder_paths, monkeypatch, tmp_path
-):
-    """
-    Test that when 'base_path' is an absolute path, each subdirectory is joined with that absolute path,
-    rather than being relative to the YAML's directory.
-    """
-    abs_base = os.path.join(str(tmp_path), "abs_base")
-    config_data = {
-        "some_absolute_folder": {
-            "base_path": abs_base,   # <-- absolute
-            "is_default": True,
-            "loras": "loras_folder",
-            "embeddings": "embeddings_folder"
-        }
-    }
-    mock_yaml_load.return_value = config_data
-
-    dummy_yaml_name = "dummy_abs.yaml"
-
-    def fake_abspath(path):
-        if path == dummy_yaml_name:
-            # If it's the YAML path, treat it like it is in tmp_path
-            return os.path.join(str(tmp_path), dummy_yaml_name)
-        return path  # For absolute base, we just return path directly
-
-    def fake_dirname(path):
-        return str(tmp_path) if path.endswith(dummy_yaml_name) else os.path.dirname(path)
-
-    monkeypatch.setattr(os.path, "abspath", fake_abspath)
-    monkeypatch.setattr(os.path, "dirname", fake_dirname)
-
-    load_extra_path_config(dummy_yaml_name)
-
-    # Expect the final paths to be <abs_base>/loras_folder and <abs_base>/embeddings_folder
-    expected_loras = os.path.join(abs_base, "loras_folder")
-    expected_embeddings = os.path.join(abs_base, "embeddings_folder")
-
-    actual_loras = folder_paths.folder_names_and_paths["loras"][0]
-    assert len(actual_loras) == 1, "Should have one path for 'loras'."
-    assert actual_loras[0] == os.path.abspath(expected_loras)
-
-    actual_embeddings = folder_paths.folder_names_and_paths["embeddings"][0]
-    assert len(actual_embeddings) == 1, "Should have one path for 'embeddings'."
-    assert actual_embeddings[0] == os.path.abspath(expected_embeddings)
-
-
-@patch("builtins.open", new_callable=mock_open, read_data="dummy yaml content")
-@patch("yaml.safe_load")
-def test_load_extra_path_config_no_base_path(
-    mock_yaml_load, _mock_file, clear_folder_paths, monkeypatch, tmp_path
-):
-    """
-    Test that if 'base_path' is not present, each path is joined
-    with the directory of the YAML file (unless it's already absolute).
-    """
-    config_data = {
-        "some_folder_without_base": {
-            "is_default": True,
-            "text_encoders": "clip",
-            "diffusion_models": "unet"
-        }
-    }
-    mock_yaml_load.return_value = config_data
-
-    dummy_yaml_name = "dummy_no_base.yaml"
-
-    def fake_abspath(path):
-        if path == dummy_yaml_name:
-            return os.path.join(str(tmp_path), dummy_yaml_name)
-        return os.path.join(str(tmp_path), path)
-
-    def fake_dirname(path):
-        return str(tmp_path) if path.endswith(dummy_yaml_name) else os.path.dirname(path)
-
-    monkeypatch.setattr(os.path, "abspath", fake_abspath)
-    monkeypatch.setattr(os.path, "dirname", fake_dirname)
-
-    load_extra_path_config(dummy_yaml_name)
-
-    expected_clip = os.path.join(str(tmp_path), "clip")
-    expected_unet = os.path.join(str(tmp_path), "unet")
-
-    actual_text_encoders = folder_paths.folder_names_and_paths["text_encoders"][0]
-    assert len(actual_text_encoders) == 1, "Should have one path for 'text_encoders'."
-    assert actual_text_encoders[0] == os.path.abspath(expected_clip)
-
-    actual_diffusion = folder_paths.folder_names_and_paths["diffusion_models"][0]
-    assert len(actual_diffusion) == 1, "Should have one path for 'diffusion_models'."
-    assert actual_diffusion[0] == os.path.abspath(expected_unet)

tests-unit/utils/json_util_test.py

@@ -1,71 +0,0 @@
-from utils.json_util import merge_json_recursive
-
-
-def test_merge_simple_dicts():
-    base = {"a": 1, "b": 2}
-    update = {"b": 3, "c": 4}
-    expected = {"a": 1, "b": 3, "c": 4}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_nested_dicts():
-    base = {"a": {"x": 1, "y": 2}, "b": 3}
-    update = {"a": {"y": 4, "z": 5}}
-    expected = {"a": {"x": 1, "y": 4, "z": 5}, "b": 3}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_lists():
-    base = {"a": [1, 2], "b": 3}
-    update = {"a": [3, 4]}
-    expected = {"a": [1, 2, 3, 4], "b": 3}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_nested_lists():
-    base = {"a": {"x": [1, 2]}}
-    update = {"a": {"x": [3, 4]}}
-    expected = {"a": {"x": [1, 2, 3, 4]}}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_mixed_types():
-    base = {"a": [1, 2], "b": {"x": 1}}
-    update = {"a": [3], "b": {"y": 2}}
-    expected = {"a": [1, 2, 3], "b": {"x": 1, "y": 2}}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_overwrite_non_dict():
-    base = {"a": 1}
-    update = {"a": {"x": 2}}
-    expected = {"a": {"x": 2}}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_empty_dicts():
-    base = {}
-    update = {"a": 1}
-    expected = {"a": 1}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_none_values():
-    base = {"a": None}
-    update = {"a": {"x": 1}}
-    expected = {"a": {"x": 1}}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_different_types():
-    base = {"a": [1, 2]}
-    update = {"a": "string"}
-    expected = {"a": "string"}
-    assert merge_json_recursive(base, update) == expected
-
-
-def test_merge_complex_nested():
-    base = {"a": [1, 2], "b": {"x": [3, 4], "y": {"p": 1}}}
-    update = {"a": [5], "b": {"x": [6], "y": {"q": 2}}}
-    expected = {"a": [1, 2, 5], "b": {"x": [3, 4, 6], "y": {"p": 1, "q": 2}}}
-    assert merge_json_recursive(base, update) == expected

tests-unit/websocket_feature_flags_test.py

@@ -1,77 +0,0 @@
-"""Simplified tests for WebSocket feature flags functionality."""
-from comfy_api import feature_flags
-
-
-class TestWebSocketFeatureFlags:
-    """Test suite for WebSocket feature flags integration."""
-
-    def test_server_feature_flags_response(self):
-        """Test server feature flags are properly formatted."""
-        features = feature_flags.get_server_features()
-
-        # Check expected server features
-        assert "supports_preview_metadata" in features
-        assert features["supports_preview_metadata"] is True
-        assert "max_upload_size" in features
-        assert isinstance(features["max_upload_size"], (int, float))
-
-    def test_progress_py_checks_feature_flags(self):
-        """Test that progress.py checks feature flags before sending metadata."""
-        # This simulates the check in progress.py
-        client_id = "test_client"
-        sockets_metadata = {"test_client": {"feature_flags": {}}}
-
-        # The actual check would be in progress.py
-        supports_metadata = feature_flags.supports_feature(
-            sockets_metadata, client_id, "supports_preview_metadata"
-        )
-
-        assert supports_metadata is False
-
-    def test_multiple_clients_different_features(self):
-        """Test handling multiple clients with different feature support."""
-        sockets_metadata = {
-            "modern_client": {
-                "feature_flags": {"supports_preview_metadata": True}
-            },
-            "legacy_client": {
-                "feature_flags": {}
-            }
-        }
-
-        # Check modern client
-        assert feature_flags.supports_feature(
-            sockets_metadata, "modern_client", "supports_preview_metadata"
-        ) is True
-
-        # Check legacy client
-        assert feature_flags.supports_feature(
-            sockets_metadata, "legacy_client", "supports_preview_metadata"
-        ) is False
-
-    def test_feature_negotiation_message_format(self):
-        """Test the format of feature negotiation messages."""
-        # Client message format
-        client_message = {
-            "type": "feature_flags",
-            "data": {
-                "supports_preview_metadata": True,
-                "api_version": "1.0.0"
-            }
-        }
-
-        # Verify structure
-        assert client_message["type"] == "feature_flags"
-        assert "supports_preview_metadata" in client_message["data"]
-
-        # Server response format (what would be sent)
-        server_features = feature_flags.get_server_features()
-        server_message = {
-            "type": "feature_flags",
-            "data": server_features
-        }
-
-        # Verify structure
-        assert server_message["type"] == "feature_flags"
-        assert "supports_preview_metadata" in server_message["data"]
-        assert server_message["data"]["supports_preview_metadata"] is True