Robert Wojciechowski
a1faf7e916
* fix: isolate local worker databases and avoid port conflicts * fix: harden local worker startup preflight and database paths Handle auto-launch before the master binds, allow TIME_WAIT restarts without accepting active listeners, resolve relative paths in the worker cwd, and round-trip SQLite URLs through SQLAlchemy. Add regressions and remove the README additions.
114 lines
5.4 KiB
Python
114 lines
5.4 KiB
Python
import importlib.util
|
|
import os
|
|
import socket
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
def load_ports():
|
|
path = Path(__file__).resolve().parents[1] / "workers" / "ports.py"
|
|
spec = importlib.util.spec_from_file_location("worker_ports_test", path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class WorkerPortTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.ports = load_ports()
|
|
|
|
def test_allocation_starts_above_actual_master_and_skips_assigned_and_occupied(self):
|
|
workers = [
|
|
{"id": "existing", "host": "localhost", "port": 8190, "enabled": False},
|
|
{"id": "remote", "host": "remote.example", "port": 8192},
|
|
]
|
|
with patch.object(self.ports, "is_port_available", side_effect=lambda port: port != 8191):
|
|
self.assertEqual(self.ports.allocate_worker_ports(8189, workers, 3), [8192, 8193, 8194])
|
|
|
|
def test_local_host_forms_reserve_ports(self):
|
|
for host in ["::1", "[::1]", "http://localhost/", "HTTPS://LOCALHOST", "0.0.0.0", None]:
|
|
with self.subTest(host=host), patch.object(self.ports, "is_port_available", return_value=True):
|
|
self.assertEqual(self.ports.allocate_worker_ports(8189, [{"id": "local", "host": host, "port": 8190}], 1), [8191])
|
|
|
|
def test_exhaustion_is_explicit_not_a_partial_allocation(self):
|
|
with patch.object(self.ports, "is_port_available", return_value=True):
|
|
with self.assertRaisesRegex(ValueError, "available.*ports"):
|
|
self.ports.allocate_worker_ports(65534, [], 2)
|
|
|
|
def test_launch_rejects_master_port_without_reassigning_configuration(self):
|
|
worker = {"id": "manual", "port": 8189}
|
|
with self.assertRaisesRegex(ValueError, "master.*8189"):
|
|
self.ports.validate_worker_port(worker, 8189, [])
|
|
self.assertEqual(worker["port"], 8189)
|
|
|
|
def test_launch_rejects_another_local_workers_reserved_port(self):
|
|
worker = {"id": "manual", "port": 8190}
|
|
other = {"id": "other", "host": None, "port": 8190, "enabled": False}
|
|
with self.assertRaisesRegex(ValueError, "other"):
|
|
self.ports.validate_worker_port(worker, 8189, [worker, other])
|
|
|
|
def test_launch_ignores_itself_and_remote_workers_with_same_port(self):
|
|
worker = {"id": "manual", "port": 8190}
|
|
other = {"id": "remote", "host": "example.com", "port": 8190}
|
|
with patch.object(self.ports, "is_port_available", return_value=True):
|
|
self.ports.validate_worker_port(worker, 8189, [worker, other])
|
|
|
|
def test_launch_rejects_an_occupied_port(self):
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
|
|
listener.bind(("127.0.0.1", 0))
|
|
listener.listen()
|
|
port = listener.getsockname()[1]
|
|
self.assertFalse(self.ports.is_port_available(port))
|
|
with self.assertRaisesRegex(ValueError, "already in use"):
|
|
self.ports.validate_worker_port({"id": "manual", "port": port}, 8189, [])
|
|
|
|
@unittest.skipIf(os.name == "nt", "asyncio does not reuse addresses on Windows")
|
|
def test_recently_closed_connection_does_not_block_worker_restart(self):
|
|
with socket.socket() as listener:
|
|
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
listener.bind(("127.0.0.1", 0))
|
|
listener.listen(1)
|
|
port = listener.getsockname()[1]
|
|
with socket.create_connection(("127.0.0.1", port), timeout=2) as client:
|
|
accepted, _ = listener.accept()
|
|
with accepted:
|
|
accepted.settimeout(2)
|
|
accepted.shutdown(socket.SHUT_WR)
|
|
self.assertEqual(client.recv(1), b"")
|
|
client.shutdown(socket.SHUT_WR)
|
|
self.assertEqual(accepted.recv(1), b"")
|
|
self.assertTrue(self.ports.is_port_available(port))
|
|
self.ports.validate_worker_port({"id": "restart", "port": port}, 1, [])
|
|
with socket.socket() as restarted:
|
|
restarted.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
restarted.bind(("127.0.0.1", port))
|
|
restarted.listen(1)
|
|
|
|
def test_active_listener_with_reuseaddr_is_still_a_conflict(self):
|
|
with socket.socket() as listener:
|
|
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
listener.bind(("127.0.0.1", 0))
|
|
listener.listen(1)
|
|
self.assertFalse(self.ports.is_port_available(listener.getsockname()[1]))
|
|
|
|
@unittest.skipUnless(socket.has_ipv6, "IPv6 unavailable")
|
|
def test_ipv6_only_listener_is_still_a_conflict(self):
|
|
with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as listener:
|
|
listener.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
|
|
try:
|
|
listener.bind(("::1", 0))
|
|
except OSError as exc:
|
|
self.skipTest(f"IPv6 loopback unavailable: {exc}")
|
|
listener.listen(1)
|
|
self.assertFalse(self.ports.is_port_available(listener.getsockname()[1]))
|
|
|
|
def test_invalid_ports_fail_clearly(self):
|
|
for port in [0, 65536, "not-a-port", None, True, 8190.5]:
|
|
with self.subTest(port=port), self.assertRaisesRegex(ValueError, "port"):
|
|
self.ports.validate_worker_port({"id": "manual", "port": port}, 8189, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|