File size: 2,394 Bytes
4f59651
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import unittest
from types import SimpleNamespace
from unittest.mock import patch

from fastapi import HTTPException
from jose import JWTError
from app.core.security import create_access_token, create_refresh_token, decode_token, decode_refresh_token
from app.api.v1.endpoints.auth import refresh_session
from app.schemas.auth import RefreshRequest


class SessionRefreshTests(unittest.TestCase):
    def setUp(self):
        self.user = SimpleNamespace(id=42, hashed_password="test-password-hash", is_active=True)
        self.token = create_refresh_token("42", self.user.hashed_password)

    def test_refresh_keeps_account(self):
        with patch("app.api.v1.endpoints.auth.get_user_by_id", return_value=self.user) as lookup:
            result = refresh_session(RefreshRequest(refresh_token=self.token), object())
        self.assertEqual(decode_token(result["access_token"])["sub"], "42")
        self.assertEqual(decode_refresh_token(result["refresh_token"])["sub"], "42")
        self.assertEqual(lookup.call_args.args[1], 42)

    def test_refresh_cannot_be_used_as_access(self):
        with self.assertRaises(JWTError):
            decode_token(self.token)

    def test_access_cannot_be_used_as_refresh(self):
        with self.assertRaises(JWTError):
            decode_refresh_token(create_access_token("42"))

    def test_expired_access_rejected(self):
        with self.assertRaises(JWTError):
            decode_token(create_access_token("42", expires_minutes=-1))

    def test_password_change_revokes_refresh(self):
        self.user.hashed_password = "changed"
        with patch("app.api.v1.endpoints.auth.get_user_by_id", return_value=self.user):
            with self.assertRaises(HTTPException) as raised:
                refresh_session(RefreshRequest(refresh_token=self.token), object())
        self.assertEqual(raised.exception.status_code, 401)

    def test_deleted_or_inactive_account_rejected(self):
        for user in (None, SimpleNamespace(id=42, hashed_password="test-password-hash", is_active=False)):
            with patch("app.api.v1.endpoints.auth.get_user_by_id", return_value=user):
                with self.assertRaises(HTTPException) as raised:
                    refresh_session(RefreshRequest(refresh_token=self.token), object())
            self.assertEqual(raised.exception.status_code, 401)


if __name__ == "__main__":
    unittest.main()