Jun 23, 2025 · Jun 23, 2025
diff --git a/testgres/impl/port_manager__generic.py b/testgres/impl/port_manager__generic.py


 class PortManager__Generic(PortManager):
    _C_MIN_PORT_NUMBER = 1024
    _C_MAX_PORT_NUMBER = 65535

    _os_ops: OsOperations
    _guard: object
    # TODO: is there better to use bitmap fot _available_ports?
    _available_ports: typing.Set[int]
    _reserved_ports: typing.Set[int]

    def __init__(self, os_ops: OsOperations):
        assert __class__._C_MIN_PORT_NUMBER <= __class__._C_MAX_PORT_NUMBER

        assert os_ops is not None
        assert isinstance(os_ops, OsOperations)
        self._os_ops = os_ops
        self._guard = threading.Lock()
        self._available_ports: typing.Set[int] = set(range(1024, 65535))
        self._reserved_ports: typing.Set[int] = set()

        self._available_ports = set(
            range(__class__._C_MIN_PORT_NUMBER, __class__._C_MAX_PORT_NUMBER + 1)
        )
        assert len(self._available_ports) == (
            (__class__._C_MAX_PORT_NUMBER - __class__._C_MIN_PORT_NUMBER) + 1
        )
        self._reserved_ports = set()
        return

    def reserve_port(self) -> int:
        assert self._guard is not None
            t = None

            for port in sampled_ports:
                assert type(port) == int  # noqa: E721
                assert not (port in self._reserved_ports)
                assert port in self._available_ports

                assert port >= __class__._C_MIN_PORT_NUMBER
                assert port <= __class__._C_MAX_PORT_NUMBER

                if not self._os_ops.is_port_free(port):
                    continue


    def release_port(self, number: int) -> None:
        assert type(number) == int  # noqa: E721
        assert number >= __class__._C_MIN_PORT_NUMBER
        assert number <= __class__._C_MAX_PORT_NUMBER

        assert self._guard is not None
        assert type(self._reserved_ports) == set  # noqa: E721
            self._reserved_ports.discard(number)
            assert not (number in self._reserved_ports)
            assert number in self._available_ports
        return
Original file line number	Diff line number	Diff line change
Expand Up		@@ -9,19 +9,31 @@


		class PortManager__Generic(PortManager):
		_C_MIN_PORT_NUMBER = 1024
		_C_MAX_PORT_NUMBER = 65535

		_os_ops: OsOperations
		_guard: object
		# TODO: is there better to use bitmap fot _available_ports?
		_available_ports: typing.Set[int]
		_reserved_ports: typing.Set[int]

		def __init__(self, os_ops: OsOperations):
		assert __class__._C_MIN_PORT_NUMBER <= __class__._C_MAX_PORT_NUMBER

		assert os_ops is not None
		assert isinstance(os_ops, OsOperations)
		self._os_ops = os_ops
		self._guard = threading.Lock()
		self._available_ports: typing.Set[int] = set(range(1024, 65535))
		self._reserved_ports: typing.Set[int] = set()

		self._available_ports = set(
		range(__class__._C_MIN_PORT_NUMBER, __class__._C_MAX_PORT_NUMBER + 1)
		)
		assert len(self._available_ports) == (
		(__class__._C_MAX_PORT_NUMBER - __class__._C_MIN_PORT_NUMBER) + 1
		)
		self._reserved_ports = set()
		return

		def reserve_port(self) -> int:
		assert self._guard is not None
Expand All		@@ -35,9 +47,13 @@ def reserve_port(self) -> int:
		t = None

		for port in sampled_ports:
		assert type(port) == int # noqa: E721
		assert not (port in self._reserved_ports)
		assert port in self._available_ports

		assert port >= __class__._C_MIN_PORT_NUMBER
		assert port <= __class__._C_MAX_PORT_NUMBER

		if not self._os_ops.is_port_free(port):
		continue

Expand All		@@ -51,6 +67,8 @@ def reserve_port(self) -> int:

		def release_port(self, number: int) -> None:
		assert type(number) == int # noqa: E721
		assert number >= __class__._C_MIN_PORT_NUMBER
		assert number <= __class__._C_MAX_PORT_NUMBER

		assert self._guard is not None
		assert type(self._reserved_ports) == set # noqa: E721
Expand All		@@ -62,3 +80,4 @@ def release_port(self, number: int) -> None:
		self._reserved_ports.discard(number)
		assert not (number in self._reserved_ports)
		assert number in self._available_ports
		return