reformat file and change create_random_port_socket test

This commit is contained in:
Uriel
2026-03-08 06:38:47 -03:00
parent 10d2908339
commit 2725720406
2 changed files with 13 additions and 32 deletions

View File

@@ -210,7 +210,7 @@ def parse_game_ports(game_ports: tuple[str | int, ...]) -> GameRangePorts:
total_length += 1 total_length += 1
weights.append(total_length) weights.append(total_length)
num = int(item) num = int(item)
parsed_ports.append(range(num, num+1)) parsed_ports.append(range(num, num + 1))
return GameRangePorts(parsed_ports, weights, ephemeral_allowed) return GameRangePorts(parsed_ports, weights, ephemeral_allowed)
@@ -492,7 +492,7 @@ def run_server_process(name: str, ponyconfig: dict, static_server_data: dict,
def run(self): def run(self):
while 1: while 1:
next_room = rooms_to_run.get(block=True, timeout=None) next_room = rooms_to_run.get(block=True, timeout=None)
gc.collect() gc.collect()
task = asyncio.run_coroutine_threadsafe(start_room(next_room), loop) task = asyncio.run_coroutine_threadsafe(start_room(next_room), loop)
self._tasks.append(task) self._tasks.append(task)

View File

@@ -1,6 +1,4 @@
import os import os
import statistics
import timeit
import unittest import unittest
from WebHostLib.customserver import parse_game_ports, create_random_port_socket, get_used_ports from WebHostLib.customserver import parse_game_ports, create_random_port_socket, get_used_ports
@@ -44,6 +42,7 @@ class TestWebDescriptions(unittest.TestCase):
parse_game_ports(tuple("f-21215")) parse_game_ports(tuple("f-21215"))
def test_random_port_socket_edge_cases(self) -> None: def test_random_port_socket_edge_cases(self) -> None:
"""Verify if edge cases on creation of random port socket is working fine"""
# Try giving an empty tuple and fail over it # Try giving an empty tuple and fail over it
with self.assertRaises(OSError) as err: with self.assertRaises(OSError) as err:
create_random_port_socket(tuple(), "127.0.0.1") create_random_port_socket(tuple(), "127.0.0.1")
@@ -59,42 +58,24 @@ class TestWebDescriptions(unittest.TestCase):
self.assertNotEqual(err.strerror, "No available ports", self.assertNotEqual(err.strerror, "No available ports",
"Raised an unexpected error string") "Raised an unexpected error string")
# @unittest.skipUnless(ci, "can't guarantee free ports outside of CI") @unittest.skipUnless(ci, "can't guarantee free ports outside of CI")
def test_random_port_socket(self) -> None: def test_random_port_socket(self) -> None:
"""Verify if returned sockets use the correct port ranges"""
sockets = [] sockets = []
for _ in range(6): for _ in range(6):
socket = create_random_port_socket(("8080-8085",), "127.0.0.1") socket = create_random_port_socket(("8080-8085",), "127.0.0.1")
sockets.append(socket) sockets.append(socket)
_, port = socket.getsockname() _, port = socket.getsockname()
self.assertIn(port, range(8080,8086), "Port of socket was not inside the expected range") self.assertIn(port, range(8080, 8086), "Port of socket was not inside the expected range")
for s in sockets: for s in sockets:
s.close() s.close()
# Compared averages were calculated with a range of 100 in a Linux machine and then rounded up
sockets.clear() sockets.clear()
time = [] for _ in range(30_000):
size = 65535 - (len(get_used_ports()) + 1024 + 4000) socket = create_random_port_socket(("30000-65535",), "127.0.0.1")
for _ in range(10): sockets.append(socket)
time.append(timeit.timeit(lambda: sockets.append( _, port = socket.getsockname()
create_random_port_socket(("1024-30000", "30001-65535"), "127.0.0.1") self.assertIn(port, range(30_000, 65536), "Port of socket was not inside the expected range")
), number=size))
for s in sockets: for s in sockets:
s.close() s.close()
self.assertLess(statistics.fmean(time), 1.2,
f"Time took to allocate {size} ports consecutively is higher than expected")
sockets.clear()
time.clear()
size = 65535 - (len(get_used_ports()) + 1024 + 5)
for _ in range(10):
time.append(timeit.timeit(lambda: sockets.append(
create_random_port_socket(("1024-30000", "30001-65535"), "127.0.0.1")
), number=size))
for s in sockets:
s.close()
self.assertLess(statistics.fmean(time), 5,
f"Time took to allocate {size} ports consecutively is higher than expected")