mirror of
https://github.com/suitenumerique/meet.git
synced 2026-10-09 23:04:31 +00:00
♻️(backend) share the user access token test helper
generate_user_access_token was copy-pasted in seven test modules (user access token authentication, and rooms create, list, retrieve, update, participants management, rename/toggle and subtitle). Extract it into core.tests.utils and reuse it everywhere. The shared version keeps the optional application and claim overrides from the authentication tests, and reads the token type claim from USER_ACCESS_TOKEN_TYPE_CLAIM instead of hardcoding it.
This commit is contained in:
committed by
aleb_the_flash
parent
191e6d49d8
commit
7fa9c3fa4f
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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,
|
||||
):
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
Reference in New Issue
Block a user