Content-Length: 5333 | pFad | http://github.com/postgrespro/testgres/pull/269.patch

thub.com From ab1d303362fbc4401866a4bee29ca31d7ff37eb5 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Mon, 23 Jun 2025 09:25:51 +0300 Subject: [PATCH] [#264] New property PostrgsNode::port_manager This patch adds the RO-property PostgresNode::port_manager. This property returns a used port manager or None. Declaration of PostgresNode::_port_manager was corrected. New tests for PostgresNode::port_manager and PostgresNode::os_ops are added. --- testgres/node.py | 7 ++- tests/test_testgres_common.py | 82 +++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) diff --git a/testgres/node.py b/testgres/node.py index 9a2f4e7..a019faf 100644 --- a/testgres/node.py +++ b/testgres/node.py @@ -146,7 +146,7 @@ class PostgresNode(object): _port: typing.Optional[int] _should_free_port: bool _os_ops: OsOperations - _port_manager: PortManager + _port_manager: typing.Optional[PortManager] def __init__(self, name=None, @@ -313,6 +313,11 @@ def os_ops(self) -> OsOperations: assert isinstance(self._os_ops, OsOperations) return self._os_ops + @property + def port_manager(self) -> typing.Optional[PortManager]: + assert self._port_manager is None or isinstance(self._port_manager, PortManager) + return self._port_manager + @property def name(self) -> str: if self._name is None: diff --git a/tests/test_testgres_common.py b/tests/test_testgres_common.py index cf203a6..72873a0 100644 --- a/tests/test_testgres_common.py +++ b/tests/test_testgres_common.py @@ -1503,6 +1503,88 @@ def test_try_to_start_node_after_free_manual_port(self, node_svc: PostgresNodeSe ): node2.start() + def test_node__os_ops(self, node_svc: PostgresNodeService): + assert type(node_svc) == PostgresNodeService # noqa: E721 + + assert node_svc.os_ops is not None + assert isinstance(node_svc.os_ops, OsOperations) + + with PostgresNode(name="node", os_ops=node_svc.os_ops, port_manager=node_svc.port_manager) as node: + # retest + assert node_svc.os_ops is not None + assert isinstance(node_svc.os_ops, OsOperations) + + assert node.os_ops is node_svc.os_ops + # one more time + assert node.os_ops is node_svc.os_ops + + def test_node__port_manager(self, node_svc: PostgresNodeService): + assert type(node_svc) == PostgresNodeService # noqa: E721 + + assert node_svc.port_manager is not None + assert isinstance(node_svc.port_manager, PortManager) + + with PostgresNode(name="node", os_ops=node_svc.os_ops, port_manager=node_svc.port_manager) as node: + # retest + assert node_svc.port_manager is not None + assert isinstance(node_svc.port_manager, PortManager) + + assert node.port_manager is node_svc.port_manager + # one more time + assert node.port_manager is node_svc.port_manager + + def test_node__port_manager_and_explicit_port(self, node_svc: PostgresNodeService): + assert type(node_svc) == PostgresNodeService # noqa: E721 + + assert isinstance(node_svc.os_ops, OsOperations) + assert node_svc.port_manager is not None + assert isinstance(node_svc.port_manager, PortManager) + + port = node_svc.port_manager.reserve_port() + assert type(port) == int # noqa: E721 + + try: + with PostgresNode(name="node", port=port, os_ops=node_svc.os_ops) as node: + # retest + assert isinstance(node_svc.os_ops, OsOperations) + assert node_svc.port_manager is not None + assert isinstance(node_svc.port_manager, PortManager) + + assert node.port_manager is None + assert node.os_ops is node_svc.os_ops + + # one more time + assert node.port_manager is None + assert node.os_ops is node_svc.os_ops + finally: + node_svc.port_manager.release_port(port) + + def test_node__no_port_manager(self, node_svc: PostgresNodeService): + assert type(node_svc) == PostgresNodeService # noqa: E721 + + assert isinstance(node_svc.os_ops, OsOperations) + assert node_svc.port_manager is not None + assert isinstance(node_svc.port_manager, PortManager) + + port = node_svc.port_manager.reserve_port() + assert type(port) == int # noqa: E721 + + try: + with PostgresNode(name="node", port=port, os_ops=node_svc.os_ops, port_manager=None) as node: + # retest + assert isinstance(node_svc.os_ops, OsOperations) + assert node_svc.port_manager is not None + assert isinstance(node_svc.port_manager, PortManager) + + assert node.port_manager is None + assert node.os_ops is node_svc.os_ops + + # one more time + assert node.port_manager is None + assert node.os_ops is node_svc.os_ops + finally: + node_svc.port_manager.release_port(port) + @staticmethod def helper__get_node( node_svc: PostgresNodeService,

pFad - (p)hone/(F)rame/(a)nonymizer/(d)eclutterfier! Saves Data!