Initial commit

tests/python/fixtures/example_node.py

@@ -0,0 +1,26 @@
+class ExampleNormalizeTextNode:
+    @classmethod
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "text": (
+                    "STRING",
+                    {
+                        "default": "",
+                        "multiline": True,
+                    },
+                )
+            }
+        }
+
+    RETURN_TYPES = ("STRING",)
+    FUNCTION = "normalize_text"
+    CATEGORY = "examples/testing"
+
+    def normalize_text(self, text: str):
+        normalized_parts = [part.strip() for part in text.splitlines() if part.strip()]
+        return (" ".join(normalized_parts),)
+
+
+NODE_CLASS_MAPPINGS = {"ExampleNormalizeText": ExampleNormalizeTextNode}
+NODE_DISPLAY_NAME_MAPPINGS = {"ExampleNormalizeText": "Example Normalize Text"}

tests/python/test_bundled_sounds.py

@@ -0,0 +1,42 @@
+import hashlib
+import wave
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+SOUNDS_DIR = REPO_ROOT / "sounds" / "bundled"
+
+
+def wav_peak_amplitude(path: Path) -> int:
+    with wave.open(str(path), "rb") as wav_file:
+        frames = wav_file.readframes(wav_file.getnframes())
+        sample_width = wav_file.getsampwidth()
+        if sample_width != 2:
+            msg = f"Unsupported sample width for {path.name}: {sample_width}"
+            raise AssertionError(msg)
+
+        peak = 0
+        for index in range(0, len(frames), sample_width):
+            sample = int.from_bytes(
+                frames[index : index + sample_width],
+                byteorder="little",
+                signed=True,
+            )
+            peak = max(peak, abs(sample))
+
+        return peak
+
+
+def file_sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def test_bundled_sounds_are_audible_and_distinct():
+    success_path = SOUNDS_DIR / "ping-success.wav"
+    failure_path = SOUNDS_DIR / "ping-failure.wav"
+
+    assert success_path.is_file()
+    assert failure_path.is_file()
+    assert wav_peak_amplitude(success_path) > 0
+    assert wav_peak_amplitude(failure_path) > 0
+    assert file_sha256(success_path) != file_sha256(failure_path)

tests/python/test_entrypoint_smoke.py

@@ -0,0 +1,50 @@
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+import json
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+ENTRYPOINT_PATH = REPO_ROOT / "__init__.py"
+PACKAGE_JSON_PATH = REPO_ROOT / "package.json"
+PYPROJECT_PATH = REPO_ROOT / "pyproject.toml"
+
+
+def load_module_from_path(module_name: str, module_path: Path):
+    spec = spec_from_file_location(module_name, module_path)
+    assert spec is not None
+    assert spec.loader is not None
+
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_template_entrypoint_exports_expected_symbols():
+    module = load_module_from_path("template_entrypoint", ENTRYPOINT_PATH)
+
+    assert module.WEB_DIRECTORY == "./dist"
+    assert module.NODE_CLASS_MAPPINGS is None
+    assert module.NODE_DISPLAY_NAME_MAPPINGS is None
+    assert module.__all__ == [
+        "NODE_CLASS_MAPPINGS",
+        "NODE_DISPLAY_NAME_MAPPINGS",
+        "WEB_DIRECTORY",
+        "comfy_entrypoint",
+    ]
+
+
+def test_project_metadata_replaces_template_defaults():
+    package_json = json.loads(PACKAGE_JSON_PATH.read_text(encoding="utf-8"))
+    pyproject_text = PYPROJECT_PATH.read_text(encoding="utf-8")
+
+    assert package_json["name"] == "comfyui-ping"
+    assert package_json["description"] == (
+        "Browser-played workflow success and failure audio notifications for ComfyUI"
+    )
+    assert 'name = "comfyui-ping"' in pyproject_text
+    assert (
+        'description = "Browser-played workflow success and failure audio notifications for ComfyUI"'
+        in pyproject_text
+    )
+    assert 'PublisherId = "pbanddev"' in pyproject_text
+    assert 'DisplayName = "comfyui-ping"' in pyproject_text

tests/python/test_example_node_pattern.py

@@ -0,0 +1,28 @@
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+
+
+FIXTURE_PATH = Path(__file__).resolve().parent / "fixtures" / "example_node.py"
+
+
+def load_fixture_module():
+    spec = spec_from_file_location("example_node_fixture", FIXTURE_PATH)
+    assert spec is not None
+    assert spec.loader is not None
+
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_example_node_shows_how_to_unit_test_pure_python_logic():
+    module = load_fixture_module()
+    node = module.ExampleNormalizeTextNode()
+
+    assert module.NODE_CLASS_MAPPINGS == {
+        "ExampleNormalizeText": module.ExampleNormalizeTextNode
+    }
+    assert module.NODE_DISPLAY_NAME_MAPPINGS == {
+        "ExampleNormalizeText": "Example Normalize Text"
+    }
+    assert node.normalize_text("  alpha \n\n beta  \n gamma ") == ("alpha beta gamma",)

tests/python/test_global_completion.py

@@ -0,0 +1,177 @@
+import sys
+import types
+from contextlib import contextmanager
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+
+from comfyui_ping.global_completion import (
+    build_global_completion_payload,
+    install_global_completion_hooks,
+    resolve_global_completion_payload,
+    should_notify_global,
+)
+
+
+@contextmanager
+def fake_execution_module(prompt_queue_cls: type):
+    original_execution = sys.modules.get("execution")
+    execution_module = types.ModuleType("execution")
+    execution_module.PromptQueue = prompt_queue_cls
+    sys.modules["execution"] = execution_module
+    try:
+        yield
+    finally:
+        if original_execution is None:
+            sys.modules.pop("execution", None)
+        else:
+            sys.modules["execution"] = original_execution
+
+
+def test_queue_drained_mode_skips_nonfinal_prompt():
+    assert should_notify_global(
+        notify_mode="queue_drained",
+        queue_size_after_completion=2,
+    ) is False
+
+
+def test_every_prompt_mode_notifies_each_completion():
+    assert should_notify_global(
+        notify_mode="every_prompt",
+        queue_size_after_completion=2,
+    ) is True
+
+
+def test_build_global_completion_payload_uses_global_event_kind():
+    payload = build_global_completion_payload(
+        status="failure",
+        global_settings={"failure_sound": "global-failure.wav"},
+    )
+
+    assert payload.event_kind == "global"
+    assert payload.status == "failure"
+    assert payload.sound_id == "bundled:global-failure.wav"
+
+
+def test_resolve_global_completion_payload_normalizes_error_status():
+    payload = resolve_global_completion_payload(
+        status="error",
+        queue_size_after_completion=0,
+        runtime_settings={
+            "enabled": True,
+            "notify_mode": "queue_drained",
+            "success_enabled": True,
+            "failure_enabled": True,
+            "success_sound": "ping-success.wav",
+            "failure_sound": "custom:uploaded.wav",
+            "volume": 0.45,
+        },
+    )
+
+    assert payload is not None
+    assert payload.status == "failure"
+    assert payload.sound_id == "custom:uploaded.wav"
+    assert payload.volume == 0.45
+
+
+def test_resolve_global_completion_payload_skips_disabled_failure_notifications():
+    payload = resolve_global_completion_payload(
+        status="error",
+        queue_size_after_completion=0,
+        runtime_settings={
+            "enabled": True,
+            "notify_mode": "queue_drained",
+            "success_enabled": True,
+            "failure_enabled": False,
+            "success_sound": "ping-success.wav",
+            "failure_sound": "ping-failure.wav",
+            "volume": 0.8,
+        },
+    )
+
+    assert payload is None
+
+
+def test_reinstall_updates_callbacks_after_noop_install():
+    emitted_payloads = []
+
+    class FakeStatus:
+        status_str = "success"
+
+    class FakePromptQueue:
+        def __init__(self):
+            self.queue = []
+            self.calls = []
+
+        def task_done(self, item_id, history_result, status):
+            self.calls.append((item_id, history_result, status.status_str))
+            return "done"
+
+    with fake_execution_module(FakePromptQueue):
+        assert install_global_completion_hooks() is True
+        assert install_global_completion_hooks(
+            emit_notification=emitted_payloads.append,
+            get_runtime_settings=lambda: {
+                "enabled": True,
+                "notify_mode": "queue_drained",
+                "success_enabled": True,
+                "failure_enabled": True,
+                "success_sound": "global-success.wav",
+                "failure_sound": "global-failure.wav",
+                "volume": 0.65,
+            },
+        ) is True
+
+        queue = FakePromptQueue()
+        result = queue.task_done("item-1", {"history": True}, FakeStatus())
+
+    assert result == "done"
+    assert emitted_payloads[0].event_kind == "global"
+    assert emitted_payloads[0].status == "success"
+    assert emitted_payloads[0].sound_id == "bundled:global-success.wav"
+    assert emitted_payloads[0].volume == 0.65
+    assert len(emitted_payloads) == 1
+
+
+def test_wrapped_task_done_accepts_process_item_keyword():
+    emitted_payloads = []
+
+    class FakeStatus:
+        status_str = "success"
+
+    class FakePromptQueue:
+        def __init__(self):
+            self.queue = []
+            self.calls = []
+
+        def task_done(self, item_id, history_result, status, process_item=None):
+            self.calls.append(process_item)
+            return "done"
+
+    with fake_execution_module(FakePromptQueue):
+        assert install_global_completion_hooks(
+            emit_notification=emitted_payloads.append,
+            get_runtime_settings=lambda: {
+                "enabled": True,
+                "notify_mode": "queue_drained",
+                "success_enabled": True,
+                "failure_enabled": True,
+                "success_sound": "global-success.wav",
+                "failure_sound": "global-failure.wav",
+                "volume": 0.8,
+            },
+        ) is True
+
+        queue = FakePromptQueue()
+        result = queue.task_done(
+            "item-1",
+            {"history": True},
+            FakeStatus(),
+            process_item=lambda prompt: prompt,
+        )
+
+    assert result == "done"
+    assert len(emitted_payloads) == 1

tests/python/test_notification_policy.py

@@ -0,0 +1,67 @@
+import sys
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+
+from comfyui_ping import build_notification_payload, resolve_sound_id
+
+
+def test_notification_payload_includes_status_and_sound_id():
+    payload = build_notification_payload(
+        event_kind="global",
+        status="failure",
+        global_settings={"failure_sound": "buzz.wav"},
+        node_overrides=None,
+    )
+
+    assert payload.status == "failure"
+    assert payload.sound_id == "bundled:buzz.wav"
+    assert payload.volume == 0.8
+    assert payload.source == "global"
+
+
+def test_notification_policy_prefers_node_override():
+    payload = build_notification_payload(
+        event_kind="node",
+        status="success",
+        global_settings={"success_sound": "global-success.wav"},
+        node_overrides={"success_sound": "node-success.wav"},
+    )
+
+    assert payload.sound_id == "bundled:node-success.wav"
+
+
+def test_invalid_custom_sound_does_not_fallback():
+    payload = build_notification_payload(
+        event_kind="global",
+        status="failure",
+        global_settings={"failure_sound": "missing.wav"},
+        node_overrides=None,
+    )
+
+    assert payload.allow_fallback is False
+
+
+def test_resolve_sound_id_uses_global_value_without_override():
+    assert (
+        resolve_sound_id(
+            status="failure",
+            global_settings={"failure_sound": "global-failure.wav"},
+            node_overrides=None,
+        )
+        == "bundled:global-failure.wav"
+    )
+
+
+def test_notification_policy_preserves_source_aware_sound_id():
+    payload = build_notification_payload(
+        event_kind="node",
+        status="failure",
+        global_settings={"failure_sound": "bundled:global-failure.wav"},
+        node_overrides={"failure_sound": "custom:alert.wav"},
+    )
+
+    assert payload.sound_id == "custom:alert.wav"

tests/python/test_ping_entrypoint.py

@@ -0,0 +1,210 @@
+import sys
+import types
+from asyncio import run
+from contextlib import contextmanager
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+ENTRYPOINT_PATH = REPO_ROOT / "__init__.py"
+
+
+@contextmanager
+def comfy_api_stubbed():
+    comfy_api_module = types.ModuleType("comfy_api")
+    v002_module = types.ModuleType("comfy_api.v0_0_2")
+    original_modules = {
+        name: sys.modules.get(name)
+        for name in (
+            "comfy_api",
+            "comfy_api.v0_0_2",
+            "comfyui_ping.extension",
+            "comfyui_ping.nodes",
+        )
+    }
+
+    class ComfyExtension:
+        async def get_node_list(self):
+            return []
+
+    class _Input:
+        def __init__(self, name: str, **kwargs):
+            self.name = name
+            self.kwargs = kwargs
+
+    class _Output:
+        def __init__(self, name: str, **kwargs):
+            self.name = name
+            self.kwargs = kwargs
+
+    class _String:
+        Input = _Input
+
+    class _Float:
+        Input = _Input
+
+    class _Boolean:
+        Input = _Input
+
+    class _AnyType:
+        Output = _Output
+
+    class Schema:
+        def __init__(self, **kwargs):
+            self.__dict__.update(kwargs)
+
+    class NodeOutput:
+        def __init__(self, *result, **kwargs):
+            self.result = result
+            self.kwargs = kwargs
+
+    class ComfyNode:
+        pass
+
+    io = types.SimpleNamespace(
+        Boolean=_Boolean,
+        ComfyNode=ComfyNode,
+        Float=_Float,
+        NodeOutput=NodeOutput,
+        Schema=Schema,
+        String=_String,
+        AnyType=_AnyType,
+    )
+
+    v002_module.ComfyExtension = ComfyExtension
+    v002_module.io = io
+    comfy_api_module.v0_0_2 = v002_module
+
+    sys.modules["comfy_api"] = comfy_api_module
+    sys.modules["comfy_api.v0_0_2"] = v002_module
+
+    try:
+        yield
+    finally:
+        for name, module in original_modules.items():
+            if module is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = module
+
+
+def load_module_from_path(module_name: str, module_path: Path):
+    spec = spec_from_file_location(module_name, module_path)
+    assert spec is not None
+    assert spec.loader is not None
+
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_root_entrypoint_exports_v3_entrypoint_and_web_directory():
+    module = load_module_from_path("ping_root_entrypoint", ENTRYPOINT_PATH)
+
+    assert module.WEB_DIRECTORY == "./dist"
+    assert callable(module.comfy_entrypoint)
+
+
+def test_comfy_entrypoint_returns_ping_extension():
+    with comfy_api_stubbed():
+        from comfyui_ping.extension import PingExtension
+
+        module = load_module_from_path("ping_root_entrypoint", ENTRYPOINT_PATH)
+        extension = module.comfy_entrypoint()
+
+        assert isinstance(extension, PingExtension)
+
+
+def test_comfy_entrypoint_works_without_repo_root_on_sys_path():
+    with comfy_api_stubbed():
+        original_sys_path = list(sys.path)
+        original_extension = sys.modules.pop("comfyui_ping.extension", None)
+        original_package = sys.modules.pop("comfyui_ping", None)
+        sys.path[:] = [path for path in sys.path if path != str(REPO_ROOT)]
+
+        try:
+            module = load_module_from_path(
+                "ping_root_entrypoint_no_repo_path",
+                ENTRYPOINT_PATH,
+            )
+            extension = module.comfy_entrypoint()
+        finally:
+            sys.path[:] = original_sys_path
+            if original_package is not None:
+                sys.modules["comfyui_ping"] = original_package
+            if original_extension is not None:
+                sys.modules["comfyui_ping.extension"] = original_extension
+
+        assert extension is not None
+
+
+def test_ping_extension_retries_sound_route_registration_on_get_node_list():
+    with comfy_api_stubbed():
+        import comfyui_ping.extension as extension_module
+
+        register_attempts: list[int] = []
+
+        def fake_install_global_completion_hooks(
+            *,
+            emit_notification,
+            get_runtime_settings,
+        ) -> bool:
+            return True
+
+        def fake_register_sound_routes() -> bool:
+            register_attempts.append(len(register_attempts) + 1)
+            return len(register_attempts) > 1
+
+        original_install = extension_module.install_global_completion_hooks
+        original_register = extension_module.register_sound_routes
+        extension_module.install_global_completion_hooks = (
+            fake_install_global_completion_hooks
+        )
+        extension_module.register_sound_routes = fake_register_sound_routes
+
+        try:
+            extension = extension_module.PingExtension()
+
+            assert extension.sound_routes_registered is False
+
+            run(extension.get_node_list())
+
+            assert extension.sound_routes_registered is True
+            assert register_attempts == [1, 2]
+        finally:
+            extension_module.install_global_completion_hooks = original_install
+            extension_module.register_sound_routes = original_register
+
+
+def test_ping_extension_retries_global_hook_installation_on_get_node_list():
+    with comfy_api_stubbed():
+        import comfyui_ping.extension as extension_module
+
+        hook_attempts: list[int] = []
+
+        def fake_install_global_completion_hooks(
+            *,
+            emit_notification,
+            get_runtime_settings,
+        ) -> bool:
+            hook_attempts.append(len(hook_attempts) + 1)
+            return len(hook_attempts) > 1
+
+        original_install = extension_module.install_global_completion_hooks
+        extension_module.install_global_completion_hooks = (
+            fake_install_global_completion_hooks
+        )
+
+        try:
+            extension = extension_module.PingExtension()
+
+            assert extension.global_completion_hooks_installed is False
+
+            run(extension.get_node_list())
+
+            assert extension.global_completion_hooks_installed is True
+            assert hook_attempts == [1, 2]
+        finally:
+            extension_module.install_global_completion_hooks = original_install

tests/python/test_ping_node.py

@@ -0,0 +1,134 @@
+import sys
+import types
+from contextlib import contextmanager
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+
+@contextmanager
+def comfy_api_stubbed():
+    comfy_api_module = types.ModuleType("comfy_api")
+    v002_module = types.ModuleType("comfy_api.v0_0_2")
+    original_modules = {
+        name: sys.modules.get(name)
+        for name in ("comfy_api", "comfy_api.v0_0_2", "comfyui_ping.nodes")
+    }
+
+    class ComfyExtension:
+        async def get_node_list(self):
+            return []
+
+    class _Input:
+        def __init__(self, name: str, **kwargs):
+            self.name = name
+            self.kwargs = kwargs
+
+    class _Output:
+        def __init__(self, name: str, **kwargs):
+            self.name = name
+            self.kwargs = kwargs
+
+    class _String:
+        Input = _Input
+
+    class _Float:
+        Input = _Input
+
+    class _Boolean:
+        Input = _Input
+
+    class _AnyType:
+        Output = _Output
+
+    class Schema:
+        def __init__(self, **kwargs):
+            self.__dict__.update(kwargs)
+
+    class NodeOutput:
+        def __init__(self, *result, **kwargs):
+            self.result = result
+            self.kwargs = kwargs
+
+    class ComfyNode:
+        pass
+
+    io = types.SimpleNamespace(
+        Boolean=_Boolean,
+        ComfyNode=ComfyNode,
+        Float=_Float,
+        NodeOutput=NodeOutput,
+        Schema=Schema,
+        String=_String,
+        AnyType=_AnyType,
+    )
+
+    v002_module.ComfyExtension = ComfyExtension
+    v002_module.io = io
+    comfy_api_module.v0_0_2 = v002_module
+
+    sys.modules["comfy_api"] = comfy_api_module
+    sys.modules["comfy_api.v0_0_2"] = v002_module
+
+    try:
+        yield
+    finally:
+        for name, module in original_modules.items():
+            if module is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = module
+
+
+def test_ping_node_schema_exposes_override_inputs():
+    with comfy_api_stubbed():
+        from comfyui_ping.nodes import PingNode
+
+        schema = PingNode.define_schema()
+        input_names = [input_def.name for input_def in schema.inputs]
+
+        assert "success_sound" in input_names
+        assert "failure_sound" in input_names
+        assert "volume" in input_names
+        assert "enabled" in input_names
+
+
+def test_ping_node_execute_emits_notification_payload():
+    with comfy_api_stubbed():
+        from comfyui_ping.nodes import PingNode
+
+        output = PingNode.execute(
+            success_sound="ding.wav",
+            failure_sound="buzz.wav",
+            volume=0.8,
+            enabled=True,
+        )
+
+        assert output is not None
+
+
+def test_ping_node_execute_emits_runtime_notification_when_enabled():
+    with comfy_api_stubbed():
+        import comfyui_ping.nodes as nodes_module
+
+        emitted_payloads = []
+        original_emit_notification = nodes_module.emit_notification
+        nodes_module.emit_notification = emitted_payloads.append
+
+        try:
+            output = nodes_module.PingNode.execute(
+                success_sound="ding.wav",
+                failure_sound="buzz.wav",
+                volume=0.35,
+                enabled=True,
+            )
+        finally:
+            nodes_module.emit_notification = original_emit_notification
+
+        assert output is not None
+        assert len(emitted_payloads) == 1
+        assert emitted_payloads[0].status == "success"
+        assert emitted_payloads[0].sound_id == "bundled:ding.wav"
+        assert emitted_payloads[0].volume == 0.35

tests/python/test_sound_routes.py

@@ -0,0 +1,353 @@
+import sys
+from asyncio import run
+from io import BytesIO
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+from comfyui_ping.routes import (
+    SOUND_CATALOG_ROUTE,
+    SOUND_FILE_ROUTE,
+    SOUND_UPLOAD_ROUTE,
+    attach_sound_routes,
+    build_sound_catalog_payload,
+    resolve_sound_path,
+    save_uploaded_sound,
+)
+from comfyui_ping.sounds import MAX_UPLOAD_BYTES
+
+
+def test_build_sound_catalog_payload_returns_sound_list():
+    payload = build_sound_catalog_payload(
+        bundled=["ping-success.wav"],
+        custom=["custom.wav"],
+    )
+
+    assert payload == {
+        "sounds": [
+            {"name": "ping-success.wav", "source": "bundled"},
+            {"name": "custom.wav", "source": "custom"},
+        ],
+    }
+
+
+def test_list_sounds_route_payload_returns_structured_catalog(tmp_path: Path):
+    bundled_dir = tmp_path / "bundled"
+    custom_dir = tmp_path / "custom"
+    legacy_custom_dir = tmp_path / "legacy"
+    bundled_dir.mkdir()
+    custom_dir.mkdir()
+    (bundled_dir / "ping-success.wav").write_bytes(b"bundled")
+    (custom_dir / "custom.wav").write_bytes(b"custom")
+
+    from comfyui_ping.routes import list_sounds_route_payload
+
+    payload = list_sounds_route_payload(
+        bundled_dir=bundled_dir,
+        custom_dir=custom_dir,
+        legacy_custom_dir=legacy_custom_dir,
+    )
+
+    assert payload == {
+        "sounds": [
+            {"name": "ping-success.wav", "source": "bundled"},
+            {"name": "custom.wav", "source": "custom"},
+        ],
+    }
+
+
+def test_resolve_sound_path_uses_requested_source():
+    bundled_dir = REPO_ROOT / "sounds" / "bundled"
+    custom_dir = REPO_ROOT / "sounds" / "custom"
+    legacy_custom_dir = REPO_ROOT / "input" / "ping-unused-test"
+    custom_dir.mkdir(parents=True, exist_ok=True)
+    custom_file = custom_dir / "ping-success.wav"
+    custom_file.write_bytes(b"custom")
+
+    try:
+        bundled_path = resolve_sound_path(
+            filename="ping-success.wav",
+            source="bundled",
+            bundled_dir=bundled_dir,
+            custom_dir=custom_dir,
+            legacy_custom_dir=legacy_custom_dir,
+        )
+        custom_path = resolve_sound_path(
+            filename="ping-success.wav",
+            source="custom",
+            bundled_dir=bundled_dir,
+            custom_dir=custom_dir,
+            legacy_custom_dir=legacy_custom_dir,
+        )
+    finally:
+        custom_file.unlink(missing_ok=True)
+
+    assert bundled_path == bundled_dir / "ping-success.wav"
+    assert custom_path == custom_file
+
+
+def test_list_sounds_route_payload_migrates_legacy_custom_uploads(tmp_path: Path):
+    bundled_dir = tmp_path / "sounds" / "bundled"
+    custom_dir = tmp_path / "sounds" / "custom"
+    legacy_custom_dir = tmp_path / "input" / "ping"
+    bundled_dir.mkdir(parents=True)
+    legacy_custom_dir.mkdir(parents=True)
+    (bundled_dir / "ping-success.wav").write_bytes(b"bundled")
+    (legacy_custom_dir / "ringtone5.mp3").write_bytes(b"legacy")
+
+    from comfyui_ping.routes import list_sounds_route_payload
+
+    payload = list_sounds_route_payload(
+        bundled_dir=bundled_dir,
+        custom_dir=custom_dir,
+        legacy_custom_dir=legacy_custom_dir,
+    )
+
+    assert payload == {
+        "sounds": [
+            {"name": "ping-success.wav", "source": "bundled"},
+            {"name": "ringtone5.mp3", "source": "custom"},
+        ],
+    }
+    assert (custom_dir / "ringtone5.mp3").read_bytes() == b"legacy"
+    assert (legacy_custom_dir / "ringtone5.mp3").exists() is False
+
+
+def test_list_sounds_route_payload_skips_legacy_files_promoted_to_bundled(
+    tmp_path: Path,
+):
+    bundled_dir = tmp_path / "sounds" / "bundled"
+    custom_dir = tmp_path / "sounds" / "custom"
+    legacy_custom_dir = tmp_path / "input" / "ping"
+    bundled_dir.mkdir(parents=True)
+    legacy_custom_dir.mkdir(parents=True)
+    (bundled_dir / "ping-ringtone.wav").write_bytes(b"bundled")
+    (legacy_custom_dir / "ping-ringtone.wav").write_bytes(b"legacy")
+
+    from comfyui_ping.routes import list_sounds_route_payload
+
+    payload = list_sounds_route_payload(
+        bundled_dir=bundled_dir,
+        custom_dir=custom_dir,
+        legacy_custom_dir=legacy_custom_dir,
+    )
+
+    assert payload == {
+        "sounds": [{"name": "ping-ringtone.wav", "source": "bundled"}],
+    }
+    assert (custom_dir / "ping-ringtone.wav").exists() is False
+    assert (legacy_custom_dir / "ping-ringtone.wav").exists() is False
+
+
+def test_save_uploaded_sound_rejects_invalid_extension(tmp_path: Path):
+    try:
+        save_uploaded_sound(
+            filename="not-a-sound.txt",
+            data=b"bad",
+            custom_dir=tmp_path,
+            legacy_custom_dir=tmp_path / "legacy",
+        )
+    except ValueError as exc:
+        assert str(exc) == "Invalid audio file format"
+    else:
+        raise AssertionError("save_uploaded_sound should reject invalid extensions")
+
+
+def test_save_uploaded_sound_rejects_oversized_file(tmp_path: Path):
+    try:
+        save_uploaded_sound(
+            filename="too-large.wav",
+            data=b"0" * (MAX_UPLOAD_BYTES + 1),
+            custom_dir=tmp_path / "custom",
+            legacy_custom_dir=tmp_path / "legacy",
+        )
+    except ValueError as exc:
+        assert str(exc) == "Sound file exceeds 10 MiB limit"
+    else:
+        raise AssertionError("save_uploaded_sound should reject oversized files")
+
+
+def test_save_uploaded_sound_rejects_bundled_name_collision(tmp_path: Path):
+    bundled_dir = tmp_path / "bundled"
+    legacy_custom_dir = tmp_path / "legacy"
+    bundled_dir.mkdir()
+    (bundled_dir / "ping-success.wav").write_bytes(b"bundled")
+
+    try:
+        save_uploaded_sound(
+            filename="ping-success.wav",
+            data=b"custom",
+            custom_dir=tmp_path / "custom",
+            bundled_dir=bundled_dir,
+            legacy_custom_dir=legacy_custom_dir,
+        )
+    except ValueError as exc:
+        assert str(exc) == "Filename collides with bundled sound"
+    else:
+        raise AssertionError("save_uploaded_sound should reject bundled name collisions")
+
+
+def test_save_uploaded_sound_rejects_existing_custom_filename(tmp_path: Path):
+    custom_dir = tmp_path / "custom"
+    legacy_custom_dir = tmp_path / "legacy"
+    custom_dir.mkdir()
+    (custom_dir / "custom.wav").write_bytes(b"old")
+
+    try:
+        save_uploaded_sound(
+            filename="custom.wav",
+            data=b"new",
+            custom_dir=custom_dir,
+            bundled_dir=tmp_path / "bundled",
+            legacy_custom_dir=legacy_custom_dir,
+        )
+    except ValueError as exc:
+        assert str(exc) == "Filename already exists in custom sounds"
+    else:
+        raise AssertionError("save_uploaded_sound should reject existing custom filenames")
+
+
+def test_resolve_sound_path_rejects_non_files(tmp_path: Path):
+    bundled_dir = tmp_path / "bundled"
+    custom_dir = tmp_path / "custom"
+    legacy_custom_dir = tmp_path / "legacy"
+    (bundled_dir / "ping-success.wav").mkdir(parents=True)
+    custom_dir.mkdir()
+
+    path = resolve_sound_path(
+        filename="ping-success.wav",
+        source="bundled",
+        bundled_dir=bundled_dir,
+        custom_dir=custom_dir,
+        legacy_custom_dir=legacy_custom_dir,
+    )
+
+    assert path is None
+
+
+class FakeRoutes:
+    def __init__(self) -> None:
+        self.handlers: dict[tuple[str, str], object] = {}
+
+    def get(self, path: str):
+        def decorator(handler):
+            self.handlers[("GET", path)] = handler
+            return handler
+
+        return decorator
+
+    def post(self, path: str):
+        def decorator(handler):
+            self.handlers[("POST", path)] = handler
+            return handler
+
+        return decorator
+
+
+class FakeWeb:
+    class HTTPNotFound(Exception):
+        pass
+
+    class Response:
+        def __init__(
+            self,
+            *,
+            payload: object | None = None,
+            path: Path | None = None,
+            status: int = 200,
+        ) -> None:
+            self.payload = payload
+            self.path = path
+            self.status = status
+
+    @staticmethod
+    def json_response(payload: object, status: int = 200) -> "FakeWeb.Response":
+        return FakeWeb.Response(payload=payload, status=status)
+
+    @staticmethod
+    def FileResponse(path: Path) -> "FakeWeb.Response":
+        return FakeWeb.Response(path=path, status=200)
+
+
+class FakeUploadField:
+    def __init__(self, *, filename: str, data: bytes) -> None:
+        self.filename = filename
+        self.file = BytesIO(data)
+
+
+class FakeRequest:
+    def __init__(
+        self,
+        *,
+        post_data: dict[str, object] | None = None,
+        match_info: dict[str, str] | None = None,
+    ) -> None:
+        self._post_data = post_data or {}
+        self.match_info = match_info or {}
+
+    async def post(self) -> dict[str, object]:
+        return self._post_data
+
+
+def test_attach_sound_routes_registers_live_handlers(tmp_path: Path):
+    routes = FakeRoutes()
+    bundled_dir = tmp_path / "bundled"
+    custom_dir = tmp_path / "custom"
+    legacy_custom_dir = tmp_path / "legacy"
+    bundled_dir.mkdir()
+    custom_dir.mkdir()
+    (bundled_dir / "ping-success.wav").write_bytes(b"bundled")
+
+    attach_sound_routes(
+        routes=routes,
+        web_module=FakeWeb,
+        bundled_dir=bundled_dir,
+        custom_dir=custom_dir,
+        legacy_custom_dir=legacy_custom_dir,
+    )
+
+    assert ("GET", SOUND_CATALOG_ROUTE) in routes.handlers
+    assert ("GET", SOUND_FILE_ROUTE) in routes.handlers
+    assert ("POST", SOUND_UPLOAD_ROUTE) in routes.handlers
+
+    list_handler = routes.handlers[("GET", SOUND_CATALOG_ROUTE)]
+    list_response = run(list_handler(FakeRequest()))
+    assert list_response.payload == {
+        "sounds": [{"name": "ping-success.wav", "source": "bundled"}]
+    }
+
+    upload_handler = routes.handlers[("POST", SOUND_UPLOAD_ROUTE)]
+    upload_response = run(
+        upload_handler(
+            FakeRequest(
+                post_data={
+                    "file": FakeUploadField(
+                        filename="custom.wav",
+                        data=b"custom",
+                    )
+                }
+            )
+        )
+    )
+    assert upload_response.status == 201
+    assert upload_response.payload == {
+        "sounds": [
+            {"name": "ping-success.wav", "source": "bundled"},
+            {"name": "custom.wav", "source": "custom"},
+        ]
+    }
+
+    serve_handler = routes.handlers[("GET", SOUND_FILE_ROUTE)]
+    serve_response = run(
+        serve_handler(
+            FakeRequest(
+                match_info={
+                    "source": "custom",
+                    "filename": "custom.wav",
+                }
+            )
+        )
+    )
+    assert serve_response.path == custom_dir / "custom.wav"

tests/python/test_sound_storage.py

@@ -0,0 +1,55 @@
+import sys
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+from comfyui_ping.sounds import (
+    ALLOWED_EXTENSIONS,
+    is_allowed_upload,
+    list_available_sounds,
+)
+
+
+def test_list_sounds_returns_bundled_and_custom_entries():
+    sounds = list_available_sounds(
+        bundled=["ping-success.wav"],
+        custom=["my.wav"],
+    )
+
+    assert sounds == [
+        {"name": "ping-success.wav", "source": "bundled"},
+        {"name": "my.wav", "source": "custom"},
+    ]
+
+
+def test_rejects_non_audio_extension():
+    assert is_allowed_upload("not-a-sound.txt") is False
+
+
+def test_accepts_expected_audio_extensions():
+    assert ".wav" in ALLOWED_EXTENSIONS
+    assert is_allowed_upload("ping-success.wav") is True
+    assert is_allowed_upload("failure.ogg") is True
+
+
+def test_bundled_sound_files_exist():
+    assert (REPO_ROOT / "sounds" / "bundled" / "ping-success.wav").exists()
+    assert (REPO_ROOT / "sounds" / "bundled" / "ping-failure.wav").exists()
+    assert (REPO_ROOT / "sounds" / "bundled" / "ping-ringtone.wav").exists()
+    assert (REPO_ROOT / "sounds" / "bundled" / "notification-soft.wav").exists()
+    assert (REPO_ROOT / "sounds" / "bundled" / "beep-ping.wav").exists()
+    assert (REPO_ROOT / "sounds" / "bundled" / "harmonic-beep.wav").exists()
+
+
+def test_catalog_preserves_duplicate_names_by_source():
+    sounds = list_available_sounds(
+        bundled=["ping-success.wav"],
+        custom=["ping-success.wav"],
+    )
+
+    assert sounds == [
+        {"name": "ping-success.wav", "source": "bundled"},
+        {"name": "ping-success.wav", "source": "custom"},
+    ]