[DOP-21482] - add test with expired access_token

Author: maxim-lixakov

syncmaster/backend/providers/auth/keycloak_provider.py

@@ -81,6 +81,8 @@ async def get_current_user(self, access_token: str, *args, **kwargs) -> Any:
             self.redirect_to_auth(request.url.path)
 
         try:
+            # if user is disabled or blocked in Keycloak after the token is issued, he will
+            # remain authorized until the token expires (not more than 15 minutes in MTS SSO)
             token_info = self.keycloak_openid.decode_token(token=access_token)
         except Exception as e:
             log.info("Access token is invalid or expired: %s", e)

tests/test_unit/test_auth/auth_fixtures/__init__.py

@@ -1,6 +1,7 @@
 from tests.test_unit.test_auth.auth_fixtures.keycloak_fixture import (
     create_session_cookie,
     mock_keycloak_realm,
+    mock_keycloak_token_refresh,
     mock_keycloak_well_known,
     rsa_keys,
 )

tests/test_unit/test_auth/auth_fixtures/keycloak_fixture.py

@@ -120,3 +120,39 @@ def mock_keycloak_realm(settings, rsa_keys):
         status=200,
         content_type="application/json",
     )
+
+
+@pytest.fixture
+def mock_keycloak_token_refresh(settings, rsa_keys):
+    server_url = settings.auth.server_url
+    realm_name = settings.auth.client_id
+    token_url = f"{server_url}/realms/{realm_name}/protocol/openid-connect/token"
+
+    # generate new access and refresh tokens
+    expires_in = int(time.time()) + 1000
+    private_pem = rsa_keys["private_pem"]
+    payload = {
+        "sub": "mock_user_id",
+        "preferred_username": "mock_username",
+        "email": "[email protected]",
+        "given_name": "Mock",
+        "middle_name": "User",
+        "family_name": "Name",
+        "exp": expires_in,
+    }
+
+    new_access_token = jwt.encode(payload, private_pem, algorithm="RS256")
+    new_refresh_token = "mock_new_refresh_token"
+
+    responses.add(
+        responses.POST,
+        token_url,
+        json={
+            "access_token": new_access_token,
+            "refresh_token": new_refresh_token,
+            "token_type": "bearer",
+            "expires_in": expires_in,
+        },
+        status=200,
+        content_type="application/json",
+    )

tests/test_unit/test_auth/test_auth_keycloak.py

@@ -1,3 +1,5 @@
+import logging
+
 import pytest
 import responses
 from httpx import AsyncClient
@@ -68,6 +70,50 @@ async def test_get_keycloak_user_authorized(
     }
 
 
+@responses.activate
+@pytest.mark.parametrize(
+    "settings",
+    [
+        {
+            "auth": {
+                "provider": KEYCLOAK_PROVIDER,
+            },
+        },
+    ],
+    indirect=True,
+)
+async def test_get_keycloak_user_expired_access_token(
+    caplog,
+    client: AsyncClient,
+    simple_user: MockUser,
+    settings: Settings,
+    create_session_cookie,
+    mock_keycloak_well_known,
+    mock_keycloak_realm,
+    mock_keycloak_token_refresh,
+):
+    session_cookie = create_session_cookie(simple_user, expire_in_msec=-100000000)  # expired access token
+    headers = {
+        "Cookie": f"session={session_cookie}",
+    }
+
+    with caplog.at_level(logging.DEBUG):
+        response = await client.get(
+            f"/v1/users/{simple_user.id}",
+            headers=headers,
+        )
+
+    assert "Access token is invalid or expired" in caplog.text
+    assert "Access token refreshed and decoded successfully" in caplog.text
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "id": simple_user.id,
+        "is_superuser": simple_user.is_superuser,
+        "username": simple_user.username,
+    }
+
+
 @responses.activate
 @pytest.mark.parametrize(
     "settings",