diff --git a/kobo/apps/subsequences/tests/test_versioning.py b/kobo/apps/subsequences/tests/test_versioning.py new file mode 100644 index 0000000000..c7946c7fbf --- /dev/null +++ b/kobo/apps/subsequences/tests/test_versioning.py @@ -0,0 +1,103 @@ +from django.test import TestCase + +from kobo.apps.kobo_auth.shortcuts import User +from kobo.apps.subsequences.constants import Action +from kobo.apps.subsequences.models import QuestionAdvancedAction +from kobo.apps.subsequences.utils.versioning import convert_qual_params +from kpi.models import Asset + + +class TestVersioning(TestCase): + def setUp(self): + self.owner = User.objects.create(username='asset_owner') + + def _create_asset(self): + return Asset.objects.create(owner=self.owner, content={'survey': []}) + + def test_convert_qual_params_create(self): + asset = self._create_asset() + qualdict = { + 'qual_survey': [ + { + 'uuid': 'q1', + 'xpath': '/a', + 'type': 'qual_text', + 'labels': {'_default': 'A1'}, + }, + { + 'uuid': 'q2', + 'xpath': '/a', + 'type': 'qual_text', + 'labels': {'_default': 'A2'}, + }, + { + 'uuid': 'q3', + 'xpath': '/b', + 'type': 'qual_integer', + 'labels': {'_default': 'B1'}, + }, + ] + } + + created = convert_qual_params(asset, qualdict) + + # Two different xpaths will create two DB rows + self.assertEqual(len(created), 2) + db_objs = QuestionAdvancedAction.objects.filter(asset=asset, action=Action.QUAL) + self.assertEqual(db_objs.count(), 2) + + qa_a = QuestionAdvancedAction.objects.get(asset=asset, question_xpath='/a') + self.assertEqual(len(qa_a.params), 2) + self.assertEqual({p['uuid'] for p in qa_a.params}, {'q1', 'q2'}) + + qa_b = QuestionAdvancedAction.objects.get(asset=asset, question_xpath='/b') + self.assertEqual(len(qa_b.params), 1) + self.assertEqual(qa_b.params[0]['uuid'], 'q3') + + def test_convert_qual_params_update(self): + asset = self._create_asset() + # pre-create an action for xpath '/a' + QuestionAdvancedAction.objects.create( + asset=asset, + action=Action.QUAL, + question_xpath='/a', + params=[{'uuid': 'old', 'type': 'qual_text'}], + ) + + qualdict = { + 'qual_survey': [ + { + 'uuid': 'q1', + 'xpath': '/a', + 'type': 'qual_text', + 'labels': {'_default': 'A1'}, + }, + ] + } + + created = convert_qual_params(asset, qualdict) + # Should return the existing object (updated) + self.assertEqual(len(created), 1) + self.assertEqual( + QuestionAdvancedAction.objects.filter( + asset=asset, action=Action.QUAL + ).count(), + 1, + ) + + qa = QuestionAdvancedAction.objects.get(asset=asset, question_xpath='/a') + self.assertEqual(len(qa.params), 1) + self.assertEqual(qa.params[0]['uuid'], 'q1') + + def test_convert_qual_params_invalid(self): + asset = self._create_asset() + + # empty dict + res = convert_qual_params(asset, {}) + self.assertEqual(res, []) + self.assertEqual(QuestionAdvancedAction.objects.filter(asset=asset).count(), 0) + + # malformed qual_survey + res = convert_qual_params(asset, {'qual_survey': 'not-a-list'}) + self.assertEqual(res, []) + self.assertEqual(QuestionAdvancedAction.objects.filter(asset=asset).count(), 0) diff --git a/kobo/apps/subsequences/utils/versioning.py b/kobo/apps/subsequences/utils/versioning.py new file mode 100644 index 0000000000..d9d3ee5e45 --- /dev/null +++ b/kobo/apps/subsequences/utils/versioning.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from typing import Any, Dict, Iterable, List + +from django.db import transaction + +from ..constants import Action +from ..models import QuestionAdvancedAction + + +def convert_qual_params( + asset: Any, qualdict: Dict[str, Any] +) -> List[QuestionAdvancedAction]: + """Convert a qual dict (from `Asset.advanced_features['qual']`) into + `QuestionAdvancedAction` objects grouped by `xpath`. + + Returns the list of created/updated `QuestionAdvancedAction` instances. + """ + if not qualdict: + return [] + + qual_survey = qualdict.get('qual_survey') + if not isinstance(qual_survey, Iterable): + return [] + + groups: Dict[str, List[Dict[str, Any]]] = {} + for item in qual_survey: + if not isinstance(item, dict): + continue + xpath = item.get('xpath') or item.get('qpath') + if not xpath: + continue + groups.setdefault(xpath, []).append(item) + + created_objs: List[QuestionAdvancedAction] = [] + with transaction.atomic(): + for xpath, items in groups.items(): + obj, _ = QuestionAdvancedAction.objects.update_or_create( + asset=asset, + action=Action.QUAL, + question_xpath=xpath, + defaults={'params': list(items)}, + ) + created_objs.append(obj) + + return created_objs