(backend) add server-to-server API endpoint to create documents

We want trusted external applications to be able to create documents
via the API on behalf of any user. The user may or may not pre-exist
in our database and should be notified of the document creation by
email.
This commit is contained in:
Samuel Paccoud - DINUM
2024-12-01 11:25:01 +01:00
committed by aleb_the_flash
parent 47e23bff90
commit 5cdd06d432
17 changed files with 1072 additions and 218 deletions

View File

@@ -171,10 +171,11 @@ def test_api_document_accesses_create_authenticated_administrator(via, mock_user
email = mail.outbox[0]
assert email.to == [other_user["email"]]
email_content = " ".join(email.body.split())
assert f"{user.full_name} shared a document with you!" in email_content
assert (
f"{user.full_name} shared a document with you: {document.title}"
in email_content
)
f"{user.full_name} ({user.email}) invited you with the role ``{role}`` "
f"on the following document: {document.title}"
) in email_content
assert "docs/" + str(document.id) + "/" in email_content
@@ -228,8 +229,9 @@ def test_api_document_accesses_create_authenticated_owner(via, mock_user_teams):
email = mail.outbox[0]
assert email.to == [other_user["email"]]
email_content = " ".join(email.body.split())
assert f"{user.full_name} shared a document with you!" in email_content
assert (
f"{user.full_name} shared a document with you: {document.title}"
in email_content
)
f"{user.full_name} ({user.email}) invited you with the role ``{role}`` "
f"on the following document: {document.title}"
) in email_content
assert "docs/" + str(document.id) + "/" in email_content

View File

@@ -402,10 +402,11 @@ def test_api_document_invitations_create_privileged_members(
email = mail.outbox[0]
assert email.to == ["guest@example.com"]
email_content = " ".join(email.body.split())
assert f"{user.full_name} shared a document with you!" in email_content
assert (
f"{user.full_name} shared a document with you: {document.title}"
in email_content
)
f"{user.full_name} ({user.email}) invited you with the role ``{invited}`` "
f"on the following document: {document.title}"
) in email_content
else:
assert models.Invitation.objects.exists() is False
@@ -452,10 +453,7 @@ def test_api_document_invitations_create_email_from_content_language():
assert email.to == ["guest@example.com"]
email_content = " ".join(email.body.split())
assert (
f"{user.full_name} a partagé un document avec vous: {document.title}"
in email_content
)
assert f"{user.full_name} a partagé un document avec vous !" in email_content
def test_api_document_invitations_create_email_from_content_language_not_supported():
@@ -494,10 +492,7 @@ def test_api_document_invitations_create_email_from_content_language_not_support
assert email.to == ["guest@example.com"]
email_content = " ".join(email.body.split())
assert (
f"{user.full_name} shared a document with you: {document.title}"
in email_content
)
assert f"{user.full_name} shared a document with you!" in email_content
def test_api_document_invitations_create_email_full_name_empty():
@@ -535,10 +530,10 @@ def test_api_document_invitations_create_email_full_name_empty():
assert email.to == ["guest@example.com"]
email_content = " ".join(email.body.split())
assert f"{user.email} shared a document with you: {document.title}" in email_content
assert f"{user.email} shared a document with you!" in email_content
assert (
f'{user.email} invited you with the role "reader" on the '
f"following document : {document.title}" in email_content
f"{user.email.capitalize()} invited you with the role ``reader`` on the "
f"following document: {document.title}" in email_content
)

View File

@@ -0,0 +1,364 @@
"""
Tests for Documents API endpoint in impress's core app: create
"""
# pylint: disable=W0621
from unittest.mock import patch
from django.core import mail
from django.test import override_settings
import pytest
from rest_framework.test import APIClient
from core import factories
from core.models import Document, Invitation, User
from core.services.converter_services import ConversionError, YdocConverter
pytestmark = pytest.mark.django_db
@pytest.fixture
def mock_convert_markdown():
"""Mock YdocConverter.convert_markdown to return a converted content."""
with patch.object(
YdocConverter,
"convert_markdown",
return_value={"content": "Converted document content"},
) as mock:
yield mock
def test_api_documents_create_for_owner_missing_token():
"""Requests with no token should not be allowed to create documents for owner."""
data = {
"title": "My Document",
"content": "Document content",
"sub": "123",
"email": "john.doe@example.com",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/", data, format="json"
)
assert response.status_code == 401
assert not Document.objects.exists()
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_invalid_token():
"""Requests with an invalid token should not be allowed to create documents for owner."""
data = {
"title": "My Document",
"content": "Document content",
"sub": "123",
"email": "john.doe@example.com",
"language": "fr",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer InvalidToken",
)
assert response.status_code == 401
assert not Document.objects.exists()
def test_api_documents_create_for_owner_authenticated_forbidden():
"""
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(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
)
assert response.status_code == 401
assert not Document.objects.exists()
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_missing_sub():
"""Requests with no sub should not be allowed to create documents for owner."""
data = {
"title": "My Document",
"content": "Document content",
"email": "john.doe@example.com",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
assert response.status_code == 400
assert not Document.objects.exists()
assert response.json() == {"sub": ["This field is required."]}
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_missing_email():
"""Requests with no email should not be allowed to create documents for owner."""
data = {
"title": "My Document",
"content": "Document content",
"sub": "123",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
assert response.status_code == 400
assert not Document.objects.exists()
assert response.json() == {"email": ["This field is required."]}
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_invalid_sub():
"""Requests with an invalid sub should not be allowed to create documents for owner."""
data = {
"title": "My Document",
"content": "Document content",
"sub": "123!!",
"email": "john.doe@example.com",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
assert response.status_code == 400
assert not Document.objects.exists()
assert response.json() == {
"sub": [
"Enter a valid sub. This value may contain only letters, "
"numbers, and @/./+/-/_/: characters."
]
}
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_existing(mock_convert_markdown):
"""It should be possible to create a document on behalf of a pre-existing user."""
user = factories.UserFactory(language="en-us")
data = {
"title": "My Document",
"content": "Document content",
"sub": str(user.sub),
"email": "irrelevant@example.com", # Should be ignored since the user already exists
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
assert response.status_code == 201
mock_convert_markdown.assert_called_once_with("Document content")
document = Document.objects.get()
assert response.json() == {"id": str(document.id)}
assert document.title == "My Document"
assert document.content == "Converted document content"
assert document.creator == user
assert document.accesses.filter(user=user, role="owner").exists()
assert Invitation.objects.exists() is False
assert len(mail.outbox) == 1
email = mail.outbox[0]
assert email.to == [user.email]
assert email.subject == "A new document was created on your behalf!"
email_content = " ".join(email.body.split())
assert "A new document was created on your behalf!" in email_content
assert (
"You have been granted ownership of a new document: My Document"
) in email_content
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_new_user(mock_convert_markdown):
"""
It should be possible to create a document on behalf of new users by
passing only their email address.
"""
data = {
"title": "My Document",
"content": "Document content",
"sub": "123",
"email": "john.doe@example.com", # Should be used to create a new user
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
assert response.status_code == 201
mock_convert_markdown.assert_called_once_with("Document content")
document = Document.objects.get()
assert response.json() == {"id": str(document.id)}
assert document.title == "My Document"
assert document.content == "Converted document content"
assert document.creator is None
assert document.accesses.exists() is False
invitation = Invitation.objects.get()
assert invitation.email == "john.doe@example.com"
assert invitation.role == "owner"
assert len(mail.outbox) == 1
email = mail.outbox[0]
assert email.to == ["john.doe@example.com"]
assert email.subject == "A new document was created on your behalf!"
email_content = " ".join(email.body.split())
assert "A new document was created on your behalf!" in email_content
assert (
"You have been granted ownership of a new document: My Document"
) in email_content
# The creator field on the document should be set when the user is created
user = User.objects.create(email="john.doe@example.com", password="!")
document.refresh_from_db()
assert document.creator == user
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_with_custom_language(mock_convert_markdown):
"""
Test creating a document with a specific language.
Useful if the remote server knows the user's language.
"""
data = {
"title": "My Document",
"content": "Document content",
"sub": "123",
"email": "john.doe@example.com",
"language": "fr-fr",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
assert response.status_code == 201
mock_convert_markdown.assert_called_once_with("Document content")
assert len(mail.outbox) == 1
email = mail.outbox[0]
assert email.to == ["john.doe@example.com"]
assert email.subject == "Un nouveau document a été créé pour vous !"
email_content = " ".join(email.body.split())
assert "Un nouveau document a été créé pour vous !" in email_content
assert (
"Vous avez été déclaré propriétaire d'un nouveau document : My Document"
) in email_content
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_with_custom_subject_and_message(
mock_convert_markdown,
):
"""It should be possible to customize the subject and message of the invitation email."""
data = {
"title": "My Document",
"content": "Document content",
"sub": "123",
"email": "john.doe@example.com",
"message": "mon message spécial",
"subject": "mon sujet spécial !",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
assert response.status_code == 201
mock_convert_markdown.assert_called_once_with("Document content")
assert len(mail.outbox) == 1
email = mail.outbox[0]
assert email.to == ["john.doe@example.com"]
assert email.subject == "Mon sujet spécial !"
email_content = " ".join(email.body.split())
assert "Mon sujet spécial !" in email_content
assert "Mon message spécial" in email_content
@override_settings(SERVER_TO_SERVER_API_TOKENS=["DummyToken"])
def test_api_documents_create_for_owner_with_converter_exception(
mock_convert_markdown,
):
"""It should be possible to customize the subject and message of the invitation email."""
mock_convert_markdown.side_effect = ConversionError("Conversion failed")
data = {
"title": "My Document",
"content": "Document content",
"sub": "123",
"email": "john.doe@example.com",
"message": "mon message spécial",
"subject": "mon sujet spécial !",
}
response = APIClient().post(
"/api/v1.0/documents/create-for-owner/",
data,
format="json",
HTTP_AUTHORIZATION="Bearer DummyToken",
)
mock_convert_markdown.assert_called_once_with("Document content")
assert response.status_code == 500
assert response.json() == {"detail": "could not convert content"}

View File

@@ -33,11 +33,8 @@ def test_models_documents_id_unique():
def test_models_documents_creator_required():
"""The "creator" field should be required."""
with pytest.raises(ValidationError) as excinfo:
models.Document.objects.create()
assert excinfo.value.message_dict["creator"] == ["This field cannot be null."]
"""No field should be required on the Document model."""
models.Document.objects.create()
def test_models_documents_title_null():
@@ -430,8 +427,8 @@ def test_models_documents__email_invitation__success():
assert len(mail.outbox) == 0
sender = factories.UserFactory(full_name="Test Sender", email="sender@example.com")
document.email_invitation(
"en", "guest@example.com", models.RoleChoices.EDITOR, sender
document.send_invitation_email(
"guest@example.com", models.RoleChoices.EDITOR, sender, "en"
)
# pylint: disable-next=no-member
@@ -444,8 +441,8 @@ def test_models_documents__email_invitation__success():
email_content = " ".join(email.body.split())
assert (
f'Test Sender (sender@example.com) invited you with the role "editor" '
f"on the following document : {document.title}" in email_content
f"Test Sender (sender@example.com) invited you with the role ``editor`` "
f"on the following document: {document.title}" in email_content
)
assert f"docs/{document.id}/" in email_content
@@ -462,11 +459,11 @@ def test_models_documents__email_invitation__success_fr():
sender = factories.UserFactory(
full_name="Test Sender2", email="sender2@example.com"
)
document.email_invitation(
"fr-fr",
document.send_invitation_email(
"guest2@example.com",
models.RoleChoices.OWNER,
sender,
"fr-fr",
)
# pylint: disable-next=no-member
@@ -479,7 +476,7 @@ def test_models_documents__email_invitation__success_fr():
email_content = " ".join(email.body.split())
assert (
f'Test Sender2 (sender2@example.com) vous a invité avec le rôle "propriétaire" '
f"Test Sender2 (sender2@example.com) vous a invité avec le rôle ``propriétaire`` "
f"sur le document suivant : {document.title}" in email_content
)
assert f"docs/{document.id}/" in email_content
@@ -498,11 +495,11 @@ def test_models_documents__email_invitation__failed(mock_logger, _mock_send_mail
assert len(mail.outbox) == 0
sender = factories.UserFactory()
document.email_invitation(
"en",
document.send_invitation_email(
"guest3@example.com",
models.RoleChoices.ADMIN,
sender,
"en",
)
# No email has been sent
@@ -514,9 +511,9 @@ def test_models_documents__email_invitation__failed(mock_logger, _mock_send_mail
(
_,
email,
emails,
exception,
) = mock_logger.call_args.args
assert email == "guest3@example.com"
assert emails == ["guest3@example.com"]
assert isinstance(exception, smtplib.SMTPException)

View File

@@ -144,7 +144,7 @@ def test_models_invitationd_new_user_filter_expired_invitations():
).exists()
@pytest.mark.parametrize("num_invitations, num_queries", [(0, 3), (1, 6), (20, 6)])
@pytest.mark.parametrize("num_invitations, num_queries", [(0, 3), (1, 7), (20, 7)])
def test_models_invitationd_new_userd_user_creation_constant_num_queries(
django_assert_num_queries, num_invitations, num_queries
):

View File

@@ -0,0 +1,145 @@
"""Test converter services."""
from unittest.mock import MagicMock, patch
import pytest
import requests
from core.services.converter_services import (
InvalidResponseError,
MissingContentError,
ServiceUnavailableError,
ValidationError,
YdocConverter,
)
def test_auth_header(settings):
"""Test authentication header generation."""
settings.CONVERSION_API_KEY = "test-key"
converter = YdocConverter()
assert converter.auth_header == "Bearer test-key"
def test_convert_markdown_empty_text():
"""Should raise ValidationError when text is empty."""
converter = YdocConverter()
with pytest.raises(ValidationError, match="Input text cannot be empty"):
converter.convert_markdown("")
@patch("requests.post")
def test_convert_markdown_service_unavailable(mock_post):
"""Should raise ServiceUnavailableError when service is unavailable."""
converter = YdocConverter()
mock_post.side_effect = requests.RequestException("Connection error")
with pytest.raises(
ServiceUnavailableError,
match="Failed to connect to conversion service",
):
converter.convert_markdown("test text")
@patch("requests.post")
def test_convert_markdown_http_error(mock_post):
"""Should raise ServiceUnavailableError when HTTP error occurs."""
converter = YdocConverter()
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("HTTP Error")
mock_post.return_value = mock_response
with pytest.raises(
ServiceUnavailableError,
match="Failed to connect to conversion service",
):
converter.convert_markdown("test text")
@patch("requests.post")
def test_convert_markdown_invalid_json_response(mock_post):
"""Should raise InvalidResponseError when response is not valid JSON."""
converter = YdocConverter()
mock_response = MagicMock()
mock_response.json.side_effect = ValueError("Invalid JSON")
mock_post.return_value = mock_response
with pytest.raises(
InvalidResponseError,
match="Could not parse conversion service response",
):
converter.convert_markdown("test text")
@patch("requests.post")
def test_convert_markdown_missing_content_field(mock_post, settings):
"""Should raise MissingContentError when response is missing required field."""
settings.CONVERSION_API_CONTENT_FIELD = "expected_field"
converter = YdocConverter()
mock_response = MagicMock()
mock_response.json.return_value = {"wrong_field": "content"}
mock_post.return_value = mock_response
with pytest.raises(
MissingContentError,
match="Response missing required field: expected_field",
):
converter.convert_markdown("test text")
@patch("requests.post")
def test_convert_markdown_full_integration(mock_post, settings):
"""Test full integration with all settings."""
settings.CONVERSION_API_URL = "http://test.com"
settings.CONVERSION_API_KEY = "test-key"
settings.CONVERSION_API_TIMEOUT = 5
settings.CONVERSION_API_CONTENT_FIELD = "content"
converter = YdocConverter()
expected_content = {"converted": "content"}
mock_response = MagicMock()
mock_response.json.return_value = {"content": expected_content}
mock_post.return_value = mock_response
result = converter.convert_markdown("test markdown")
assert result == expected_content
mock_post.assert_called_once_with(
"http://test.com",
json={"content": "test markdown"},
headers={
"Authorization": "Bearer test-key",
"Content-Type": "application/json",
},
timeout=5,
)
@patch("requests.post")
def test_convert_markdown_timeout(mock_post):
"""Should raise ServiceUnavailableError when request times out."""
converter = YdocConverter()
mock_post.side_effect = requests.Timeout("Request timed out")
with pytest.raises(
ServiceUnavailableError,
match="Failed to connect to conversion service",
):
converter.convert_markdown("test text")
def test_convert_markdown_none_input():
"""Should raise ValidationError when input is None."""
converter = YdocConverter()
with pytest.raises(ValidationError, match="Input text cannot be empty"):
converter.convert_markdown(None)