KYC Service Layer

corehq/apps/integration/kyc/models.py

@@ -1,8 +1,10 @@
+from django.conf import settings
 from django.db import models
 from django.utils.translation import gettext as _
 
 import jsonfield
 
+from corehq.motech.const import OAUTH2_CLIENT
 from corehq.motech.models import ConnectionSettings
 
 
@@ -18,6 +20,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 +31,39 @@ 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 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

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

@@ -0,0 +1,49 @@
+from django.test import TestCase
+
+import pytest
+
+from corehq.apps.integration.kyc.models import KycConfig, UserDataStore
+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()

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