PostgresNodePortManager__Generic is added

testgres/node.py

@@ -100,6 +100,8 @@
     options_string, \
     clean_on_error
 
+from .helpers.port_manager import PortForException
+
 from .backup import NodeBackup
 
 from .operations.os_ops import ConnectionParams
@@ -172,6 +174,52 @@ def release_port(self, number: int) -> None:
         return utils.release_port(number)
 
 
+class PostgresNodePortManager__Generic(PostgresNodePortManager):
+    _os_ops: OsOperations
+    _allocated_ports_guard: object
+    _allocated_ports: set[int]
+
+    def __init__(self, os_ops: OsOperations):
+        assert os_ops is not None
+        assert isinstance(os_ops, OsOperations)
+        self._os_ops = os_ops
+        self._allocated_ports_guard = threading.Lock()
+        self._allocated_ports = set[int]()
+
+    def reserve_port(self) -> int:
+        ports = set(range(1024, 65535))
+        assert type(ports) == set  # noqa: E721
+
+        assert self._allocated_ports_guard is not None
+        assert type(self._allocated_ports) == set  # noqa: E721
+
+        with self._allocated_ports_guard:
+            ports.difference_update(self._allocated_ports)
+
+            sampled_ports = random.sample(tuple(ports), min(len(ports), 100))
+
+            for port in sampled_ports:
+                assert not (port in self._allocated_ports)
+
+                if not self._os_ops.is_port_free(port):
+                    continue
+
+                self._allocated_ports.add(port)
+                return port
+
+        raise PortForException("Can't select a port")
+
+    def release_port(self, number: int) -> None:
+        assert type(number) == int  # noqa: E721
+
+        assert self._allocated_ports_guard is not None
+        assert type(self._allocated_ports) == set  # noqa: E721
+
+        with self._allocated_ports_guard:
+            assert number in self._allocated_ports
+            self._allocated_ports.discard(number)
+
+
 class PostgresNode(object):
     # a max number of node start attempts
     _C_MAX_START_ATEMPTS = 5
@@ -308,8 +356,11 @@ def _get_port_manager(os_ops: OsOperations) -> PostgresNodePortManager:
         assert os_ops is not None
         assert isinstance(os_ops, OsOperations)
 
-        # [2025-04-03] It is our old, incorrected behaviour
-        return PostgresNodePortManager__ThisHost()
+        if isinstance(os_ops, LocalOperations):
+            return PostgresNodePortManager__ThisHost()
+
+        # TODO: Throw exception "Please define a port manager."
+        return PostgresNodePortManager__Generic(os_ops)
 
     def clone_with_new_name_and_base_dir(self, name: str, base_dir: str):
         assert name is None or type(name) == str  # noqa: E721

testgres/operations/local_ops.py

@@ -6,6 +6,7 @@
 import subprocess
 import tempfile
 import time
+import socket
 
 import psutil
 
@@ -436,6 +437,16 @@ def get_process_children(self, pid):
         assert type(pid) == int  # noqa: E721
         return psutil.Process(pid).children()
 
+    def is_port_free(self, number: int) -> bool:
+        assert type(number) == int  # noqa: E721
+
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            try:
+                s.bind(("", number))
+                return True
+            except OSError:
+                return False
+
     # Database control
     def db_connect(self, dbname, user, password=None, host="localhost", port=5432):
         conn = pglib.connect(

testgres/operations/os_ops.py

@@ -127,6 +127,10 @@ def get_pid(self):
     def get_process_children(self, pid):
         raise NotImplementedError()
 
+    def is_port_free(self, number: int):
+        assert type(number) == int  # noqa: E721
+        raise NotImplementedError()
+
     # Database control
     def db_connect(self, dbname, user, password=None, host="localhost", port=5432):
         raise NotImplementedError()

testgres/operations/remote_ops.py

@@ -629,6 +629,44 @@ def get_process_children(self, pid):
 
         raise ExecUtilException(f"Error in getting process children. Error: {result.stderr}")
 
+    def is_port_free(self, number: int) -> bool:
+        assert type(number) == int  # noqa: E721
+
+        cmd = ["nc", "-w", "5", "-z", "-v", "localhost", str(number)]
+
+        exit_status, output, error = self.exec_command(cmd=cmd, encoding=get_default_encoding(), ignore_errors=True, verbose=True)
+
+        assert type(output) == str  # noqa: E721
+        assert type(error) == str  # noqa: E721
+
+        if exit_status == 0:
+            return __class__.helper__is_port_free__process_0(output)
+
+        if exit_status == 1:
+            return __class__.helper__is_port_free__process_1(error)
+
+        errMsg = "nc returns an unknown result code: {0}".format(exit_status)
+
+        RaiseError.CommandExecutionError(
+            cmd=cmd,
+            exit_code=exit_status,
+            msg_arg=errMsg,
+            error=error,
+            out=output
+        )
+
+    @staticmethod
+    def helper__is_port_free__process_0(output: str) -> bool:
+        assert type(output) == str  # noqa: E721
+        # TODO: check output message
+        return False
+
+    @staticmethod
+    def helper__is_port_free__process_1(error: str) -> bool:
+        assert type(error) == str  # noqa: E721
+        # TODO: check error message
+        return True
+
     # Database control
     def db_connect(self, dbname, user, password=None, host="localhost", port=5432):
         conn = pglib.connect(

tests/test_testgres_remote.py

@@ -4,7 +4,6 @@
 import subprocess
 
 import pytest
-import psutil
 import logging
 
 from .helpers.os_ops_descrs import OsOpsDescrs
@@ -27,8 +26,6 @@
     get_pg_config
 
 # NOTE: those are ugly imports
-from ..testgres import bound_ports
-from ..testgres.node import ProcessProxy
 
 
 def util_exists(util):
@@ -259,70 +256,6 @@ def test_unix_sockets(self):
                 assert (res_exec == [(1,)])
                 assert (res_psql == b'1\n')
 
-    def test_ports_management(self):
-        assert bound_ports is not None
-        assert type(bound_ports) == set  # noqa: E721
-
-        if len(bound_ports) != 0:
-            logging.warning("bound_ports is not empty: {0}".format(bound_ports))
-
-        stage0__bound_ports = bound_ports.copy()
-
-        with __class__.helper__get_node() as node:
-            assert bound_ports is not None
-            assert type(bound_ports) == set  # noqa: E721
-
-            assert node.port is not None
-            assert type(node.port) == int  # noqa: E721
-
-            logging.info("node port is {0}".format(node.port))
-
-            assert node.port in bound_ports
-            assert node.port not in stage0__bound_ports
-
-            assert stage0__bound_ports <= bound_ports
-            assert len(stage0__bound_ports) + 1 == len(bound_ports)
-
-            stage1__bound_ports = stage0__bound_ports.copy()
-            stage1__bound_ports.add(node.port)
-
-            assert stage1__bound_ports == bound_ports
-
-        # check that port has been freed successfully
-        assert bound_ports is not None
-        assert type(bound_ports) == set  # noqa: E721
-        assert bound_ports == stage0__bound_ports
-
-    # TODO: Why does not this test work with remote host?
-    def test_child_process_dies(self):
-        nAttempt = 0
-
-        while True:
-            if nAttempt == 5:
-                raise Exception("Max attempt number is exceed.")
-
-            nAttempt += 1
-
-            logging.info("Attempt #{0}".format(nAttempt))
-
-            # test for FileNotFound exception during child_processes() function
-            with subprocess.Popen(["sleep", "60"]) as process:
-                r = process.poll()
-
-                if r is not None:
-                    logging.warning("process.pool() returns an unexpected result: {0}.".format(r))
-                    continue
-
-                assert r is None
-                # collect list of processes currently running
-                children = psutil.Process(os.getpid()).children()
-                # kill a process, so received children dictionary becomes invalid
-                process.kill()
-                process.wait()
-                # try to handle children list -- missing processes will have ptype "ProcessType.Unknown"
-                [ProcessProxy(p) for p in children]
-                break
-
     @staticmethod
     def helper__get_node(name=None):
         assert isinstance(__class__.sm_os_ops, OsOperations)