diff --git a/kobo/apps/subsequences/actions/base.py b/kobo/apps/subsequences/actions/base.py index e0ddbf87ad..d0311f7b6b 100644 --- a/kobo/apps/subsequences/actions/base.py +++ b/kobo/apps/subsequences/actions/base.py @@ -4,6 +4,7 @@ from typing import Optional import jsonschema +from constance import config from django.conf import settings from django.utils import timezone @@ -193,7 +194,11 @@ def attach_action_dependency(self, action_data: dict): def check_limits(self, user: User): - if not settings.STRIPE_ENABLED or not self._is_usage_limited: + if ( + not settings.STRIPE_ENABLED + or not self._is_usage_limited + or not config.USAGE_LIMIT_ENFORCEMENT + ): return calculator = ServiceUsageCalculator(user) @@ -399,7 +404,6 @@ def revise_data( `submission` argument for future use by subclasses this method might need to be made more friendly for overriding """ - self.validate_data(action_data) self.raise_for_any_leading_underscore_key(action_data) diff --git a/kobo/apps/subsequences/models.py b/kobo/apps/subsequences/models.py index 9d664f073f..1131201eba 100644 --- a/kobo/apps/subsequences/models.py +++ b/kobo/apps/subsequences/models.py @@ -131,6 +131,9 @@ def retrieve_data( same column, the most recently accepted action result is used as the value """ + + from .utils.versioning import migrate_submission_supplementals + if (submission_root_uuid is None) == (prefetched_supplement is None): raise ValueError( 'Specify either `submission_root_uuid` or `prefetched_supplement`' @@ -139,10 +142,10 @@ def retrieve_data( if submission_root_uuid: submission_uuid = remove_uuid_prefix(submission_root_uuid) try: - supplemental_data = SubmissionExtras.objects.get( + supplemental_data = SubmissionSupplement.objects.get( asset=asset, submission_uuid=submission_uuid ).content - except SubmissionExtras.DoesNotExist: + except SubmissionSupplement.DoesNotExist: supplemental_data = None else: supplemental_data = prefetched_supplement @@ -150,7 +153,7 @@ def retrieve_data( if not supplemental_data: return {} - schema_version = supplemental_data.pop('_version') + schema_version = supplemental_data.pop('_version', None) if schema_version not in SCHEMA_VERSIONS: # TODO: raise error. Unknown version @@ -161,6 +164,7 @@ def retrieve_data( if migrated_data is None: raise InvalidAction supplemental_data = migrated_data + schema_version = supplemental_data.pop('_version') retrieved_supplemental_data = {} data_for_output = {} diff --git a/kobo/apps/subsequences/tests/api/v2/test_api.py b/kobo/apps/subsequences/tests/api/v2/test_api.py new file mode 100644 index 0000000000..ececf9ee55 --- /dev/null +++ b/kobo/apps/subsequences/tests/api/v2/test_api.py @@ -0,0 +1,837 @@ +import uuid +from datetime import datetime +from unittest.mock import MagicMock, patch +from zoneinfo import ZoneInfo + +import pytest +from constance.test import override_config +from ddt import data, ddt, unpack +from django.conf import settings +from django.urls import reverse +from django.utils import timezone +from freezegun import freeze_time +from rest_framework import status + +from kobo.apps.openrosa.apps.logger.models import Instance +from kobo.apps.openrosa.apps.logger.xform_instance_parser import add_uuid_prefix +from kobo.apps.organizations.constants import UsageType +from kobo.apps.subsequences.actions.automatic_google_transcription import ( + AutomaticGoogleTranscriptionAction, +) +from kobo.apps.subsequences.models import QuestionAdvancedFeature, SubmissionSupplement +from kobo.apps.subsequences.tests.api.v2.base import SubsequenceBaseTestCase +from kobo.apps.subsequences.tests.constants import QUESTION_SUPPLEMENT +from kpi.utils.xml import ( + edit_submission_xml, + fromstring_preserve_root_xmlns, + xml_tostring, +) + + +class SubmissionSupplementAPITestCase(SubsequenceBaseTestCase): + def setUp(self): + super().setUp() + + def _simulate_completed_transcripts(self): + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_transcription', + params=[{'language': 'en'}], + ) + + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='manual_transcription', + params=[{'language': 'en'}], + ) + + # Simulate a completed transcription, first. + mock_submission_supplement = {'_version': '20250820', 'q1': QUESTION_SUPPLEMENT} + SubmissionSupplement.objects.create( + submission_uuid=self.submission_uuid, + content=mock_submission_supplement, + asset=self.asset, + ) + + def test_get_submission_with_nonexistent_instance_404s(self): + non_existent_supplement_details_url = reverse( + self._get_endpoint('submission-supplement'), + args=[self.asset.uid, 'bad-uuid'], + ) + rr = self.client.get(non_existent_supplement_details_url) + assert rr.status_code == 404 + + def test_patch_submission_with_nonexistent_instance_404s(self): + payload = { + '_version': '20250820', + 'q1': { + 'manual_transcription': { + 'language': 'en', + 'value': 'Hello world', + }, + }, + } + non_existent_supplement_details_url = reverse( + self._get_endpoint('submission-supplement'), + args=[self.asset.uid, 'bad-uuid'], + ) + rr = self.client.patch( + non_existent_supplement_details_url, data=payload, format='json' + ) + assert rr.status_code == 404 + + def test_get_submission_after_edit(self): + # Simulate edit + instance = Instance.objects.only('pk').get(root_uuid=self.submission_uuid) + deployment = self.asset.deployment + new_uuid = str(uuid.uuid4()) + xml_parsed = fromstring_preserve_root_xmlns(instance.xml) + edit_submission_xml( + xml_parsed, + deployment.SUBMISSION_DEPRECATED_UUID_XPATH, + add_uuid_prefix(self.submission_uuid), + ) + edit_submission_xml( + xml_parsed, + deployment.SUBMISSION_ROOT_UUID_XPATH, + add_uuid_prefix(instance.root_uuid), + ) + edit_submission_xml( + xml_parsed, + deployment.SUBMISSION_CURRENT_UUID_XPATH, + add_uuid_prefix(new_uuid), + ) + instance.xml = xml_tostring(xml_parsed) + instance.uuid = new_uuid + instance.save() + assert instance.root_uuid == self.submission_uuid + + # Retrieve advanced submission schema for edited submission + rr = self.client.get(self.supplement_details_url) + assert rr.status_code == status.HTTP_200_OK + + def test_get_submission_with_null_root_uuid(self): + # Simulate an old submission (never edited) where `root_uuid` was not yet set + Instance.objects.filter(root_uuid=self.submission_uuid).update(root_uuid=None) + rr = self.client.get(self.supplement_details_url) + assert rr.status_code == status.HTTP_200_OK + + def test_asset_post_submission_extra_with_transcript(self): + payload = { + '_version': '20250820', + 'q1': { + 'manual_transcription': { + 'language': 'en', + 'value': 'Hello world', + }, + }, + } + + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='manual_transcription', + params=[{'language': 'en'}], + ) + + now = timezone.now() + now_iso = now.isoformat().replace('+00:00', 'Z') + with freeze_time(now): + with patch( + 'kobo.apps.subsequences.actions.base.uuid.uuid4', return_value='uuid1' + ): + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == status.HTTP_200_OK + expected_data = { + '_version': '20250820', + 'q1': { + 'manual_transcription': { + '_dateCreated': now_iso, + '_dateModified': now_iso, + '_versions': [ + { + '_data': { + 'language': 'en', + 'value': 'Hello world', + }, + '_dateAccepted': now_iso, + '_dateCreated': now_iso, + '_uuid': 'uuid1', + } + ], + }, + }, + } + assert response.data == expected_data + + def test_valid_manual_transcription(self): + payload = { + '_version': '20250820', + 'q1': { + 'manual_transcription': { + 'language': 'en', + 'value': 'hello world', + }, + }, + } + + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='manual_transcription', + params=[{'language': 'en'}], + ) + + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + + assert response.status_code == status.HTTP_200_OK + + def test_valid_manual_translation(self): + self._simulate_completed_transcripts() + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='manual_translation', + params=[{'language': 'es'}], + ) + + payload = { + '_version': '20250820', + 'q1': { + 'manual_translation': { + 'language': 'es', + 'value': 'hola el mundo', + }, + }, + } + + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + + assert response.status_code == status.HTTP_200_OK + + def test_valid_automatic_transcription(self): + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_transcription', + params=[{'language': 'en'}], + ) + + payload = { + '_version': '20250820', + 'q1': { + 'automatic_google_transcription': { + 'language': 'en', + }, + }, + } + + # Mock GoogleTranscriptionService and simulate completed transcription + mock_service = MagicMock() + mock_service.process_data.return_value = { + 'status': 'complete', + 'value': 'hello world', + } + + with patch( + 'kobo.apps.subsequences.actions.automatic_google_transcription.GoogleTranscriptionService', # noqa + return_value=mock_service, + ): + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + + assert response.status_code == status.HTTP_200_OK + + def test_valid_automatic_translation(self): + self._simulate_completed_transcripts() + # Set up the asset to allow automatic google translation + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_translation', + params=[{'language': 'es'}], + ) + + payload = { + '_version': '20250820', + 'q1': { + 'automatic_google_translation': { + 'language': 'es', + }, + }, + } + + # Mock GoogleTranslationService and simulate in progress translation + mock_service = MagicMock() + mock_service.process_data.return_value = { + 'status': 'complete', + 'value': 'hola el mundo', + } + + with patch( + 'kobo.apps.subsequences.actions.automatic_google_translation.GoogleTranslationService', # noqa + return_value=mock_service, + ): + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + + assert response.status_code == status.HTTP_200_OK + + def test_cannot_set_value_with_automatic_actions(self): + self._simulate_completed_transcripts() + # Set up the asset to allow automatic actions + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_translation', + params=[{'language': 'fr'}], + ) + + automatic_actions = self.asset.advanced_features_set.filter( + question_xpath='q1' + ).values_list('action', flat=True) + for automatic_action in automatic_actions: + payload = { + '_version': '20250820', + 'q1': { + automatic_action: { + 'language': 'es', + 'value': 'some text', # forbidden field + } + }, + } + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Invalid payload' in str(response.data) + + def test_cannot_accept_incomplete_automatic_translation(self): + self._simulate_completed_transcripts() + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_translation', + params=[{'language': 'fr'}], + ) + + # Try to set 'accepted' status when translation is not complete + payload = { + '_version': '20250820', + 'q1': { + 'automatic_google_translation': { + 'language': 'fr', + 'accepted': True, + } + }, + } + + # Mock GoogleTranscriptionService and simulate in progress translation + mock_service = MagicMock() + mock_service.process_data.return_value = {'status': 'in_progress'} + + with patch( + 'kobo.apps.subsequences.actions.automatic_google_translation.GoogleTranslationService', # noqa + return_value=mock_service, + ): + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Invalid payload' in str(response.data) + + def test_retrieve_does_migrate_data(self): + """ + The migration utils are already covered by other tests (`test_versioning.py), + but we need to test that the correct value is return when fetching data from + the API endpoint. + """ + self.asset.known_cols = [ + 'q1:transcript_auto_google:en', + 'q1:transcript:en', + 'q1:translation_auto_google:fr', + 'q1:translation:fr', + ] + + self.asset.advanced_features = { + 'qual': { + 'qual_survey': [ + { + 'type': 'qual_select_multiple', + 'uuid': 'b0bce6b0-9bf3-4f0f-a76e-4b3b4e9ba0e8', + 'scope': 'by_question#survey', + 'xpath': 'q1', + 'labels': {'_default': 'Multiple Choice'}, + 'choices': [ + { + 'uuid': '35793589-556e-4872-b5eb-3b75e4dc4a99', + 'labels': {'_default': 'Day'}, + }, + { + 'uuid': 'efceb7be-c120-43b4-9d6c-48c3c8d393bc', + 'labels': {'_default': 'Night'}, + }, + ], + }, + { + 'type': 'qual_select_one', + 'uuid': 'c52ba63d-3202-44bc-8f55-159983e7f0d9', + 'scope': 'by_question#survey', + 'xpath': 'q1', + 'labels': {'_default': 'Single choice'}, + 'choices': [ + { + 'uuid': '83212060-fd18-445a-b121-ad82c2e5811d', + 'labels': {'_default': 'yes'}, + }, + { + 'uuid': '394e7c6e-1468-4964-8d04-8d9bdd0d1746', + 'labels': {'_default': 'no'}, + }, + ], + }, + { + 'type': 'qual_text', + 'uuid': 'fd61cafc-9516-4063-8498-5eace89146a5', + 'scope': 'by_question#survey', + 'xpath': 'audio', + 'labels': {'_default': 'Question?'}, + }, + ] + }, + 'transcript': {'languages': ['en']}, + 'translation': {'languages': ['fr']}, + } + + old_supplement_data = { + 'q1': { + 'qual': [ + { + 'val': 'Answer', + 'type': 'qual_text', + 'uuid': 'fd61cafc-9516-4063-8498-5eace89146a5', + }, + { + 'val': '83212060-fd18-445a-b121-ad82c2e5811d', + 'type': 'qual_select_one', + 'uuid': 'c52ba63d-3202-44bc-8f55-159983e7f0d9', + }, + { + 'val': [ + '35793589-556e-4872-b5eb-3b75e4dc4a99', + 'efceb7be-c120-43b4-9d6c-48c3c8d393bc', + ], + 'type': 'qual_select_multiple', + 'uuid': 'b0bce6b0-9bf3-4f0f-a76e-4b3b4e9ba0e8', + }, + ], + 'googlets': { + 'value': 'Hello world', + 'status': 'complete', + 'regionCode': 'en-CA', + 'languageCode': 'en', + }, + 'googletx': { + 'value': 'Bonjour le monde', + 'source': 'en', + 'status': 'complete', + 'languageCode': 'fr', + }, + 'transcript': { + 'value': 'Hello world!', + 'revisions': [ + { + 'value': 'Hello world', + 'dateModified': '2025-12-11 23:57:21', + 'languageCode': 'en', + } + ], + 'dateCreated': '2025-12-12 00:03:23', + 'dateModified': '2025-12-12 00:03:23', + 'languageCode': 'en', + }, + 'translation': { + 'fr': { + 'value': 'Bonjour le monde!', + 'revisions': [], + 'dateCreated': '2025-12-12T00:04:38Z', + 'dateModified': '2025-12-12T00:04:38Z', + 'languageCode': 'fr', + } + }, + } + } + + # Simulate old data + self.asset.save( + update_fields=['advanced_features', 'known_cols'], + create_version=False, + adjust_content=False, + ) + + SubmissionSupplement.objects.create( + asset=self.asset, + submission_uuid=self.submission_uuid, + content=old_supplement_data, + ) + + frozen_datetime_now = datetime( + year=2025, + month=12, + day=15, + hour=22, + minute=22, + second=0, + tzinfo=ZoneInfo('UTC'), + ) + + result_mock_uuid_sequence = [ + 'a9a817c0-7208-4063-bab6-93c0a3a7615b', + '61d23cd7-ce2c-467b-ab26-0839226c714d', + '20dd5185-ee43-451f-8759-2f5185c3c912', + '409c690e-d148-4d80-8c73-51be941b33b0', + '49fbd509-e042-44ce-843c-db04485a0096', + '5799f662-76d7-49ab-9a1c-ae2c7d502a78', + 'c4fa8263-50c0-4252-9c9b-216ca338be13', + '64e59cc1-adaf-47a3-a068-550854d8f98f', + '909c62cf-d544-4926-8839-7f035c6c7483', + '15ccc864-0e83-48f2-be1d-dc2adb9297f4', + 'f2b4c6b1-3c6a-4a7f-9e55-1a8c2a0a7c91', + '8c9a8e44-7a3d-4c58-b7bb-5f2a1c6e5c3a', + ] + + uuid_list = [uuid.UUID(u) for u in result_mock_uuid_sequence] + + with patch('uuid.uuid4', side_effect=uuid_list): + with freeze_time(frozen_datetime_now): + response = self.client.get(self.supplement_details_url, format='json') + + assert response.status_code == status.HTTP_200_OK + + expected_response = { + 'q1': { + 'manual_transcription': { + '_dateCreated': '2025-12-12 00:03:23', + '_dateModified': '2025-12-12 00:03:23', + '_versions': [ + { + '_dateCreated': '2025-12-12 00:03:23', + '_data': { + 'language': 'en', + 'value': 'Hello world!', + }, + '_uuid': 'c4fa8263-50c0-4252-9c9b-216ca338be13', + '_dateAccepted': '2025-12-15T22:22:00+00:00', + } + ], + }, + 'automatic_google_transcription': { + '_dateCreated': '2025-12-11 23:57:21', + '_dateModified': '2025-12-11 23:57:21', + '_versions': [ + { + '_dateCreated': '2025-12-11 23:57:21', + '_data': { + 'language': 'en', + 'value': 'Hello world', + 'status': 'complete', + }, + '_uuid': '64e59cc1-adaf-47a3-a068-550854d8f98f', + '_dateAccepted': '2025-12-15T22:22:00+00:00', + } + ], + }, + 'manual_translation': { + 'fr': { + '_dateCreated': '2025-12-12T00:04:38Z', + '_dateModified': '2025-12-12T00:04:38Z', + '_versions': [ + { + '_dateCreated': '2025-12-12T00:04:38Z', + '_data': { + 'language': 'fr', + 'value': 'Bonjour le monde!', + }, + '_uuid': '909c62cf-d544-4926-8839-7f035c6c7483', + '_dateAccepted': '2025-12-15T22:22:00+00:00', + '_dependency': { + '_uuid': 'c4fa8263-50c0-4252-9c9b-216ca338be13', + '_actionId': 'manual_transcription', + }, + } + ], + } + }, + 'qual': { + 'fd61cafc-9516-4063-8498-5eace89146a5': { + '_dateCreated': '2025-12-15T22:22:00+00:00', + '_dateModified': '2025-12-15T22:22:00+00:00', + '_versions': [ + { + '_data': { + 'uuid': 'fd61cafc-9516-4063-8498-5eace89146a5', + 'value': 'Answer', + }, + '_dateCreated': '2025-12-15T22:22:00+00:00', + '_dateAccepted': '2025-12-15T22:22:00+00:00', + '_uuid': '15ccc864-0e83-48f2-be1d-dc2adb9297f4', + } + ], + }, + 'c52ba63d-3202-44bc-8f55-159983e7f0d9': { + '_dateCreated': '2025-12-15T22:22:00+00:00', + '_dateModified': '2025-12-15T22:22:00+00:00', + '_versions': [ + { + '_data': { + 'uuid': 'c52ba63d-3202-44bc-8f55-159983e7f0d9', + 'value': '83212060-fd18-445a-b121-ad82c2e5811d', + }, + '_dateCreated': '2025-12-15T22:22:00+00:00', + '_dateAccepted': '2025-12-15T22:22:00+00:00', + '_uuid': 'f2b4c6b1-3c6a-4a7f-9e55-1a8c2a0a7c91', + } + ], + }, + 'b0bce6b0-9bf3-4f0f-a76e-4b3b4e9ba0e8': { + '_dateCreated': '2025-12-15T22:22:00+00:00', + '_dateModified': '2025-12-15T22:22:00+00:00', + '_versions': [ + { + '_data': { + 'uuid': 'b0bce6b0-9bf3-4f0f-a76e-4b3b4e9ba0e8', + 'value': [ + '35793589-556e-4872-b5eb-3b75e4dc4a99', + 'efceb7be-c120-43b4-9d6c-48c3c8d393bc', + ], + }, + '_dateCreated': '2025-12-15T22:22:00+00:00', + '_dateAccepted': '2025-12-15T22:22:00+00:00', + '_uuid': '8c9a8e44-7a3d-4c58-b7bb-5f2a1c6e5c3a', + } + ], + }, + }, + }, + '_version': '20250820', + } + assert response.data == expected_response + + +class SubmissionSupplementAPIValidationTestCase(SubsequenceBaseTestCase): + + def test_cannot_patch_if_question_has_no_configured_actions(self): + payload = { + '_version': '20250820', + 'q1': { + 'manual_translation': { + 'language': 'es', + 'value': 'buenas noches', + }, + }, + } + + # No actions activated at the asset level for any questions + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Invalid action' in str(response.data) + + def test_cannot_patch_if_action_is_invalid(self): + # Activate manual transcription (even though payload asks for translation) + payload = { + '_version': '20250820', + 'q1': { + 'manual_translation': { + 'language': 'es', + 'value': 'buenas noches', + }, + }, + } + + # No actions activated for q1 + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Invalid action' in str(response.data) + + # Activate manual transcription (even if payload asks for translation) + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='manual_transcription', + params=[{'language': 'es'}], + ) + + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Invalid action' in str(response.data) + + def test_cannot_patch_with_invalid_payload(self): + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='manual_transcription', + params=[{'language': 'es'}], + ) + + payload = { + '_version': '20250820', + 'q1': { + 'manual_translation': { + 'languageCode': 'es', # wrong attribute + 'value': 'buenas noches', + }, + }, + } + + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Invalid action' in str(response.data) + + def test_cannot_accept_incomplete_automatic_transcription(self): + # Set up the asset to allow automatic google transcription + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_transcription', + params=[{'language': 'es'}], + ) + + # Try to set 'accepted' status when translation is not complete + payload = { + '_version': '20250820', + 'q1': { + 'automatic_google_transcription': { + 'language': 'es', + 'accepted': True, + }, + }, + } + + # Mock GoogleTranscriptionService and simulate in progress transcription + mock_service = MagicMock() + mock_service.process_data.return_value = {'status': 'in_progress'} + + with patch( + 'kobo.apps.subsequences.actions.automatic_google_transcription.GoogleTranscriptionService', # noqa + return_value=mock_service, + ): + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Invalid payload' in str(response.data) + + def test_cannot_request_translation_without_transcription(self): + # Set up the asset to allow automatic google actions + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_transcription', + params=[{'language': 'en'}], + ) + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_translation', + params=[{'language': 'fr'}], + ) + # Try to ask for translation + payload = { + '_version': '20250820', + 'q1': { + 'automatic_google_translation': { + 'language': 'fr', + } + }, + } + + # Mock GoogleTranscriptionService and simulate in progress translation + mock_service = MagicMock() + mock_service.process_data.return_value = {'status': 'in_progress'} + + with patch( + 'kobo.apps.subsequences.actions.automatic_google_translation.GoogleTranslationService', # noqa + return_value=mock_service, + ): + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'Cannot translate without transcription' in str(response.data) + + +@ddt +class SubmissionSupplementAPIUsageLimitsTestCase(SubsequenceBaseTestCase): + def setUp(self): + super().setUp() + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_transcription', + params=[{'language': 'en'}], + ) + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='q1', + action='automatic_google_translation', + params=[{'language': 'es'}], + ) + + @pytest.mark.skipif( + not settings.STRIPE_ENABLED, reason='Requires stripe functionality' + ) + @data( + (True, status.HTTP_402_PAYMENT_REQUIRED), + (False, status.HTTP_200_OK), + ) + @unpack + def test_google_services_usage_limit_checks( + self, usage_limit_enforcement, expected_result_code + ): + payload = { + '_version': '20250820', + 'q1': { + 'automatic_google_transcription': { + 'language': 'en', + }, + }, + } + mock_balances = { + UsageType.ASR_SECONDS: {'exceeded': True}, + UsageType.MT_CHARACTERS: {'exceeded': True}, + } + + with patch( + 'kobo.apps.subsequences.actions.base.ServiceUsageCalculator.get_usage_balances', # noqa + return_value=mock_balances, + ): + with override_config(USAGE_LIMIT_ENFORCEMENT=usage_limit_enforcement): + with patch.object( + AutomaticGoogleTranscriptionAction, + 'run_external_process', + return_value=None, # noqa + ): + response = self.client.patch( + self.supplement_details_url, data=payload, format='json' + ) + assert response.status_code == expected_result_code diff --git a/kobo/apps/subsequences/tests/api/v2/test_permissions.py b/kobo/apps/subsequences/tests/api/v2/test_permissions.py index 99d39220a1..2886a9414a 100644 --- a/kobo/apps/subsequences/tests/api/v2/test_permissions.py +++ b/kobo/apps/subsequences/tests/api/v2/test_permissions.py @@ -96,7 +96,7 @@ def test_can_read(self, username, shared, status_code): False, status.HTTP_404_NOT_FOUND, ), - # regular user with view permission + # regular user with change permission ( 'anotheruser', True, @@ -108,7 +108,7 @@ def test_can_read(self, username, shared, status_code): False, status.HTTP_200_OK, ), - # admin user with view permissions + # admin user with change permissions ( 'adminuser', True, diff --git a/kobo/apps/subsequences/tests/api/v2/test_validation.py b/kobo/apps/subsequences/tests/api/v2/test_validation.py deleted file mode 100644 index 0d902d25b2..0000000000 --- a/kobo/apps/subsequences/tests/api/v2/test_validation.py +++ /dev/null @@ -1,340 +0,0 @@ -from unittest.mock import MagicMock, patch - -from rest_framework import status - -from kobo.apps.subsequences.models import QuestionAdvancedFeature, SubmissionSupplement -from kobo.apps.subsequences.tests.api.v2.base import SubsequenceBaseTestCase -from kobo.apps.subsequences.tests.constants import QUESTION_SUPPLEMENT - - -class SubmissionSupplementAPITestCase(SubsequenceBaseTestCase): - - def _simulate_completed_transcripts(self): - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_transcription', - params=[{'language': 'en'}], - ) - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='manual_transcription', - params=[{'language': 'en'}], - ) - - # Simulate a completed transcription, first. - mock_submission_supplement = {'_version': '20250820', 'q1': QUESTION_SUPPLEMENT} - - SubmissionSupplement.objects.create( - submission_uuid=self.submission_uuid, - content=mock_submission_supplement, - asset=self.asset, - ) - - def test_valid_manual_transcription(self): - payload = { - '_version': '20250820', - 'q1': { - 'manual_transcription': { - 'language': 'en', - 'value': 'hello world', - } - }, - } - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='manual_transcription', - params=[{'language': 'en'}], - ) - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_200_OK - - def test_valid_manual_translation(self): - self._simulate_completed_transcripts() - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='manual_translation', - params=[{'language': 'es'}], - ) - payload = { - '_version': '20250820', - 'q1': { - 'manual_translation': { - 'language': 'es', - 'value': 'hola el mundo', - } - }, - } - - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_200_OK - - def test_valid_automatic_transcription(self): - # Set up the asset to allow automatic google transcription - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_transcription', - params=[{'language': 'en'}], - ) - - payload = { - '_version': '20250820', - 'q1': { - 'automatic_google_transcription': { - 'language': 'en', - } - }, - } - - # Mock GoogleTranscriptionService and simulate completed transcription - mock_service = MagicMock() - mock_service.process_data.return_value = { - 'status': 'complete', - 'value': 'hello world', - } - - with patch( - 'kobo.apps.subsequences.actions.automatic_google_transcription.GoogleTranscriptionService', # noqa - return_value=mock_service, - ): - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_200_OK - - def test_valid_automatic_translation(self): - self._simulate_completed_transcripts() - # Set up the asset to allow automatic google translation - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_translation', - params=[{'language': 'es'}], - ) - - payload = { - '_version': '20250820', - 'q1': { - 'automatic_google_translation': { - 'language': 'es', - } - }, - } - - # Mock GoogleTranslationService and simulate in progress translation - mock_service = MagicMock() - mock_service.process_data.return_value = { - 'status': 'complete', - 'value': 'hola el mundo', - } - - with patch( - 'kobo.apps.subsequences.actions.automatic_google_translation.GoogleTranslationService', # noqa - return_value=mock_service, - ): - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_200_OK - - def test_cannot_patch_if_action_is_invalid(self): - payload = { - '_version': '20250820', - 'q1': { - 'manual_translation': { - 'language': 'es', - 'value': 'buenas noches', - } - }, - } - - # No actions activated for q1 - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_400_BAD_REQUEST - assert 'Invalid action' in str(response.data) - - # Activate manual transcription (even if payload asks for translation) - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='manual_transcription', - params=[{'language': 'es'}], - ) - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_400_BAD_REQUEST - assert 'Invalid action' in str(response.data) - - def test_cannot_patch_with_invalid_payload(self): - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='manual_transcription', - params=[{'language': 'es'}], - ) - - payload = { - '_version': '20250820', - 'q1': { - 'manual_translation': { - 'languageCode': 'es', # wrong attribute - 'value': 'buenas noches', - } - }, - } - - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - - assert response.status_code == status.HTTP_400_BAD_REQUEST - assert 'Invalid action' in str(response.data) - - def test_cannot_set_value_with_automatic_actions(self): - self._simulate_completed_transcripts() - # Set up the asset to allow automatic actions - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_translation', - params=[{'language': 'fr'}], - ) - - automatic_actions = self.asset.advanced_features_set.filter( - question_xpath='q1' - ).values_list('action', flat=True) - for automatic_action in automatic_actions: - payload = { - '_version': '20250820', - 'q1': { - automatic_action: { - 'language': 'es', - 'value': 'some text', # forbidden field - } - }, - } - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_400_BAD_REQUEST - assert 'Invalid payload' in str(response.data) - - - def test_cannot_accept_incomplete_automatic_transcription(self): - # Set up the asset to allow automatic google transcription - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_transcription', - params=[{'language': 'es'}], - ) - - # Try to set 'accepted' status when translation is not complete - payload = { - '_version': '20250820', - 'q1': { - 'automatic_google_transcription': { - 'language': 'es', - 'accepted': True, - } - }, - } - - # Mock GoogleTranscriptionService and simulate in progress transcription - mock_service = MagicMock() - mock_service.process_data.return_value = {'status': 'in_progress'} - - with patch( - 'kobo.apps.subsequences.actions.automatic_google_transcription.GoogleTranscriptionService', # noqa - return_value=mock_service, - ): - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_400_BAD_REQUEST - assert 'Invalid payload' in str(response.data) - - def test_cannot_accept_incomplete_automatic_translation(self): - self._simulate_completed_transcripts() - # Set up the asset to allow automatic google translation - - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_translation', - params=[{'language': 'fr'}], - ) - - # Try to set 'accepted' status when translation is not complete - payload = { - '_version': '20250820', - 'q1': { - 'automatic_google_translation': { - 'language': 'fr', - 'accepted': True, - } - }, - } - - # Mock GoogleTranscriptionService and simulate in progress translation - mock_service = MagicMock() - mock_service.process_data.return_value = {'status': 'in_progress'} - - with patch( - 'kobo.apps.subsequences.actions.automatic_google_translation.GoogleTranslationService', # noqa - return_value=mock_service, - ): - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_400_BAD_REQUEST - assert 'Invalid payload' in str(response.data) - - def test_cannot_request_translation_without_transcription(self): - # Set up the asset to allow automatic google actions - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_transcription', - params=[{'language': 'en'}], - ) - QuestionAdvancedFeature.objects.create( - asset=self.asset, - question_xpath='q1', - action='automatic_google_translation', - params=[{'language': 'fr'}], - ) - # Try to ask for translation - payload = { - '_version': '20250820', - 'q1': { - 'automatic_google_translation': { - 'language': 'fr', - } - }, - } - - # Mock GoogleTranscriptionService and simulate in progress translation - mock_service = MagicMock() - mock_service.process_data.return_value = {'status': 'in_progress'} - - with patch( - 'kobo.apps.subsequences.actions.automatic_google_translation.GoogleTranslationService', # noqa - return_value=mock_service, - ): - response = self.client.patch( - self.supplement_details_url, data=payload, format='json' - ) - assert response.status_code == status.HTTP_400_BAD_REQUEST - assert 'Cannot translate without transcription' in str(response.data) diff --git a/kobo/apps/subsequences/tests/test_proj_advanced_features.py b/kobo/apps/subsequences/tests/test_proj_advanced_features.py new file mode 100644 index 0000000000..565327effd --- /dev/null +++ b/kobo/apps/subsequences/tests/test_proj_advanced_features.py @@ -0,0 +1,50 @@ +from django.conf import settings +from django.test import TestCase +from django.urls import reverse +from django.utils import timezone +from model_bakery import baker +from rest_framework import status + +from kpi.models import Asset + + +class ProjectAdvancedFeaturesRefactoredTestCase(TestCase): + + fixtures = ['test_data'] + + def setUp(self): + user = baker.make( + settings.AUTH_USER_MODEL, + username='johndoe', + date_joined=timezone.now(), + ) + self.asset = Asset.objects.create( + owner=user, content={'survey': [{'type': 'audio', 'name': 'q1'}]} + ) + + def sample_asset(self, advanced_features=None): + if advanced_features is not None: + self.asset.advanced_features = advanced_features + return self.asset + + def test_qpath_to_xpath_with_renamed_question(self): + asset = self.sample_asset( + advanced_features={ + '_version': 'v1', + 'translation': { + 'languages': ['en', 'fr'], + }, + } + ) + + # Simulate known_cols with a legacy (renamed or deleted) question + asset.known_cols = [ + 'group_ia0id17-q1:translation:en', + 'group_ia0id17-q1:translation:fr', + ] + asset.save() + + self.client.force_login(asset.owner) + asset_detail_url = reverse('asset-detail', kwargs={'uid_asset': asset.uid}) + response = self.client.get(asset_detail_url) + assert response.status_code == status.HTTP_200_OK diff --git a/kobo/apps/subsequences/tests/test_submission_stream.py b/kobo/apps/subsequences/tests/test_submission_stream.py new file mode 100644 index 0000000000..102f2aea77 --- /dev/null +++ b/kobo/apps/subsequences/tests/test_submission_stream.py @@ -0,0 +1,183 @@ +import uuid +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.test import TestCase + +from kobo.apps.openrosa.apps.logger.exceptions import ConflictingSubmissionUUIDError +from kobo.apps.subsequences.constants import SUPPLEMENT_KEY +from kobo.apps.subsequences.models import QuestionAdvancedFeature, SubmissionSupplement +from kobo.apps.subsequences.utils.supplement_data import stream_with_supplements +from kpi.models import Asset + + +class TestSubmissionStream(TestCase): + def setUp(self): + self._create_asset() + self._create_submission_extras() + + def test_stream_with_supplements_handles_duplicated_submission_uuids(self): + submissions = self._create_submissions() + QuestionAdvancedFeature.objects.create( + asset=self.asset, + question_xpath='Tell_me_a_story', + action='qual', + params=[ + { + 'labels': {'_default': 'What is the quality score?'}, + 'type': 'qualText', + 'uuid': '4dcf9c9f-e503-4e5c-81f5-74250b295001', + } + ], + ) + + with patch.object( + self.asset.deployment, + 'get_submissions', + return_value=iter(submissions), + ): + with self.assertRaises(ConflictingSubmissionUUIDError): + self.asset.deployment.mock_submissions(submissions) + + output = list( + stream_with_supplements( + asset=self.asset, + submission_stream=self.asset.deployment.get_submissions( + user=self.asset.owner + ), + for_output=False, + ) + ) + + self.assertEqual(len(output), 2) + for submission in output: + self.assertIn(SUPPLEMENT_KEY, submission) + + supplemental_details = submission[SUPPLEMENT_KEY] + self.assertIn('Tell_me_a_story', supplemental_details) + + qual_data = supplemental_details.get('Tell_me_a_story').get('qual') + if '_versions' in qual_data: + for version_entry in qual_data['_versions']: + uuid_field = version_entry.get('_uuid') + self.assertIsInstance(uuid_field, str) + + def test_stream_with_extras_ignores_empty_qual_responses(self): + submission_extras = SubmissionSupplement.objects.get( + submission_uuid='1c05898e-b43c-491d-814c-79595eb84e81' + ) + content = submission_extras.content + content['Tell_me_a_story']['qual'] = {} + submission_extras.content = content + submission_extras.save() + + output = list( + stream_with_supplements( + asset=self.asset, + submission_stream=self.asset.deployment.get_submissions( + user=self.asset.owner + ), + for_output=False, + ) + ) + + for submission in output: + self.assertIn(SUPPLEMENT_KEY, submission) + + supplemental_details = submission[SUPPLEMENT_KEY] + self.assertIn('Tell_me_a_story', supplemental_details) + + qual_data = supplemental_details.get('Tell_me_a_story').get('qual') + self.assertEqual(qual_data, {}) + + def _create_asset(self): + owner = get_user_model().objects.create(username='nlp_owner') + + self.asset = Asset.objects.create( + owner=owner, + content={ + 'schema': '1', + 'survey': [ + { + 'type': 'text', + 'label': ["What's your name?"], + '$xpath': 'What_s_your_name', + 'required': False, + }, + { + 'type': 'audio', + 'label': ['Tell me a story!'], + '$xpath': 'Tell_me_a_story', + 'required': False, + }, + ], + 'settings': {}, + }, + advanced_features={ + '_version': 'v1', + 'qual': { + 'questions': [ + { + 'uuid': '4dcf9c9f-e503-4e5c-81f5-74250b295001', + 'type': 'qualInteger', + 'labels': {'_default': 'Quality Score'}, + } + ] + }, + }, + ) + self.asset.deploy(backend='mock', active=True) + self.asset.save() + + def _create_submission_extras(self): + qual_action_data = { + '_dateCreated': '2024-04-08T15:27:00Z', + '_dateModified': '2024-04-08T15:27:00Z', + '_versions': [ + { + '_data': { + 'uuid': '4dcf9c9f-e503-4e5c-81f5-74250b295001', + 'value': 'What is the quality score?', + }, + '_uuid': str(uuid.uuid4()), + '_dateCreated': '2024-04-08T15:27:00Z', + '_dateAccepted': '2024-04-08T15:29:00Z', + } + ], + } + SubmissionSupplement.objects.create( + submission_uuid='1c05898e-b43c-491d-814c-79595eb84e81', + asset=self.asset, + content={ + '_version': '20250820', + 'Tell_me_a_story': { + 'qual': qual_action_data, + }, + }, + ) + SubmissionSupplement.objects.create( + submission_uuid='1c05898e-b43c-491d-814c-79595eb84e82', + asset=self.asset, + content={ + '_version': '20250820', + 'Tell_me_a_story': { + 'qual': qual_action_data, + }, + }, + ) + + def _create_submissions(self): + return [ + { + 'What_s_your_name': 'Ed', + 'Tell_me_a_story': 'ed-18_6_24.ogg', + 'meta/rootUuid': 'uuid:1c05898e-b43c-491d-814c-79595eb84e81', + '_uuid': '1c05898e-b43c-491d-814c-79595eb84e81', + }, + { + 'What_s_your_name': 'Ed', + 'Tell_me_a_story': 'ed-18_6_44.ogg', + 'meta/rootUuid': 'uuid:1c05898e-b43c-491d-814c-79595eb84e82', + '_uuid': '1c05898e-b43c-491d-814c-79595eb84e81', + }, + ]