43 lines
2.3 KiB
Python
43 lines
2.3 KiB
Python
"""Refresh locking must work on PostgreSQL despite nullable role relations."""
|
|
from django.db import connection
|
|
from django.test import skipUnlessDBFeature
|
|
from django.test.utils import CaptureQueriesContext
|
|
|
|
from apps.cuser.models import CuserRole
|
|
from apps.jwt_auth.models import JwtToken
|
|
from apps.jwt_auth.test_regressions import RegressionFixtures
|
|
|
|
|
|
class RefreshLockingTests(RegressionFixtures):
|
|
@skipUnlessDBFeature('has_select_for_update_of')
|
|
def test_refresh_locks_only_session_and_rejects_reuse(self):
|
|
tokens = self.use_role()
|
|
with CaptureQueriesContext(connection) as queries:
|
|
response = self.post('token/refresh/', {'refresh': tokens['refresh']})
|
|
self.assertEqual(response.status_code, 200)
|
|
locking = [query['sql'] for query in queries if 'FOR UPDATE' in query['sql'].upper()]
|
|
self.assertEqual(len(locking), 1)
|
|
# The actual endpoint SQL retains LEFT JOINs for nullable relations but
|
|
# targets only the JWT row in its PostgreSQL locking clause.
|
|
self.assertIn('LEFT OUTER JOIN', locking[0].upper())
|
|
expected = 'FOR UPDATE OF ' + connection.ops.quote_name(JwtToken._meta.db_table)
|
|
self.assertEqual(locking[0].split('FOR UPDATE', 1)[1].strip(), expected.split('FOR UPDATE', 1)[1].strip())
|
|
self.assertNotEqual(response.data['refresh'], tokens['refresh'])
|
|
self.assertEqual(JwtToken.objects.count(), 1)
|
|
self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
|
|
|
|
def test_refresh_for_role_without_workspace_is_unauthorized(self):
|
|
tokens = self.use_role()
|
|
CuserRole.objects.filter(pk=self.role.pk).update(sys_ws=None)
|
|
response = self.post('token/refresh/', {'refresh': tokens['refresh']})
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertIn('فضای کاری', str(response.data))
|
|
self.assertEqual(JwtToken.objects.get().refresh, tokens['refresh'])
|
|
|
|
def test_refresh_for_role_without_user_is_unauthorized(self):
|
|
tokens = self.use_role()
|
|
CuserRole.objects.filter(pk=self.role.pk).update(cuser=None)
|
|
response = self.post('token/refresh/', {'refresh': tokens['refresh']})
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(JwtToken.objects.get().refresh, tokens['refresh'])
|