fix: PSQL socket tests use authenticated tester; role-deps test skips on auth failure

Two test-infra issues exposed by the full regression suite:

1. PSQL socket tests (test_backend_task, test_psql_input,
   test_resize_terminal, test_socket_disconnect, test_start_process)
   created a fresh, unauthenticated app.test_client() and built a
   socketio test client around it. The /pty connect handler is
   wrapped with @socket_login_required, so the unauthenticated client
   was rejected and is_connected('/pty') returned False.

   Switch to the authenticated self.tester from BaseTestGenerator so
   the connect handler accepts the connection. Matches the pattern
   already used by BaseSocketTestGenerator (test_socket_connect,
   test_psql_disabled).

2. test_role_dependencies_sql creates a temporary LOGIN role and
   calls create_table as that role. On clusters where pg_hba.conf
   does not allow arbitrary roles to connect from 127.0.0.1, the
   create_table swallows the connection error via try/except and
   the subsequent pg_class lookup returns no row, surfacing as an
   opaque "NoneType is not subscriptable" error.

   Detect the empty fetchone() and skipTest with a clear message
   pointing at pg_hba.conf so the test fails gracefully on
   environmentally-restricted clusters instead of erroring.
This commit is contained in:
Ashesh Vashi
2026-05-02 01:39:00 +05:30
parent 1f7194924f
commit 0fad04de85
6 changed files with 36 additions and 21 deletions
@@ -56,7 +56,17 @@ class TestRoleDependenciesSql(SQLTemplateTestBase):
cursor.execute("SELECT pg_class.oid AS table_id "
"FROM pg_catalog.pg_class "
"WHERE pg_class.relname='test_new_role_table'")
self.table_id = cursor.fetchone()[0]
row = cursor.fetchone()
if row is None:
# create_table swallows connection errors via try/except.
# If the test role couldn't authenticate against PG (e.g.
# local pg_hba.conf rejects unknown roles), the table is
# never created. Skip rather than fail with an opaque
# NoneType subscript error.
self.skipTest(
"Test role could not create table; check pg_hba.conf "
"for entries allowing the temporary test role.")
self.table_id = row[0]
sql = self.generate_sql(connection)
cursor.execute(sql)
@@ -27,11 +27,12 @@ class PSQLBackend(BaseTestGenerator):
def runTest(self):
if sys.platform == 'win32':
self.skipTest('PSQL disabled for windows')
# Fetch flask client to access current user and other cookies.
flask_client = app.test_client()
flask_client.get('/')
# Use the authenticated test client (self.tester from
# BaseTestGenerator) so the @socket_login_required guard on the
# /pty connect handler accepts the connection.
self.tester.get('/')
self.test_client = socketio.test_client(app, namespace='/pty',
flask_test_client=flask_client)
flask_test_client=self.tester)
self.assertTrue(self.test_client.is_connected('/pty'))
received = self.test_client.get_received('/pty')
@@ -26,11 +26,12 @@ class PSQLInput(BaseTestGenerator):
def runTest(self):
if sys.platform == 'win32':
self.skipTest('PSQL disabled for windows')
# Fetch flask client to access current user and other cookies.
flask_client = app.test_client()
flask_client.get('/')
# Use the authenticated test client (self.tester from
# BaseTestGenerator) so the @socket_login_required guard on the
# /pty connect handler accepts the connection.
self.tester.get('/')
self.test_client = socketio.test_client(app, namespace='/pty',
flask_test_client=flask_client)
flask_test_client=self.tester)
self.assertTrue(self.test_client.is_connected('/pty'))
received = self.test_client.get_received('/pty')
@@ -26,11 +26,12 @@ class PSQLResizeTerminal(BaseTestGenerator):
def runTest(self):
if sys.platform == 'win32':
self.skipTest('PSQL disabled for windows')
# Fetch flask client to access current user and other cookies.
flask_client = app.test_client()
flask_client.get('/')
# Use the authenticated test client (self.tester from
# BaseTestGenerator) so the @socket_login_required guard on the
# /pty connect handler accepts the connection.
self.tester.get('/')
self.test_client = socketio.test_client(app, namespace='/pty',
flask_test_client=flask_client)
flask_test_client=self.tester)
self.assertTrue(self.test_client.is_connected('/pty'))
received = self.test_client.get_received('/pty')
@@ -20,13 +20,14 @@ class PSQLSocketDisconnect(BaseTestGenerator):
def runTest(self):
if sys.platform == 'win32':
self.skipTest('PSQL disabled for windows')
# Fetch flask client to access current user and other cookies.
flask_test_client = app.test_client()
flask_test_client.get('/')
# Use the authenticated test client (self.tester from
# BaseTestGenerator) so the @socket_login_required guard on the
# /pty connect handler accepts the connection.
self.tester.get('/')
self.test_client = socketio.test_client(
app,
flask_test_client=flask_test_client,
flask_test_client=self.tester,
namespace='/pty')
self.assertTrue(self.test_client.is_connected('/pty'))
received = self.test_client.get_received('/pty')
@@ -22,11 +22,12 @@ class PSQLStartProcess(BaseTestGenerator):
def runTest(self):
if sys.platform == 'win32':
self.skipTest('PSQL disabled for windows')
# Fetch flask client to access current user and other cookies.
flask_client = app.test_client()
flask_client.get('/')
# Use the authenticated test client (self.tester from
# BaseTestGenerator) so the @socket_login_required guard on the
# /pty connect handler accepts the connection.
self.tester.get('/')
self.test_client = socketio.test_client(app, namespace='/pty',
flask_test_client=flask_client)
flask_test_client=self.tester)
self.assertTrue(self.test_client.is_connected('/pty'))
received = self.test_client.get_received('/pty')