Keep open Query Tool connections when the Object Explorer connects their server. #10742

web/pgadmin/browser/server_groups/servers/__init__.py

@@ -1842,9 +1842,14 @@ def connect(self, gid, sid, is_qt=False, server=None):
 
         # Connect the Server
         manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
-        # Update the manager with the server details if not connected and
-        # the API call is not made from SQL Editor or View/Edit Data tool
-        if not manager.connection().connected() and not is_qt:
+        # Update the manager with the server details only if none of its
+        # connections is open and the API call is not made from SQL Editor
+        # or View/Edit Data tool. update() releases every connection of the
+        # server, including those of open Query Tool and View/Edit Data
+        # tabs: tabs restored at login connect before the Object Explorer
+        # does, and would be left polling a transaction without a
+        # connection ("Transaction status check failed.").
+        if not manager.has_live_connection() and not is_qt:
             manager.update(server)
         conn = manager.connection()
 

web/pgadmin/browser/server_groups/servers/tests/test_connect_keeps_tool_connections.py

@@ -0,0 +1,91 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""
+Connecting a server from the Object Explorer must not release the
+connections of open Query Tool and View/Edit Data tabs (#10742).
+
+Restored tabs connect to their database before the Object Explorer
+connects the server, so the maintenance database connection is still
+closed when the tree's connect request arrives. ServerNode.connect() used
+to call ServerManager.update() in that case, which releases every
+connection of the server: each restored tab then answered its status poll
+with "Transaction status check failed.".
+"""
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from flask import Flask
+
+from pgadmin.browser.server_groups.servers import ServerNode
+from pgadmin.utils.driver.psycopg3.server_manager import ServerManager
+from pgadmin.utils.exception import CryptKeyMissing
+from pgadmin.utils.route import BaseTestGenerator
+
+MODULE = 'pgadmin.browser.server_groups.servers'
+
+
+def connection(connected):
+    return MagicMock(**{'connected.return_value': connected})
+
+
+class TestConnectKeepsToolConnections(BaseTestGenerator):
+    """ServerNode.connect() re-reads the server details with
+    ServerManager.update() only while none of the server's connections is
+    open, and never for the Query Tool."""
+
+    scenarios = [
+        ('Nothing connected: update the manager', dict(
+            tool_connected=False, is_qt=False, expect_update=True)),
+        ('Restored tool tab connected, tree not yet: keep its connection',
+         dict(tool_connected=True, is_qt=False, expect_update=False)),
+        ('Request from the Query Tool: never update', dict(
+            tool_connected=False, is_qt=True, expect_update=False)),
+    ]
+
+    def setUp(self):
+        # Pure logic test with mocked connections: skip the server setup.
+        unittest.TestCase.setUp(self)
+
+    def runTest(self):
+        maintenance_conn = connection(False)
+
+        manager = ServerManager.__new__(ServerManager)
+        manager.connections = {
+            'DB:postgres': maintenance_conn,
+            'DB:appdb': connection(self.tool_connected),
+            'CONN:4584253': connection(self.tool_connected),
+        }
+        manager.connection = MagicMock(return_value=maintenance_conn)
+        manager.update = MagicMock()
+
+        driver = MagicMock()
+        driver.connection_manager.return_value = manager
+        server = SimpleNamespace(id=1, username='postgres', service=None,
+                                 shared=False, use_ssh_tunnel=0)
+
+        app = Flask(__name__)
+        with app.test_request_context(method='POST', data='{}'), \
+                patch(MODULE + '.get_driver', return_value=driver), \
+                patch(MODULE + '._is_non_owner', return_value=False), \
+                patch(MODULE + '.current_user', SimpleNamespace(id=1)), \
+                patch(MODULE + '.User'), \
+                patch(MODULE + '.get_crypt_key', return_value=(False, None)):
+            # A missing crypt key ends the request right after the
+            # update decision, before any connection is attempted.
+            with self.assertRaises(CryptKeyMissing):
+                ServerNode.connect(ServerNode.__new__(ServerNode), 1, 1,
+                                   is_qt=self.is_qt, server=server)
+
+        if self.expect_update:
+            manager.update.assert_called_once_with(server)
+        else:
+            manager.update.assert_not_called()

web/pgadmin/utils/driver/psycopg3/server_manager.py

@@ -223,6 +223,14 @@ def patch_version(self):
             return int(int(self.sversion / 100) / 100)
         raise InternalServerError(self._INFORMATION_MSG)
 
+    def has_live_connection(self):
+        """
+        Whether any connection of this server is open: the Object
+        Explorer's, or one of a Query Tool or View/Edit Data tab.
+        """
+        return any(conn.connected()
+                   for conn in list(self.connections.values()))
+
     def connection(self, **kwargs):
         database = kwargs.get('database', None)
         conn_id = kwargs.get('conn_id', None)

web/pgadmin/utils/driver/psycopg3/tests/test_manager_has_live_connection.py

@@ -0,0 +1,82 @@
+##########################################################################
+#
+# pgAdmin 4 - PostgreSQL Tools
+#
+# Copyright (C) 2013 - 2026, The pgAdmin Development Team
+# This software is released under the PostgreSQL Licence
+#
+##########################################################################
+
+"""
+Unit tests for ServerManager.has_live_connection, which decides whether
+connecting a server from the Object Explorer may re-read the server
+details with ServerManager.update() — that releases every connection,
+including those of open Query Tool and View/Edit Data tabs.
+
+Run as plain unittest TestCases without a Postgres server connection.
+"""
+
+import unittest
+from unittest.mock import MagicMock
+
+from pgadmin.utils.route import BaseTestGenerator
+from pgadmin.utils.driver.psycopg3.server_manager import ServerManager
+
+
+def make_manager(**connections):
+    """A ServerManager with the given connections (conn_id -> connected),
+    built without a Server row."""
+    manager = ServerManager.__new__(ServerManager)
+    manager.connections = {
+        conn_id: MagicMock(**{'connected.return_value': connected})
+        for conn_id, connected in connections.items()
+    }
+    return manager
+
+
+class _PureUnitTestSetupMixin:
+    """setUp here calls unittest.TestCase.setUp directly, skipping
+    BaseTestGenerator.setUp's Postgres connection."""
+
+    def setUp(self):
+        unittest.TestCase.setUp(self)
+
+
+class TestHasLiveConnectionWithoutConnections(
+        _PureUnitTestSetupMixin, BaseTestGenerator):
+    """A manager that never connected has no live connection."""
+
+    scenarios = [('default', dict())]
+
+    def runTest(self):
+        self.assertFalse(make_manager().has_live_connection())
+
+
+class TestHasLiveConnectionAllClosed(
+        _PureUnitTestSetupMixin, BaseTestGenerator):
+    """Connections that are all closed (e.g. after a disconnect) do not
+    count."""
+
+    scenarios = [('default', dict())]
+
+    def runTest(self):
+        manager = make_manager(**{'DB:postgres': False, 'CONN:42': False})
+        self.assertFalse(manager.has_live_connection())
+
+
+class TestHasLiveConnectionQueryToolOnly(
+        _PureUnitTestSetupMixin, BaseTestGenerator):
+    """A View/Edit Data tab restored at login opens its own connection
+    and the one to its database, while the Object Explorer's maintenance
+    database connection is still closed: the manager has a live
+    connection, so update() must not run."""
+
+    scenarios = [('default', dict())]
+
+    def runTest(self):
+        manager = make_manager(**{
+            'DB:postgres': False,       # Object Explorer, not yet connected
+            'DB:appdb': True,           # database of the tab
+            'CONN:4584253': True,       # the tab's own connection
+        })
+        self.assertTrue(manager.has_live_connection())