(project) add CRUD API endpoints for Rooms and ResourceAccess models

Introduce CRUD API endpoints for the Rooms and ResourceAccess models.
The code follows the Magnify logic, with the exception that token generation
has been removed and replaced by a TODO item with a mocked value.

Proper integration of LiveKit will be added in future commits.

With the removal of group logic, some complex query sets can be simplified.
Previously, we checked for both direct and indirect access to a room.
Indirect access meant a room was shared with a group, and the user was a
member of that group. I haven’t simplified those query set, as I preferred
isolate changes in dedicated commits.

Additionally, all previous tests are still passing, although tests related
to groups have been removed.
This commit is contained in:
lebaudantoine
2024-06-25 00:21:36 +02:00
parent 2e6feede31
commit c90a92d5c9
13 changed files with 2033 additions and 0 deletions
@@ -0,0 +1,71 @@
"""
Test rooms API endpoints in the impress core app: create.
"""
import pytest
from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory
from ...models import Room
pytestmark = pytest.mark.django_db
def test_api_rooms_create_anonymous():
"""Anonymous users should not be allowed to create rooms."""
client = APIClient()
response = client.post(
"/api/v1.0/rooms/",
{
"name": "my room",
},
)
assert response.status_code == 401
assert Room.objects.exists() is False
def test_api_rooms_create_authenticated():
"""
Authenticated users should be able to create rooms and should automatically be declared
as owner of the newly created room.
"""
user = UserFactory()
client = APIClient()
client.force_login(user)
response = client.post(
"/api/v1.0/rooms/",
{
"name": "my room",
},
)
assert response.status_code == 201
room = Room.objects.get()
assert room.name == "my room"
assert room.slug == "my-room"
assert room.accesses.filter(role="owner", user=user).exists() is True
def test_api_rooms_create_authenticated_existing_slug():
"""
A user trying to create a room with a name that translates to a slug that already exists
should receive a 400 error.
"""
RoomFactory(name="my room")
user = UserFactory()
client = APIClient()
client.force_login(user)
response = client.post(
"/api/v1.0/rooms/",
{
"name": "My Room!",
},
)
assert response.status_code == 400
assert response.json() == {"slug": ["Room with this Slug already exists."]}
@@ -0,0 +1,101 @@
"""
Test rooms API endpoints in the impress core app: delete.
"""
import pytest
from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory
from ...models import Room
pytestmark = pytest.mark.django_db
def test_api_rooms_delete_anonymous():
"""Anonymous users should not be allowed to destroy a room."""
room = RoomFactory()
client = APIClient()
response = client.delete(
f"/api/v1.0/rooms/{room.id!s}/",
)
assert response.status_code == 401
assert Room.objects.count() == 1
def test_api_rooms_delete_authenticated():
"""
Authenticated users should not be allowed to delete a room to which they are not
related.
"""
room = RoomFactory()
user = UserFactory()
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/rooms/{room.id!s}/",
)
assert response.status_code == 403
assert Room.objects.count() == 1
def test_api_rooms_delete_members():
"""
Authenticated users should not be allowed to delete a room for which they are
only a member.
"""
user = UserFactory()
room = RoomFactory(
users=[(user, "member")]
) # as user declared in the room but not administrator
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/rooms/{room.id}/",
)
assert response.status_code == 403
assert Room.objects.count() == 1
def test_api_rooms_delete_administrators():
"""
Authenticated users should not be allowed to delete a room for which they are
administrator.
"""
user = UserFactory()
room = RoomFactory(users=[(user, "administrator")])
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/rooms/{room.id}/",
)
assert response.status_code == 403
assert Room.objects.count() == 1
def test_api_rooms_delete_owners():
"""
Authenticated users should be able to delete a room for which they are directly
owner.
"""
user = UserFactory()
room = RoomFactory(users=[(user, "owner")])
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/rooms/{room.id}/",
)
assert response.status_code == 204
assert Room.objects.exists() is False
@@ -0,0 +1,116 @@
"""
Test rooms API endpoints in the impress core app: list.
"""
from unittest import mock
import pytest
from rest_framework.pagination import PageNumberPagination
from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory
pytestmark = pytest.mark.django_db
def test_api_rooms_list_anonymous():
"""Anonymous users should not be able to list rooms."""
RoomFactory(is_public=False)
RoomFactory(is_public=True)
client = APIClient()
response = client.get("/api/v1.0/rooms/")
assert response.status_code == 200
results = response.json()["results"]
assert len(results) == 0
def test_api_rooms_list_authenticated():
"""
Authenticated users listing rooms, should only see the rooms
to which they are related.
"""
user = UserFactory()
client = APIClient()
client.force_login(user)
other_user = UserFactory()
RoomFactory(is_public=False)
RoomFactory(is_public=True)
room_user_accesses = RoomFactory(is_public=False, users=[user])
RoomFactory(is_public=False, users=[other_user])
response = client.get(
"/api/v1.0/rooms/",
)
assert response.status_code == 200
results = response.json()["results"]
assert len(results) == 1
expected_ids = {
str(room_user_accesses.id),
}
results_id = {result["id"] for result in results}
assert expected_ids == results_id
@mock.patch.object(PageNumberPagination, "get_page_size", return_value=2)
def test_api_rooms_list_pagination(_mock_page_size):
"""Pagination should work as expected."""
user = UserFactory()
client = APIClient()
client.force_login(user)
rooms = RoomFactory.create_batch(3, users=[user])
room_ids = [str(room.id) for room in rooms]
response = client.get(
"/api/v1.0/rooms/",
)
assert response.status_code == 200
content = response.json()
assert content["count"] == 3
assert content["next"] == "http://testserver/api/v1.0/rooms/?page=2"
assert content["previous"] is None
assert len(content["results"]) == 2
for item in content["results"]:
room_ids.remove(item["id"])
# Get page 2
response = client.get(
"/api/v1.0/rooms/?page=2",
)
assert response.status_code == 200
content = response.json()
assert content["count"] == 3
assert content["next"] is None
assert content["previous"], "http://testserver/api/v1.0/rooms/"
assert len(content["results"]) == 1
room_ids.remove(content["results"][0]["id"])
assert room_ids == []
def test_api_rooms_list_authenticated_distinct():
"""A public room with several related users should only be listed once."""
user = UserFactory()
other_user = UserFactory()
client = APIClient()
client.force_login(user)
room = RoomFactory(is_public=True, users=[user, other_user])
response = client.get(
"/api/v1.0/rooms/",
)
assert response.status_code == 200
content = response.json()
assert len(content["results"]) == 1
assert content["results"][0]["id"] == str(room.id)
@@ -0,0 +1,344 @@
"""
Test rooms API endpoints in the impress core app: retrieve.
"""
import random
from django.test.utils import override_settings
import pytest
from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory, UserResourceAccessFactory
pytestmark = pytest.mark.django_db
def test_api_rooms_retrieve_anonymous_private_pk():
"""
Anonymous users should be allowed to retrieve a private room but should not be
given any token.
"""
room = RoomFactory(is_public=False)
client = APIClient()
response = client.get(f"/api/v1.0/rooms/{room.id!s}/")
assert response.status_code == 200
assert response.json() == {
"id": str(room.id),
"is_administrable": False,
"is_public": False,
"name": room.name,
"slug": room.slug,
}
def test_api_rooms_retrieve_anonymous_private_pk_no_dashes():
"""It should be possible to get a room by its id stripped of its dashes."""
room = RoomFactory(is_public=False)
id_no_dashes = str(room.id).replace("-", "")
client = APIClient()
response = client.get(f"/api/v1.0/rooms/{id_no_dashes:s}/")
assert response.status_code == 200
assert response.json() == {
"id": str(room.id),
"is_administrable": False,
"is_public": False,
"name": room.name,
"slug": room.slug,
}
def test_api_rooms_retrieve_anonymous_private_slug():
"""It should be possible to get a room by its slug."""
room = RoomFactory(is_public=False)
client = APIClient()
response = client.get(f"/api/v1.0/rooms/{room.slug!s}/")
assert response.status_code == 200
assert response.json() == {
"id": str(room.id),
"is_administrable": False,
"is_public": False,
"name": room.name,
"slug": room.slug,
}
def test_api_rooms_retrieve_anonymous_private_slug_not_normalized():
"""Getting a room by a slug that is not normalized should work."""
room = RoomFactory(name="Réunion", is_public=False)
client = APIClient()
response = client.get("/api/v1.0/rooms/Réunion/")
assert response.status_code == 200
assert response.json() == {
"id": str(room.id),
"is_administrable": False,
"is_public": False,
"name": room.name,
"slug": room.slug,
}
@override_settings(ALLOW_UNREGISTERED_ROOMS=True)
def test_api_rooms_retrieve_anonymous_unregistered_allowed():
"""
Retrieving an unregistered room should return a Livekit token
if unregistered rooms are allowed.
"""
client = APIClient()
response = client.get("/api/v1.0/rooms/unregistered-room/")
assert response.status_code == 200
assert response.json() == {
"id": None,
"livekit": {
"room": "unregistered-room",
"token": "foo",
},
}
# todo - assert generate_token has been called
@override_settings(ALLOW_UNREGISTERED_ROOMS=True)
def test_api_rooms_retrieve_anonymous_unregistered_allowed_not_normalized():
"""
Getting an unregistered room by a slug that is not normalized should work
and use the Livekit room on the url-safe name.
"""
client = APIClient()
response = client.get("/api/v1.0/rooms/Réunion/")
assert response.status_code == 200
assert response.json() == {
"id": None,
"livekit": {
"room": "reunion",
"token": "foo",
},
}
# todo - assert generate_token has been called
@override_settings(ALLOW_UNREGISTERED_ROOMS=False)
def test_api_rooms_retrieve_anonymous_unregistered_not_allowed():
"""
Retrieving an unregistered room should return a 404 if unregistered rooms are not allowed.
"""
client = APIClient()
response = client.get("/api/v1.0/rooms/unregistered-room/")
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
def test_api_rooms_retrieve_anonymous_public():
"""
Anonymous users should be able to retrieve a room with a token provided it is public.
"""
room = RoomFactory(is_public=True)
client = APIClient()
response = client.get(f"/api/v1.0/rooms/{room.id!s}/")
assert response.status_code == 200
assert response.json() == {
"id": str(room.id),
"is_administrable": False,
"is_public": True,
"livekit": {
"room": "foo",
"token": "foo",
},
"name": room.name,
"slug": room.slug,
}
# todo - assert generate_token has been called
def test_api_rooms_retrieve_authenticated_public():
"""
Authenticated users should be allowed to retrieve a room and get a token for a room to
which they are not related, provided the room is public.
They should not see related users.
"""
room = RoomFactory(is_public=True)
user = UserFactory()
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/rooms/{room.id!s}/",
)
assert response.status_code == 200
assert response.json() == {
"id": str(room.id),
"is_administrable": False,
"is_public": True,
"livekit": {
"room": "foo",
"token": "foo",
},
"name": room.name,
"slug": room.slug,
}
# todo - assert generate_token has been called
def test_api_rooms_retrieve_authenticated():
"""
Authenticated users should be allowed to retrieve a private room to which they
are not related but should not be given any token.
"""
room = RoomFactory(is_public=False)
user = UserFactory()
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/rooms/{room.id!s}/",
)
assert response.status_code == 200
assert response.json() == {
"id": str(room.id),
"is_administrable": False,
"is_public": False,
"name": room.name,
"slug": room.slug,
}
def test_api_rooms_retrieve_members(django_assert_num_queries):
"""
Users who are members of a room should be allowed to see related users.
"""
user = UserFactory()
other_user = UserFactory()
room = RoomFactory()
user_access = UserResourceAccessFactory(resource=room, user=user, role="member")
other_user_access = UserResourceAccessFactory(
resource=room, user=other_user, role="member"
)
client = APIClient()
client.force_login(user)
with django_assert_num_queries(4):
response = client.get(
f"/api/v1.0/rooms/{room.id!s}/",
)
assert response.status_code == 200
content_dict = response.json()
assert sorted(content_dict.pop("accesses"), key=lambda x: x["id"]) == sorted(
[
{
"id": str(user_access.id),
"user": {
"id": str(user_access.user.id),
"email": user_access.user.email,
},
"resource": str(room.id),
"role": user_access.role,
},
{
"id": str(other_user_access.id),
"user": {
"id": str(other_user_access.user.id),
"email": other_user_access.user.email,
},
"resource": str(room.id),
"role": other_user_access.role,
},
],
key=lambda x: x["id"],
)
assert content_dict == {
"id": str(room.id),
"is_administrable": False,
"is_public": room.is_public,
"livekit": {
"room": "foo",
"token": "foo",
},
"name": room.name,
"slug": room.slug,
}
# todo - assert generate_token has been called
def test_api_rooms_retrieve_administrators(django_assert_num_queries):
"""
A user who is an administrator or owner of a room should be allowed
to see related users.
"""
user = UserFactory()
other_user = UserFactory()
room = RoomFactory()
user_access = UserResourceAccessFactory(
resource=room, user=user, role=random.choice(["administrator", "owner"])
)
other_user_access = UserResourceAccessFactory(
resource=room, user=other_user, role="member"
)
client = APIClient()
client.force_login(user)
with django_assert_num_queries(4):
response = client.get(
f"/api/v1.0/rooms/{room.id!s}/",
)
assert response.status_code == 200
content_dict = response.json()
assert sorted(content_dict.pop("accesses"), key=lambda x: x["id"]) == sorted(
[
{
"id": str(other_user_access.id),
"user": {
"id": str(other_user_access.user.id),
"email": other_user_access.user.email,
},
"resource": str(room.id),
"role": other_user_access.role,
},
{
"id": str(user_access.id),
"user": {
"id": str(user_access.user.id),
"email": user_access.user.email,
},
"resource": str(room.id),
"role": user_access.role,
},
],
key=lambda x: x["id"],
)
assert content_dict == {
"id": str(room.id),
"is_administrable": True,
"is_public": room.is_public,
"configuration": {},
"livekit": {
"room": "foo",
"token": "foo",
},
"name": room.name,
"slug": room.slug,
}
# todo - assert generate_token has been called
@@ -0,0 +1,123 @@
"""
Test rooms API endpoints in the impress core app: update.
"""
import random
import pytest
from rest_framework.test import APIClient
from ...factories import RoomFactory, UserFactory
pytestmark = pytest.mark.django_db
def test_api_rooms_update_anonymous():
"""Anonymous users should not be allowed to update a room."""
room = RoomFactory(name="Old name")
client = APIClient()
response = client.put(
f"/api/v1.0/rooms/{room.id!s}/",
{
"name": "New name",
},
)
assert response.status_code == 401
room.refresh_from_db()
assert room.name == "Old name"
assert room.slug == "old-name"
def test_api_rooms_update_authenticated():
"""Authenticated users should not be allowed to update a room."""
user = UserFactory()
room = RoomFactory(name="Old name")
client = APIClient()
client.force_login(user)
response = client.put(
f"/api/v1.0/rooms/{room.id!s}/",
{
"name": "New name",
},
)
assert response.status_code == 403
room.refresh_from_db()
assert room.name == "Old name"
assert room.slug == "old-name"
def test_api_rooms_update_members():
"""
Users who are members of a room but not administrators should
not be allowed to update it.
"""
user = UserFactory()
room = RoomFactory(name="Old name", users=[(user, "member")])
client = APIClient()
client.force_login(user)
new_is_public = not room.is_public
response = client.put(
f"/api/v1.0/rooms/{room.id!s}/",
{
"name": "New name",
"slug": "should-be-ignored",
"is_public": new_is_public,
"configuration": {"the_key": "the_value"},
},
format="json",
)
assert response.status_code, 403
room.refresh_from_db()
assert room.name == "Old name"
assert room.slug == "old-name"
assert room.is_public != new_is_public
assert room.configuration == {}
def test_api_rooms_update_administrators():
"""Administrators or owners of a room should be allowed to update it."""
user = UserFactory()
room = RoomFactory(users=[(user, random.choice(["administrator", "owner"]))])
client = APIClient()
client.force_login(user)
new_is_public = not room.is_public
response = client.put(
f"/api/v1.0/rooms/{room.id!s}/",
{
"name": "New name",
"slug": "should-be-ignored",
"is_public": new_is_public,
"configuration": {"the_key": "the_value"},
},
format="json",
)
assert response.status_code == 200
room.refresh_from_db()
assert room.name == "New name"
assert room.slug == "new-name"
assert room.is_public == new_is_public
assert room.configuration == {"the_key": "the_value"}
def test_api_rooms_update_administrators_of_another():
"""
Being administrator or owner of a room should not grant authorization to update
another room.
"""
user = UserFactory()
RoomFactory(users=[(user, random.choice(["administrator", "owner"]))])
other_room = RoomFactory(name="Old name")
client = APIClient()
client.force_login(user)
response = client.put(
f"/api/v1.0/rooms/{other_room.id!s}/",
{"name": "New name", "slug": "should-be-ignored"},
)
assert response.status_code, 403
other_room.refresh_from_db()
assert other_room.name == "Old name"
assert other_room.slug == "old-name"