|
3 | 3 | import parameterized
|
4 | 4 | import rest_framework.status
|
5 | 5 | import rest_framework.test
|
| 6 | +import rest_framework_simplejwt.token_blacklist.models as tb_models |
6 | 7 |
|
7 | 8 | import user.models
|
8 | 9 |
|
@@ -386,3 +387,127 @@ def test_signin_success(self):
|
386 | 387 | response.status_code,
|
387 | 388 | rest_framework.status.HTTP_200_OK,
|
388 | 389 | )
|
| 390 | + |
| 391 | + |
| 392 | +class JWTTests(rest_framework.test.APITestCase): |
| 393 | + def setUp(self): |
| 394 | + |
| 395 | + self.signin_url = django.urls.reverse('api-user:sign-in') |
| 396 | + self.protected_url = django.urls.reverse('api-core:protected') |
| 397 | + self.refresh_url = django.urls.reverse('api-user:token_refresh') |
| 398 | + user.models.User.objects.create_user( |
| 399 | + name='John', |
| 400 | + surname='Doe', |
| 401 | + |
| 402 | + password='SuperStrongPassword2000!', |
| 403 | + other={'age': 25, 'country': 'us'}, |
| 404 | + ) |
| 405 | + self.user_data = { |
| 406 | + |
| 407 | + 'password': 'SuperStrongPassword2000!', |
| 408 | + } |
| 409 | + |
| 410 | + super(JWTTests, self).setUp() |
| 411 | + |
| 412 | + def tearDown(self): |
| 413 | + user.models.User.objects.all().delete() |
| 414 | + |
| 415 | + super(JWTTests, self).tearDown() |
| 416 | + |
| 417 | + def test_access_protected_view_with_valid_token(self): |
| 418 | + response = self.client.post( |
| 419 | + self.signin_url, |
| 420 | + self.user_data, |
| 421 | + format='json', |
| 422 | + ) |
| 423 | + |
| 424 | + token = response.data['access'] |
| 425 | + |
| 426 | + self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + token) |
| 427 | + response = self.client.get(self.protected_url) |
| 428 | + self.assertEqual(response.status_code, 200) |
| 429 | + self.assertEqual(response.data['status'], 'request was permitted') |
| 430 | + |
| 431 | + def test_refresh_token_invalidation_after_new_login(self): |
| 432 | + |
| 433 | + first_login_response = self.client.post( |
| 434 | + self.signin_url, |
| 435 | + self.user_data, |
| 436 | + format='json', |
| 437 | + ) |
| 438 | + refresh_token_v1 = first_login_response.data['refresh'] |
| 439 | + |
| 440 | + second_login_response = self.client.post( |
| 441 | + self.signin_url, |
| 442 | + self.user_data, |
| 443 | + format='json', |
| 444 | + ) |
| 445 | + refresh_token_v2 = second_login_response.data['refresh'] |
| 446 | + |
| 447 | + refresh_response_v1 = self.client.post( |
| 448 | + self.refresh_url, |
| 449 | + {'refresh': refresh_token_v1}, |
| 450 | + format='json', |
| 451 | + ) |
| 452 | + self.assertEqual( |
| 453 | + refresh_response_v1.status_code, |
| 454 | + rest_framework.status.HTTP_401_UNAUTHORIZED, |
| 455 | + ) |
| 456 | + self.assertEqual(refresh_response_v1.data['code'], 'token_not_valid') |
| 457 | + self.assertEqual( |
| 458 | + str(refresh_response_v1.data['detail']), |
| 459 | + 'Token is blacklisted', |
| 460 | + ) |
| 461 | + |
| 462 | + refresh_response_v2 = self.client.post( |
| 463 | + self.refresh_url, |
| 464 | + {'refresh': refresh_token_v2}, |
| 465 | + format='json', |
| 466 | + ) |
| 467 | + self.assertEqual( |
| 468 | + refresh_response_v2.status_code, |
| 469 | + rest_framework.status.HTTP_200_OK, |
| 470 | + ) |
| 471 | + self.assertIn('access', refresh_response_v2.data) |
| 472 | + |
| 473 | + self.client.credentials( |
| 474 | + HTTP_AUTHORIZATION='Bearer ' + first_login_response.data['access'], |
| 475 | + ) |
| 476 | + protected_response = self.client.get(self.protected_url) |
| 477 | + self.assertEqual( |
| 478 | + protected_response.status_code, |
| 479 | + rest_framework.status.HTTP_401_UNAUTHORIZED, |
| 480 | + ) |
| 481 | + |
| 482 | + def test_blacklist_storage(self): |
| 483 | + |
| 484 | + self.client.post(self.signin_url, self.user_data, format='json') |
| 485 | + |
| 486 | + self.client.post(self.signin_url, self.user_data, format='json') |
| 487 | + |
| 488 | + self.assertEqual( |
| 489 | + (tb_models.BlacklistedToken.objects.count()), |
| 490 | + 1, |
| 491 | + ) |
| 492 | + self.assertEqual( |
| 493 | + (tb_models.OutstandingToken.objects.count()), |
| 494 | + 2, |
| 495 | + ) |
| 496 | + |
| 497 | + def test_token_version_increment(self): |
| 498 | + response1 = self.client.post( |
| 499 | + self.signin_url, |
| 500 | + self.user_data, |
| 501 | + format='json', |
| 502 | + ) |
| 503 | + self.assertEqual(response1.data['token_version'], 1) |
| 504 | + |
| 505 | + response2 = self.client.post( |
| 506 | + self.signin_url, |
| 507 | + self.user_data, |
| 508 | + format='json', |
| 509 | + ) |
| 510 | + self.assertEqual(response2.data['token_version'], 2) |
| 511 | + |
| 512 | + user_ = user.models.User.objects.get(email=self.user_data['email']) |
| 513 | + self.assertEqual(user_.token_version, 2) |
0 commit comments