mirror of
https://github.com/suitenumerique/docs.git
synced 2026-09-30 05:25:06 +02:00
✨(backend) add resource server api
Adds a resource server API similar to the one that already exists for Drive.
This commit is contained in:
Sylvain Boissel
@@ -0,0 +1,772 @@
|
||||
"""
|
||||
Tests for the Resource Server API for documents.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
from datetime import timedelta
|
||||
from io import BytesIO
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.test import override_settings
|
||||
from django.utils import timezone
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
from core.services import mime_types
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_documents_retrieve_anonymous_public_standalone():
|
||||
"""
|
||||
Anonymous users SHOULD NOT be allowed to retrieve a document from external
|
||||
API if resource server is not enabled.
|
||||
"""
|
||||
document = factories.DocumentFactory(link_reach="public")
|
||||
|
||||
response = APIClient().get(f"/external_api/v1.0/documents/{document.id!s}/")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_external_api_documents_list_connected_not_resource_server():
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to list documents if resource server is not enabled.
|
||||
"""
|
||||
user = factories.UserFactory()
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(document=document, user=user, role="reader")
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_external_api_documents_list_connected_resource_server(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""Connected users should be allowed to list documents from a resource server."""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role="reader"
|
||||
)
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/")
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_external_api_documents_list_connected_resource_server_with_invalid_token(
|
||||
user_token, resource_server_backend
|
||||
):
|
||||
"""A user with an invalid sub SHOULD NOT be allowed to retrieve documents
|
||||
from a resource server."""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_external_api_documents_retrieve_connected_resource_server_with_wrong_abilities(
|
||||
user_token, user_specific_sub, resource_server_backend
|
||||
):
|
||||
"""
|
||||
A user with wrong abilities SHOULD NOT be allowed to retrieve a document from
|
||||
a resource server.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_external_api_documents_retrieve_connected_resource_server_using_access_token(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
A user with an access token SHOULD be allowed to retrieve a document from
|
||||
a resource server.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.LinkRoleChoices.READER
|
||||
)
|
||||
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/")
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_external_api_documents_create_root_success(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users with an access token should be able to create a root document through the resource
|
||||
server and should automatically be declared as the owner of the newly created document.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
response = client.post(
|
||||
"/external_api/v1.0/documents/",
|
||||
{
|
||||
"title": "Test Root Document",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
|
||||
data = response.json()
|
||||
document = models.Document.objects.get(id=data["id"])
|
||||
|
||||
assert document.title == "Test Root Document"
|
||||
assert document.creator == user_specific_sub
|
||||
assert document.accesses.filter(role="owner", user=user_specific_sub).exists()
|
||||
|
||||
|
||||
def test_external_api_documents_create_subdocument_owner_success(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users with an access token SHOULD BE able to create a sub-document through the resource
|
||||
server when they have OWNER permissions on the parent document.
|
||||
The creator is set to the authenticated user, and permissions are inherited
|
||||
from the parent document.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
# Create a parent document first
|
||||
parent_document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=parent_document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{parent_document.id}/children/",
|
||||
{
|
||||
"title": "Test Sub Document",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
|
||||
data = response.json()
|
||||
document = models.Document.objects.get(id=data["id"])
|
||||
|
||||
assert document.title == "Test Sub Document"
|
||||
assert document.creator == user_specific_sub
|
||||
assert document.get_parent() == parent_document
|
||||
# Child documents inherit permissions from parent, no direct access needed
|
||||
assert not document.accesses.exists()
|
||||
|
||||
|
||||
def test_external_api_documents_create_subdocument_editor_success(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users with an access token SHOULD BE able to create a sub-document through the resource
|
||||
server when they have EDITOR permissions on the parent document.
|
||||
Permissions are inherited from the parent document.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
# Create a parent document first
|
||||
parent_document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=parent_document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.EDITOR,
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{parent_document.id}/children/",
|
||||
{
|
||||
"title": "Test Sub Document",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
|
||||
data = response.json()
|
||||
document = models.Document.objects.get(id=data["id"])
|
||||
|
||||
assert document.title == "Test Sub Document"
|
||||
assert document.creator == user_specific_sub
|
||||
assert document.get_parent() == parent_document
|
||||
# Child documents inherit permissions from parent, no direct access needed
|
||||
assert not document.accesses.exists()
|
||||
|
||||
|
||||
def test_external_api_documents_create_subdocument_reader_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users with an access token SHOULD NOT be able to create a sub-document through the resource
|
||||
server when they have READER permissions on the parent document.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
# Create a parent document first
|
||||
parent_document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=parent_document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.READER,
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{parent_document.id}/children/",
|
||||
{
|
||||
"title": "Test Sub Document",
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@patch("core.services.converter_services.Converter.convert")
|
||||
def test_external_api_documents_create_with_markdown_file_success(
|
||||
mock_convert, user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users with an access token should be able to create documents through the resource
|
||||
server by uploading a Markdown file and should automatically be declared as the owner
|
||||
of the newly created document.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
# Mock the conversion
|
||||
converted_yjs = "base64encodedyjscontent"
|
||||
mock_convert.return_value = converted_yjs
|
||||
|
||||
# Create a fake Markdown file
|
||||
file_content = b"# Test Document\n\nThis is a test."
|
||||
file = BytesIO(file_content)
|
||||
file.name = "readme.md"
|
||||
|
||||
response = client.post(
|
||||
"/external_api/v1.0/documents/",
|
||||
{
|
||||
"file": file,
|
||||
},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
|
||||
data = response.json()
|
||||
document = models.Document.objects.get(id=data["id"])
|
||||
|
||||
assert document.title == "readme.md"
|
||||
assert document.content == converted_yjs
|
||||
assert document.accesses.filter(role="owner", user=user_specific_sub).exists()
|
||||
|
||||
# Verify the converter was called correctly
|
||||
mock_convert.assert_called_once_with(
|
||||
file_content,
|
||||
content_type=mime_types.MARKDOWN,
|
||||
accept=mime_types.YJS,
|
||||
)
|
||||
|
||||
|
||||
def test_external_api_documents_list_with_multiple_roles(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
List all documents accessible to a user with different roles and verify
|
||||
that associated permissions are correctly returned in the response.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
# Create documents with different roles for the user
|
||||
owner_document = factories.DocumentFactory(
|
||||
title="Owner Document",
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=owner_document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
editor_document = factories.DocumentFactory(
|
||||
title="Editor Document",
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=editor_document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.EDITOR,
|
||||
)
|
||||
|
||||
reader_document = factories.DocumentFactory(
|
||||
title="Reader Document",
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=reader_document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.READER,
|
||||
)
|
||||
|
||||
# Create a document the user should NOT have access to
|
||||
other_document = factories.DocumentFactory(
|
||||
title="Other Document",
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
)
|
||||
other_user = factories.UserFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=other_document,
|
||||
user=other_user,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Verify the response contains results
|
||||
assert "results" in data
|
||||
results = data["results"]
|
||||
|
||||
# Verify user can see exactly 3 documents (owner, editor, reader)
|
||||
result_ids = {result["id"] for result in results}
|
||||
assert len(results) == 3
|
||||
assert str(owner_document.id) in result_ids
|
||||
assert str(editor_document.id) in result_ids
|
||||
assert str(reader_document.id) in result_ids
|
||||
assert str(other_document.id) not in result_ids
|
||||
|
||||
# Verify each document has correct user_role field indicating permission level
|
||||
for result in results:
|
||||
if result["id"] == str(owner_document.id):
|
||||
assert result["title"] == "Owner Document"
|
||||
assert result["user_role"] == models.RoleChoices.OWNER
|
||||
elif result["id"] == str(editor_document.id):
|
||||
assert result["title"] == "Editor Document"
|
||||
assert result["user_role"] == models.RoleChoices.EDITOR
|
||||
elif result["id"] == str(reader_document.id):
|
||||
assert result["title"] == "Reader Document"
|
||||
assert result["user_role"] == models.RoleChoices.READER
|
||||
|
||||
|
||||
def test_external_api_documents_duplicate_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users CAN DUPLICATE a document from a resource server
|
||||
when they have the required permissions on the document,
|
||||
as this action bypasses the permission checks.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/duplicate/",
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
# NOT allowed actions on resource server.
|
||||
|
||||
|
||||
def test_external_api_documents_put_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to PUT a document from a resource server.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.put(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/", {"title": "new title"}
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_external_api_document_delete_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to delete a document from a resource server.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.delete(f"/external_api/v1.0/documents/{document.id!s}/")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_external_api_documents_move_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to MOVE a document from a resource server.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
new_parent = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=new_parent,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/move/",
|
||||
{"target_document_id": new_parent.id},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_external_api_documents_restore_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to restore a document from a resource server.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.post(f"/external_api/v1.0/documents/{document.id!s}/restore/")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.parametrize("role", models.LinkRoleChoices.values)
|
||||
@pytest.mark.parametrize("reach", models.LinkReachChoices.values)
|
||||
def test_external_api_documents_trashbin_not_allowed(
|
||||
role, reach, user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to list documents from the trashbin,
|
||||
regardless of the document link reach and user role, from a resource server.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=reach,
|
||||
creator=user_specific_sub,
|
||||
deleted_at=timezone.now(),
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=role,
|
||||
)
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/trashbin/")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_external_api_documents_create_for_owner_not_allowed():
|
||||
"""
|
||||
Authenticated users SHOULD NOT be allowed to call create documents
|
||||
on behalf of other users.
|
||||
This API endpoint is reserved for server-to-server calls.
|
||||
"""
|
||||
user = factories.UserFactory()
|
||||
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
data = {
|
||||
"title": "My Document",
|
||||
"content": "Document content",
|
||||
"sub": "123",
|
||||
"email": "john.doe@example.com",
|
||||
}
|
||||
|
||||
response = client.post(
|
||||
"/external_api/v1.0/documents/create-for-owner/",
|
||||
data,
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not models.Document.objects.exists()
|
||||
|
||||
|
||||
# Test overrides
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "children", "trashbin"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_trashbin_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to list soft deleted documents from a resource server
|
||||
when the trashbin action is enabled in EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
document.soft_delete()
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/trashbin/")
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
content = response.json()
|
||||
results = content.pop("results")
|
||||
assert content == {
|
||||
"count": 1,
|
||||
"next": None,
|
||||
"previous": None,
|
||||
}
|
||||
assert len(results) == 1
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "children", "destroy"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_delete_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to delete a document from a resource server
|
||||
when the delete action is enabled in EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.delete(f"/external_api/v1.0/documents/{document.id!s}/")
|
||||
|
||||
assert response.status_code == 204
|
||||
# Verify the document is soft deleted
|
||||
document.refresh_from_db()
|
||||
assert document.deleted_at is not None
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"update",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_update_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to update a document from a resource server
|
||||
when the update action is enabled in EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
original_title = document.title
|
||||
response = client.put(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/", {"title": "new title"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Verify the document is updated
|
||||
document.refresh_from_db()
|
||||
assert document.title == "new title"
|
||||
assert document.title != original_title
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "children", "move"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_move_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to move a document from a resource server
|
||||
when the move action is enabled in EXTERNAL_API settings and they
|
||||
have the required permissions on the document and the target location.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
parent = factories.DocumentFactory(
|
||||
users=[(user_specific_sub, "owner")], teams=[("lasuite", "owner")]
|
||||
)
|
||||
# A document with no owner
|
||||
document = factories.DocumentFactory(
|
||||
parent=parent, users=[(user_specific_sub, "reader")]
|
||||
)
|
||||
target = factories.DocumentFactory()
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/move/",
|
||||
data={"target_document_id": str(target.id), "position": "first-sibling"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"message": "Document moved successfully."}
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "children", "restore"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_restore_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to restore a recently soft-deleted document
|
||||
from a resource server when the restore action is enabled in EXTERNAL_API
|
||||
settings and they have the required permissions on the document.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
now = timezone.now() - timedelta(days=15)
|
||||
document = factories.DocumentFactory(deleted_at=now)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role="owner"
|
||||
)
|
||||
|
||||
response = client.post(f"/external_api/v1.0/documents/{document.id!s}/restore/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"detail": "Document has been successfully restored."}
|
||||
|
||||
document.refresh_from_db()
|
||||
assert document.deleted_at is None
|
||||
assert document.ancestors_deleted_at is None
|
||||
@@ -0,0 +1,681 @@
|
||||
"""
|
||||
Tests for the Resource Server API for documents accesses.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
import responses
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
from core.api import serializers
|
||||
from core.tests.utils.urls import reload_urls
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_document_accesses_anonymous_public_standalone():
|
||||
"""
|
||||
Anonymous users SHOULD NOT be allowed to list document accesses
|
||||
from external API if resource server is not enabled.
|
||||
"""
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
)
|
||||
|
||||
response = APIClient().get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/"
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_external_api_document_accesses_list_connected_not_resource_server():
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to list document accesses
|
||||
if resource server is not enabled.
|
||||
"""
|
||||
user = factories.UserFactory()
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
|
||||
response = APIClient().get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/"
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_list_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to list the accesses of
|
||||
a document from a resource server.
|
||||
"""
|
||||
reload_urls()
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/accesses/")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_retrieve_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to retrieve a specific access of
|
||||
a document from a resource server.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
access = factories.UserDocumentAccessFactory(document=document)
|
||||
|
||||
response = client.get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/"
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_accesses_create_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to create an access for a document
|
||||
from a resource server.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
other_user = factories.UserFactory()
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/",
|
||||
{"user_id": other_user.id, "role": models.RoleChoices.READER},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_update_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to update an access for a
|
||||
document from a resource server through PUT.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
other_user = factories.UserFactory()
|
||||
access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=other_user, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
response = client.put(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
|
||||
{"role": models.RoleChoices.EDITOR},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_partial_update_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to update an access
|
||||
for a document from a resource server through PATCH.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
other_user = factories.UserFactory()
|
||||
access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=other_user, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
response = client.patch(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
|
||||
{"role": models.RoleChoices.EDITOR},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_accesses_delete_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to delete an access for
|
||||
a document from a resource server.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
access = factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.delete(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
# Overrides
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_list_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to list the accesses of a document from a resource server
|
||||
when the list action is enabled in EXTERNAL_API document_access settings.
|
||||
"""
|
||||
|
||||
reload_urls()
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED, creator=user_specific_sub
|
||||
)
|
||||
user_access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
# Create additional accesses
|
||||
other_user = factories.UserFactory()
|
||||
other_access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=other_user, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/accesses/")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
access_ids = [entry["id"] for entry in data]
|
||||
assert str(user_access.id) in access_ids
|
||||
assert str(other_access.id) in access_ids
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_retrieve_can_be_allowed(
|
||||
user_token,
|
||||
resource_server_backend,
|
||||
user_specific_sub,
|
||||
):
|
||||
"""
|
||||
A user who is related to a document SHOULD be allowed to retrieve the
|
||||
associated document user accesses.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
|
||||
)
|
||||
data = response.json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert data["id"] == str(access.id)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "create"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_create_can_be_allowed(
|
||||
user_token,
|
||||
resource_server_backend,
|
||||
user_specific_sub,
|
||||
):
|
||||
"""
|
||||
A user who is related to a document SHOULD be allowed to create
|
||||
a user access for the document.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
other_user = factories.UserFactory()
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/",
|
||||
data={"user_id": other_user.id, "role": models.RoleChoices.READER},
|
||||
)
|
||||
data = response.json()
|
||||
|
||||
assert response.status_code == 201
|
||||
assert data["role"] == models.RoleChoices.READER
|
||||
assert str(data["user"]["id"]) == str(other_user.id)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "update"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_update_can_be_allowed(
|
||||
user_token,
|
||||
resource_server_backend,
|
||||
user_specific_sub,
|
||||
settings,
|
||||
):
|
||||
"""
|
||||
A user who is related to a document SHOULD be allowed to update
|
||||
a user access for the document through PUT.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
other_user = factories.UserFactory()
|
||||
access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=other_user, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
# Add the reset-connections endpoint to the existing mock
|
||||
settings.COLLABORATION_API_URL = "http://example.com/"
|
||||
settings.COLLABORATION_SERVER_SECRET = "secret-token"
|
||||
endpoint_url = (
|
||||
f"{settings.COLLABORATION_API_URL}reset-connections/?room={document.id}"
|
||||
)
|
||||
resource_server_backend.add(
|
||||
responses.POST,
|
||||
endpoint_url,
|
||||
json={},
|
||||
status=200,
|
||||
)
|
||||
|
||||
old_values = serializers.DocumentAccessSerializer(instance=access).data
|
||||
|
||||
# Update only the role field
|
||||
response = client.put(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
|
||||
{**old_values, "role": models.RoleChoices.EDITOR}, # type: ignore
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["role"] == models.RoleChoices.EDITOR
|
||||
assert str(data["user"]["id"]) == str(other_user.id)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "partial_update"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_document_accesses_partial_update_can_be_allowed(
|
||||
user_token,
|
||||
resource_server_backend,
|
||||
user_specific_sub,
|
||||
settings,
|
||||
):
|
||||
"""
|
||||
A user who is related to a document SHOULD be allowed to update
|
||||
a user access for the document through PATCH.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
other_user = factories.UserFactory()
|
||||
access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=other_user, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
# Add the reset-connections endpoint to the existing mock
|
||||
settings.COLLABORATION_API_URL = "http://example.com/"
|
||||
settings.COLLABORATION_SERVER_SECRET = "secret-token"
|
||||
endpoint_url = (
|
||||
f"{settings.COLLABORATION_API_URL}reset-connections/?room={document.id}"
|
||||
)
|
||||
resource_server_backend.add(
|
||||
responses.POST,
|
||||
endpoint_url,
|
||||
json={},
|
||||
status=200,
|
||||
)
|
||||
|
||||
response = client.patch(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{access.id!s}/",
|
||||
data={"role": models.RoleChoices.EDITOR},
|
||||
)
|
||||
data = response.json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert data["role"] == models.RoleChoices.EDITOR
|
||||
assert str(data["user"]["id"]) == str(other_user.id)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_access": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "destroy"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_accesses_delete_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub, settings
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to delete an access for
|
||||
a document from a resource server when the destroy action is
|
||||
enabled in settings.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
other_user = factories.UserFactory()
|
||||
other_access = factories.UserDocumentAccessFactory(
|
||||
document=document, user=other_user, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
# Add the reset-connections endpoint to the existing mock
|
||||
settings.COLLABORATION_API_URL = "http://example.com/"
|
||||
settings.COLLABORATION_SERVER_SECRET = "secret-token"
|
||||
endpoint_url = (
|
||||
f"{settings.COLLABORATION_API_URL}reset-connections/?room={document.id}"
|
||||
)
|
||||
resource_server_backend.add(
|
||||
responses.POST,
|
||||
endpoint_url,
|
||||
json={},
|
||||
status=200,
|
||||
)
|
||||
|
||||
response = client.delete(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/accesses/{other_access.id!s}/",
|
||||
)
|
||||
|
||||
assert response.status_code == 204
|
||||
@@ -0,0 +1,273 @@
|
||||
"""
|
||||
Tests for the Resource Server API for document AI features.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
from core.tests.documents.test_api_documents_ai_proxy import ( # pylint: disable=unused-import
|
||||
ai_settings,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_documents_ai_transform_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to access AI transform endpoints
|
||||
from a resource server by default.
|
||||
"""
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/ai-transform/",
|
||||
{"text": "hello", "action": "prompt"},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {
|
||||
"detail": "You do not have permission to perform this action."
|
||||
}
|
||||
|
||||
|
||||
def test_external_api_documents_ai_translate_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to access AI translate endpoints
|
||||
from a resource server by default.
|
||||
"""
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/ai-translate/",
|
||||
{"text": "hello", "language": "es"},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {
|
||||
"detail": "You do not have permission to perform this action."
|
||||
}
|
||||
|
||||
|
||||
def test_external_api_documents_ai_proxy_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to access AI proxy endpoints
|
||||
from a resource server by default.
|
||||
"""
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/ai-proxy/",
|
||||
b"{}",
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {
|
||||
"detail": "You do not have permission to perform this action."
|
||||
}
|
||||
|
||||
|
||||
# Overrides
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"ai_transform",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
@pytest.mark.usefixtures("ai_settings")
|
||||
@patch("openai.resources.chat.completions.Completions.create")
|
||||
def test_external_api_documents_ai_transform_can_be_allowed(
|
||||
mock_create, user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users SHOULD be allowed to transform a document using AI when the
|
||||
corresponding action is enabled via EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED, favorited_by=[user_specific_sub]
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
mock_create.return_value = MagicMock(
|
||||
choices=[MagicMock(message=MagicMock(content="Salut"))]
|
||||
)
|
||||
|
||||
url = f"/external_api/v1.0/documents/{document.id!s}/ai-transform/"
|
||||
response = client.post(url, {"text": "Hello", "action": "prompt"})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"answer": "Salut"}
|
||||
# pylint: disable=line-too-long
|
||||
mock_create.assert_called_once_with(
|
||||
model="llama",
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": (
|
||||
"Answer the prompt using markdown formatting for structure and emphasis. "
|
||||
"Return the content directly without wrapping it in code blocks or markdown delimiters. "
|
||||
"Preserve the language and markdown formatting. "
|
||||
"Do not provide any other information. "
|
||||
"Preserve the language."
|
||||
),
|
||||
},
|
||||
{"role": "user", "content": "Hello"},
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"ai_translate",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
@pytest.mark.usefixtures("ai_settings")
|
||||
@patch("openai.resources.chat.completions.Completions.create")
|
||||
def test_external_api_documents_ai_translate_can_be_allowed(
|
||||
mock_create, user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users SHOULD be allowed to translate a document using AI when the
|
||||
corresponding action is enabled via EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED, favorited_by=[user_specific_sub]
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
mock_create.return_value = MagicMock(
|
||||
choices=[MagicMock(message=MagicMock(content="Salut"))]
|
||||
)
|
||||
|
||||
url = f"/external_api/v1.0/documents/{document.id!s}/ai-translate/"
|
||||
response = client.post(url, {"text": "Hello", "language": "es-co"})
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"answer": "Salut"}
|
||||
mock_create.assert_called_once_with(
|
||||
model="llama",
|
||||
messages=[
|
||||
{
|
||||
"role": "system",
|
||||
"content": (
|
||||
"Keep the same html structure and formatting. "
|
||||
"Translate the content in the html to the "
|
||||
"specified language Colombian Spanish. "
|
||||
"Check the translation for accuracy and make any necessary corrections. "
|
||||
"Do not provide any other information."
|
||||
),
|
||||
},
|
||||
{"role": "user", "content": "Hello"},
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"ai_proxy",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
@pytest.mark.usefixtures("ai_settings")
|
||||
@patch("core.services.ai_services.AIService.stream")
|
||||
def test_external_api_documents_ai_proxy_can_be_allowed(
|
||||
mock_stream, user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users SHOULD be allowed to use the AI proxy endpoint when the
|
||||
corresponding action is enabled via EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED, creator=user_specific_sub
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
mock_stream.return_value = iter(["data: response\n"])
|
||||
|
||||
url = f"/external_api/v1.0/documents/{document.id!s}/ai-proxy/"
|
||||
response = client.post(
|
||||
url,
|
||||
b"{}",
|
||||
content_type="application/json",
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response["Content-Type"] == "text/event-stream" # type: ignore
|
||||
mock_stream.assert_called_once()
|
||||
@@ -0,0 +1,121 @@
|
||||
"""
|
||||
Tests for the Resource Server API for document attachments.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
import re
|
||||
import uuid
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_documents_attachment_upload_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to upload attachments to a document
|
||||
from a resource server.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
pixel = (
|
||||
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00"
|
||||
b"\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\xf8\xff\xff?\x00\x05\xfe\x02\xfe"
|
||||
b"\xa7V\xbd\xfa\x00\x00\x00\x00IEND\xaeB`\x82"
|
||||
)
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
file = SimpleUploadedFile(name="test.png", content=pixel, content_type="image/png")
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/attachment-upload/",
|
||||
{"file": file},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {
|
||||
"detail": "You do not have permission to perform this action."
|
||||
}
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"attachment_upload",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_attachment_upload_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to upload attachments to a document
|
||||
from a resource server when the attachment-upload action is enabled in EXTERNAL_API settings.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
pixel = (
|
||||
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00"
|
||||
b"\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\xf8\xff\xff?\x00\x05\xfe\x02\xfe"
|
||||
b"\xa7V\xbd\xfa\x00\x00\x00\x00IEND\xaeB`\x82"
|
||||
)
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
file = SimpleUploadedFile(name="test.png", content=pixel, content_type="image/png")
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/attachment-upload/",
|
||||
{"file": file},
|
||||
format="multipart",
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
|
||||
pattern = re.compile(rf"^{document.id!s}/attachments/(.*)\.png")
|
||||
url_parsed = urlparse(response.json()["file"])
|
||||
assert url_parsed.path == f"/api/v1.0/documents/{document.id!s}/media-check/"
|
||||
query = parse_qs(url_parsed.query)
|
||||
assert query["key"][0] is not None
|
||||
file_path = query["key"][0]
|
||||
match = pattern.search(file_path)
|
||||
file_id = match.group(1) # type: ignore
|
||||
|
||||
# Validate that file_id is a valid UUID
|
||||
uuid.UUID(file_id)
|
||||
@@ -0,0 +1,157 @@
|
||||
"""
|
||||
Tests for the Resource Server API for document favorites.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_documents_favorites_list_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to list their favorites
|
||||
from a resource server, as favorite_list() bypasses permissions.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.UserDocumentAccessFactory(
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.READER,
|
||||
document__favorited_by=[user_specific_sub],
|
||||
).document
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/favorite_list/")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["count"] == 1
|
||||
assert data["results"][0]["id"] == str(document.id)
|
||||
|
||||
|
||||
def test_external_api_documents_favorite_add_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
By default the "favorite" action is not permitted on the external API.
|
||||
POST to the endpoint must return 403.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.post(f"/external_api/v1.0/documents/{document.id!s}/favorite/")
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_external_api_documents_favorite_delete_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
By default the "favorite" action is not permitted on the external API.
|
||||
DELETE to the endpoint must return 403.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.delete(f"/external_api/v1.0/documents/{document.id!s}/favorite/")
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
# Overrides
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"favorite",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_favorite_add_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users SHOULD be allowed to POST to the favorite endpoint when the
|
||||
corresponding action is enabled via EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.post(f"/external_api/v1.0/documents/{document.id!s}/favorite/")
|
||||
assert response.status_code == 201
|
||||
assert models.DocumentFavorite.objects.filter(
|
||||
document=document, user=user_specific_sub
|
||||
).exists()
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"favorite",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_favorite_delete_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Users SHOULD be allowed to DELETE from the favorite endpoint when the
|
||||
corresponding action is enabled via EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED, favorited_by=[user_specific_sub]
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.delete(f"/external_api/v1.0/documents/{document.id!s}/favorite/")
|
||||
assert response.status_code == 204
|
||||
assert not models.DocumentFavorite.objects.filter(
|
||||
document=document, user=user_specific_sub
|
||||
).exists()
|
||||
@@ -0,0 +1,474 @@
|
||||
"""
|
||||
Tests for the Resource Server API for invitations.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
from core.tests.utils.urls import reload_urls
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_document_invitations_anonymous_public_standalone():
|
||||
"""
|
||||
Anonymous users SHOULD NOT be allowed to list invitations from external
|
||||
API if resource server is not enabled.
|
||||
"""
|
||||
invitation = factories.InvitationFactory()
|
||||
response = APIClient().get(
|
||||
f"/external_api/v1.0/documents/{invitation.document.id!s}/invitations/"
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_external_api_document_invitations_list_connected_not_resource_server():
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to list document invitations
|
||||
if resource server is not enabled.
|
||||
"""
|
||||
user = factories.UserFactory()
|
||||
client = APIClient()
|
||||
client.force_login(user)
|
||||
|
||||
invitation = factories.InvitationFactory()
|
||||
response = APIClient().get(
|
||||
f"/external_api/v1.0/documents/{invitation.document.id!s}/invitations/"
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_list_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to list document invitations
|
||||
by default.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
invitation = factories.InvitationFactory()
|
||||
response = client.get(
|
||||
f"/external_api/v1.0/documents/{invitation.document.id!s}/invitations/"
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_retrieve_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to retrieve a document invitation
|
||||
by default.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
invitation = factories.InvitationFactory()
|
||||
document = invitation.document
|
||||
|
||||
response = client.get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/{invitation.id!s}/"
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": [],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_create_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to create a document invitation
|
||||
by default.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/",
|
||||
{"email": "invited@example.com", "role": models.RoleChoices.READER},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve"],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_partial_update_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to partially update a document invitation
|
||||
by default.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
invitation = factories.InvitationFactory(
|
||||
document=document, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
response = client.patch(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/{invitation.id!s}/",
|
||||
{"role": models.RoleChoices.EDITOR},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve"],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_delete_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to delete a document invitation
|
||||
by default.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
invitation = factories.InvitationFactory(document=document)
|
||||
|
||||
response = client.delete(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/{invitation.id!s}/",
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
# Overrides
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve"],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_list_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to list document invitations
|
||||
when the action is explicitly enabled.
|
||||
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
invitation = factories.InvitationFactory(document=document)
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/invitations/")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["count"] == 1
|
||||
assert data["results"][0]["id"] == str(invitation.id)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve"],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_retrieve_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to retrieve a document invitation
|
||||
when the action is explicitly enabled.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
invitation = factories.InvitationFactory(document=document)
|
||||
|
||||
response = client.get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/{invitation.id!s}/"
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == str(invitation.id)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "create"],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_create_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to create a document invitation
|
||||
when the create action is explicitly enabled.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/",
|
||||
{"email": "invited@example.com", "role": models.RoleChoices.READER},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["email"] == "invited@example.com"
|
||||
assert data["role"] == models.RoleChoices.READER
|
||||
assert str(data["document"]) == str(document.id)
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "partial_update"],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_partial_update_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to partially update a document invitation
|
||||
when the partial_update action is explicitly enabled.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
invitation = factories.InvitationFactory(
|
||||
document=document, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
response = client.patch(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/{invitation.id!s}/",
|
||||
{"role": models.RoleChoices.EDITOR},
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["role"] == models.RoleChoices.EDITOR
|
||||
assert data["email"] == invitation.email
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
],
|
||||
},
|
||||
"document_invitation": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "destroy"],
|
||||
},
|
||||
},
|
||||
)
|
||||
def test_external_api_document_invitations_delete_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to delete a document invitation
|
||||
when the destroy action is explicitly enabled.
|
||||
"""
|
||||
reload_urls()
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory()
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
invitation = factories.InvitationFactory(document=document)
|
||||
|
||||
response = client.delete(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/invitations/{invitation.id!s}/",
|
||||
)
|
||||
|
||||
assert response.status_code == 204
|
||||
+105
@@ -0,0 +1,105 @@
|
||||
"""
|
||||
Tests for the Resource Server API for document link configurations.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_documents_link_configuration_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to update the link configuration of a document
|
||||
from a resource server.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.put(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/link-configuration/"
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"link_configuration",
|
||||
],
|
||||
},
|
||||
},
|
||||
COLLABORATION_API_URL="http://example.com/",
|
||||
COLLABORATION_SERVER_SECRET="secret-token",
|
||||
)
|
||||
@patch("core.services.collaboration_services.CollaborationService.reset_connections")
|
||||
def test_external_api_documents_link_configuration_can_be_allowed(
|
||||
mock_reset, user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to update the link configuration of a document
|
||||
from a resource server when the corresponding action is enabled in EXTERNAL_API settings.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
# attempt to change reach/role to a valid combination
|
||||
new_data = {
|
||||
"link_reach": models.LinkReachChoices.PUBLIC,
|
||||
"link_role": models.LinkRoleChoices.EDITOR,
|
||||
}
|
||||
|
||||
response = client.put(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/link-configuration/",
|
||||
new_data,
|
||||
format="json",
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# verify the document was updated in the database
|
||||
document.refresh_from_db()
|
||||
assert document.link_reach == models.LinkReachChoices.PUBLIC
|
||||
assert document.link_role == models.LinkRoleChoices.EDITOR
|
||||
@@ -0,0 +1,94 @@
|
||||
"""
|
||||
Tests for the Resource Server API for document media authentication.
|
||||
|
||||
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 uuid import uuid4
|
||||
|
||||
from django.core.files.storage import default_storage
|
||||
from django.test import override_settings
|
||||
from django.utils import timezone
|
||||
|
||||
import pytest
|
||||
from freezegun import freeze_time
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
from core.enums import DocumentAttachmentStatus
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_documents_media_auth_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to access media auth endpoints
|
||||
from a resource server by default.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
response = client.get("/external_api/v1.0/documents/media-auth/")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"media_auth",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_media_auth_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to access media auth endpoints
|
||||
from a resource server when the media-auth action is enabled in EXTERNAL_API settings.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document_id = uuid4()
|
||||
filename = f"{uuid4()!s}.jpg"
|
||||
key = f"{document_id!s}/attachments/{filename:s}"
|
||||
media_url = f"http://localhost/media/{key:s}"
|
||||
|
||||
default_storage.connection.meta.client.put_object(
|
||||
Bucket=default_storage.bucket_name,
|
||||
Key=key,
|
||||
Body=BytesIO(b"my prose"),
|
||||
ContentType="text/plain",
|
||||
Metadata={"status": DocumentAttachmentStatus.READY},
|
||||
)
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
id=document_id, link_reach=models.LinkReachChoices.RESTRICTED, attachments=[key]
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.READER
|
||||
)
|
||||
|
||||
now = timezone.now()
|
||||
with freeze_time(now):
|
||||
response = client.get(
|
||||
"/external_api/v1.0/documents/media-auth/", HTTP_X_ORIGINAL_URL=media_url
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
@@ -0,0 +1,163 @@
|
||||
"""
|
||||
Tests for the Resource Server API for document versions.
|
||||
|
||||
Not testing external API endpoints that are already tested in the /api
|
||||
because the resource server viewsets inherit from the api viewsets.
|
||||
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
from django.test import override_settings
|
||||
|
||||
import pytest
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from core import factories, models
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
def test_external_api_documents_versions_list_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to list the versions of a document
|
||||
from a resource server by default.
|
||||
"""
|
||||
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(
|
||||
link_reach=models.LinkReachChoices.RESTRICTED,
|
||||
creator=user_specific_sub,
|
||||
)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document,
|
||||
user=user_specific_sub,
|
||||
role=models.RoleChoices.OWNER,
|
||||
)
|
||||
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/versions/")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_external_api_documents_versions_detail_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to retrieve a specific version of a document
|
||||
from a resource server by default.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/versions/1234/"
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
# Overrides
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": ["list", "retrieve", "children", "versions_list"],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_versions_list_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to list version of a document from a resource server
|
||||
when the versions action is enabled in EXTERNAL_API settings.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
# Add new versions to the document
|
||||
for i in range(3):
|
||||
document.content = f"new content {i:d}"
|
||||
document.save()
|
||||
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/versions/")
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
content = response.json()
|
||||
assert content["count"] == 2
|
||||
|
||||
|
||||
@override_settings(
|
||||
EXTERNAL_API={
|
||||
"documents": {
|
||||
"enabled": True,
|
||||
"actions": [
|
||||
"list",
|
||||
"retrieve",
|
||||
"children",
|
||||
"versions_list",
|
||||
"versions_detail",
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
def test_external_api_documents_versions_detail_can_be_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD be allowed to retrieve a specific version of a document
|
||||
from a resource server when the versions_detail action is enabled.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
document = factories.DocumentFactory(link_reach=models.LinkReachChoices.RESTRICTED)
|
||||
factories.UserDocumentAccessFactory(
|
||||
document=document, user=user_specific_sub, role=models.RoleChoices.OWNER
|
||||
)
|
||||
|
||||
# ensure access datetime is earlier than versions (minio precision is one second)
|
||||
time.sleep(1)
|
||||
|
||||
# create several versions, spacing them out to get distinct LastModified values
|
||||
for i in range(3):
|
||||
document.content = f"new content {i:d}"
|
||||
document.save()
|
||||
time.sleep(1)
|
||||
|
||||
# call the list endpoint and verify basic structure
|
||||
response = client.get(f"/external_api/v1.0/documents/{document.id!s}/versions/")
|
||||
assert response.status_code == 200
|
||||
|
||||
content = response.json()
|
||||
# count should reflect two saved versions beyond the original
|
||||
assert content.get("count") == 2
|
||||
|
||||
# pick the first version returned by the list (should be accessible)
|
||||
version_id = content.get("versions")[0]["version_id"]
|
||||
|
||||
detailed_response = client.get(
|
||||
f"/external_api/v1.0/documents/{document.id!s}/versions/{version_id}/"
|
||||
)
|
||||
assert detailed_response.status_code == 200
|
||||
assert detailed_response.json()["content"] == "new content 1"
|
||||
@@ -0,0 +1,158 @@
|
||||
"""
|
||||
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_external_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_external_api_users_me_connected_not_allowed():
|
||||
"""
|
||||
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_external_api_users_me_connected_allowed(
|
||||
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_external_api_users_me_connected_with_invalid_token_not_allowed(
|
||||
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_external_api_users_list_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be 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_external_api_users_retrieve_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to retrieve a specific user 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_external_api_users_put_patch_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to update or patch a user from a resource server.
|
||||
"""
|
||||
client = APIClient()
|
||||
client.credentials(HTTP_AUTHORIZATION=f"Bearer {user_token}")
|
||||
|
||||
other_user = factories.UserFactory()
|
||||
|
||||
new_user_values = {
|
||||
k: v
|
||||
for k, v in serializers.UserSerializer(
|
||||
instance=factories.UserFactory()
|
||||
).data.items()
|
||||
if v is not None
|
||||
}
|
||||
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_external_api_users_delete_not_allowed(
|
||||
user_token, resource_server_backend, user_specific_sub
|
||||
):
|
||||
"""
|
||||
Connected users SHOULD NOT be allowed to delete a user 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
|
||||
Reference in New Issue
Block a user