diff --git a/src/backend/core/factories.py b/src/backend/core/factories.py index 7fe300e8..deab843b 100644 --- a/src/backend/core/factories.py +++ b/src/backend/core/factories.py @@ -65,3 +65,48 @@ class RoomFactory(ResourceFactory): name = factory.Faker("catch_phrase") slug = factory.LazyAttribute(lambda o: slugify(o.name)) + + +class RecordingFactory(factory.django.DjangoModelFactory): + """Create fake recording for testing.""" + + class Meta: + model = models.Recording + + room = factory.SubFactory(RoomFactory) + status = models.RecordingStatusChoices.INITIATED + worker_id = None + + @factory.post_generation + def users(self, create, extracted, **kwargs): + """Add users to recording from a given list of users with or without roles.""" + if create and extracted: + for item in extracted: + if isinstance(item, models.User): + UserRecordingAccessFactory(recording=self, user=item) + else: + UserRecordingAccessFactory( + recording=self, user=item[0], role=item[1] + ) + + +class UserRecordingAccessFactory(factory.django.DjangoModelFactory): + """Create fake recording user accesses for testing.""" + + class Meta: + model = models.RecordingAccess + + recording = factory.SubFactory(RecordingFactory) + user = factory.SubFactory(UserFactory) + role = factory.fuzzy.FuzzyChoice(models.RoleChoices.values) + + +class TeamRecordingAccessFactory(factory.django.DjangoModelFactory): + """Create fake recording team accesses for testing.""" + + class Meta: + model = models.RecordingAccess + + recording = factory.SubFactory(RecordingFactory) + team = factory.Sequence(lambda n: f"team{n}") + role = factory.fuzzy.FuzzyChoice(models.RoleChoices.values) diff --git a/src/backend/core/migrations/0005_recording_recordingaccess_and_more.py b/src/backend/core/migrations/0005_recording_recordingaccess_and_more.py new file mode 100644 index 00000000..cdc61e62 --- /dev/null +++ b/src/backend/core/migrations/0005_recording_recordingaccess_and_more.py @@ -0,0 +1,67 @@ +# Generated by Django 5.1.1 on 2024-11-06 14:31 + +import django.db.models.deletion +import uuid +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0004_alter_user_language'), + ] + + operations = [ + migrations.CreateModel( + name='Recording', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')), + ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')), + ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')), + ('status', models.CharField(choices=[('initiated', 'Initiated'), ('active', 'Active'), ('stopped', 'Stopped'), ('saved', 'Saved'), ('aborted', 'Aborted'), ('failed_to_start', 'Failed to Start'), ('failed_to_stop', 'Failed to Stop')], default='initiated', max_length=20)), + ('worker_id', models.CharField(blank=True, help_text='Enter an identifier for the worker recording.This ID is retained even when the worker stops, allowing for easy tracking.', max_length=255, null=True, verbose_name='Worker ID')), + ('room', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='recordings', to='core.room', verbose_name='Room')), + ], + options={ + 'verbose_name': 'Recording', + 'verbose_name_plural': 'Recordings', + 'db_table': 'meet_recording', + 'ordering': ('-created_at',), + }, + ), + migrations.CreateModel( + name='RecordingAccess', + fields=[ + ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')), + ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')), + ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')), + ('team', models.CharField(blank=True, max_length=100)), + ('role', models.CharField(choices=[('member', 'Member'), ('administrator', 'Administrator'), ('owner', 'Owner')], default='member', max_length=20)), + ('recording', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='accesses', to='core.recording')), + ('user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)), + ], + options={ + 'verbose_name': 'Recording/user relation', + 'verbose_name_plural': 'Recording/user relations', + 'db_table': 'meet_recording_access', + 'ordering': ('-created_at',), + }, + ), + migrations.AddConstraint( + model_name='recording', + constraint=models.UniqueConstraint(condition=models.Q(('status__in', ['active', 'initiated'])), fields=('room',), name='unique_initiated_or_active_recording_per_room'), + ), + migrations.AddConstraint( + model_name='recordingaccess', + constraint=models.UniqueConstraint(condition=models.Q(('user__isnull', False)), fields=('user', 'recording'), name='unique_recording_user', violation_error_message='This user is already in this recording.'), + ), + migrations.AddConstraint( + model_name='recordingaccess', + constraint=models.UniqueConstraint(condition=models.Q(('team__gt', '')), fields=('team', 'recording'), name='unique_recording_team', violation_error_message='This team is already in this recording.'), + ), + migrations.AddConstraint( + model_name='recordingaccess', + constraint=models.CheckConstraint(condition=models.Q(models.Q(('team', ''), ('user__isnull', False)), models.Q(('team__gt', ''), ('user__isnull', True)), _connector='OR'), name='check_recording_access_either_user_or_team', violation_error_message='Either user or team must be set, not both.'), + ), + ] diff --git a/src/backend/core/models.py b/src/backend/core/models.py index a0be2de6..b6271770 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -4,6 +4,7 @@ Declare and configure the models for the Meet core application import uuid from logging import getLogger +from typing import List from django.conf import settings from django.contrib.auth import models as auth_models @@ -38,6 +39,39 @@ class RoleChoices(models.TextChoices): return role == cls.OWNER +class RecordingStatusChoices(models.TextChoices): + """Enumeration of possible states for a recording operation.""" + + INITIATED = "initiated", _("Initiated") + ACTIVE = "active", _("Active") + STOPPED = "stopped", _("Stopped") + SAVED = "saved", _("Saved") + ABORTED = "aborted", _("Aborted") + FAILED_TO_START = "failed_to_start", _("Failed to Start") + FAILED_TO_STOP = "failed_to_stop", _("Failed to Stop") + + @classmethod + def is_final(cls, status): + """Determine if the recording status represents a final state. + + A final status indicates the recording flow has completed, either + successfully or unsuccessfully. + """ + + return status in { + cls.STOPPED, + cls.SAVED, + cls.ABORTED, + cls.FAILED_TO_START, + cls.FAILED_TO_STOP, + } + + @classmethod + def is_unsuccessful(cls, status): + """Determine if the recording status represents an unsuccessful state.""" + return status in {cls.ABORTED, cls.FAILED_TO_START, cls.FAILED_TO_STOP} + + class BaseModel(models.Model): """ Serves as an abstract base model for other models, ensuring that records are validated @@ -171,6 +205,34 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin): return f"***@{self.email.split('@')[1]}" +def get_resource_roles(resource: models.Model, user: User) -> List[str]: + """ + Get all roles assigned to a user for a specific resource, including team-based roles. + + Args: + resource: The resource to check permissions for + user: The user to get roles for + + Returns: + List of role strings assigned to the user + """ + if not user.is_authenticated: + return [] + + # Use pre-annotated roles if available from viewset optimization + if hasattr(resource, "user_roles"): + return resource.user_roles or [] + + try: + return list( + resource.accesses.filter_user(user) + .values_list("role", flat=True) + .distinct() + ) + except (IndexError, models.ObjectDoesNotExist): + return [] + + class Resource(BaseModel): """Model to define access control""" @@ -325,3 +387,206 @@ class Room(Resource): else: raise ValidationError({"name": f'Room name "{self.name:s}" is reserved.'}) super().clean_fields(exclude=exclude) + + +class BaseAccessManager(models.Manager): + """Base manager for handling resource access control.""" + + def filter_user(self, user): + """Filter accesses for a given user, including both direct and team-based access.""" + return self.filter(models.Q(user=user) | models.Q(team__in=user.get_teams())) + + +class BaseAccess(BaseModel): + """Base model for accesses to handle resources.""" + + user = models.ForeignKey( + User, + on_delete=models.CASCADE, + null=True, + blank=True, + ) + team = models.CharField(max_length=100, blank=True) + role = models.CharField( + max_length=20, choices=RoleChoices.choices, default=RoleChoices.MEMBER + ) + + objects = BaseAccessManager() + + class Meta: + abstract = True + + def _get_abilities(self, resource, user): + """ + Compute and return abilities for a given user taking into account + the current state of the object. + """ + + roles = get_resource_roles(resource, user) + + is_owner = RoleChoices.OWNER in roles + has_privileges = is_owner or RoleChoices.ADMIN in roles + + # Default values for unprivileged users + set_role_to = set() + can_delete = False + + # Special handling when modifying an owner's access + if self.role == RoleChoices.OWNER: + # Prevent orphaning the resource + can_delete = ( + is_owner + and resource.accesses.filter(role=RoleChoices.OWNER).count() > 1 + ) + if can_delete: + set_role_to = {RoleChoices.ADMIN, RoleChoices.OWNER, RoleChoices.MEMBER} + elif has_privileges: + can_delete = True + set_role_to = {RoleChoices.ADMIN, RoleChoices.MEMBER} + if is_owner: + set_role_to.add(RoleChoices.OWNER) + + # Remove the current role as we don't want to propose it as an option + set_role_to.discard(self.role) + + return { + "destroy": can_delete, + "update": bool(set_role_to), + "partial_update": bool(set_role_to), + "retrieve": bool(roles), + "set_role_to": sorted(r.value for r in set_role_to), + } + + +class Recording(BaseModel): + """Model for recordings that take place in a room""" + + room = models.ForeignKey( + Room, + on_delete=models.CASCADE, + related_name="recordings", + verbose_name=_("Room"), + ) + status = models.CharField( + max_length=20, + choices=RecordingStatusChoices.choices, + default=RecordingStatusChoices.INITIATED, + ) + worker_id = models.CharField( + max_length=255, + null=True, + blank=True, + verbose_name=_("Worker ID"), + help_text=_( + "Enter an identifier for the worker recording." + "This ID is retained even when the worker stops, allowing for easy tracking." + ), + ) + + class Meta: + db_table = "meet_recording" + ordering = ("-created_at",) + verbose_name = _("Recording") + verbose_name_plural = _("Recordings") + constraints = [ + models.UniqueConstraint( + fields=["room"], + condition=models.Q( + status__in=[ + RecordingStatusChoices.ACTIVE, + RecordingStatusChoices.INITIATED, + ] + ), + name="unique_initiated_or_active_recording_per_room", + ) + ] + + def __str__(self): + return f"Recording {self.id} ({self.status})" + + def get_abilities(self, user): + """Compute and return abilities for a given user on the recording.""" + + roles = set(get_resource_roles(self, user)) + + is_owner_or_admin = bool( + roles.intersection({RoleChoices.OWNER, RoleChoices.ADMIN}) + ) + + is_final_status = RecordingStatusChoices.is_final(self.status) + + return { + "destroy": is_owner_or_admin and is_final_status, + "partial_update": False, + "retrieve": is_owner_or_admin, + "stop": is_owner_or_admin and not is_final_status, + "update": False, + } + + def is_savable(self) -> bool: + """Determine if the recording can be saved based on its current status.""" + + is_unsuccessful = RecordingStatusChoices.is_unsuccessful(self.status) + is_already_saved = self.status == RecordingStatusChoices.SAVED + + return not is_unsuccessful and not is_already_saved + + +class RecordingAccess(BaseAccess): + """Relation model to give access to a recording for a user or a team with a role. + + Recording Status Flow: + 1. INITIATED: Initial state when recording is requested + 2. ACTIVE: Recording is currently in progress + 3. STOPPED: Recording has been stopped by user/system + 4. SAVED: Recording has been successfully processed and stored + + Error States: + - FAILED_TO_START: Worker failed to initialize recording + - FAILED_TO_STOP: Worker failed during stop operation + - ABORTED: Recording was terminated before completion + + Warning: Worker failures may lead to database inconsistency between the actual + recording state and its status in the database. + """ + + recording = models.ForeignKey( + Recording, + on_delete=models.CASCADE, + related_name="accesses", + ) + + class Meta: + db_table = "meet_recording_access" + ordering = ("-created_at",) + verbose_name = _("Recording/user relation") + verbose_name_plural = _("Recording/user relations") + constraints = [ + models.UniqueConstraint( + fields=["user", "recording"], + condition=models.Q(user__isnull=False), # Exclude null users + name="unique_recording_user", + violation_error_message=_("This user is already in this recording."), + ), + models.UniqueConstraint( + fields=["team", "recording"], + condition=models.Q(team__gt=""), # Exclude empty string teams + name="unique_recording_team", + violation_error_message=_("This team is already in this recording."), + ), + models.CheckConstraint( + condition=models.Q(user__isnull=False, team="") + | models.Q(user__isnull=True, team__gt=""), + name="check_recording_access_either_user_or_team", + violation_error_message=_("Either user or team must be set, not both."), + ), + ] + + def __str__(self): + return f"{self.user!s} is {self.role:s} in {self.recording!s}" + + def get_abilities(self, user): + """ + Compute and return abilities for a given user on the recording access. + """ + return self._get_abilities(self.recording, user) diff --git a/src/backend/core/tests/test_models_recording.py b/src/backend/core/tests/test_models_recording.py new file mode 100644 index 00000000..8b71bd08 --- /dev/null +++ b/src/backend/core/tests/test_models_recording.py @@ -0,0 +1,212 @@ +""" +Unit tests for the Recording model +""" + +from django.core.exceptions import ValidationError + +import pytest + +from core.factories import ( + RecordingFactory, + RoomFactory, + UserFactory, + UserRecordingAccessFactory, +) +from core.models import Recording, RecordingStatusChoices + +pytestmark = pytest.mark.django_db + + +def test_models_recording_str(): + """The str representation should be the recording ID.""" + recording = RecordingFactory() + assert str(recording) == f"Recording {recording.id} (initiated)" + + +def test_models_recording_ordering(): + """Recordings should be returned ordered by created_at in descending order.""" + RecordingFactory.create_batch(3) + recordings = Recording.objects.all() + assert recordings[0].created_at >= recordings[1].created_at + assert recordings[1].created_at >= recordings[2].created_at + + +def test_models_recording_room_relationship(): + """It should maintain proper relationship with room.""" + room = RoomFactory() + recording = RecordingFactory(room=room) + assert recording.room == room + assert recording in room.recordings.all() + + +def test_models_recording_default_status(): + """A new recording should have INITIATED status by default.""" + recording = RecordingFactory() + assert recording.status == RecordingStatusChoices.INITIATED + + +def test_models_recording_unique_initiated_or_active_per_room(): + """Only one initiated or active recording should be allowed per room.""" + room = RoomFactory() + RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE) + + with pytest.raises(ValidationError): + RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE) + + with pytest.raises(ValidationError): + RecordingFactory(room=room, status=RecordingStatusChoices.INITIATED) + + +def test_models_recording_multiple_finished_allowed(): + """Multiple finished recordings should be allowed in the same room.""" + room = RoomFactory() + RecordingFactory(room=room, status=RecordingStatusChoices.SAVED) + RecordingFactory(room=room, status=RecordingStatusChoices.SAVED) + assert room.recordings.count() == 2 + + +# Test get_abilities method + + +@pytest.mark.parametrize( + "role", + ["owner", "administrator"], +) +def test_models_recording_get_abilities_privileges_active(role): + """Test abilities for active recording and privileged user.""" + + user = UserFactory() + access = UserRecordingAccessFactory(role=role, user=user) + access.recording.status = RecordingStatusChoices.ACTIVE + abilities = access.recording.get_abilities(user) + + assert abilities == { + "destroy": False, # Not final status + "partial_update": False, + "retrieve": True, # Privileged users can always retrieve + "stop": True, # Not final status, so can stop + "update": False, + } + + +def test_models_recording_get_abilities_member_active(): + """Test abilities for a user who is unprivileged.""" + + user = UserFactory() + access = UserRecordingAccessFactory(role="member", user=user) + access.recording.status = RecordingStatusChoices.ACTIVE + abilities = access.recording.get_abilities(user) + + assert abilities == { + "destroy": False, + "partial_update": False, + "retrieve": False, + "stop": False, + "update": False, + } + + +@pytest.mark.parametrize( + "role", + ["owner", "administrator"], +) +def test_models_recording_get_abilities_privileges_final(role): + """Test abilities for active recording and privileged user.""" + + user = UserFactory() + access = UserRecordingAccessFactory(role=role, user=user) + access.recording.status = RecordingStatusChoices.SAVED + abilities = access.recording.get_abilities(user) + + assert abilities == { + "destroy": True, + "partial_update": False, + "retrieve": True, # Privileged users can always retrieve + "stop": False, # In final status, so can not stop + "update": False, + } + + +def test_models_recording_get_abilities_member_final(): + """Test abilities for a user who is unprivileged.""" + + user = UserFactory() + access = UserRecordingAccessFactory(role="member", user=user) + access.recording.status = RecordingStatusChoices.SAVED + abilities = access.recording.get_abilities(user) + + assert abilities == { + "destroy": False, + "partial_update": False, + "retrieve": False, + "stop": False, + "update": False, + } + + +# Test is_savable method + + +def test_models_recording_is_savable_normal(): + """Test is_savable for normal recording status.""" + recording = RecordingFactory(status=RecordingStatusChoices.ACTIVE) + assert recording.is_savable() is True + + +@pytest.mark.parametrize( + "status", + [ + RecordingStatusChoices.FAILED_TO_STOP, + RecordingStatusChoices.FAILED_TO_START, + RecordingStatusChoices.ABORTED, + ], +) +def test_models_recording_is_savable_error(status): + """Test is_savable for error status.""" + recording = RecordingFactory(status=status) + assert recording.is_savable() is False + + +def test_models_recording_is_savable_already_saved(): + """Test is_savable for already saved recording.""" + recording = RecordingFactory(status=RecordingStatusChoices.SAVED) + assert recording.is_savable() is False + + +# Test few corner cases + + +def test_models_recording_worker_id_optional(): + """Worker ID should be optional.""" + recording = RecordingFactory(worker_id=None) + assert recording.worker_id is None + + recording = RecordingFactory(worker_id="worker-123") + assert recording.worker_id == "worker-123" + + +def test_models_recording_invalid_status(): + """Test that setting an invalid status raises an error.""" + recording = RecordingFactory() + recording.status = "INVALID_STATUS" + with pytest.raises(ValidationError): + recording.save() + + +def test_models_recording_room_deletion(): + """Test that deleting a room cascades to its recordings.""" + room = RoomFactory() + recording = RecordingFactory(room=room) + room.delete() + assert not Recording.objects.filter(id=recording.id).exists() + + +def test_models_recording_worker_id_very_long(): + """Test worker_id with maximum length.""" + long_id = "w" * 255 + recording = RecordingFactory(worker_id=long_id) + assert recording.worker_id == long_id + + too_long_id = "w" * 256 + with pytest.raises(ValidationError): + RecordingFactory(worker_id=too_long_id) diff --git a/src/backend/core/tests/test_models_recording_accesses.py b/src/backend/core/tests/test_models_recording_accesses.py new file mode 100644 index 00000000..2f7f278d --- /dev/null +++ b/src/backend/core/tests/test_models_recording_accesses.py @@ -0,0 +1,312 @@ +""" +Unit tests for the RecordingAccess model +""" + +from django.contrib.auth.models import AnonymousUser +from django.core.exceptions import ValidationError + +import pytest + +from core import factories + +pytestmark = pytest.mark.django_db + + +def test_models_recording_accesses_str(): + """ + The str representation should include user email, recording ID and role. + """ + user = factories.UserFactory(email="david.bowman@example.com") + access = factories.UserRecordingAccessFactory( + role="member", + user=user, + ) + assert ( + str(access) + == f"david.bowman@example.com is member in Recording {access.recording.id} (initiated)" + ) + + +def test_models_recording_accesses_unique_user(): + """Recording accesses should be unique for a given couple of user and recording.""" + access = factories.UserRecordingAccessFactory() + + with pytest.raises( + ValidationError, + match="This user is already in this recording.", + ): + factories.UserRecordingAccessFactory( + user=access.user, recording=access.recording + ) + + +def test_models_recording_accesses_several_empty_teams(): + """A recording can have several recording accesses with an empty team.""" + access = factories.UserRecordingAccessFactory() + factories.UserRecordingAccessFactory(recording=access.recording) + + +def test_models_recording_accesses_unique_team(): + """Recording accesses should be unique for a given couple of team and recording.""" + access = factories.TeamRecordingAccessFactory() + + with pytest.raises( + ValidationError, + match="This team is already in this recording.", + ): + factories.TeamRecordingAccessFactory( + team=access.team, recording=access.recording + ) + + +def test_models_recording_accesses_several_null_users(): + """A recording can have several recording accesses with a null user.""" + access = factories.TeamRecordingAccessFactory() + factories.TeamRecordingAccessFactory(recording=access.recording) + + +def test_models_recording_accesses_user_and_team_set(): + """User and team can't both be set on a recording access.""" + with pytest.raises( + ValidationError, + match="Either user or team must be set, not both.", + ): + factories.UserRecordingAccessFactory(team="my-team") + + +def test_models_recording_accesses_user_and_team_empty(): + """User and team can't both be empty on a recording access.""" + with pytest.raises( + ValidationError, + match="Either user or team must be set, not both.", + ): + factories.UserRecordingAccessFactory(user=None) + + +# get_abilities + + +def test_models_recording_access_get_abilities_anonymous(): + """Check abilities returned for an anonymous user.""" + access = factories.UserRecordingAccessFactory() + abilities = access.get_abilities(AnonymousUser()) + assert abilities == { + "destroy": False, + "retrieve": False, + "update": False, + "partial_update": False, + "set_role_to": [], + } + + +def test_models_recording_access_get_abilities_authenticated(): + """Check abilities returned for an authenticated user.""" + access = factories.UserRecordingAccessFactory() + user = factories.UserFactory() + abilities = access.get_abilities(user) + assert abilities == { + "destroy": False, + "retrieve": False, + "update": False, + "partial_update": False, + "set_role_to": [], + } + + +# - for owner + + +def test_models_recording_access_get_abilities_for_owner_of_self_allowed(): + """ + Check abilities of self access for the owner of a recording when + there is more than one owner left. + """ + access = factories.UserRecordingAccessFactory(role="owner") + factories.UserRecordingAccessFactory(recording=access.recording, role="owner") + abilities = access.get_abilities(access.user) + assert abilities == { + "destroy": True, + "retrieve": True, + "update": True, + "partial_update": True, + "set_role_to": ["administrator", "member"], + } + + +def test_models_recording_access_get_abilities_for_owner_of_self_last(): + """ + Check abilities of self access for the owner of a recording when there is only one owner left. + """ + access = factories.UserRecordingAccessFactory(role="owner") + abilities = access.get_abilities(access.user) + assert abilities == { + "destroy": False, + "retrieve": True, + "update": False, + "partial_update": False, + "set_role_to": [], + } + + +def test_models_recording_access_get_abilities_for_owner_of_owner(): + """Check abilities of owner access for the owner of a recording.""" + access = factories.UserRecordingAccessFactory(role="owner") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="owner" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": True, + "retrieve": True, + "update": True, + "partial_update": True, + "set_role_to": ["administrator", "member"], + } + + +def test_models_recording_access_get_abilities_for_owner_of_administrator(): + """Check abilities of administrator access for the owner of a recording.""" + access = factories.UserRecordingAccessFactory(role="administrator") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="owner" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": True, + "retrieve": True, + "update": True, + "partial_update": True, + "set_role_to": ["member", "owner"], + } + + +def test_models_recording_access_get_abilities_for_owner_of_member(): + """Check abilities of member access for the owner of a recording.""" + access = factories.UserRecordingAccessFactory(role="member") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="owner" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": True, + "retrieve": True, + "update": True, + "partial_update": True, + "set_role_to": ["administrator", "owner"], + } + + +# - for administrator + + +def test_models_recording_access_get_abilities_for_administrator_of_owner(): + """Check abilities of owner access for the administrator of a recording.""" + access = factories.UserRecordingAccessFactory(role="owner") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="administrator" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": False, + "retrieve": True, + "update": False, + "partial_update": False, + "set_role_to": [], + } + + +def test_models_recording_access_get_abilities_for_administrator_of_administrator(): + """Check abilities of administrator access for the administrator of a recording.""" + access = factories.UserRecordingAccessFactory(role="administrator") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="administrator" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": True, + "retrieve": True, + "update": True, + "partial_update": True, + "set_role_to": ["member"], + } + + +def test_models_recording_access_get_abilities_for_administrator_of_member(): + """Check abilities of member access for the administrator of a recording.""" + access = factories.UserRecordingAccessFactory(role="member") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="administrator" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": True, + "retrieve": True, + "update": True, + "partial_update": True, + "set_role_to": ["administrator"], + } + + +# - for member + + +def test_models_recording_access_get_abilities_for_member_of_owner(): + """Check abilities of owner access for the member of a recording.""" + access = factories.UserRecordingAccessFactory(role="owner") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="member" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": False, + "retrieve": True, + "update": False, + "partial_update": False, + "set_role_to": [], + } + + +def test_models_recording_access_get_abilities_for_member_of_administrator(): + """Check abilities of administrator access for the member of a recording.""" + access = factories.UserRecordingAccessFactory(role="administrator") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="member" + ).user + abilities = access.get_abilities(user) + assert abilities == { + "destroy": False, + "retrieve": True, + "update": False, + "partial_update": False, + "set_role_to": [], + } + + +def test_models_recording_access_get_abilities_for_member_of_member_user( + django_assert_num_queries, +): + """Check abilities of member access for the member of a recording.""" + access = factories.UserRecordingAccessFactory(role="member") + factories.UserRecordingAccessFactory(recording=access.recording) # another one + user = factories.UserRecordingAccessFactory( + recording=access.recording, role="member" + ).user + + with django_assert_num_queries(1): + abilities = access.get_abilities(user) + + assert abilities == { + "destroy": False, + "retrieve": True, + "update": False, + "partial_update": False, + "set_role_to": [], + }