diff --git a/src/backend/core/tests/rooms/test_api_rooms_create.py b/src/backend/core/tests/rooms/test_api_rooms_create.py index 50c4976b2..745a638f0 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_create.py +++ b/src/backend/core/tests/rooms/test_api_rooms_create.py @@ -2,13 +2,10 @@ Test rooms API endpoints in the Meet core app: create. """ -from datetime import datetime, timedelta, timezone - # pylint: disable=redefined-outer-name,unused-argument from django.conf import settings from django.core.cache import cache -import jwt import pytest from rest_framework.test import APIClient @@ -16,8 +13,9 @@ from ...api.throttling import ( RoomCreationDailyUserRateThrottle, RoomCreationUserRateThrottle, ) -from ...factories import ApplicationFactory, RoomFactory, UserFactory -from ...models import ApplicationScope, Room, RoomAccessLevel +from ...factories import RoomFactory, UserFactory +from ...models import Room, RoomAccessLevel +from ..utils import generate_user_access_token pytestmark = pytest.mark.django_db @@ -463,29 +461,6 @@ def test_api_rooms_create_daily_throttle_does_not_limit_other_actions( assert response.status_code == 200 -def generate_user_access_token(user): - """Generate a valid user access JWT signed with the token secret.""" - now = datetime.now(timezone.utc) - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": settings.USER_ACCESS_TOKEN_ISSUER, - "aud": settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_token", - "client_id": application.client_id, - "scope": "user:access", - } - - return jwt.encode( - payload, - settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=settings.USER_ACCESS_TOKEN_ALG, - ) - - def test_api_rooms_create_authenticated_with_user_access_token(): """A user access token should create a room exactly like a session would.""" user = UserFactory() diff --git a/src/backend/core/tests/rooms/test_api_rooms_list.py b/src/backend/core/tests/rooms/test_api_rooms_list.py index 8e1503676..87af67f28 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_list.py +++ b/src/backend/core/tests/rooms/test_api_rooms_list.py @@ -2,18 +2,15 @@ Test rooms API endpoints in the Meet core app: list. """ -from datetime import datetime, timedelta, timezone from unittest import mock -from django.conf import settings as django_settings - -import jwt import pytest from rest_framework.pagination import PageNumberPagination from rest_framework.test import APIClient -from ...factories import ApplicationFactory, RoomFactory, UserFactory -from ...models import ApplicationScope, RoomAccessLevel +from ...factories import RoomFactory, UserFactory +from ...models import RoomAccessLevel +from ..utils import generate_user_access_token pytestmark = pytest.mark.django_db @@ -162,29 +159,6 @@ def test_api_rooms_list_pagination_page_size(): assert content["previous"] is None -def generate_user_access_token(user): - """Generate a valid user access JWT signed with the token secret.""" - now = datetime.now(timezone.utc) - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": django_settings.USER_ACCESS_TOKEN_ISSUER, - "aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_token", - "client_id": application.client_id, - "scope": "user:access", - } - - return jwt.encode( - payload, - django_settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=django_settings.USER_ACCESS_TOKEN_ALG, - ) - - def test_api_rooms_list_authenticated_with_user_access_token(): """A user access token should list rooms exactly like a session would.""" user = UserFactory() diff --git a/src/backend/core/tests/rooms/test_api_rooms_participants_management.py b/src/backend/core/tests/rooms/test_api_rooms_participants_management.py index d95691c15..83d4bfba5 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_participants_management.py +++ b/src/backend/core/tests/rooms/test_api_rooms_participants_management.py @@ -5,16 +5,13 @@ Test rooms API endpoints in the Meet core app: participants management. # pylint: disable=redefined-outer-name,unused-argument,protected-access,no-name-in-module,too-many-lines import random -from datetime import datetime, timedelta, timezone from unittest import mock from uuid import uuid4 -from django.conf import settings as django_settings from django.contrib.auth.models import AnonymousUser from django.core.exceptions import SuspiciousOperation from django.urls import reverse -import jwt import pytest from livekit.api import TwirpError, UpdateParticipantRequest from livekit.protocol.models import ParticipantInfo @@ -23,13 +20,12 @@ from rest_framework.test import APIClient from core import utils from core.factories import ( - ApplicationFactory, RoomFactory, UserFactory, UserResourceAccessFactory, ) -from core.models import ApplicationScope from core.services.lobby import LobbyParticipant, LobbyParticipantStatus, LobbyService +from core.tests.utils import generate_user_access_token pytestmark = pytest.mark.django_db @@ -1040,29 +1036,6 @@ def test_remove_participant_not_found(mock_livekit_client): mock_livekit_client.aclose.assert_called_once() -def generate_user_access_token(user): - """Generate a valid user access JWT signed with the token secret.""" - now = datetime.now(timezone.utc) - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": django_settings.USER_ACCESS_TOKEN_ISSUER, - "aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_access", - "client_id": application.client_id, - "scope": "user:access", - } - - return jwt.encode( - payload, - django_settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=django_settings.USER_ACCESS_TOKEN_ALG, - ) - - def test_mute_participant_bearer_scheme_defers_to_next_authentication( mock_livekit_client, ): diff --git a/src/backend/core/tests/rooms/test_api_rooms_rename_toggle.py b/src/backend/core/tests/rooms/test_api_rooms_rename_toggle.py index 48aacdaa8..dfbb297f0 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_rename_toggle.py +++ b/src/backend/core/tests/rooms/test_api_rooms_rename_toggle.py @@ -4,15 +4,12 @@ Test rooms API endpoints: toggle hand and rename participant. # pylint: disable=redefined-outer-name,unused-argument,protected-access -from datetime import datetime, timedelta, timezone from unittest import mock from uuid import uuid4 -from django.conf import settings as django_settings from django.contrib.auth.models import AnonymousUser from django.urls import reverse -import jwt import pytest from freezegun import freeze_time from livekit.api import TwirpError @@ -21,12 +18,11 @@ from rest_framework.test import APIClient from core import utils from core.factories import ( - ApplicationFactory, RoomFactory, UserFactory, UserResourceAccessFactory, ) -from core.models import ApplicationScope +from core.tests.utils import generate_user_access_token pytestmark = pytest.mark.django_db @@ -702,25 +698,7 @@ def test_rename_participant_not_found(mock_livekit_client, room, token): @pytest.fixture def user_access_token(user): """Generate a valid user access JWT, sent with the "X-LiveKit-Token" scheme.""" - now = datetime.now(timezone.utc) - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": django_settings.USER_ACCESS_TOKEN_ISSUER, - "aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_access", - "client_id": application.client_id, - "scope": "user:access", - } - - return jwt.encode( - payload, - django_settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=django_settings.USER_ACCESS_TOKEN_ALG, - ) + return generate_user_access_token(user) def test_toggle_hand_bearer_scheme_defers_to_next_authentication( diff --git a/src/backend/core/tests/rooms/test_api_rooms_retrieve.py b/src/backend/core/tests/rooms/test_api_rooms_retrieve.py index 7eb2953cc..f40693303 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_retrieve.py +++ b/src/backend/core/tests/rooms/test_api_rooms_retrieve.py @@ -3,25 +3,22 @@ Test rooms API endpoints in the Meet core app: retrieve. """ import random -from datetime import datetime, timedelta, timezone from unittest import mock -from django.conf import settings as django_settings from django.contrib.auth.models import AnonymousUser from django.test.utils import override_settings from django.utils import timezone as dj_timezone -import jwt import pytest from rest_framework.test import APIClient from ...factories import ( - ApplicationFactory, RoomFactory, UserFactory, UserResourceAccessFactory, ) -from ...models import ApplicationScope, RoleChoices, RoomAccessLevel +from ...models import RoleChoices, RoomAccessLevel +from ..utils import generate_user_access_token pytestmark = pytest.mark.django_db @@ -535,29 +532,6 @@ def test_api_rooms_retrieve_last_started_at_not_exposed(role, access_level): assert "last_started_at" not in response.json() -def generate_user_access_token(user): - """Generate a valid user access JWT signed with the token secret.""" - now = datetime.now(timezone.utc) - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": django_settings.USER_ACCESS_TOKEN_ISSUER, - "aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_token", - "client_id": application.client_id, - "scope": "user:access", - } - - return jwt.encode( - payload, - django_settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=django_settings.USER_ACCESS_TOKEN_ALG, - ) - - def test_api_rooms_retrieve_authenticated_with_user_access_token(): """A user access token should retrieve a room exactly like a session would.""" user = UserFactory() diff --git a/src/backend/core/tests/rooms/test_api_rooms_subtitle.py b/src/backend/core/tests/rooms/test_api_rooms_subtitle.py index 52bd1aa3c..5e5e6a155 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_subtitle.py +++ b/src/backend/core/tests/rooms/test_api_rooms_subtitle.py @@ -4,18 +4,16 @@ Test rooms API endpoints in the Meet core app: start subtitle. # pylint: disable=W0621 import uuid -from datetime import datetime, timedelta, timezone from unittest import mock from django.conf import settings -import jwt import pytest from livekit.api import AccessToken, TwirpError, VideoGrants from rest_framework.test import APIClient -from core.factories import ApplicationFactory, RoomFactory, UserFactory -from core.models import ApplicationScope +from core.factories import RoomFactory, UserFactory +from core.tests.utils import generate_user_access_token pytestmark = pytest.mark.django_db @@ -236,26 +234,7 @@ def test_start_subtitle_wrong_signature(settings, mock_livekit_token): @pytest.fixture def user_access_token(): """Generate a valid user access JWT, sent with the "Bearer" scheme.""" - user = UserFactory() - now = datetime.now(timezone.utc) - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": settings.USER_ACCESS_TOKEN_ISSUER, - "aud": settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_access", - "client_id": application.client_id, - "scope": "user:access", - } - - return jwt.encode( - payload, - settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=settings.USER_ACCESS_TOKEN_ALG, - ) + return generate_user_access_token(UserFactory()) def test_start_subtitle_bearer_scheme_defers_to_next_authentication( diff --git a/src/backend/core/tests/rooms/test_api_rooms_update.py b/src/backend/core/tests/rooms/test_api_rooms_update.py index 8c9f63c59..afaee8419 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_update.py +++ b/src/backend/core/tests/rooms/test_api_rooms_update.py @@ -3,23 +3,22 @@ Test rooms API endpoints in the Meet core app: update. """ import random -from datetime import datetime, timedelta, timezone +from datetime import timedelta from unittest.mock import patch -from django.conf import settings as django_settings -from django.utils import timezone as dj_timezone +from django.utils import timezone -import jwt import pytest from rest_framework.test import APIClient -from ...factories import ApplicationFactory, RoomFactory, UserFactory -from ...models import ApplicationScope, RoomAccessLevel +from ...factories import RoomFactory, UserFactory +from ...models import RoomAccessLevel from ...services.room_management import ( RoomManagement, RoomManagementException, RoomNotFoundException, ) +from ..utils import generate_user_access_token pytestmark = pytest.mark.django_db @@ -239,7 +238,7 @@ def test_api_rooms_update_last_started_at_ignored(method): not be able to keep a room alive by postponing its last start date. """ user = UserFactory() - last_started_at = dj_timezone.now() - timedelta(days=30) + last_started_at = timezone.now() - timedelta(days=30) room = RoomFactory( name="Old name", last_started_at=last_started_at, @@ -250,7 +249,7 @@ def test_api_rooms_update_last_started_at_ignored(method): response = getattr(client, method)( f"/api/v1.0/rooms/{room.id!s}/", - {"name": "New name", "last_started_at": dj_timezone.now().isoformat()}, + {"name": "New name", "last_started_at": timezone.now().isoformat()}, format="json", ) @@ -450,29 +449,6 @@ def test_api_rooms_update_livekit_sync_failure(mock_update_metadata, exception): ) -def generate_user_access_token(user): - """Generate a valid user access JWT signed with the token secret.""" - now = datetime.now(timezone.utc) - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": django_settings.USER_ACCESS_TOKEN_ISSUER, - "aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_token", - "client_id": application.client_id, - "scope": "user:access", - } - - return jwt.encode( - payload, - django_settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=django_settings.USER_ACCESS_TOKEN_ALG, - ) - - @pytest.mark.parametrize("privileged_role", ["administrator", "owner"]) def test_api_rooms_update_authenticated_with_user_access_token(privileged_role): """Role-based permissions apply unchanged with a user access token.""" diff --git a/src/backend/core/tests/test_api_user_access_token_authentication.py b/src/backend/core/tests/test_api_user_access_token_authentication.py index 9eb61f227..57cfef4dc 100644 --- a/src/backend/core/tests/test_api_user_access_token_authentication.py +++ b/src/backend/core/tests/test_api_user_access_token_authentication.py @@ -17,37 +17,11 @@ from rest_framework.test import APIClient from core.factories import ApplicationFactory, RoomFactory, UserFactory from core.models import ApplicationScope, RoleChoices +from core.tests.utils import generate_user_access_token pytestmark = pytest.mark.django_db -def generate_user_access_token(user, application=None, **overrides): - """Generate a valid user access JWT signed with the token secret.""" - now = datetime.now(timezone.utc) - - if application is None: - application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) - - payload = { - "iss": django_settings.USER_ACCESS_TOKEN_ISSUER, - "aud": django_settings.USER_ACCESS_TOKEN_AUDIENCE, - "iat": now, - "exp": now + timedelta(seconds=django_settings.USER_ACCESS_TOKEN_TTL), - "user_id": str(user.id), - "token_type": "user_token", - "client_id": application.client_id, - "scope": "user:access", - } - payload.update(overrides) - payload = {key: value for key, value in payload.items() if value is not None} - - return jwt.encode( - payload, - django_settings.USER_ACCESS_TOKEN_SECRET_KEY, - algorithm=django_settings.USER_ACCESS_TOKEN_ALG, - ) - - def test_user_access_token_users_me(): """A user access token should authenticate the user on /users/me/.""" user = UserFactory() diff --git a/src/backend/core/tests/utils.py b/src/backend/core/tests/utils.py new file mode 100644 index 000000000..fdfe453ed --- /dev/null +++ b/src/backend/core/tests/utils.py @@ -0,0 +1,41 @@ +"""Shared helpers for tests in the Meet core application""" + +from datetime import datetime, timedelta, timezone + +from django.conf import settings + +import jwt + +from core.factories import ApplicationFactory +from core.models import ApplicationScope + + +def generate_user_access_token(user, application=None, **overrides): + """Generate a valid user access JWT signed with the token secret. + + Claims can be overridden through keyword arguments; passing None for a + claim removes it from the payload. + """ + now = datetime.now(timezone.utc) + + if application is None: + application = ApplicationFactory(scopes=[ApplicationScope.USERS_SESSION]) + + payload = { + "iss": settings.USER_ACCESS_TOKEN_ISSUER, + "aud": settings.USER_ACCESS_TOKEN_AUDIENCE, + "iat": now, + "exp": now + timedelta(seconds=settings.USER_ACCESS_TOKEN_TTL), + "user_id": str(user.id), + "token_type": settings.USER_ACCESS_TOKEN_TYPE_CLAIM, + "client_id": application.client_id, + "scope": "user:access", + } + payload.update(overrides) + payload = {key: value for key, value in payload.items() if value is not None} + + return jwt.encode( + payload, + settings.USER_ACCESS_TOKEN_SECRET_KEY, + algorithm=settings.USER_ACCESS_TOKEN_ALG, + )