_row_data() -> KycConfig.get_user_objects()

corehq/apps/integration/kyc/models.py

@@ -4,6 +4,9 @@
 
 import jsonfield
 
+from corehq.apps.es.case_search import CaseSearchES
+from corehq.apps.users.models import CommCareUser
+from corehq.form_processor.models import CommCareCase
 from corehq.motech.const import OAUTH2_CLIENT
 from corehq.motech.models import ConnectionSettings
 
@@ -65,3 +68,24 @@ def get_connection_settings(self):
             else:
                 raise ValueError(f'Unable to determine connection settings for KYC provider {self.provider!r}.')
         return self.connection_settings
+
+    def get_user_objects(self):
+        """
+        Returns all CommCareUser or CommCareCase instances based on the
+        user data store.
+        """
+        if self.user_data_store in (
+            UserDataStore.CUSTOM_USER_DATA,
+            UserDataStore.USER_CASE,
+        ):
+            return CommCareUser.by_domain(self.domain)
+        elif self.user_data_store == UserDataStore.OTHER_CASE_TYPE:
+            # assert self.other_case_type
+            case_ids = (
+                CaseSearchES()
+                .domain(self.domain)
+                .case_type(self.other_case_type)
+            ).get_ids()
+            if not case_ids:
+                return []
+            return CommCareCase.objects.get_cases(case_ids, self.domain)

corehq/apps/integration/kyc/tests/test_models.py

@@ -2,7 +2,12 @@
 
 import pytest
 
+from corehq.apps.app_manager.const import USERCASE_TYPE
+from corehq.apps.es.case_search import case_search_adapter
+from corehq.apps.es.tests.utils import es_test
 from corehq.apps.integration.kyc.models import KycConfig, UserDataStore
+from corehq.apps.users.models import CommCareUser
+from corehq.form_processor.tests.utils import create_case
 from corehq.motech.models import ConnectionSettings
 
 DOMAIN = 'test-domain'
@@ -47,3 +52,76 @@ def test_bad_config(self):
                   "'invalid'.$",
         ):
             config.get_connection_settings()
+
+
+class TestGetUserObjectsUsers(TestCase):
+
+    def setUp(self):
+        self.commcare_user = CommCareUser.create(
+            DOMAIN, f'test.user@{DOMAIN}.commcarehq.org', 'Passw0rd!',
+            None, None,
+            user_data={'custom_field': 'custom_value'},
+        )
+        self.addCleanup(self.commcare_user.delete, DOMAIN, deleted_by=None)
+        self.user_case = create_case(
+            DOMAIN,
+            case_type=USERCASE_TYPE,
+            user_id=self.commcare_user._id,
+            name='test.user',
+            external_id=self.commcare_user._id,
+            save=True,
+            case_json={'user_case_property': 'user_case_value'},
+        )
+
+    def test_custom_user_data(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.CUSTOM_USER_DATA,
+        )
+        commcare_users = config.get_user_objects()
+        assert len(commcare_users) == 1
+        user_data = commcare_users[0].get_user_data(DOMAIN).to_dict()
+        assert user_data == {
+            'commcare_profile': '',
+            'custom_field': 'custom_value',
+        }
+
+    def test_user_case(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.USER_CASE,
+        )
+        commcare_users = config.get_user_objects()
+        assert len(commcare_users) == 1
+        user_case = commcare_users[0].get_usercase()
+        assert user_case.case_json == {
+            'user_case_property': 'user_case_value',
+        }
+
+    def test_assert_other_case_type(self):
+        pass
+
+
+@es_test(requires=[case_search_adapter])
+class TestGetUserObjectsCases(TestCase):
+
+    def setUp(self):
+        self.other_case = create_case(
+            DOMAIN,
+            case_type='other_case_type',
+            save=True,
+            case_json={'other_case_property': 'other_case_value'},
+        )
+        case_search_adapter.bulk_index([self.other_case], refresh=True)
+
+    def test_other_case_type(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.OTHER_CASE_TYPE,
+            other_case_type='other_case_type',
+        )
+        commcare_cases = config.get_user_objects()
+        assert len(commcare_cases) == 1
+        assert commcare_cases[0].case_json == {
+            'other_case_property': 'other_case_value',
+        }

corehq/apps/integration/kyc/views.py

@@ -5,18 +5,13 @@
 from corehq import toggles
 from corehq.apps.domain.decorators import login_required
 from corehq.apps.domain.views.base import BaseDomainView
-from corehq.apps.es.case_search import CaseSearchES
 from corehq.apps.hqwebapp.decorators import use_bootstrap5
 from corehq.apps.hqwebapp.tables.pagination import SelectablePaginatedTableView
 from corehq.apps.integration.kyc.forms import KycConfigureForm
-from corehq.apps.integration.kyc.models import (
-    KycConfig,
-    UserDataStore,
-)
-from corehq.apps.integration.kyc.tables import KycVerifyTable
+from corehq.apps.integration.kyc.models import KycConfig
 from corehq.apps.integration.kyc.services import get_user_data_for_api
+from corehq.apps.integration.kyc.tables import KycVerifyTable
 from corehq.apps.users.models import CommCareUser
-from corehq.form_processor.models import CommCareCase
 from corehq.util.htmx_action import HqHtmxActionMixin, hq_hx_action
 
 
@@ -82,24 +77,8 @@ class KycVerificationTableView(HqHtmxActionMixin, SelectablePaginatedTableView):
 
     def get_queryset(self):
         kyc_config = KycConfig.objects.get(domain=self.request.domain)
-        row_objs = self._row_data(kyc_config)
-        rows = []
-        for row_obj in row_objs:
-            rows.append(
-                self._parse_row(row_obj, kyc_config)
-            )
-        return rows
-
-    def _row_data(self, kyc_config):
-        if kyc_config.user_data_store in [UserDataStore.CUSTOM_USER_DATA, UserDataStore.USER_CASE]:
-            return CommCareUser.by_domain(self.request.domain)
-        elif kyc_config.user_data_store == UserDataStore.OTHER_CASE_TYPE:
-            case_ids = (
-                CaseSearchES()
-                .domain(self.request.domain)
-                .case_type(kyc_config.other_case_type)
-            ).get_ids()
-            return CommCareCase.objects.get_cases(case_ids, self.request.domain)
+        row_objs = kyc_config.get_user_objects()
+        return [self._parse_row(row_obj, kyc_config) for row_obj in row_objs]
 
     def _parse_row(self, row_obj, config):
         user_data = get_user_data_for_api(row_obj, config)