test(subsequences): port old unit tests DEV-1231

kobo/apps/subsequences/actions/base.py

@@ -368,7 +368,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)
 
@@ -614,7 +613,7 @@ def external_data_schema(self) -> dict:
         Schema rules:
 
         - The field `status` is always required and must be one of:
-          ["requested", "in_progress", "complete", "failed"].
+          ["deleted", "in_progress", "complete", "failed"].
         - If `status` == "complete":
             * The field `value` becomes required and must be a string.
         - If `status` == "failed":

kobo/apps/subsequences/models.py

@@ -1,9 +1,10 @@
+from constance import config
 from django.db import models
 
 from kobo.apps.openrosa.apps.logger.xform_instance_parser import remove_uuid_prefix
 from kpi.models.abstract_models import AbstractTimeStampedModel
 from .actions import ACTION_IDS_TO_CLASSES
-from .constants import SUBMISSION_UUID_FIELD, SCHEMA_VERSIONS
+from .constants import SCHEMA_VERSIONS, SUBMISSION_UUID_FIELD
 from .exceptions import InvalidAction, InvalidXPath
 from .schemas import validate_submission_supplement
 
@@ -83,7 +84,9 @@ def revise_data(asset: 'kpi.Asset', submission: dict, incoming_data: dict) -> di
                     raise InvalidAction from e
 
                 action = action_class(question_xpath, action_params, asset)
-                action.check_limits(asset.owner)
+
+                if config.USAGE_LIMIT_ENFORCEMENT:
+                    action.check_limits(asset.owner)
 
                 question_supplemental_data = supplemental_data.setdefault(
                     question_xpath, {}

kobo/apps/subsequences/tests/api/v2/test_validation.py → kobo/apps/subsequences/tests/api/v2/test_api.py

@@ -1,15 +1,155 @@
+import uuid
 from unittest.mock import MagicMock, patch
 
+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 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()
+        self.set_asset_advanced_features(
+            {
+                '_version': '20250820',
+                '_actionConfigs': {
+                    'q1': {
+                        'manual_transcription': [
+                            {'language': 'en'},
+                        ]
+                    }
+                },
+            }
+        )
 
-    def test_cannot_patch_if_action_is_invalid(self):
+    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',
+                }
+            },
+        }
+        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': [
+                        {
+                            '_dateAccepted': now_iso,
+                            '_dateCreated': now_iso,
+                            '_uuid': 'uuid1',
+                            'language': 'en',
+                            'value': 'Hello world',
+                        }
+                    ],
+                }
+            },
+        }
+        assert response.data == expected_data
+
+
+class SubmissionSupplementAPIValidationTestCase(SubsequenceBaseTestCase):
+
+    def test_cannot_patch_if_question_has_no_configured_actions(self):
         payload = {
             '_version': '20250820',
             'q1': {
@@ -20,14 +160,24 @@ def test_cannot_patch_if_action_is_invalid(self):
             },
         }
 
-        # No actions activated at the asset level
+        # 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)
+        assert 'Invalid question' in str(response.data)
 
-        # Activate manual transcription (even if payload asks for translation)
+    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',
+                }
+            },
+        }
         self.set_asset_advanced_features(
             {
                 '_version': '20250820',
@@ -88,17 +238,14 @@ def test_cannot_set_value_with_automatic_actions(self):
                     ],
                     'automatic_google_translation': [
                         {'language': 'fr'},
-                    ]
+                    ],
                 }
             },
         }
         self.set_asset_advanced_features(advanced_features)
 
         # Simulate a completed transcription, first.
-        mock_submission_supplement = {
-            '_version': '20250820',
-            'q1': QUESTION_SUPPLEMENT
-        }
+        mock_submission_supplement = {'_version': '20250820', 'q1': QUESTION_SUPPLEMENT}
 
         SubmissionSupplement.objects.create(
             submission_uuid=self.submission_uuid,
@@ -122,7 +269,6 @@ def test_cannot_set_value_with_automatic_actions(self):
             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
         self.set_asset_advanced_features(
@@ -175,17 +321,14 @@ def test_cannot_accept_incomplete_automatic_translation(self):
                         ],
                         'automatic_google_translation': [
                             {'language': 'fr'},
-                        ]
+                        ],
                     }
                 },
             }
         )
 
         # Simulate a completed transcription, first.
-        mock_submission_supplement = {
-            '_version': '20250820',
-            'q1': QUESTION_SUPPLEMENT
-        }
+        mock_submission_supplement = {'_version': '20250820', 'q1': QUESTION_SUPPLEMENT}
         SubmissionSupplement.objects.create(
             submission_uuid=self.submission_uuid,
             content=mock_submission_supplement,
@@ -229,7 +372,7 @@ def test_cannot_request_translation_without_transcription(self):
                         ],
                         'automatic_google_translation': [
                             {'language': 'fr'},
-                        ]
+                        ],
                     }
                 },
             }
@@ -258,3 +401,63 @@ def test_cannot_request_translation_without_transcription(self):
             )
             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()
+        self.set_asset_advanced_features(
+            {
+                '_version': '20250820',
+                '_actionConfigs': {
+                    'q1': {
+                        'automatic_google_transcription': [
+                            {'language': 'en'},
+                        ],
+                        'automatic_google_translation': [
+                            {'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

kobo/apps/subsequences/tests/api/v2/test_permissions.py

@@ -93,7 +93,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,
@@ -105,7 +105,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,

kobo/apps/subsequences/tests/test_automatic_google_translation.py

@@ -6,8 +6,8 @@
 import pytest
 
 from ..actions.automatic_google_translation import AutomaticGoogleTranslationAction
-from .constants import EMPTY_SUBMISSION, EMPTY_SUPPLEMENT, QUESTION_SUPPLEMENT
 from ..exceptions import TranscriptionNotFound
+from .constants import EMPTY_SUBMISSION, EMPTY_SUPPLEMENT, QUESTION_SUPPLEMENT
 
 
 def test_valid_params_pass_validation():