mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-06-17 21:39:45 +08:00
Adds port availability check before loading custom nodes to prevent port conflicts that only surface after lengthy custom node loading. Changes: - Added socket import for port checking - Added port availability check in main.py before creating PromptServer - Created port_check.py utility module with helper functions - Created test_port_check.py with comprehensive tests The check will: 1. Verify the server port is available before any heavy loading 2. Exit early with clear error message if port is in use 3. Provide helpful instructions for resolving the conflict Fixes #8935 [Bounty]
62 lines
1.7 KiB
Python
62 lines
1.7 KiB
Python
"""
|
|
Test for port check functionality in ComfyUI
|
|
"""
|
|
|
|
import socket
|
|
import threading
|
|
import time
|
|
import sys
|
|
sys.path.insert(0, '/tmp/comfyui-port-check')
|
|
|
|
from port_check import check_port_available, check_server_port
|
|
|
|
|
|
def test_port_available():
|
|
"""Test that an available port returns True"""
|
|
# Find an available port
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(('127.0.0.1', 0))
|
|
port = s.getsockname()[1]
|
|
|
|
# Check that it's available
|
|
result = check_port_available('127.0.0.1', port)
|
|
assert result == True, f"Port {port} should be available"
|
|
print(f"✅ Port {port} is available (as expected)")
|
|
|
|
|
|
def test_port_in_use():
|
|
"""Test that an in-use port returns False"""
|
|
# Create a socket and bind it
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.bind(('127.0.0.1', 0))
|
|
port = sock.getsockname()[1]
|
|
sock.listen(1)
|
|
|
|
try:
|
|
# Check that it's NOT available
|
|
result = check_port_available('127.0.0.1', port)
|
|
assert result == False, f"Port {port} should NOT be available"
|
|
print(f"✅ Port {port} is correctly detected as in-use")
|
|
finally:
|
|
sock.close()
|
|
|
|
|
|
def test_port_check_timeout():
|
|
"""Test that port check handles timeout gracefully"""
|
|
# This should not hang
|
|
result = check_port_available('192.0.2.1', 12345, timeout=0.1)
|
|
# Result may vary depending on network, but should not hang
|
|
print(f"✅ Port check completed without hanging")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Testing port check functionality...")
|
|
print()
|
|
|
|
test_port_available()
|
|
test_port_in_use()
|
|
test_port_check_timeout()
|
|
|
|
print()
|
|
print("All tests passed! ✅")
|