diff --git a/app/signals/apps/api/generics/permissions.py b/app/signals/apps/api/generics/permissions.py index 725035253..7f7963c99 100644 --- a/app/signals/apps/api/generics/permissions.py +++ b/app/signals/apps/api/generics/permissions.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: MPL-2.0 -# Copyright (C) 2019 - 2021 Gemeente Amsterdam +# Copyright (C) 2019 - 2023 Gemeente Amsterdam from django.views import View from rest_framework import exceptions from rest_framework.permissions import BasePermission, DjangoModelPermissions @@ -160,3 +160,22 @@ def has_object_permission(self, request: Request, view: View, obj: Reporter) -> user=request.user, permission='signals.sia_can_view_all_categories', ) + + +class CanCreateI18NextTranslationFile(BasePermission): + def has_permission(self, request: Request, *args: set, **kwargs: dict) -> bool: + """ + Check if the user has permission to create an I18Next translation file. + + Args: + request (Request): The incoming request. + **kwargs (dict): Additional keyword arguments. + + Returns: + bool: True if the user has permission, False otherwise. + """ + # Allow access to root user or users with the specific permission + return ( + request.user.is_superuser or + request.user.has_perm('signals.sia_add_i18next_translation_file') + ) diff --git a/app/signals/apps/api/tests/test_translation_endpoints.py b/app/signals/apps/api/tests/test_translation_endpoints.py new file mode 100644 index 000000000..e34228837 --- /dev/null +++ b/app/signals/apps/api/tests/test_translation_endpoints.py @@ -0,0 +1,260 @@ +# SPDX-License-Identifier: MPL-2.0 +# Copyright (C) 2023 Gemeente Amsterdam +import json +import tempfile +from typing import Any + +from django.contrib.auth.models import Permission, User +from django.core.files.base import ContentFile +from django.core.files.storage import default_storage +from rest_framework.test import APITestCase + +from signals.apps.api.views.translations import I18NEXT_TRANSLATION_FILE_PATH + +TEST_TRANSLATION_DATA: dict[str, Any] = { + "en": { + "common": { + "welcome": "Welcome to our website!", + "language": "Language", + "english": "English", + "dutch": "Dutch", + }, + "errors": { + "notFound": "Page not found", + "serverError": "Server error occurred" + }, + }, + "nl": { + "common": { + "welcome": "Welkom op onze website!", + "language": "Taal", + "english": "Engels", + "dutch": "nederlands", + }, + "errors": { + "notFound": "Pagina niet gevonden", + "serverError": "Er is een serverfout opgetreden" + }, + }, +} + + +class PrivateCreateI18NextTranslationFileView(APITestCase): + endpoint = '/signals/v1/private/translations/' + + def setUp(self) -> None: + # Create a user with the necessary permission + self.add_i18next_translation_file_permission: Permission = Permission.objects.get( + codename='sia_add_i18next_translation_file' + ) + self.user: User = User.objects.create_user( + username='testuser', + password='testpassword' + ) + self.user.user_permissions.add( + self.add_i18next_translation_file_permission + ) + + self.client.force_authenticate(user=self.user) + + def tearDown(self) -> None: + self.client.logout() + + def test_post_translation_new_file(self) -> None: + """ + Posting a translation file with permission. + + 1. Check that no file called "i18next/translations.js" + exists initially + 2. Post the JSON blob to the specified URL + 3. Check that a file has been created in storage + at "i18next/translations.js" + 4. Check that the contents of the file match the posted JSON blob + """ + with tempfile.TemporaryDirectory() as tmpdir: + with self.settings(MEDIA_ROOT=tmpdir): + + # Check that no file called "i18next/translations.js" + # exists initially + self.assertFalse( + default_storage.exists(I18NEXT_TRANSLATION_FILE_PATH) + ) + + # Post the JSON blob + response = self.client.post( + self.endpoint, + data=json.dumps(TEST_TRANSLATION_DATA), + content_type="application/json" + ) + + # Check that a file has been created + self.assertEqual(response.status_code, 201) + self.assertTrue( + default_storage.exists(I18NEXT_TRANSLATION_FILE_PATH) + ) + + # Check that the contents of the file match the posted JSON blob + with default_storage.open(I18NEXT_TRANSLATION_FILE_PATH) as file: + file_contents: str = file.read().decode() + self.assertEqual( + json.loads(file_contents), + TEST_TRANSLATION_DATA + ) + + def test_post_translation_file_already_exists(self) -> None: + """ + Posting a translation file with an existing file. + + 1. Create an existing translation file + 2. Post the JSON blob to the specified URL + 3. Check that a file still exists at "i18next/translations.js" + 4. Check that the contents of the file match the posted JSON blob + 5. Check that the contents of the file is no longer the same + as before + """ + with tempfile.TemporaryDirectory() as tmpdir: + with self.settings(MEDIA_ROOT=tmpdir): + + # Create an existing translation file + default_storage.save( + I18NEXT_TRANSLATION_FILE_PATH, + ContentFile("Existing file content") + ) + + # Post the JSON blob + response = self.client.post( + self.endpoint, + data=json.dumps(TEST_TRANSLATION_DATA), + content_type="application/json" + ) + + # Check that a file still exists + self.assertEqual(response.status_code, 201) + self.assertTrue( + default_storage.exists(I18NEXT_TRANSLATION_FILE_PATH) + ) + + # Check that the contents of the file match the posted JSON blob + with default_storage.open(I18NEXT_TRANSLATION_FILE_PATH) as file: + file_contents: str = file.read().decode() + self.assertEqual( + json.loads(file_contents), + TEST_TRANSLATION_DATA + ) + + # Check that the contents of the file are no longer the + # same as before + self.assertNotEqual( + file_contents, + "Existing file content" + ) + + def test_post_no_user_logged_in(self) -> None: + """ + Posting without a logged-in user. + + 1. Log the user out + 2. Post the JSON blob to the specified URL + 3. Check that the request is not allowed because the user is + not logged in (HTTP 401 Unauthorized) + """ + with tempfile.TemporaryDirectory() as tmpdir: + with self.settings(MEDIA_ROOT=tmpdir): + + # Log the user out + self.client.logout() + + # Post the JSON blob + response = self.client.post( + self.endpoint, + data=json.dumps(TEST_TRANSLATION_DATA), + content_type="application/json" + ) + + # Check that the request is not allowed because the user is + # not logged in + self.assertEqual(response.status_code, 401) + + def test_post_no_permission(self) -> None: + """ + Posting without the required permission. + + 1. Remove the "add_i18next_translation_file" permission from + the user + 2. Post the JSON blob to the specified URL + 3. Check that the request is not allowed because the user doesn't + have the required permission (HTTP 403 Forbidden) + """ + with tempfile.TemporaryDirectory() as tmpdir: + with self.settings(MEDIA_ROOT=tmpdir): + + # Remove the permission from the user + self.user.user_permissions.remove( + self.add_i18next_translation_file_permission + ) + + # Post the JSON blob + response = self.client.post( + self.endpoint, + data=json.dumps(TEST_TRANSLATION_DATA), + content_type="application/json" + ) + + # Check that the request is not allowed because the user doesn't + # have the required permission + self.assertEqual(response.status_code, 403) + + +class TestPublicRetrieveI18NextTranslationFileView(APITestCase): + endpoint = '/signals/v1/public/translations.json' + + def test_retrieve_translation_file(self) -> None: + """ + Retrieving the translation file + + 1. Generate test file + 3. Retrieve the contents of the translation file + 4. Check that the retrieved contents match the posted JSON blob + """ + with tempfile.TemporaryDirectory() as tmpdir: + with self.settings(MEDIA_ROOT=tmpdir): + + # Create an existing translation file + default_storage.save( + I18NEXT_TRANSLATION_FILE_PATH, + ContentFile(json.dumps(TEST_TRANSLATION_DATA)) + ) + + # Retrieve the contents of the file + response_retrieve = self.client.get(self.endpoint) + + self.assertEqual(response_retrieve.status_code, 200) + + # Check that the retrieved content matches the posted JSON blob + file_contents = response_retrieve.streaming_content + content_file = ContentFile(b"".join(file_contents)) + + self.assertEqual( + json.loads(content_file.read().decode()), + TEST_TRANSLATION_DATA + ) + + def test_retrieve_file_does_no_exist(self) -> None: + """ + Retrieving the translation file does not exist. + + 1. Attempt to retrieve a file that does not exist + 2. Check that a 404 response is raised (HTTP 404 Not Found) + """ + with tempfile.TemporaryDirectory() as tmpdir: + with self.settings(MEDIA_ROOT=tmpdir): + + # Delete the translation file if it already exists + if default_storage.exists(I18NEXT_TRANSLATION_FILE_PATH): + default_storage.delete(I18NEXT_TRANSLATION_FILE_PATH) + + # Attempt to retrieve a nonexistent file + response = self.client.get(self.endpoint) + + # Check that a 404 response is raised + self.assertEqual(response.status_code, 404) diff --git a/app/signals/apps/api/urls.py b/app/signals/apps/api/urls.py index f24dd3221..4a08dccef 100644 --- a/app/signals/apps/api/urls.py +++ b/app/signals/apps/api/urls.py @@ -34,6 +34,10 @@ StatusMessagesCategoryPositionViewSet, StatusMessagesViewSet ) +from signals.apps.api.views.translations import ( + PrivateCreateI18NextTranslationFileView, + PublicRetrieveI18NextTranslationFileView +) from signals.apps.search.rest_framework.views import SearchView, StatusMessageSearchView from signals.apps.users.rest_framework.views import ( AutocompleteUsernameListView, @@ -101,6 +105,9 @@ PublicSignalAttachmentsViewSet.as_view({'post': 'create'}), name='public-signals-attachments'), re_path(r'questions/?$', PublicQuestionViewSet.as_view({'get': 'list'}), name='question-detail'), path('reporter/verify-email', EmailVerificationView.as_view(), name='email-verification'), + path('translations.json', PublicRetrieveI18NextTranslationFileView.as_view(), + name='public-i18next-latest-translations'), + ])), # Private additions @@ -144,7 +151,9 @@ 'put': 'update', 'patch': 'update', 'delete': 'destroy'}), - name='private-category-icon') + name='private-category-icon'), + + path('translations/', PrivateCreateI18NextTranslationFileView.as_view(), name='private-i18next-translations') ])), # Feedback diff --git a/app/signals/apps/api/views/translations.py b/app/signals/apps/api/views/translations.py new file mode 100644 index 000000000..e15a52eab --- /dev/null +++ b/app/signals/apps/api/views/translations.py @@ -0,0 +1,72 @@ +# SPDX-License-Identifier: MPL-2.0 +# Copyright (C) 2023 Gemeente Amsterdam +import json +import mimetypes + +from django.core.files.base import ContentFile +from django.core.files.storage import default_storage +from django.http import FileResponse +from rest_framework.request import Request +from rest_framework.response import Response +from rest_framework.status import HTTP_201_CREATED, HTTP_404_NOT_FOUND +from rest_framework.views import APIView + +from signals.apps.api.generics.permissions import CanCreateI18NextTranslationFile +from signals.auth.backend import JWTAuthBackend + +I18NEXT_TRANSLATION_FILE_PATH = 'i18next/translations.json' + + +class PrivateCreateI18NextTranslationFileView(APIView): + authentication_classes = (JWTAuthBackend, ) + permission_classes = (CanCreateI18NextTranslationFile, ) + + def post(self, request: Request) -> Response: + """ + Create an I18Next translation file from the JSON data. + + Args: + request (Request): The incoming request containing JSON data. + + Returns: + Response: A success response with the created JSON data. + """ + # Create ContentFile from JSON data and save to storage + latest_file_content = json.dumps(request.data) + + # Ensure new_content is already in bytes format (if not, you can encode it) + if not isinstance(latest_file_content, bytes): + latest_file_content = latest_file_content.encode('utf-8') + + if default_storage.exists(I18NEXT_TRANSLATION_FILE_PATH): + # Overwrite the JSON data to storage + with default_storage.open(I18NEXT_TRANSLATION_FILE_PATH, 'wb') as translation_file: + translation_file.write(latest_file_content) + else: + # Save JSON data to storage + default_storage.save(I18NEXT_TRANSLATION_FILE_PATH, ContentFile(latest_file_content)) + + return Response(data=request.data, status=HTTP_201_CREATED) + + +class PublicRetrieveI18NextTranslationFileView(APIView): + def get(self, request: Request, *args: set, **kwargs: dict) -> Response | FileResponse: + """ + Retrieve the latest I18Next translation file. + + Args: + request (Request): The incoming request. + *args (set): Additional positional arguments. + **kwargs (dict): Additional keyword arguments. + + Returns: + Response | FileResponse: A FileResponse with the file content or Response (404) if the file doesn't exist. + """ + # Check if the file exists in storage + if not default_storage.exists(I18NEXT_TRANSLATION_FILE_PATH): + return Response('Translation file not found.', status=HTTP_404_NOT_FOUND) + + # Determine the file's content type + content_type, _ = mimetypes.guess_type(I18NEXT_TRANSLATION_FILE_PATH) + response = FileResponse(default_storage.open(I18NEXT_TRANSLATION_FILE_PATH), content_type=content_type) + return response diff --git a/app/signals/apps/signals/models/permission.py b/app/signals/apps/signals/models/permission.py index 077601334..6bf393532 100644 --- a/app/signals/apps/signals/models/permission.py +++ b/app/signals/apps/signals/models/permission.py @@ -39,7 +39,7 @@ def sync_custom_permissions(sender, **kwargs): custom_permissions = [ # Custom permission to allow users to create i18next translation files. - ('add_i18next_translation_file', 'Can create i18next translation file') + ('sia_add_i18next_translation_file', 'Can create i18next translation file') ] for codename, name in custom_permissions: diff --git a/app/signals/apps/users/tests/rest_framework/test_permissions.py b/app/signals/apps/users/tests/rest_framework/test_permissions.py index 8f08d9806..9243d7a71 100644 --- a/app/signals/apps/users/tests/rest_framework/test_permissions.py +++ b/app/signals/apps/users/tests/rest_framework/test_permissions.py @@ -12,8 +12,8 @@ def test_get_permissions(self): self.assertEqual(response.status_code, 200) data = response.json() - self.assertEqual(data['count'], 26) - self.assertEqual(len(data['results']), 26) + self.assertEqual(data['count'], 27) + self.assertEqual(len(data['results']), 27) def test_get_permission(self): self.client.force_authenticate(user=self.sia_read_user)