|
| 1 | +import django.test |
| 2 | +import django.urls |
| 3 | +import rest_framework.status |
| 4 | +import rest_framework.test |
| 5 | +import rest_framework_simplejwt.token_blacklist.models as tb_models |
| 6 | + |
| 7 | +import user.models |
| 8 | + |
| 9 | + |
| 10 | +class JWTTests(rest_framework.test.APITestCase): |
| 11 | + def setUp(self): |
| 12 | + self.signup_url = django.urls.reverse('api-user:sign-up') |
| 13 | + self.signin_url = django.urls.reverse('api-user:sign-in') |
| 14 | + self.protected_url = django.urls.reverse('api-core:protected') |
| 15 | + self.refresh_url = django.urls.reverse('api-user:token_refresh') |
| 16 | + user.models.User.objects.create_user( |
| 17 | + name='John', |
| 18 | + surname='Doe', |
| 19 | + |
| 20 | + password='SuperStrongPassword2000!', |
| 21 | + other={'age': 25, 'country': 'us'}, |
| 22 | + ) |
| 23 | + self.user_data = { |
| 24 | + |
| 25 | + 'password': 'SuperStrongPassword2000!', |
| 26 | + } |
| 27 | + |
| 28 | + super(JWTTests, self).setUp() |
| 29 | + |
| 30 | + def tearDown(self): |
| 31 | + user.models.User.objects.all().delete() |
| 32 | + |
| 33 | + super(JWTTests, self).tearDown() |
| 34 | + |
| 35 | + def test_access_protected_view_with_valid_token(self): |
| 36 | + response = self.client.post( |
| 37 | + self.signin_url, |
| 38 | + self.user_data, |
| 39 | + format='json', |
| 40 | + ) |
| 41 | + |
| 42 | + token = response.data['access'] |
| 43 | + |
| 44 | + self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + token) |
| 45 | + response = self.client.get(self.protected_url) |
| 46 | + self.assertEqual(response.status_code, 200) |
| 47 | + self.assertEqual(response.data['status'], 'request was permitted') |
| 48 | + |
| 49 | + def test_registration_token_invalid_after_login(self): |
| 50 | + data = { |
| 51 | + |
| 52 | + 'password': 'StrongPass123!cd', |
| 53 | + 'name': 'John', |
| 54 | + 'surname': 'Doe', |
| 55 | + 'other': {'age': 22, 'country': 'us'}, |
| 56 | + } |
| 57 | + response = self.client.post( |
| 58 | + self.signup_url, |
| 59 | + data, |
| 60 | + format='json', |
| 61 | + ) |
| 62 | + reg_access_token = response.data['access'] |
| 63 | + |
| 64 | + self.client.credentials( |
| 65 | + HTTP_AUTHORIZATION=f'Bearer {reg_access_token}', |
| 66 | + ) |
| 67 | + response = self.client.get(self.protected_url) |
| 68 | + self.assertEqual(response.status_code, 200) |
| 69 | + |
| 70 | + login_data = {'email': data['email'], 'password': data['password']} |
| 71 | + response = self.client.post( |
| 72 | + self.signin_url, |
| 73 | + login_data, |
| 74 | + format='json', |
| 75 | + ) |
| 76 | + login_access_token = response.data['access'] |
| 77 | + |
| 78 | + self.client.credentials( |
| 79 | + HTTP_AUTHORIZATION=f'Bearer {reg_access_token}', |
| 80 | + ) |
| 81 | + response = self.client.get(self.protected_url) |
| 82 | + self.assertEqual(response.status_code, 401) |
| 83 | + |
| 84 | + self.client.credentials( |
| 85 | + HTTP_AUTHORIZATION=f'Bearer {login_access_token}', |
| 86 | + ) |
| 87 | + response = self.client.get(self.protected_url) |
| 88 | + self.assertEqual(response.status_code, 200) |
| 89 | + |
| 90 | + def test_refresh_token_invalidation_after_new_login(self): |
| 91 | + first_login_response = self.client.post( |
| 92 | + self.signin_url, |
| 93 | + self.user_data, |
| 94 | + format='json', |
| 95 | + ) |
| 96 | + |
| 97 | + refresh_token_v1 = first_login_response.data['refresh'] |
| 98 | + |
| 99 | + second_login_response = self.client.post( |
| 100 | + self.signin_url, |
| 101 | + self.user_data, |
| 102 | + format='json', |
| 103 | + ) |
| 104 | + refresh_token_v2 = second_login_response.data['refresh'] |
| 105 | + |
| 106 | + refresh_response_v1 = self.client.post( |
| 107 | + self.refresh_url, |
| 108 | + {'refresh': refresh_token_v1}, |
| 109 | + format='json', |
| 110 | + ) |
| 111 | + self.assertEqual( |
| 112 | + refresh_response_v1.status_code, |
| 113 | + rest_framework.status.HTTP_401_UNAUTHORIZED, |
| 114 | + ) |
| 115 | + self.assertEqual(refresh_response_v1.data['code'], 'token_not_valid') |
| 116 | + self.assertEqual( |
| 117 | + str(refresh_response_v1.data['detail']), |
| 118 | + 'Token is blacklisted', |
| 119 | + ) |
| 120 | + |
| 121 | + refresh_response_v2 = self.client.post( |
| 122 | + self.refresh_url, |
| 123 | + {'refresh': refresh_token_v2}, |
| 124 | + format='json', |
| 125 | + ) |
| 126 | + self.assertEqual( |
| 127 | + refresh_response_v2.status_code, |
| 128 | + rest_framework.status.HTTP_200_OK, |
| 129 | + ) |
| 130 | + self.assertIn('access', refresh_response_v2.data) |
| 131 | + |
| 132 | + self.client.credentials( |
| 133 | + HTTP_AUTHORIZATION='Bearer ' + first_login_response.data['access'], |
| 134 | + ) |
| 135 | + protected_response = self.client.get(self.protected_url) |
| 136 | + self.assertEqual( |
| 137 | + protected_response.status_code, |
| 138 | + rest_framework.status.HTTP_401_UNAUTHORIZED, |
| 139 | + ) |
| 140 | + |
| 141 | + def test_blacklist_storage(self): |
| 142 | + |
| 143 | + self.client.post(self.signin_url, self.user_data, format='json') |
| 144 | + |
| 145 | + self.client.post(self.signin_url, self.user_data, format='json') |
| 146 | + |
| 147 | + self.assertEqual( |
| 148 | + (tb_models.BlacklistedToken.objects.count()), |
| 149 | + 1, |
| 150 | + ) |
| 151 | + self.assertEqual( |
| 152 | + (tb_models.OutstandingToken.objects.count()), |
| 153 | + 2, |
| 154 | + ) |
0 commit comments