Apr 6, 2025 · Apr 3, 2025 · Apr 4, 2025 · Apr 4, 2025 · Apr 4, 2025 · Apr 4, 2025
diff --git a/testgres/node.py b/testgres/node.py
 from queue import Queue

 import time
 import typing

 try:
    from collections.abc import Iterable
                                                 repr(self.process))


 class PostgresNodePortManager:
    def reserve_port(self) -> int:
        raise NotImplementedError("PostManager::reserve_port is not implemented.")

    def release_port(self, number: int) -> None:
        assert type(number) == int  # noqa: E721
        raise NotImplementedError("PostManager::release_port is not implemented.")


 class PostgresNodePortManager__Global(PostgresNodePortManager):
    def __init__(self):
        pass

    def reserve_port(self) -> int:
        return utils.reserve_port()

    def release_port(self, number: int) -> None:
        assert type(number) == int  # noqa: E721
        return utils.release_port(number)


 class PostgresNode(object):
    # a max number of node start attempts
    _C_MAX_START_ATEMPTS = 5

    def __init__(self, name=None, base_dir=None, port=None, conn_params: ConnectionParams = ConnectionParams(),
                 bin_dir=None, prefix=None, os_ops=None):
    _name: typing.Optional[str]
    _host: typing.Optional[str]
    _port: typing.Optional[int]
    _should_free_port: bool
    _os_ops: OsOperations
    _port_manager: PostgresNodePortManager

    def __init__(self,
                 name=None,
                 base_dir=None,
                 port: typing.Optional[int] = None,
                 conn_params: ConnectionParams = ConnectionParams(),
                 bin_dir=None,
                 prefix=None,
                 os_ops: typing.Optional[OsOperations] = None,
                 port_manager: typing.Optional[PostgresNodePortManager] = None):
        """
        PostgresNode constructor.

            port: port to accept connections.
            base_dir: path to node's data directory.
            bin_dir: path to node's binary directory.
            os_ops: None or correct OS operation object.
            port_manager: None or correct port manager object.
        """
        assert port is None or type(port) == int  # noqa: E721
        assert os_ops is None or isinstance(os_ops, OsOperations)
        assert port_manager is None or isinstance(port_manager, PostgresNodePortManager)

        # private
        if os_ops is None:
 os_ops = __class__._get_os_ops(conn_params)
 self._os_ops = __class__._get_os_ops(conn_params)
        else:
            assert conn_params is None
            assert isinstance(os_ops, OsOperations)
            self._os_ops = os_ops
            pass

        assert os_ops is not None
        assert isinstance(os_ops, OsOperations)
        self._os_ops = os_ops
        assert self._os_ops is not None
        assert isinstance(self._os_ops, OsOperations)

        self._pg_version = PgVer(get_pg_version2(os_ops, bin_dir))
        self._should_free_port = port is None
        self._pg_version = PgVer(get_pg_version2(self._os_ops, bin_dir))
        self._base_dir = base_dir
        self._bin_dir = bin_dir
        self._prefix = prefix
        self._logger = None
        self._master = None

        # basic
        self.name = name or generate_app_name()
        self._name = name or generate_app_name()
        self._host = self._os_ops.host

        self.host = os_ops.host
        self.port = port or utils.reserve_port()
        if port is not None:
            assert type(port) == int  # noqa: E721
            assert port_manager is None
            self._port = port
            self._should_free_port = False
            self._port_manager = None
        else:
            if port_manager is not None:
                assert isinstance(port_manager, PostgresNodePortManager)
                self._port_manager = port_manager
            else:
                self._port_manager = __class__._get_port_manager(self._os_ops)

            assert self._port_manager is not None
            assert isinstance(self._port_manager, PostgresNodePortManager)

            self._port = self._port_manager.reserve_port()  # raises
            assert type(self._port) == int  # noqa: E721
            self._should_free_port = True

        self.ssh_key = os_ops.ssh_key
 assert type(self._port) == int  # noqa: E721

        # defaults for __exit__()
        self.cleanup_on_good_exit = testgres_config.node_cleanup_on_good_exit

    def __repr__(self):
        return "{}(name='{}', port={}, base_dir='{}')".format(
            self.__class__.__name__, self.name, self.port, self.base_dir)
            self.__class__.__name__,
            self.name,
            str(self._port) if self._port is not None else "None",
            self.base_dir
        )

    @staticmethod
    def _get_os_ops(conn_params: ConnectionParams) -> OsOperations:

        return LocalOperations(conn_params)

    @staticmethod
    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__Global()

    def clone_with_new_name_and_base_dir(self, name: str, base_dir: str):
        assert name is None or type(name) == str  # noqa: E721
        assert base_dir is None or type(base_dir) == str  # noqa: E721

        assert __class__ == PostgresNode

        if self._port_manager is None:
            raise InvalidOperationException("PostgresNode without PortManager can't be cloned.")

        assert self._port_manager is not None
        assert isinstance(self._port_manager, PostgresNodePortManager)
        assert self._os_ops is not None
        assert isinstance(self._os_ops, OsOperations)

        node = PostgresNode(
            name=name,
            base_dir=base_dir,
        assert isinstance(self._os_ops, OsOperations)
        return self._os_ops

    @property
    def name(self) -> str:
        if self._name is None:
            raise InvalidOperationException("PostgresNode name is not defined.")
        assert type(self._name) == str  # noqa: E721
        return self._name

    @property
    def host(self) -> str:
        if self._host is None:
            raise InvalidOperationException("PostgresNode host is not defined.")
        assert type(self._host) == str  # noqa: E721
        return self._host

    @property
    def port(self) -> int:
        if self._port is None:
            raise InvalidOperationException("PostgresNode port is not defined.")

        assert type(self._port) == int  # noqa: E721
        return self._port

    @property
    def ssh_key(self) -> typing.Optional[str]:
        assert self._os_ops is not None
        assert isinstance(self._os_ops, OsOperations)
        return self._os_ops.ssh_key

    @property
    def pid(self):
        """
        if self.is_started:
            return self

        if self._port is None:
            raise InvalidOperationException("Can't start PostgresNode. Port is node defined.")

        assert type(self._port) == int  # noqa: E721

        _params = [self._get_bin_path("pg_ctl"),
                   "-D", self.data_dir,
                   "-l", self.pg_log_file,
                LOCAL__raise_cannot_start_node__std(e)
        else:
            assert self._should_free_port
            assert self._port_manager is not None
            assert isinstance(self._port_manager, PostgresNodePortManager)
            assert __class__._C_MAX_START_ATEMPTS > 1

            log_files0 = self._collect_log_files()

                    log_files0 = log_files1
                    logging.warning(
                        "Detected a conflict with using the port {0}. Trying another port after a {1}-second sleep...".format(self.port, timeout)
                        "Detected a conflict with using the port {0}. Trying another port after a {1}-second sleep...".format(self._port, timeout)
                    )
                    time.sleep(timeout)
                    timeout = min(2 * timeout, 5)
                    cur_port = self.port
                    new_port =utils.reserve_port()  # can raise
                    cur_port = self._port
                    new_port =self._port_manager.reserve_port()  # can raise
                    try:
                        options = {'port': new_port}
                        self.set_auto_conf(options)
                    except:  # noqa: E722
 utils.release_port(new_port)
 self._port_manager.release_port(new_port)
                        raise
                    self.port = new_port
 utils.release_port(cur_port)
                    self._port = new_port
 self._port_manager.release_port(cur_port)
                    continue
                break
        self._maybe_start_logger()
        """

        if self._should_free_port:
            port = self.port
            assert type(self._port) == int  # noqa: E721

            assert self._port_manager is not None
            assert isinstance(self._port_manager, PostgresNodePortManager)

            port = self._port
            self._should_free_port = False
            self.port = None
 utils.release_port(port)
            self._port = None
 self._port_manager.release_port(port)

    def cleanup(self, max_attempts=3, full=False):
        """
Original file line number	Diff line number	Diff line change
Expand Up		@@ -10,6 +10,7 @@
		from queue import Queue

		import time
		import typing

		try:
		from collections.abc import Iterable
Expand DownExpand Up		@@ -131,12 +132,47 @@ def __repr__(self):
		repr(self.process))


		class PostgresNodePortManager:
		def reserve_port(self) -> int:
		raise NotImplementedError("PostManager::reserve_port is not implemented.")

		def release_port(self, number: int) -> None:
		assert type(number) == int # noqa: E721
		raise NotImplementedError("PostManager::release_port is not implemented.")


		class PostgresNodePortManager__Global(PostgresNodePortManager):
		def __init__(self):
		pass

		def reserve_port(self) -> int:
		return utils.reserve_port()

		def release_port(self, number: int) -> None:
		assert type(number) == int # noqa: E721
		return utils.release_port(number)


		class PostgresNode(object):
		# a max number of node start attempts
		_C_MAX_START_ATEMPTS = 5

		def __init__(self, name=None, base_dir=None, port=None, conn_params: ConnectionParams = ConnectionParams(),
		bin_dir=None, prefix=None, os_ops=None):
		_name: typing.Optional[str]
		_host: typing.Optional[str]
		_port: typing.Optional[int]
		_should_free_port: bool
		_os_ops: OsOperations
		_port_manager: PostgresNodePortManager

		def __init__(self,
		name=None,
		base_dir=None,
		port: typing.Optional[int] = None,
		conn_params: ConnectionParams = ConnectionParams(),
		bin_dir=None,
		prefix=None,
		os_ops: typing.Optional[OsOperations] = None,
		port_manager: typing.Optional[PostgresNodePortManager] = None):
		"""
		PostgresNode constructor.

Expand All		@@ -145,34 +181,57 @@ def __init__(self, name=None, base_dir=None, port=None, conn_params: ConnectionP
		port: port to accept connections.
		base_dir: path to node's data directory.
		bin_dir: path to node's binary directory.
		os_ops: None or correct OS operation object.
		port_manager: None or correct port manager object.
		"""
		assert port is None or type(port) == int # noqa: E721
		assert os_ops is None or isinstance(os_ops, OsOperations)
		assert port_manager is None or isinstance(port_manager, PostgresNodePortManager)

		# private
		if os_ops is None:
		os_ops = __class__._get_os_ops(conn_params)
		self._os_ops = __class__._get_os_ops(conn_params)
		else:
		assert conn_params is None
		assert isinstance(os_ops, OsOperations)
		self._os_ops = os_ops
		pass

		assert os_ops is not None
		assert isinstance(os_ops, OsOperations)
		self._os_ops = os_ops
		assert self._os_ops is not None
		assert isinstance(self._os_ops, OsOperations)

		self._pg_version = PgVer(get_pg_version2(os_ops, bin_dir))
		self._should_free_port = port is None
		self._pg_version = PgVer(get_pg_version2(self._os_ops, bin_dir))
		self._base_dir = base_dir
		self._bin_dir = bin_dir
		self._prefix = prefix
		self._logger = None
		self._master = None

		# basic
		self.name = name or generate_app_name()
		self._name = name or generate_app_name()
		self._host = self._os_ops.host

		self.host = os_ops.host
		self.port = port or utils.reserve_port()
		if port is not None:
		assert type(port) == int # noqa: E721
		assert port_manager is None
		self._port = port
		self._should_free_port = False
		self._port_manager = None
		else:
		if port_manager is not None:
		assert isinstance(port_manager, PostgresNodePortManager)
		self._port_manager = port_manager
		else:
		self._port_manager = __class__._get_port_manager(self._os_ops)

		assert self._port_manager is not None
		assert isinstance(self._port_manager, PostgresNodePortManager)

		self._port = self._port_manager.reserve_port() # raises
		assert type(self._port) == int # noqa: E721
		self._should_free_port = True

		self.ssh_key = os_ops.ssh_key
		assert type(self._port) == int # noqa: E721

		# defaults for __exit__()
		self.cleanup_on_good_exit = testgres_config.node_cleanup_on_good_exit
Expand DownExpand Up		@@ -207,7 +266,11 @@ def __exit__(self, type, value, traceback):

		def __repr__(self):
		return "{}(name='{}', port={}, base_dir='{}')".format(
		self.__class__.__name__, self.name, self.port, self.base_dir)
		self.__class__.__name__,
		self.name,
		str(self._port) if self._port is not None else "None",
		self.base_dir
		)

		@staticmethod
		def _get_os_ops(conn_params: ConnectionParams) -> OsOperations:
Expand All		@@ -221,12 +284,28 @@ def _get_os_ops(conn_params: ConnectionParams) -> OsOperations:

		return LocalOperations(conn_params)

		@staticmethod
		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__Global()

		def clone_with_new_name_and_base_dir(self, name: str, base_dir: str):
		assert name is None or type(name) == str # noqa: E721
		assert base_dir is None or type(base_dir) == str # noqa: E721

		assert __class__ == PostgresNode

		if self._port_manager is None:
		raise InvalidOperationException("PostgresNode without PortManager can't be cloned.")

		assert self._port_manager is not None
		assert isinstance(self._port_manager, PostgresNodePortManager)
		assert self._os_ops is not None
		assert isinstance(self._os_ops, OsOperations)

		node = PostgresNode(
		name=name,
		base_dir=base_dir,
Expand All		@@ -243,6 +322,34 @@ def os_ops(self) -> OsOperations:
		assert isinstance(self._os_ops, OsOperations)
		return self._os_ops

		@property
		def name(self) -> str:
		if self._name is None:
		raise InvalidOperationException("PostgresNode name is not defined.")
		assert type(self._name) == str # noqa: E721
		return self._name

		@property
		def host(self) -> str:
		if self._host is None:
		raise InvalidOperationException("PostgresNode host is not defined.")
		assert type(self._host) == str # noqa: E721
		return self._host

		@property
		def port(self) -> int:
		if self._port is None:
		raise InvalidOperationException("PostgresNode port is not defined.")

		assert type(self._port) == int # noqa: E721
		return self._port

		@property
		def ssh_key(self) -> typing.Optional[str]:
		assert self._os_ops is not None
		assert isinstance(self._os_ops, OsOperations)
		return self._os_ops.ssh_key

		@property
		def pid(self):
		"""
Expand DownExpand Up		@@ -993,6 +1100,11 @@ def start(self, params=[], wait=True):
		if self.is_started:
		return self

		if self._port is None:
		raise InvalidOperationException("Can't start PostgresNode. Port is node defined.")

		assert type(self._port) == int # noqa: E721

		_params = [self._get_bin_path("pg_ctl"),
		"-D", self.data_dir,
		"-l", self.pg_log_file,
Expand DownExpand Up		@@ -1023,6 +1135,8 @@ def LOCAL__raise_cannot_start_node__std(from_exception):
		LOCAL__raise_cannot_start_node__std(e)
		else:
		assert self._should_free_port
		assert self._port_manager is not None
		assert isinstance(self._port_manager, PostgresNodePortManager)
		assert __class__._C_MAX_START_ATEMPTS > 1

		log_files0 = self._collect_log_files()
Expand All		@@ -1048,20 +1162,20 @@ def LOCAL__raise_cannot_start_node__std(from_exception):

		log_files0 = log_files1
		logging.warning(
		"Detected a conflict with using the port {0}. Trying another port after a {1}-second sleep...".format(self.port, timeout)
		"Detected a conflict with using the port {0}. Trying another port after a {1}-second sleep...".format(self._port, timeout)
		)
		time.sleep(timeout)
		timeout = min(2 * timeout, 5)
		cur_port = self.port
		new_port =utils.reserve_port() # can raise
		cur_port = self._port
		new_port =self._port_manager.reserve_port() # can raise
		try:
		options = {'port': new_port}
		self.set_auto_conf(options)
		except: # noqa: E722
		utils.release_port(new_port)
		self._port_manager.release_port(new_port)
		raise
		self.port = new_port
		utils.release_port(cur_port)
		self._port = new_port
		self._port_manager.release_port(cur_port)
		continue
		break
		self._maybe_start_logger()
Expand DownExpand Up		@@ -1226,10 +1340,15 @@ def free_port(self):
		"""

		if self._should_free_port:
		port = self.port
		assert type(self._port) == int # noqa: E721

		assert self._port_manager is not None
		assert isinstance(self._port_manager, PostgresNodePortManager)

		port = self._port
		self._should_free_port = False
		self.port = None
		utils.release_port(port)
		self._port = None
		self._port_manager.release_port(port)

		def cleanup(self, max_attempts=3, full=False):
		"""
Expand Down