KYC Service Layer

corehq/apps/integration/kyc/forms.py

@@ -101,12 +101,3 @@ def __init__(self, *args, **kwargs):
                 }),
             )
         )
-
-    def clean(self):
-        user_data_store = self.cleaned_data['user_data_store']
-        other_case_type = self.cleaned_data['other_case_type']
-        if user_data_store == UserDataStore.OTHER_CASE_TYPE and not other_case_type:
-            self.add_error('other_case_type', _('Please specify a value'))
-        elif user_data_store != UserDataStore.OTHER_CASE_TYPE:
-            self.cleaned_data['other_case_type'] = None
-        return self.cleaned_data

corehq/apps/integration/kyc/models.py

@@ -1,8 +1,14 @@
+from django.conf import settings
+from django.core.exceptions import ValidationError
 from django.db import models
 from django.utils.translation import gettext as _
 
 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
 
 
@@ -18,6 +24,9 @@ class UserDataStore(object):
 
 
 class KycProviders(models.TextChoices):
+    # When adding a new provider:
+    # 1. Add connection settings to `settings.py` if necessary
+    # 2. Add it to `KycConfig.get_connections_settings()`
     MTN_KYC = 'mtn_kyc', _('MTN KYC')
 
 
@@ -26,10 +35,75 @@ class KycConfig(models.Model):
     user_data_store = models.CharField(max_length=25, choices=UserDataStore.CHOICES)
     other_case_type = models.CharField(max_length=126, null=True)
     api_field_to_user_data_map = jsonfield.JSONField(default=list)
-    connection_settings = models.ForeignKey(ConnectionSettings, on_delete=models.PROTECT)
-    provider = models.CharField(max_length=25, choices=KycProviders.choices, default=KycProviders.MTN_KYC)
+    provider = models.CharField(
+        max_length=25,
+        choices=KycProviders.choices,
+        default=KycProviders.MTN_KYC,
+    )
+    connection_settings = models.ForeignKey(
+        ConnectionSettings,
+        on_delete=models.PROTECT,
+        # Assumes we can determine connection settings for provider
+        null=True,
+        blank=True,
+    )
 
     class Meta:
         constraints = [
             models.UniqueConstraint(fields=['domain', 'provider'], name='unique_domain_provider'),
         ]
+
+    def clean(self):
+        super().clean()
+        if (
+            self.user_data_store == UserDataStore.OTHER_CASE_TYPE
+            and not self.other_case_type
+        ):
+            raise ValidationError({
+                'other_case_type': _(
+                    'This field is required when "User Data Store" is set to '
+                    '"Other Case Type".'
+                )
+            })
+        elif self.user_data_store != UserDataStore.OTHER_CASE_TYPE:
+            self.other_case_type = None
+
+    def get_connection_settings(self):
+        if not self.connection_settings_id:
+            if self.provider == KycProviders.MTN_KYC:
+                kyc_settings = settings.MTN_KYC_CONNECTION_SETTINGS
+                self.connection_settings = ConnectionSettings.objects.create(
+                    domain=self.domain,
+                    name=KycProviders.MTN_KYC.label,
+                    url=kyc_settings['url'],
+                    auth_type=OAUTH2_CLIENT,
+                    client_id=kyc_settings['client_id'],
+                    client_secret=kyc_settings['client_secret'],
+                    token_url=kyc_settings['token_url'],
+                )
+                self.save()
+            # elif self.provider == KycProviders.NEW_PROVIDER_HERE: ...
+            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

@@ -0,0 +1,141 @@
+from django.core.exceptions import ValidationError
+from django.test import TestCase
+
+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'
+
+
+class TestGetConnectionSettings(TestCase):
+
+    def test_valid_without_connection_settings(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.USER_CASE,
+        )
+        # Does not raise `django.db.utils.IntegrityError`
+        config.save()
+
+    def test_get_connection_settings(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.USER_CASE,
+        )
+        assert ConnectionSettings.objects.count() == 0
+
+        connx = config.get_connection_settings()
+        assert isinstance(connx, ConnectionSettings)
+        # First call creates ConnectionSettings
+        assert ConnectionSettings.objects.count() == 1
+
+        connx = config.get_connection_settings()
+        assert isinstance(connx, ConnectionSettings)
+        # Subsequent calls get existing ConnectionSettings
+        assert ConnectionSettings.objects.count() == 1
+
+    def test_bad_config(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.USER_CASE,
+            provider='invalid',
+        )
+        with pytest.raises(
+            ValueError,
+            match="^Unable to determine connection settings for KYC provider "
+                  "'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',
+        }
+
+
+@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',
+        }
+
+    def test_other_case_type_clean(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.OTHER_CASE_TYPE,
+        )
+        with pytest.raises(ValidationError):
+            config.clean()
+
+    def test_assert_other_case_type(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.OTHER_CASE_TYPE,
+        )
+        with pytest.raises(AssertionError):
+            config.get_user_objects()

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)

corehq/apps/integration/migrations/0009_alter_kycconfig_connection_settings.py

@@ -0,0 +1,25 @@
+# Generated by Django 4.2.18 on 2025-02-16 22:53
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("motech", "0017_connectionsettings_use_aes_cbc_encryption"),
+        ("integration", "0008_kycconfig_provider_kycconfig_unique_domain_provider"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="kycconfig",
+            name="connection_settings",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.PROTECT,
+                to="motech.connectionsettings",
+            ),
+        ),
+    ]

migrations.lock

@@ -652,6 +652,7 @@ integration
  0006_kycconfig
  0007_alter_kycconfig_api_field_to_user_data_map
  0008_kycconfig_provider_kycconfig_unique_domain_provider
+ 0009_alter_kycconfig_connection_settings
 ivr
  0001_initial
  0002_call_app_id

settings.py

@@ -1172,6 +1172,14 @@ def _pkce_required(client_id):
 # used by periodic tasks that delete soft deleted data older than PERMANENT_DELETION_WINDOW days
 PERMANENT_DELETION_WINDOW = 30  # days
 
+# Used by `corehq.apps.integration.kyc`. Override in localsettings.py
+MTN_KYC_CONNECTION_SETTINGS = {
+    'url': 'https://dev.api.chenosis.io/',
+    'token_url': 'https://dev.api.chenosis.io/oauth/client/accesstoken',
+    'client_id': 'test',
+    'client_secret': 'password',
+}
+
 
 try:
     # try to see if there's an environmental variable set for local_settings