API call to verify user

corehq/apps/integration/kyc/schemas/kyc-verify-v1.json

@@ -0,0 +1,44 @@
+{
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "required": [
+        "firstName",
+        "lastName",
+        "phoneNumber",
+        "emailAddress",
+        "nationalIdNumber",
+        "streetAddress",
+        "city",
+        "postCode",
+        "country"
+    ],
+    "properties": {
+        "firstName": {
+            "type": "string"
+        },
+        "lastName": {
+            "type": "string"
+        },
+        "phoneNumber": {
+            "type": "string"
+        },
+        "emailAddress": {
+            "type": "string"
+        },
+        "nationalIdNumber": {
+            "type": "string"
+        },
+        "streetAddress": {
+            "type": "string"
+        },
+        "city": {
+            "type": "string"
+        },
+        "postCode": {
+            "type": "string"
+        },
+        "country": {
+            "type": "string"
+        }
+    }
+}

corehq/apps/integration/kyc/services.py

@@ -1,10 +1,141 @@
-from corehq.apps.integration.kyc.models import UserDataStore
+import json
+import os
+import re
+from datetime import datetime
+
+from django.conf import settings
+from django.utils.text import camel_case_to_spaces
+
+import jsonschema
+
+from corehq.apps.hqcase.utils import update_case
+from corehq.apps.integration.kyc.models import KycConfig, UserDataStore
 
 
 class UserCaseNotFound(Exception):
     pass
 
 
+def verify_all(domain):
+    kyc_config = KycConfig.objects.get(domain=domain)
+    user_objs = kyc_config.get_user_objects()
+    return verify_users(user_objs, kyc_config)
+
+
+def verify_selected_ids(domain, selected_ids):
+    # kyc_config = KycConfig.objects.get(domain=domain)
+    # ...
+    pass
+
+
+def verify_users(user_objs, config):
+    # TODO: An endpoint to verify a group of users does not seem to be
+    #       available using Chenosis. If we have to do this with
+    #       multiple calls, consider using Celery gevent workers.
+    results = []
+    for user_obj in user_objs:
+        is_verified = verify_user(user_obj, config)
+        save_result(user_obj, config, is_verified)
+        results.append(is_verified)
+    return results
+
+
+def verify_user(user_obj, config):
+    """
+    Verify a user using the Chenosis MTN KYC API.
+
+    Returns True if all user data is accurate above its required
+    threshold, otherwise False.
+    """
+
+    # Documentation: https://apiportal.chenosis.io/en/apis/kyc-verify#endpoints
+    #
+    # Example request:
+    #
+    #     POST https://dev.api.chenosis.io/kycVerify/v1/customers/{customerId}
+    #     {
+    #         "firstName": "John",
+    #         "lastName": "Doe",
+    #         "phoneNumber": "27650600900",
+    #         "emailAddress": "[email protected]",
+    #         "nationalIdNumber": "123456789",
+    #         "streetAddress": "1st Park Avenue, Mzansi, Johannesburg",
+    #         "city": "Johannesburg",
+    #         "postCode": "20200",
+    #         "country": "South Africa"
+    #     }
+    #
+    # Example 200 response:
+    #
+    #     {
+    #         "statusCode": "0000",
+    #         "statusMessage": "Success.",
+    #         "customerId": "27650600900",
+    #         "transactionId": "232TXYZ-212",
+    #         "data": {
+    #             "firstName": 100,
+    #             "lastName": 100,
+    #             "phoneNumber": 100,
+    #             "emailAddress": 100,
+    #             "nationalIdNumber": 100,
+    #             "streetAddress": 100,
+    #             "city": 100,
+    #             "postCode": 100,
+    #             "country": 100
+    #         }
+    #     }
+
+    # TODO: Determine what thresholds we want
+    required_thresholds = {
+        'firstName': 100,
+        'lastName': 100,
+        'phoneNumber': 100,
+        'emailAddress': 100,
+        'nationalIdNumber': 100,
+        'streetAddress': 80,  # Allow streetAddress to be less accurate
+        'city': 100,
+        'postCode': 100,
+        'country': 0,  # e.g "Ivory Coast" vs. "Republic of Côte d'Ivoire" vs. "CIV"?
+    }
+
+    user_data = get_user_data_for_api(user_obj, config)
+    _validate_schema('kycVerify/v1', user_data)
+    requests = config.get_connection_settings().get_requests()
+    response = requests.post(
+        f'/kycVerify/v1/customers/{user_data["phoneNumber"]}',
+        json=user_data,
+    )
+    response.raise_for_status()
+    field_scores = response.json().get('data', {})
+    return all(v >= required_thresholds[k] for k, v in field_scores.items())
+
+
+def save_result(user_obj, config, is_verified):
+    update = {
+        'kyc_provider': config.provider.value,  # TODO: Or config.provider.label?
+        'kyc_last_verified_at': datetime.utcnow(),  # TODO: UTC or project timezone?
+        'kyc_is_verified': is_verified,
+    }
+    if config.user_data_store == UserDataStore.CUSTOM_USER_DATA:
+        user_data = user_obj.get_user_data(config.domain)
+        user_data.update(update)
+    elif config.user_data_store == UserDataStore.USER_CASE:
+        case = user_obj.get_usercase()
+        update_case(
+            config.domain,
+            case.case_id,
+            case_properties=update,
+            device_id='corehq.apps.integration.kyc.services.save_result',
+        )
+    else:  # UserDataStore.OTHER_CASE_TYPE
+        update_case(
+            config.domain,
+            user_obj.case_id,
+            case_properties=update,
+            device_id='corehq.apps.integration.kyc.services.save_result',
+        )
+
+
 def get_user_data_for_api(source, config):
     """
         Returns a dictionary of user data for the API.
@@ -26,11 +157,13 @@ def get_user_data_for_api(source, config):
 
 def _get_source_data(source, config):
     """
-        Returns a dictionary of source data.
-        ``source`` is a CommCareUser or a CommCareCase.
+    Returns a dictionary of source data.
+
+    :param source: A CommCareUser or a CommCareCase.
+    :param config: A KycConfig instance.
     """
     if config.user_data_store == UserDataStore.CUSTOM_USER_DATA:
-        source_data = source.get_user_data(config.domain)
+        source_data = source.get_user_data(config.domain).to_dict()
     elif config.user_data_store == UserDataStore.USER_CASE:
         custom_user_case = source.get_usercase()
         if not custom_user_case:
@@ -39,3 +172,29 @@ def _get_source_data(source, config):
     else:
         source_data = source.case_json
     return source_data
+
+
+def _validate_schema(endpoint, data):
+    """
+    Validate the data against the schema for the given endpoint
+    """
+    schema_filename = f'{_kebab_case(endpoint)}.json'
+    schema_path = os.path.join(
+        settings.BASE_DIR, 'corehq', 'apps', 'integration', 'kyc', 'schemas',
+        schema_filename,
+    )
+    with open(schema_path) as f:
+        schema = json.load(f)
+    jsonschema.validate(data, schema)
+
+
+def _kebab_case(value):
+    """
+    Convert a string to kebab-case
+
+    >>> _kebab_case('Hello, World!')
+    'hello-world'
+    """
+    value = camel_case_to_spaces(value)
+    value = re.sub(r"[^\w\s-]", "-", value.lower())
+    return re.sub(r"[-\s]+", "-", value).strip("-_")

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

@@ -0,0 +1,110 @@
+import doctest
+
+from django.test import TestCase
+
+import jsonschema
+import pytest
+
+from corehq.apps.app_manager.const import USERCASE_TYPE
+from corehq.apps.integration.kyc.models import KycConfig, UserDataStore
+from corehq.apps.integration.kyc.services import (
+    _get_source_data,
+    _validate_schema,
+)
+from corehq.apps.users.models import CommCareUser
+from corehq.form_processor.tests.utils import create_case
+
+DOMAIN = 'test-domain'
+
+
+def test_doctests():
+    import corehq.apps.integration.kyc.services as module
+    results = doctest.testmod(module)
+    assert results.failed == 0
+
+
+class TestSaveResult(TestCase):
+    # TODO: ...
+    pass
+
+
+class TestGetSourceData(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'},
+        )
+        self.other_case = create_case(
+            DOMAIN,
+            case_type='other_case_type',
+            save=True,
+            case_json={'other_case_property': 'other_case_value'},
+        )
+
+    def test_custom_user_data(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.CUSTOM_USER_DATA,
+        )
+        source_data = _get_source_data(self.commcare_user, config)
+        assert source_data == {
+            'custom_field': 'custom_value',
+            'commcare_profile': '',
+        }
+
+    def test_user_case(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.USER_CASE,
+        )
+        source_data = _get_source_data(self.commcare_user, config)
+        assert source_data == {'user_case_property': 'user_case_value'}
+
+    def test_other_case_type(self):
+        config = KycConfig(
+            domain=DOMAIN,
+            user_data_store=UserDataStore.OTHER_CASE_TYPE,
+            other_case_type='other_case_type',
+        )
+        source_data = _get_source_data(self.other_case, config)
+        assert source_data == {'other_case_property': 'other_case_value'}
+
+
+class TestValidateSchema(TestCase):
+
+    def test_valid_schema(self):
+        endpoint = 'kycVerify/v1'
+        data = {
+            'firstName': 'John',
+            'lastName': 'Doe',
+            'phoneNumber': '27650600900',
+            'emailAddress': '[email protected]',
+            'nationalIdNumber': '123456789',
+            'streetAddress': '1st Park Avenue, Mzansi, Johannesburg',
+            'city': 'Johannesburg',
+            'postCode': '20200',
+            'country': 'South Africa',
+        }
+        _validate_schema(endpoint, data)  # Should not raise an exception
+
+    def test_invalid_schema(self):
+        endpoint = 'kycVerify/v1'
+        data = {
+            'firstName': 'John',
+            'lastName': 'Doe',
+            # Missing required fields
+        }
+        with pytest.raises(jsonschema.ValidationError):
+            _validate_schema(endpoint, data)

corehq/apps/integration/kyc/views.py

@@ -9,7 +9,11 @@
 from corehq.apps.hqwebapp.tables.pagination import SelectablePaginatedTableView
 from corehq.apps.integration.kyc.forms import KycConfigureForm
 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.services import (
+    get_user_data_for_api,
+    verify_all,
+    verify_selected_ids,
+)
 from corehq.apps.integration.kyc.tables import KycVerifyTable
 from corehq.apps.users.models import CommCareUser
 from corehq.util.htmx_action import HqHtmxActionMixin, hq_hx_action
@@ -107,19 +111,14 @@ def _parse_row(self, row_obj, config):
 
     @hq_hx_action('post')
     def verify_rows(self, request, *args, **kwargs):
-        verify_all = request.POST.get('verify_all')
-        verify_success = True
-        success_count = 0
-        fail_count = 0
-        if verify_all:
-            # TODO: Need to get all IDS. Could take inspiration from _row_data to fetch all IDs
-            # TODO: Verify all rows
-            pass
+        if request.POST.get('verify_all'):
+            results = verify_all(request.domain)
         else:
-            pass
-            # TODO: Verify selected rows
-            # selected_ids = request.POST.getlist('selected_ids')
-
+            selected_ids = request.POST.getlist('selected_ids')
+            results = verify_selected_ids(request.domain, selected_ids)
+        verify_success = bool(results)
+        success_count = sum(1 for result in results if result)
+        fail_count = sum(1 for result in results if not result)
         context = {
             'verify_success': verify_success,
             'success_count': success_count,

corehq/apps/integration/tests/test_kyc_integration.py

@@ -9,7 +9,8 @@
 from corehq.motech.models import ConnectionSettings
 
 
-class TestKycGetUserDataAPI(TestCase):
+# TODO: Move to test_services
+class TestGetUserDataForAPI(TestCase):
     domain = 'test-kyc-integration'
 
     @classmethod