(back) add resource server api tests

Add test for the /external_api routes, making sure the
implementation works.
This commit is contained in:
Nathan Vasse
2025-10-31 10:50:19 +01:00
parent 1ba02d7c43
commit 51821dfb59
6 changed files with 674 additions and 12 deletions
+94
View File
@@ -1,10 +1,15 @@
"""Fixtures for tests in the drive core application"""
import base64
from unittest import mock
from django.core.cache import cache
import pytest
import responses
from core import factories
from core.tests.utils.urls import reload_urls
USER = "user"
TEAM = "team"
@@ -25,3 +30,92 @@ def mock_user_teams():
"core.models.User.teams", new_callable=mock.PropertyMock
) as mock_teams:
yield mock_teams
def resource_server_backend_setup(settings):
"""
A fixture to create a user token for testing.
"""
assert (
settings.OIDC_RS_BACKEND_CLASS
== "lasuite.oidc_resource_server.backend.ResourceServerBackend"
)
settings.OIDC_RESOURCE_SERVER_ENABLED = True
settings.OIDC_RS_CLIENT_ID = "some_client_id"
settings.OIDC_RS_CLIENT_SECRET = "some_client_secret"
settings.OIDC_OP_URL = "https://oidc.example.com"
settings.OIDC_VERIFY_SSL = False
settings.OIDC_TIMEOUT = 5
settings.OIDC_PROXY = None
settings.OIDC_OP_JWKS_ENDPOINT = "https://oidc.example.com/jwks"
settings.OIDC_OP_INTROSPECTION_ENDPOINT = "https://oidc.example.com/introspect"
settings.OIDC_RS_SCOPES = ["openid", "groups"]
settings.OIDC_RS_ALLOWED_AUDIENCES = ["some_service_provider"]
@pytest.fixture
def resource_server_backend_conf(settings):
"""
A fixture to create a user token for testing.
"""
resource_server_backend_setup(settings)
reload_urls()
@pytest.fixture
def resource_server_backend(settings):
"""
A fixture to create a user token for testing.
Including a mocked introspection endpoint.
"""
resource_server_backend_setup(settings)
reload_urls()
with responses.RequestsMock() as rsps:
rsps.add(
responses.POST,
"https://oidc.example.com/introspect",
json={
"iss": "https://oidc.example.com",
"aud": "some_client_id", # settings.OIDC_RS_CLIENT_ID
"sub": "very-specific-sub",
"client_id": "some_service_provider",
"scope": "openid groups",
"active": True,
},
)
yield rsps
@pytest.fixture
def user_specific_sub():
"""
A fixture to create a user token for testing.
"""
user = factories.UserFactory(sub="very-specific-sub")
yield user
def build_authorization_bearer(token):
"""
Build an Authorization Bearer header value from a token.
This can be used like this:
client.post(
...
HTTP_AUTHORIZATION=f"Bearer {build_authorization_bearer('some_token')}",
)
"""
return base64.b64encode(token.encode("utf-8")).decode("utf-8")
@pytest.fixture
def user_token():
"""
A fixture to create a user token for testing.
"""
return build_authorization_bearer("some_token")
@@ -0,0 +1,406 @@
"""
Tests for the Resource Server API for items.
Not testing external API endpoints that are already tested in the /api
because the resource server viewsets inherit from the api viewsets.
"""
from io import BytesIO
from django.core.files.storage import default_storage
import pytest
from rest_framework.test import APIClient
from core import factories, models
pytestmark = pytest.mark.django_db
# pylint: disable=unused-argument
def test_api_items_retrieve_anonymous_public_standalone():
"""
Anonymous users should not be allowed to retrieve an item from external
API if resource server is not enabled.
"""
item = factories.ItemFactory(link_reach="public")
response = APIClient().get(f"/external_api/v1.0/items/{item.id!s}/")
assert response.status_code == 404
def test_api_items_list_connected_not_resource_server():
"""
Connected users should not be allowed to list items if resource server is not enabled.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(item=item, user=user, role="reader")
response = client.get("/external_api/v1.0/items/")
assert response.status_code == 404
def test_api_items_list_connected_resource_server(
user_token, resource_server_backend, user_specific_sub
):
"""Connected users should be allowed to list items from a resource server."""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(item=item, user=user_specific_sub, role="reader")
response = client.get("/external_api/v1.0/items/")
assert response.status_code == 200
def test_api_items_list_connected_resource_server_with_invalid_token(
user_token, resource_server_backend
):
"""User with an invalid sub should not be allowed to retrieve items from a resource server."""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
response = client.get("/external_api/v1.0/items/")
assert response.status_code == 401
def test_api_items_retrieve_connected_resource_server_with_wrong_abilities(
user_token, user_specific_sub, resource_server_backend
):
"""
User with wrong abilities should not be allowed to retrieve an item from
a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
response = client.get(f"/external_api/v1.0/items/{item.id!s}/")
assert response.status_code == 403
def test_api_items_retrieve_connected_resource_server_using_access_token(
user_token, resource_server_backend, user_specific_sub
):
"""
User with an access token should be allowed to retrieve an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.LinkRoleChoices.READER
)
response = client.get(f"/external_api/v1.0/items/{item.id!s}/")
assert response.status_code == 200
def test_api_items_upload_resource_server_using_access_token(
user_token, resource_server_backend, user_specific_sub
):
"""
User with an access token should be allowed to upload an item to a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(
link_reach=models.LinkReachChoices.RESTRICTED,
type=models.ItemTypeChoices.FOLDER,
)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.post(
f"/external_api/v1.0/items/{item.id!s}/children/",
{
"type": models.ItemTypeChoices.FILE,
"filename": "file.txt",
},
)
assert response.status_code == 201
data = response.json()
assert data["type"] == models.ItemTypeChoices.FILE
assert data["filename"] == "file.txt"
assert "policy" in data
child = models.Item.objects.get(id=data["id"])
default_storage.save(
child.file_key,
BytesIO(b"my prose"),
)
response = client.post(f"/external_api/v1.0/items/{child.id!s}/upload-ended/")
assert response.status_code == 200
# Non allowed actions on resource server.
def test_api_items_delete_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to delete an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.delete(f"/external_api/v1.0/items/{item.id!s}/")
assert response.status_code == 403
def test_api_items_hard_delete_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to hard delete an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.delete(f"/external_api/v1.0/items/{item.id!s}/hard-delete/")
assert response.status_code == 403
def test_api_items_patch_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to patch an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.patch(
f"/external_api/v1.0/items/{item.id!s}/", {"title": "new title"}
)
assert response.status_code == 403
def test_api_items_put_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should not be allowed to put an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.put(
f"/external_api/v1.0/items/{item.id!s}/", {"title": "new title"}
)
assert response.status_code == 403
def test_api_items_move_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should not be allowed to move an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.post(
f"/external_api/v1.0/items/{item.id!s}/move/",
{
"target_item_id": factories.ItemFactory(
link_reach=models.LinkReachChoices.RESTRICTED
).id
},
)
assert response.status_code == 403
def test_api_items_restore_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should not be allowed to restore an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.post(f"/external_api/v1.0/items/{item.id!s}/restore/")
assert response.status_code == 403
def test_api_items_trashbin_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to list the trashbin from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
response = client.get("/external_api/v1.0/items/trashbin/")
assert response.status_code == 403
def test_api_items_link_configuration_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should not be allowed to update the link configuration
of an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.put(
f"/external_api/v1.0/items/{item.id!s}/link-configuration/",
{
"link_reach": models.LinkReachChoices.RESTRICTED,
"link_role": models.LinkRoleChoices.READER,
},
)
assert response.status_code == 403
def test_api_items_accesses_resource_server_not_allowed(
user_token, resource_server_backend_conf, user_specific_sub
):
"""
Connected users should not be allowed to list the accesses of
an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.get(f"/external_api/v1.0/items/{item.id!s}/accesses/")
assert response.status_code == 404
def test_api_items_accesses_create_resource_server_not_allowed(
user_token, resource_server_backend_conf, user_specific_sub
):
"""
Connected users should not be allowed to create an access for an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.post(
f"/external_api/v1.0/items/{item.id!s}/accesses/",
{"user_id": factories.UserFactory().id, "role": models.RoleChoices.READER},
)
assert response.status_code == 404
def test_api_items_invitations_resource_server_not_allowed(
user_token, resource_server_backend_conf, user_specific_sub
):
"""
Connected users should not be allowed to list the invitations of an item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.get(f"/external_api/v1.0/items/{item.id!s}/invitations/")
assert response.status_code == 404
def test_api_items_invitations_create_resource_server_not_allowed(
user_token, resource_server_backend_conf, user_specific_sub
):
"""
Connected users should not be allowed to create an invitation for an
item from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
factories.UserItemAccessFactory(
item=item, user=user_specific_sub, role=models.RoleChoices.OWNER
)
response = client.post(
f"/external_api/v1.0/items/{item.id!s}/invitations/",
{"email": "test@example.com", "role": models.RoleChoices.READER},
)
assert response.status_code == 404
@@ -0,0 +1,152 @@
"""
Tests for the Resource Server API for users.
Not testing external API endpoints that are already tested in the /api
because the resource server viewsets inherit from the api viewsets.
"""
import pytest
from rest_framework.test import APIClient
from core import factories
from core.api import serializers
from core.tests.utils.urls import reload_urls
pytestmark = pytest.mark.django_db
# pylint: disable=unused-argument
def test_api_users_me_anonymous_public_standalone():
"""
Anonymous users should not be allowed to retrieve their own user information from external
API if resource server is not enabled.
"""
reload_urls()
response = APIClient().get("/external_api/v1.0/users/me/")
assert response.status_code == 404
def test_api_users_me_connected_not_resource_server():
"""
Connected users should not be allowed to retrieve their own user information from external
API if resource server is not enabled.
"""
reload_urls()
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
response = client.get("/external_api/v1.0/users/me/")
assert response.status_code == 404
def test_api_users_me_connected_resource_server(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should be allowed to retrieve their own user information from external API
if resource server is enabled.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
response = client.get("/external_api/v1.0/users/me/")
assert response.status_code == 200
data = response.json()
assert data["id"] == str(user_specific_sub.id)
assert data["email"] == user_specific_sub.email
def test_api_users_me_connected_resource_server_with_invalid_token(
user_token, resource_server_backend
):
"""
Connected users should not be allowed to retrieve their own user information from external API
if resource server is enabled with an invalid token.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
response = client.get("/external_api/v1.0/users/me/")
assert response.status_code == 401
# Non allowed actions on resource server.
def test_api_users_list_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to list users from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
response = client.get("/external_api/v1.0/users/")
assert response.status_code == 403
def test_api_users_retrieve_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to list users from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
other_user = factories.UserFactory()
response = client.get(f"/external_api/v1.0/users/{other_user.id!s}/")
assert response.status_code == 403
def test_api_users_put_patch_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to list users from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
other_user = factories.UserFactory()
new_user_values = serializers.UserSerializer(instance=factories.UserFactory()).data
response = client.put(
f"/external_api/v1.0/users/{other_user.id!s}/", new_user_values
)
assert response.status_code == 403
response = client.patch(
f"/external_api/v1.0/users/{other_user.id!s}/",
{"email": "new_email@example.com"},
)
assert response.status_code == 403
def test_api_users_delete_resource_server_not_allowed(
user_token, resource_server_backend, user_specific_sub
):
"""
Connected users should notbe allowed to list users from a resource server.
"""
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
other_user = factories.UserFactory()
response = client.delete(f"/external_api/v1.0/users/{other_user.id!s}/")
assert response.status_code == 403
+20
View File
@@ -0,0 +1,20 @@
"""Utils for testing URLs."""
import importlib
from django.urls import clear_url_caches
def reload_urls():
"""
Reload the URLs. Since the url are loaded based on a
settings value, we need to reload the urls to make the
URL settings based condition effective.
"""
import core.urls # pylint:disable=import-outside-toplevel # noqa: PLC0415
import drive.urls # pylint:disable=import-outside-toplevel # noqa: PLC0415
importlib.reload(core.urls)
importlib.reload(drive.urls)
clear_url_caches()
+2 -12
View File
@@ -2,26 +2,16 @@
Test e2e API endpoints.
"""
import importlib
from django.test.utils import override_settings
from django.urls import clear_url_caches
import pytest
from rest_framework.test import APIClient
from core.tests.utils.urls import reload_urls
pytestmark = pytest.mark.django_db
def reload_urls():
"""Reload the URLs to test the e2e URLs. Since the url are loaded based on a
settings value, we need to reload the urls to test the e2e urls."""
import drive.urls # pylint:disable=import-outside-toplevel # noqa: PLC0415
importlib.reload(drive.urls)
clear_url_caches()
def test_api_e2e_user_auth_no_urls():
"""E2E URLs not enabled should 404."""
client = APIClient()