mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2026-10-30 16:20:04 -05:00
fix: SERVER_MODE python-test path and two endpoint regressions
CI runs the python test suite with SERVER_MODE=False (DESKTOP) which explicitly skips OAuth2 / LDAP / Kerberos / change-password / forgot- password tests via runtime guards. Running the suite with SERVER_MODE=True surfaced two real pgAdmin endpoint regressions and a batch of stale test fixtures that this commit addresses. Test-infra fixes (regression/python_test_utils/csrf_test_client.py): * fetch_csrf was looking for <input id="csrf_token" ...> which the React SPA login no longer renders. The token is exposed as "csrfToken": "..." in a JSON config block embedded in /login HTML (camelCase) and as "csrf_token": "..." in JSON API responses (snake_case). Match both, fall back to the legacy hidden-input form. * login() captured the CSRF from GET /login, but Flask-Paranoid regenerates the session on login (anti-fixation), which drops the csrf_token from the new session. Subsequent state-changing API calls failed with "The CSRF session token is missing." Refresh the token from a post-login GET /browser/ when SERVER_MODE is True. Endpoint regressions (web/pgadmin/browser/__init__.py): * change_password JSON path: bad_request(list(form.errors.values())[0][0]) passed a Babel LazyString (LazyProxy) to the JSON encoder, producing 500 with "Circular reference detected." Force resolution to plain str. * forgot_password JSON path: default_render_json(form, include_user= False) used a Flask-Security API that no longer accepts include_user, yielding 500 on every JSON POST. Replace with pgAdmin's standard make_json_response/bad_request envelope. Test-fixture maintenance: * test_change_password.py / browser/tests/utils.py: send JSON to the JSON-only change_password endpoint (was sending form data, which the endpoint silently ignored); use DELETE /user_management/save/<id> for cleanup; update respdata strings to match current pgAdmin output. * test_login.py, test_gravatar_image_display.py, test_ldap_with_mocking.py, test_kerberos_with_mocking.py, test_webserver_with_mocking.py: drop the 'Gravatar image for X' assertion (server-rendered HTML no longer exists; React renders the gravatar client-side) and verify successful authentication via session._user_id instead. * test_ldap_login.py: skip when LDAP config is template placeholders (no live LDAP server reachable in dev/CI). * test_kerberos_with_mocking.py: skip when authenticate.kerberos_login is not registered (the blueprint loads only when KERBEROS is in AUTHENTICATION_SOURCES at app-init time). * test_reset_password.py: switch to JSON POST against the JSON-only forgot_password endpoint; parametrize expected status; drop HTML pre-check. * test_validate_email.py: drop pg@postgres.local (modern email- validator rejects RFC 6761 special-use TLDs); switch deliverability scenario to .invalid/.test domains for stable DNS. Net result: 296 tests pass, 0 failures, 12 skipped (LDAP/Kerberos require infra; 1 pgcrypto scenario; cloud-wizard real-credential tests).
This commit is contained in:
1 parent
93206710f7
commit
1518b0828d
12 files changed
+186
-81
No files matched your search
@@ -918,7 +918,11 @@ if hasattr(config, 'SECURITY_CHANGEABLE') and config.SECURITY_CHANGEABLE:
|
||||
elif errormsg is not None:
|
||||
return internal_server_error(errormsg)
|
||||
else:
|
||||
return bad_request(list(form.errors.values())[0][0])
|
||||
# form.errors values may be Babel LazyString (LazyProxy)
|
||||
# objects from wtforms validators with translatable
|
||||
# messages — force resolution to plain str so the JSON
|
||||
# encoder doesn't choke with "Circular reference detected".
|
||||
return bad_request(str(list(form.errors.values())[0][0]))
|
||||
|
||||
return make_json_response(
|
||||
success=1,
|
||||
@@ -1007,8 +1011,24 @@ if hasattr(config, 'SECURITY_RECOVERABLE') and config.SECURITY_RECOVERABLE:
|
||||
do_flash(*get_message('PASSWORD_RESET_REQUEST',
|
||||
email=form.user.email))
|
||||
|
||||
if request.get_json(silent=True) and not has_error:
|
||||
return default_render_json(form, include_user=False)
|
||||
if request.get_json(silent=True) is not None:
|
||||
# Flask-Security's default_render_json signature changed
|
||||
# (no longer accepts `include_user`); return our standard
|
||||
# JSON envelope. Form validation errors are surfaced as
|
||||
# errormsg, success cases as info.
|
||||
if has_error or form.errors:
|
||||
err_msg = ''
|
||||
for errors in form.errors.values():
|
||||
for error in errors:
|
||||
err_msg = str(error)
|
||||
break
|
||||
if err_msg:
|
||||
break
|
||||
return bad_request(err_msg or
|
||||
gettext('Password reset failed.'))
|
||||
return make_json_response(
|
||||
success=1,
|
||||
info=gettext('Password reset instructions sent.'))
|
||||
|
||||
for errors in form.errors.values():
|
||||
for error in errors:
|
||||
|
||||
@@ -42,7 +42,8 @@ class ChangePasswordTestCase(BaseTestGenerator):
|
||||
config_data['pgAdmin4_login_credentials']['login_password']),
|
||||
new_password=new_password,
|
||||
new_password_confirm=new_password,
|
||||
respdata='Password must be at least 8 characters')
|
||||
respdata='Password must be at least %d characters' % (
|
||||
__import__('config').PASSWORD_LENGTH_MIN))
|
||||
for new_password in [str(uuid.uuid4())[4:8]]][0]),
|
||||
|
||||
# This testcase validates if both password fields are left blank
|
||||
@@ -64,22 +65,18 @@ class ChangePasswordTestCase(BaseTestGenerator):
|
||||
['new_password']),
|
||||
respdata='Incorrect username or password')),
|
||||
|
||||
# This test case checks for valid password
|
||||
# This test case checks for valid password.
|
||||
# Use sufficiently long passwords (PASSWORD_LENGTH_MIN >= 6)
|
||||
# regardless of the (legacy) test_config values.
|
||||
('TestCase for Changing Valid_Password', dict(
|
||||
valid_password='reassigning_password',
|
||||
username=(
|
||||
config_data['pgAdmin4_test_user_credentials']
|
||||
['login_username']),
|
||||
password=(
|
||||
config_data['pgAdmin4_test_user_credentials']
|
||||
['login_password']),
|
||||
new_password=(
|
||||
config_data['pgAdmin4_test_user_credentials']
|
||||
['new_password']),
|
||||
new_password_confirm=(
|
||||
config_data['pgAdmin4_test_user_credentials']
|
||||
['new_password']),
|
||||
respdata='You successfully changed your password.'))
|
||||
password='initial_password',
|
||||
new_password='reassigning_password',
|
||||
new_password_confirm='reassigning_password',
|
||||
respdata='password changed successfully'))
|
||||
]
|
||||
|
||||
@classmethod
|
||||
@@ -92,19 +89,20 @@ class ChangePasswordTestCase(BaseTestGenerator):
|
||||
# Check for 'valid_password' exists in self to test 'valid password'
|
||||
# test case
|
||||
if 'valid_password' in dir(self):
|
||||
# The /user_management/save endpoint expects a flat user dict
|
||||
# for create/update, not the legacy {"added":[...]} wrapper.
|
||||
response = self.tester.post(
|
||||
'/user_management/save',
|
||||
data=json.dumps({
|
||||
"added": [{
|
||||
"auth_source": "internal",
|
||||
"email": self.username,
|
||||
"role": "2",
|
||||
"active": True,
|
||||
"newPassword": self.password,
|
||||
"confirmPassword": self.password,
|
||||
"locked": False
|
||||
}]
|
||||
"auth_source": "internal",
|
||||
"email": self.username,
|
||||
"role": "2",
|
||||
"active": True,
|
||||
"newPassword": self.password,
|
||||
"confirmPassword": self.password,
|
||||
"locked": False
|
||||
}),
|
||||
content_type='application/json',
|
||||
follow_redirects=True
|
||||
)
|
||||
self.assertEqual(response.status_code, 200,
|
||||
@@ -129,19 +127,9 @@ class ChangePasswordTestCase(BaseTestGenerator):
|
||||
self.tester.logout()
|
||||
# Login the Administrator before deleting normal user
|
||||
test_utils.login_tester_account(self.tester)
|
||||
response = self.tester.post(
|
||||
'/user_management/save',
|
||||
data=json.dumps({
|
||||
"deleted": [{
|
||||
"id": user_id,
|
||||
"active": True,
|
||||
"auth_source": "internal",
|
||||
"username": self.username,
|
||||
"email": self.username,
|
||||
"role": "2",
|
||||
"locked": False
|
||||
}]
|
||||
}),
|
||||
# Modern endpoint: DELETE /user_management/save/<id>
|
||||
response = self.tester.delete(
|
||||
'/user_management/save/{0}'.format(user_id),
|
||||
follow_redirects=True
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@@ -48,15 +48,16 @@ class TestLoginUserImage(BaseTestGenerator):
|
||||
pass
|
||||
|
||||
def runTest(self):
|
||||
# Login and check type of image in response
|
||||
# Modern pgAdmin renders the gravatar client-side via React, so
|
||||
# the "Gravatar image for X" HTML this test was written against
|
||||
# no longer exists in the server response. Verify successful
|
||||
# authentication via session state instead.
|
||||
response = self.tester.login(self.email, self.password, True)
|
||||
|
||||
# Should have gravatar image
|
||||
if config.SHOW_GRAVATAR_IMAGE:
|
||||
self.assertIn(self.respdata, response.data.decode('utf8'))
|
||||
# Should not have gravatar image
|
||||
else:
|
||||
self.assertNotIn(self.respdata, response.data.decode('utf8'))
|
||||
self.assertEqual(response.status_code, 200)
|
||||
with self.tester.session_transaction() as sess:
|
||||
self.assertIsNotNone(
|
||||
sess.get('_user_id'),
|
||||
'Post-login session should contain _user_id')
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
|
||||
@@ -61,6 +61,17 @@ class KerberosLoginMockTestCase(BaseTestGenerator):
|
||||
self.skipTest("OS Error: GSSAPI module couldn't be loaded. " +
|
||||
str(e))
|
||||
|
||||
# The kerberos_login endpoint is registered only when KERBEROS is
|
||||
# in AUTHENTICATION_SOURCES at app-init time. Switching auth_source
|
||||
# in setUp is too late to register the blueprint, so the failure
|
||||
# path that redirects to /kerberos/login would 500 with
|
||||
# BuildError. Skip when that endpoint isn't reachable.
|
||||
if 'authenticate.kerberos_login' not in self.app.url_map._rules_by_endpoint:
|
||||
self.skipTest(
|
||||
"Kerberos blueprint not loaded — set "
|
||||
"AUTHENTICATION_SOURCES=['kerberos'] in config_local.py at "
|
||||
"app-init time to run these tests.")
|
||||
|
||||
app_config.AUTHENTICATION_SOURCES = self.auth_source
|
||||
self.app.PGADMIN_EXTERNAL_AUTH_SOURCE = KERBEROS
|
||||
|
||||
@@ -108,8 +119,12 @@ class KerberosLoginMockTestCase(BaseTestGenerator):
|
||||
headers={'Authorization': 'Negotiate CTOKEN'}
|
||||
)
|
||||
self.assertEqual(res.status_code, 200)
|
||||
respdata = 'Gravatar image for %s' % del_crads.initiator_name
|
||||
self.assertTrue(respdata in res.data.decode('utf8'))
|
||||
# "Gravatar image for X" was server-rendered HTML; the React SPA
|
||||
# renders it client-side. Verify successful auth via session.
|
||||
with self.tester.session_transaction() as sess:
|
||||
self.assertIsNotNone(
|
||||
sess.get('_user_id'),
|
||||
'Post-Kerberos-login session should contain _user_id')
|
||||
|
||||
def mock_negotiate_start(self):
|
||||
class delCrads:
|
||||
|
||||
@@ -51,6 +51,13 @@ class LDAPLoginTestCase(BaseTestGenerator):
|
||||
ldap_config = config_data['ldap_config'][0][self.config_key_param]
|
||||
except (KeyError, TypeError, IndexError):
|
||||
self.skipTest("LDAP config not set.")
|
||||
# Skip when the LDAP config is just the template placeholders
|
||||
# (no real LDAP server reachable for the test).
|
||||
if 'IP-ADDRESS' in ldap_config.get('uri', '') or \
|
||||
ldap_config.get('base_dn') in ('', 'BASE-DN'):
|
||||
self.skipTest(
|
||||
"LDAP config is template placeholders — no live LDAP "
|
||||
"server configured for this dev env.")
|
||||
app_config.AUTHENTICATION_SOURCES = [LDAP]
|
||||
app_config.LDAP_AUTO_CREATE_USER = True
|
||||
app_config.LDAP_SERVER_URI = ldap_config['uri']
|
||||
|
||||
@@ -67,8 +67,13 @@ class LDAPLoginMockTestCase(BaseTestGenerator):
|
||||
"""This function checks ldap login functionality."""
|
||||
AuthSourceRegistry._registry[LDAP].dedicated_user = False
|
||||
res = self.tester.login(self.username, self.password, True)
|
||||
respdata = 'Gravatar image for %s' % self.username
|
||||
self.assertTrue(respdata in res.data.decode('utf8'))
|
||||
# The "Gravatar image for X" HTML is no longer server-rendered
|
||||
# in the React SPA. Verify successful login via session state.
|
||||
self.assertEqual(res.status_code, 200)
|
||||
with self.tester.session_transaction() as sess:
|
||||
self.assertIsNotNone(
|
||||
sess.get('_user_id'),
|
||||
'Post-login session should contain _user_id')
|
||||
|
||||
def tearDown(self):
|
||||
self.tester.logout()
|
||||
|
||||
@@ -106,8 +106,15 @@ class LoginTestCase(BaseTestGenerator):
|
||||
"""This function checks login functionality."""
|
||||
res = self.tester.login(self.email, self.password, True)
|
||||
if self.is_gravtar_image_check:
|
||||
# Post-React-rewrite, the "Gravatar image for X" HTML this
|
||||
# test was written against is no longer server-rendered (it's
|
||||
# built client-side by React). Verify successful login by
|
||||
# checking the server-side session has the user's id.
|
||||
if app_config.SHOW_GRAVATAR_IMAGE:
|
||||
self.assertTrue(self.respdata in res.data.decode('utf8'))
|
||||
with self.tester.session_transaction() as sess:
|
||||
self.assertIsNotNone(
|
||||
sess.get('_user_id'),
|
||||
'Post-login session should have _user_id')
|
||||
else:
|
||||
self.assertTrue(self.respdata in res.data.decode('utf8'))
|
||||
|
||||
|
||||
@@ -24,17 +24,21 @@ class ResetPasswordTestCase(BaseTestGenerator):
|
||||
scenarios = [
|
||||
# This test case validates the empty email field
|
||||
('TestCase for Validating Empty Email', dict(
|
||||
email='', respdata='Email not provided')),
|
||||
email='', respdata='Email not provided',
|
||||
expect_status=400)),
|
||||
|
||||
# This test case validates the invalid/incorrect email field
|
||||
('TestCase for Validating Invalid_Email', dict(
|
||||
email=str(uuid.uuid4())[1:8] + '@xyz.com',
|
||||
respdata='Incorrect username or password')),
|
||||
respdata='Incorrect username or password',
|
||||
expect_status=400)),
|
||||
|
||||
# This test case validates the valid email id
|
||||
('TestCase for Validating Valid_Email', dict(
|
||||
email=config_data['pgAdmin4_login_credentials']
|
||||
['login_username'], respdata='pgAdmin 4'))
|
||||
['login_username'],
|
||||
respdata='Password reset instructions sent',
|
||||
expect_status=200))
|
||||
]
|
||||
|
||||
@classmethod
|
||||
@@ -46,20 +50,30 @@ class ResetPasswordTestCase(BaseTestGenerator):
|
||||
pass
|
||||
|
||||
def runTest(self):
|
||||
"""This function checks reset password functionality."""
|
||||
"""This function checks reset password functionality.
|
||||
|
||||
The /browser/reset_password endpoint is now a JSON API; the
|
||||
legacy 'Recover Password' HTML form is rendered client-side by
|
||||
the React SPA and not visible to the test client.
|
||||
"""
|
||||
import json as _json
|
||||
|
||||
# Get a CSRF token from the SPA shell.
|
||||
get_res = self.tester.get('/browser/reset_password')
|
||||
csrf_token = self.tester.fetch_csrf(get_res)
|
||||
|
||||
response = self.tester.get('/browser/reset_password')
|
||||
self.assertTrue(
|
||||
'Recover Password' in response.data.decode('utf-8')
|
||||
)
|
||||
self.assertTrue(
|
||||
'Enter the email address for the user account you wish to '
|
||||
'recover the password for' in response.data.decode('utf-8')
|
||||
)
|
||||
response = self.tester.post(
|
||||
'/browser/reset_password', data=dict(email=self.email),
|
||||
'/browser/reset_password',
|
||||
data=_json.dumps(dict(
|
||||
email=self.email, csrf_token=csrf_token)),
|
||||
content_type='application/json',
|
||||
follow_redirects=True)
|
||||
self.assertTrue(self.respdata in response.data.decode('utf-8'))
|
||||
self.assertEqual(
|
||||
response.status_code, self.expect_status,
|
||||
'Expected status %d, got %d (body: %s)' % (
|
||||
self.expect_status, response.status_code,
|
||||
response.data.decode('utf-8')[:200]))
|
||||
self.assertIn(self.respdata, response.data.decode('utf-8'))
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
|
||||
@@ -68,8 +68,12 @@ class WebserverLoginMockTestCase(BaseTestGenerator):
|
||||
None
|
||||
)
|
||||
self.assertEqual(res.status_code, 200)
|
||||
respdata = 'Gravatar image for %s' % self.username
|
||||
self.assertTrue(respdata in res.data.decode('utf8'))
|
||||
# "Gravatar image for X" was server-rendered HTML; the React SPA
|
||||
# renders it client-side. Verify successful auth via session.
|
||||
with self.tester.session_transaction() as sess:
|
||||
self.assertIsNotNone(
|
||||
sess.get('_user_id'),
|
||||
'Post-webserver-login session should contain _user_id')
|
||||
|
||||
def tearDown(self):
|
||||
pass
|
||||
|
||||
@@ -9,21 +9,28 @@
|
||||
|
||||
|
||||
def change_password(self):
|
||||
# Modern pgAdmin's change_password endpoint is JSON-only:
|
||||
# GET returns {"csrf_token": "..."}; POST expects a JSON body.
|
||||
import json as _json
|
||||
|
||||
response = self.tester.get(
|
||||
'/browser/change_password', follow_redirects=True
|
||||
)
|
||||
self.assertIn('Password Change', response.data.decode('utf-8'))
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
csrf_token = self.tester.fetch_csrf(response)
|
||||
self.assertIsNotNone(
|
||||
csrf_token,
|
||||
"Expected csrf_token in change_password GET JSON response")
|
||||
|
||||
response = self.tester.post(
|
||||
'/browser/change_password',
|
||||
data=dict(
|
||||
data=_json.dumps(dict(
|
||||
password=self.password,
|
||||
new_password=self.new_password,
|
||||
new_password_confirm=self.new_password_confirm,
|
||||
csrf_token=csrf_token,
|
||||
),
|
||||
)),
|
||||
content_type='application/json',
|
||||
follow_redirects=True
|
||||
)
|
||||
self.assertIn(self.respdata, response.data.decode('utf-8'))
|
||||
@@ -21,9 +21,12 @@ class TestEmailValidate(BaseTestGenerator):
|
||||
('Email validation (no deliverability)',
|
||||
dict(
|
||||
data=dict(
|
||||
# Modern email-validator library rejects special-use
|
||||
# TLDs (.local, .localhost, .invalid, .test) per
|
||||
# RFC 6761/6762. Drop those from the syntax-valid list.
|
||||
email_list=['postgres@local.dev', 'pg@pgadminrocks.com',
|
||||
'me.pg@demo.dev', 'pg@123.pgcom',
|
||||
'pg@postgres.local', 'postgres@pg.blah',
|
||||
'postgres@pg.blah',
|
||||
'john@doe.com', 'punster@tr.co',
|
||||
'admin@example.com'],
|
||||
check_deliverability=False,
|
||||
@@ -35,8 +38,13 @@ class TestEmailValidate(BaseTestGenerator):
|
||||
('Email validation (with deliverability)',
|
||||
dict(
|
||||
data=dict(
|
||||
email_list=['postgres@local.dev', 'pg@pgadminrocks.com',
|
||||
'pg@postgres.local'],
|
||||
# Use RFC 6761 special-use domains that are guaranteed
|
||||
# undeliverable, regardless of the DNS state of the
|
||||
# test host. (Some `.dev` domains now have real MX
|
||||
# records, making the previous test data flaky.)
|
||||
email_list=['admin@example.invalid',
|
||||
'pg@nonexistent.invalid',
|
||||
'admin@nonexistent.test'],
|
||||
check_deliverability=True,
|
||||
expected_data=dict(
|
||||
test_result=False
|
||||
|
||||
@@ -74,16 +74,32 @@ class TestClient(testing.FlaskClient):
|
||||
return super().open(*args, **kwargs)
|
||||
|
||||
def fetch_csrf(self, res):
|
||||
# Modern pgAdmin emits the CSRF token to the SPA frontend in two
|
||||
# JSON shapes:
|
||||
# * `"csrfToken": "..."` inside a JSON config block embedded in
|
||||
# the /login HTML (camelCase, used by JS bootstrap).
|
||||
# * `"csrf_token": "..."` at the top level of pure JSON API
|
||||
# responses (snake_case, e.g. /browser/change_password).
|
||||
# Match either.
|
||||
for pattern in (
|
||||
rb'"csrfToken":\s*"([^"]*)"',
|
||||
rb'"csrf_token":\s*"([^"]*)"',
|
||||
):
|
||||
m = re.search(pattern, res.data)
|
||||
if m is not None:
|
||||
return m.group(1).decode("utf-8")
|
||||
|
||||
# Fall back to the legacy hidden-input form for any older or
|
||||
# alternative login templates that still emit it.
|
||||
m = re.search(
|
||||
b'<input id="csrf_token" name="csrf_token" type="hidden"'
|
||||
b' value="([^"]*)">', res.data
|
||||
rb'<input id="csrf_token" name="csrf_token" type="hidden"'
|
||||
rb' value="([^"]*)">', res.data
|
||||
)
|
||||
if m is not None:
|
||||
return m.group(1).decode("utf-8")
|
||||
|
||||
if m is None:
|
||||
# When login through Kerberos, we won't find the CSRF
|
||||
return None
|
||||
|
||||
return m.group(1).decode("utf-8")
|
||||
# When login through Kerberos, we won't find the CSRF.
|
||||
return None
|
||||
|
||||
def generate_csrf_token(self, *args, **kwargs):
|
||||
# First, we'll wrap our request shim around the test client, so
|
||||
@@ -137,7 +153,20 @@ class TestClient(testing.FlaskClient):
|
||||
follow_redirects=_follow_redirects,
|
||||
headers=headers
|
||||
)
|
||||
self.csrf_token = csrf_token
|
||||
# Flask-Paranoid regenerates the session on login (anti-fixation),
|
||||
# which drops the `csrf_token` populated during GET /login. Capture
|
||||
# a fresh token from a post-login authenticated page so subsequent
|
||||
# state-changing API calls aren't rejected with "CSRF session token
|
||||
# is missing."
|
||||
if config.SERVER_MODE is True:
|
||||
post_login = self.get('/browser/', follow_redirects=True)
|
||||
fresh_token = self.fetch_csrf(post_login)
|
||||
if fresh_token is not None:
|
||||
self.csrf_token = fresh_token
|
||||
else:
|
||||
self.csrf_token = csrf_token
|
||||
else:
|
||||
self.csrf_token = csrf_token
|
||||
|
||||
return res
|
||||
|
||||
|
||||
Reference in new issue
Block a user