RDI Access change for retrieving data from Kobo Reinstated

development_tools/.env.example

@@ -41,6 +41,7 @@ OPENAPI_URL=127.0.0.1:8080/api/rest/
 KOBO_API_URL=https://kobo.humanitarianresponse.info
 KOBO_KF_URL=https://kobo.humanitarianresponse.info
 KOBO_KC_URL=https://kobo.humanitarianresponse.info
+KOBO_PROJECT_VIEWS_ID=
 #MAILJET_API_KEY=
 #MAILJET_SECRET_KEY=
 #CATCH_ALL_EMAIL=

src/hct_mis_api/apps/core/kobo/api.py

@@ -2,28 +2,24 @@
 import time
 import typing
 from io import BytesIO
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple
 from urllib.parse import urlparse
 
 from django.conf import settings
 
 import requests
+from requests import Response
 from requests.adapters import HTTPAdapter
 from requests.exceptions import RetryError
 from requests.packages.urllib3.util.retry import Retry
 
-from hct_mis_api.apps.core.kobo.common import filter_by_owner
-from hct_mis_api.apps.core.models import BusinessArea, XLSXKoboTemplate
+from hct_mis_api.apps.core.models import XLSXKoboTemplate
 from hct_mis_api.apps.utils.exceptions import log_and_raise
 
 logger = logging.getLogger(__name__)
 
 
-class TokenNotProvided(Exception):
-    pass
-
-
-class TokenInvalid(Exception):
+class CountryCodeNotProvided(Exception):
     pass
 
 
@@ -32,100 +28,74 @@ class KoboRequestsSession(requests.Session):
 
     def should_strip_auth(self, old_url: str, new_url: str) -> bool:
         new_parsed = urlparse(new_url)
-        if new_parsed.hostname in KoboRequestsSession.AUTH_DOMAINS:
+        if new_parsed.hostname in KoboRequestsSession.AUTH_DOMAINS:  # pragma: no cover
             return False
-        return super().should_strip_auth(old_url, new_url)  # type: ignore # FIXME: Call to untyped function "should_strip_auth" in typed context
+        return super().should_strip_auth(
+            old_url, new_url
+        )  # type: ignore # FIXME: Call to untyped function "should_strip_auth" in typed context
 
 
 class KoboAPI:
-    def __init__(self, business_area_slug: Optional[str] = None):
-        if business_area_slug is not None:
-            self.business_area = BusinessArea.objects.get(slug=business_area_slug)
-            self.KPI_URL = self.business_area.kobo_url or settings.KOBO_KF_URL
-        else:
-            self.business_area = None
-            self.KPI_URL = settings.KOBO_KF_URL
+    LIMIT = 30_000
+    FORMAT = "json"
 
-        self._get_token()
+    def __init__(
+        self, kpi_url: Optional[str] = None, token: Optional[str] = None, project_views_id: Optional[str] = None
+    ) -> None:
+        self._kpi_url = kpi_url or settings.KOBO_KF_URL
+        self._token = token or settings.KOBO_MASTER_API_TOKEN
+        self._project_views_id = project_views_id or settings.KOBO_PROJECT_VIEWS_ID
 
-    def _handle_paginated_results(self, url: str) -> List[Dict]:
+        self._client = KoboRequestsSession()
+        self._set_token()
+
+    def _set_token(self) -> None:
+        retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504], allowed_methods=False)
+        self._client.mount(self._kpi_url, HTTPAdapter(max_retries=retries))
+        self._client.headers.update({"Authorization": f"token {self._token}"})
+
+    def _get_paginated_request(self, url: str) -> List[Dict]:
         next_url = url
         results: List = []
 
-        # if there will be more than 30000 results,
-        # we need to make additional queries
         while next_url:
-            data = self._handle_request(next_url)
+            response = self._get_request(next_url)
+            data = response.json()
             next_url = data["next"]
             results.extend(data["results"])
         return results
 
-    def _get_url(
-        self,
-        endpoint: str,
-        append_api: bool = True,
-        add_limit: bool = True,
-        additional_query_params: Optional[Any] = None,
-    ) -> str:
-        endpoint.strip("/")
-        if endpoint != "token" and append_api is True:
-            endpoint = f"api/v2/{endpoint}"
-        # According to the Kobo API documentation,
-        # the maximum limit per page is 30000
-        query_params = f"format=json{'&limit=30000' if add_limit else ''}"
-        if additional_query_params is not None:
-            query_params += f"&{additional_query_params}"
-        return f"{self.KPI_URL}/{endpoint}?{query_params}"
-
-    def _get_token(self) -> None:
-        self._client = KoboRequestsSession()
-        retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504], allowed_methods=False)
-        self._client.mount(self.KPI_URL, HTTPAdapter(max_retries=retries))
-
-        if self.business_area is None:
-            token = settings.KOBO_MASTER_API_TOKEN
-        else:
-            token = self.business_area.kobo_token
-
-        if not token:
-            msg = f"KOBO Token is not set for business area {self.business_area}"
-            logger.error(msg)
-            raise TokenNotProvided(msg)
-
-        self._client.headers.update({"Authorization": f"token {token}"})
-
-    def _handle_request(self, url: str) -> Dict:
+    def _get_request(self, url: str) -> Response:
         response = self._client.get(url=url)
         try:
             response.raise_for_status()
-        except requests.exceptions.HTTPError as e:
+        except requests.exceptions.HTTPError as e:  # pragma: no cover
             logger.exception(e)
             raise
-        return response.json()
+        return response
 
     def _post_request(
         self, url: str, data: Optional[Dict] = None, files: Optional[typing.IO] = None
-    ) -> requests.Response:
+    ) -> Response:  # pragma: no cover
         return self._client.post(url=url, data=data, files=files)
 
-    def _patch_request(
-        self, url: str, data: Optional[Dict] = None, files: Optional[typing.IO] = None
-    ) -> requests.Response:
-        return self._client.patch(url=url, data=data, files=files)
-
     def create_template_from_file(
-        self, bytes_io_file: Optional[typing.IO], xlsx_kobo_template_object: XLSXKoboTemplate, template_id: str = ""
-    ) -> Optional[Tuple[Dict, str]]:
-        data = {
-            "name": "Untitled",
-            "asset_type": "template",
-            "description": "",
-            "sector": "",
-            "country": "",
-            "share-metadata": False,
-        }
+        self, bytes_io_file: typing.IO, xlsx_kobo_template_object: XLSXKoboTemplate, template_id: str = ""
+    ) -> Optional[Tuple[Dict, str]]:  # pragma: no cover
+        # TODO: not sure if this actually works
         if not template_id:
-            asset_response = self._post_request(url=self._get_url("assets/", add_limit=False), data=data)
+            data = {
+                "name": "Untitled",
+                "asset_type": "template",
+                "description": "",
+                "sector": "",
+                "country": "",
+                "share-metadata": False,
+            }
+            endpoint = "api/v2/assets"
+            query_params = f"format={self.FORMAT}"
+            url = f"{self._kpi_url}/{endpoint}?{query_params}"
+            asset_response = self._post_request(url=url, data=data)
             try:
                 asset_response.raise_for_status()
             except requests.exceptions.HTTPError as e:
@@ -135,12 +105,13 @@ def create_template_from_file(
             asset_uid = asset_response_dict.get("uid")
         else:
             asset_uid = template_id
+
         file_import_data = {
             "assetUid": asset_uid,
-            "destination": self._get_url(f"assets/{asset_uid}/", append_api=False, add_limit=False),
+            "destination": f"{self._kpi_url}/assets/{asset_uid}?format={self.FORMAT}",
         }
         file_import_response = self._post_request(
-            url=self._get_url("imports/", append_api=False, add_limit=False),
+            url=f"{self._kpi_url}/imports?format={self.FORMAT}",
             data=file_import_data,
             files={"file": bytes_io_file},  # type: ignore # FIXME
         )
@@ -149,7 +120,8 @@ def create_template_from_file(
 
         attempts = 5
         while attempts >= 0:
-            response_dict = self._handle_request(url)
+            response = self._get_request(url)
+            response_dict = response.json()
             import_status = response_dict.get("status")
             if import_status == "processing":
                 xlsx_kobo_template_object.status = XLSXKoboTemplate.PROCESSING
@@ -162,36 +134,31 @@ def create_template_from_file(
         log_and_raise("Fetching import data took too long", error_type=RetryError)
         return None
 
-    def get_all_projects_data(self) -> List:
-        if not self.business_area:
-            logger.error("Business area is not provided")
-            raise ValueError("Business area is not provided")
-        projects_url = self._get_url("assets/")
-
-        results = self._handle_paginated_results(projects_url)
-        return filter_by_owner(results, self.business_area)
+    def get_all_projects_data(self, country_code: str) -> List:
+        if not country_code:
+            raise CountryCodeNotProvided("No country code provided")
+        endpoint = f"api/v2/project-views/{self._project_views_id}/assets/"
+        query_params = f"format={self.FORMAT}&limit={self.LIMIT}"
+        query_params += f"&q=settings__country_codes__icontains:{country_code.upper()}"
+        url = f"{self._kpi_url}/{endpoint}?{query_params}"
+        return self._get_paginated_request(url)
 
     def get_single_project_data(self, uid: str) -> Dict:
-        projects_url = self._get_url(f"assets/{uid}")
-
-        return self._handle_request(projects_url)
+        endpoint = f"api/v2/assets/{uid}/"
+        query_params = f"format={self.FORMAT}&limit={self.LIMIT}"
+        url = f"{self._kpi_url}/{endpoint}?{query_params}"
+        response = self._get_request(url)
+        return response.json()
 
-    def get_project_submissions(self, uid: str, only_active_submissions: bool) -> List:
-        additional_query_params = None
+    def get_project_submissions(self, uid: str, only_active_submissions: bool) -> List[Dict]:
+        endpoint = f"api/v2/assets/{uid}/data/"
+        query_params = f"format={self.FORMAT}&limit={self.LIMIT}"
         if only_active_submissions:
             additional_query_params = 'query={"_validation_status.uid":"validation_status_approved"}'
-        submissions_url = self._get_url(
-            f"assets/{uid}/data/",
-            additional_query_params=additional_query_params,
-        )
-
-        return self._handle_paginated_results(submissions_url)
+            query_params += f"&{additional_query_params}"
+        url = f"{self._kpi_url}/{endpoint}?{query_params}"
+        return self._get_paginated_request(url)
 
-    def get_attached_file(self, url: str) -> BytesIO:
-        response = self._client.get(url=url)
-        try:
-            response.raise_for_status()
-        except requests.exceptions.HTTPError as e:
-            logger.exception(e)
-            raise
+    def get_attached_file(self, url: str) -> BytesIO:  # pragma: no cover
+        response = self._get_request(url)
         return BytesIO(response.content)

src/hct_mis_api/apps/core/kobo/common.py

@@ -111,13 +111,6 @@ def count_population(results: list, business_area: BusinessArea) -> tuple[int, i
     return total_households_count, total_individuals_count
 
 
-def filter_by_owner(data: List, business_area: BusinessArea) -> List:
-    kobo_username = business_area.kobo_username
-    if data:
-        return [element for element in data if element["owner__username"] == kobo_username]
-    return []
-
-
 def get_submission_metadata(household_data_dict: Dict) -> Dict:
     meta_fields_mapping = {
         "_uuid": "kobo_submission_uuid",

src/hct_mis_api/apps/core/models.py

@@ -73,9 +73,11 @@ class BusinessArea(NaturalKeyModel, TimeStampedUUIDModel):
     long_name = models.CharField(max_length=255)
     region_code = models.CharField(max_length=8)
     region_name = models.CharField(max_length=8)
+    # TODO: deprecated to remove in the next release
     kobo_username = models.CharField(max_length=255, null=True, blank=True)
     kobo_token = models.CharField(max_length=255, null=True, blank=True)
     kobo_url = models.URLField(max_length=255, null=True, blank=True)
+
     rapid_pro_host = models.URLField(null=True, blank=True)
     rapid_pro_payment_verification_token = models.CharField(max_length=40, null=True, blank=True)
     rapid_pro_messages_token = models.CharField(max_length=40, null=True, blank=True)

src/hct_mis_api/apps/core/schema.py

@@ -14,7 +14,7 @@
 
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import models
-from django.db.models import Q
+from django.db.models import F, Q
 
 import graphene
 from constance import config
@@ -28,7 +28,7 @@
 from hct_mis_api.apps.core.extended_connection import ExtendedConnection
 from hct_mis_api.apps.core.field_attributes.core_fields_attributes import FieldFactory
 from hct_mis_api.apps.core.field_attributes.fields_types import FILTERABLE_TYPES, Scope
-from hct_mis_api.apps.core.kobo.api import KoboAPI
+from hct_mis_api.apps.core.kobo.api import CountryCodeNotProvided, KoboAPI
 from hct_mis_api.apps.core.kobo.common import reduce_asset, reduce_assets_list
 from hct_mis_api.apps.core.languages import Language, Languages
 from hct_mis_api.apps.core.models import (
@@ -312,11 +312,11 @@ def get_collector_fields_attr_generator() -> Generator:
 
 def resolve_asset(business_area_slug: str, uid: str) -> Dict:
     try:
-        assets = KoboAPI(business_area_slug).get_single_project_data(uid)
+        assets = KoboAPI().get_single_project_data(uid)
     except ObjectDoesNotExist as e:
         logger.exception(f"Provided business area: {business_area_slug}, does not exist.")
         raise GraphQLError("Provided business area does not exist.") from e
-    except AttributeError as error:
+    except AttributeError as error:  # pragma: no cover
         logger.exception(error)
         raise GraphQLError(str(error)) from error
 
@@ -325,13 +325,18 @@ def resolve_asset(business_area_slug: str, uid: str) -> Dict:
 
 def resolve_assets_list(business_area_slug: str, only_deployed: bool = False) -> List:
     try:
-        assets = KoboAPI(business_area_slug).get_all_projects_data()
+        business_area = BusinessArea.objects.annotate(country_code=F("countries__iso_code3")).get(
+            slug=business_area_slug
+        )
+        assets = KoboAPI().get_all_projects_data(business_area.country_code)
     except ObjectDoesNotExist as e:
         logger.exception(f"Provided business area: {business_area_slug}, does not exist.")
         raise GraphQLError("Provided business area does not exist.") from e
-    except AttributeError as error:
+    except AttributeError as error:  # pragma: no cover
         logger.exception(error)
         raise GraphQLError(str(error)) from error
+    except CountryCodeNotProvided:
+        raise GraphQLError(f"Business area {business_area_slug} does not have a country code.")
 
     return reduce_assets_list(assets, only_deployed=only_deployed)
 

src/hct_mis_api/apps/registration_datahub/tasks/pull_kobo_submissions.py

@@ -22,11 +22,11 @@ class PullKoboSubmissions:
     def execute(self, kobo_import_data: KoboImportData, program: Program) -> Dict:
         kobo_import_data.status = KoboImportData.STATUS_RUNNING
         kobo_import_data.save()
-        kobo_api = KoboAPI(kobo_import_data.business_area_slug)
+        business_area = BusinessArea.objects.get(slug=kobo_import_data.business_area_slug)
+        kobo_api = KoboAPI()
         submissions = kobo_api.get_project_submissions(
             kobo_import_data.kobo_asset_id, kobo_import_data.only_active_submissions
         )
-        business_area = BusinessArea.objects.get(slug=kobo_import_data.business_area_slug)
         validator = KoboProjectImportDataInstanceValidator(program)
         skip_validate_pictures = kobo_import_data.pull_pictures is False
         validation_errors = validator.validate_everything(submissions, business_area, skip_validate_pictures)

src/hct_mis_api/apps/registration_datahub/tasks/rdi_kobo_create.py

@@ -98,7 +98,7 @@ def _handle_image_field(self, value: Any, is_flex_field: bool) -> Optional[Union
             return None
         current_download_url = attachment.get("download_url", "")
         download_url = current_download_url.replace("?format=json", "")
-        api = KoboAPI(self.business_area.slug)
+        api = KoboAPI()
         image_bytes = api.get_attached_file(download_url)
         file = File(image_bytes, name=value)
         if is_flex_field:

src/hct_mis_api/config/env.py

@@ -35,6 +35,7 @@
     "KOBO_KF_URL": (str, "https://kf-hope.unitst.org"),
     "KOBO_KC_URL": (str, "https://kc-hope.unitst.org"),
     "KOBO_MASTER_API_TOKEN": (str, "KOBO_TOKEN"),
+    "KOBO_PROJECT_VIEWS_ID": (str, ""),
     "AZURE_CLIENT_ID": (str, ""),
     "AZURE_CLIENT_SECRET": (str, ""),
     "AZURE_TENANT_KEY": (str, ""),

src/hct_mis_api/config/settings.py

@@ -100,6 +100,7 @@
 KOBO_KF_URL = env("KOBO_KF_URL")
 KOBO_KC_URL = env("KOBO_KC_URL")
 KOBO_MASTER_API_TOKEN = env("KOBO_MASTER_API_TOKEN")
+KOBO_PROJECT_VIEWS_ID = env("KOBO_PROJECT_VIEWS_ID")
 
 # Get the ENV setting. Needs to be set in .bashrc or similar.
 ENV = env("ENV")

src/hct_mis_api/migrations_script/main.py

@@ -17,6 +17,7 @@ def export_migration_info_to_csv(filename="migrations_info.csv"):
         cursor.execute("SELECT * FROM django_migrations")
         rows = cursor.fetchall()
         column_names = [desc[0] for desc in cursor.description]
+    print(rows)
 
     with open(filename, mode="w", newline="") as file:
         writer = csv.writer(file)

tests/unit/apps/core/kobo/cassettes/TestKoboAPI.test_get_project_submissions_only_active_submissions.yaml

@@ -0,0 +1,48 @@
+interactions:
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate, br
+      Authorization:
+      - token test-token
+      Connection:
+      - keep-alive
+      User-Agent:
+      - python-requests/2.31.0
+    method: GET
+    uri: https://kf.hope.unicef.org/api/v2/assets/aWnA2d5YBBDgQ5WZXpbaRe/data/?format=json&limit=10&query=%7B%22_validation_status.uid%22:%22validation_status_approved%22%7D
+  response:
+    body:
+      string: '{"count":0,"next":null,"previous":null,"results":[]}'
+    headers:
+      Allow:
+      - GET, HEAD, OPTIONS
+      Connection:
+      - keep-alive
+      Content-Language:
+      - en
+      Content-Length:
+      - '52'
+      Content-Type:
+      - application/json
+      Date:
+      - Thu, 16 May 2024 13:43:25 GMT
+      Referrer-Policy:
+      - same-origin
+      Server:
+      - nginx/1.23.0
+      Vary:
+      - Accept, Accept-Language, Cookie, Origin
+      X-Content-Type-Options:
+      - nosniff
+      X-Frame-Options:
+      - DENY
+      X-KoBoNaUt:
+      - afg_admin
+    status:
+      code: 200
+      message: OK
+version: 1

tests/unit/apps/core/kobo/test_api.py

@@ -0,0 +1,46 @@
+from typing import Any
+
+import pytest
+
+from hct_mis_api.apps.core.kobo.api import CountryCodeNotProvided, KoboAPI
+
+
+class TestKoboAPI:
+    @pytest.fixture(autouse=True)
+    def use_kobo_master_token(self, settings: Any) -> None:
+        settings.KOBO_MASTER_API_TOKEN = "test-token"
+        settings.KOBO_KF_URL = "https://kf.hope.unicef.org"
+        settings.KOBO_PROJECT_VIEWS_ID = "pvEsUUfAgYyyV7jpR6i3FvM"
+
+    def test_get_all_projects_with_not_provided_country_code(self) -> None:
+        service = KoboAPI()
+
+        with pytest.raises(CountryCodeNotProvided):
+            service.get_all_projects_data(None)  # type: ignore[arg-type]
+
+    @pytest.mark.vcr()
+    def test_get_all_projects_filter_by_country_code(self) -> None:
+        service = KoboAPI()
+        projects = service.get_all_projects_data("AFG")
+        assert len(projects) == 117
+        assert "AFG" in projects[0]["settings"]["country_codes"]
+
+    @pytest.mark.vcr()
+    def test_get_single_project_data(self) -> None:
+        service = KoboAPI()
+        project_data = service.get_single_project_data("aWnA2d5YBBDgQ5WZXpbaRe")
+        assert project_data
+
+    @pytest.mark.vcr()
+    def test_get_project_submissions(self) -> None:
+        service = KoboAPI()
+        service.LIMIT = 10
+        submissions = service.get_project_submissions("aWnA2d5YBBDgQ5WZXpbaRe", only_active_submissions=False)
+        assert len(submissions) == 10
+
+    @pytest.mark.vcr()
+    def test_get_project_submissions_only_active_submissions(self) -> None:
+        service = KoboAPI()
+        service.LIMIT = 10
+        submissions = service.get_project_submissions("aWnA2d5YBBDgQ5WZXpbaRe", only_active_submissions=True)
+        assert len(submissions) == 0

tests/unit/apps/core/kobo/test_resolve_asset.py

@@ -0,0 +1,35 @@
+import datetime
+from typing import Any
+
+import pytest
+from dateutil.tz import tzlocal
+
+from hct_mis_api.apps.core.fixtures import create_afghanistan
+from hct_mis_api.apps.core.schema import resolve_asset
+
+pytestmark = pytest.mark.django_db(transaction=True)
+
+
+class TestResolveAsset:
+    @pytest.fixture(autouse=True)
+    def use_kobo_master_token(self, settings: Any) -> None:
+        settings.KOBO_MASTER_API_TOKEN = "token-from-env"
+        settings.KOBO_KF_URL = "https://kf.hope.unicef.org"
+        settings.KOBO_PROJECT_VIEWS_ID = "pvEsUUfAgYyyV7jpR6i3FvM"
+
+    @pytest.mark.vcr()
+    def test_resolve_asset(self) -> None:
+        create_afghanistan()
+
+        result = resolve_asset("afghanistan", "aWnA2d5YBBDgQ5WZXpbaRe")
+        assert result == {
+            "asset_type": "survey",
+            "country": "Afghanistan",
+            "date_modified": datetime.datetime(2023, 10, 4, 10, 38, 23, 610534, tzinfo=tzlocal()),
+            "deployment_active": True,
+            "has_deployment": True,
+            "id": "aWnA2d5YBBDgQ5WZXpbaRe",
+            "name": "PMU-REG-Cash_for Education-Nooristan-ACTED-Dec-2022",
+            "sector": "Humanitarian - Coordination / Information Management",
+            "xls_link": "https://kf.hope.unicef.org/api/v2/assets/aWnA2d5YBBDgQ5WZXpbaRe/?format=json.xls",
+        }

tests/unit/apps/core/kobo/test_resolve_assets_list.py

@@ -0,0 +1,49 @@
+import datetime
+from typing import Any
+
+import pytest
+from dateutil.tz import tzlocal
+from graphql import GraphQLError
+
+from hct_mis_api.apps.core.fixtures import create_afghanistan
+from hct_mis_api.apps.core.schema import resolve_assets_list
+from hct_mis_api.apps.geo.fixtures import CountryFactory
+
+pytestmark = pytest.mark.django_db(transaction=True)
+
+
+class TestResolveAssetsList:
+    @pytest.fixture(autouse=True)
+    def use_kobo_master_token(self, settings: Any) -> None:
+        settings.KOBO_MASTER_API_TOKEN = "test-token"
+        settings.KOBO_KF_URL = "https://kf.hope.unicef.org"
+        settings.KOBO_PROJECT_VIEWS_ID = "pvEsUUfAgYyyV7jpR6i3FvM"
+
+    def test_resolve_assets_list_wrong_business_area(self) -> None:
+        with pytest.raises(GraphQLError):
+            resolve_assets_list("wrong_business_area")
+
+    def test_resolve_assets_list_without_country_code(self) -> None:
+        create_afghanistan()
+        with pytest.raises(GraphQLError):
+            resolve_assets_list("afghanistan")
+
+    @pytest.mark.vcr()
+    def test_resolve_assets_list(self) -> None:
+        business_area = create_afghanistan()
+        country = CountryFactory()
+        business_area.countries.add(country)
+
+        result = resolve_assets_list("afghanistan")
+        assert len(result) == 68
+        assert result[0] == {
+            "asset_type": "survey",
+            "country": "Afghanistan",
+            "date_modified": datetime.datetime(2023, 10, 4, 10, 38, 23, 610534, tzinfo=tzlocal()),
+            "deployment_active": True,
+            "has_deployment": True,
+            "id": "aWnA2d5YBBDgQ5WZXpbaRe",
+            "name": "PMU-REG-Cash_for Education-Nooristan-ACTED-Dec-2022",
+            "sector": "Humanitarian - Coordination / Information Management",
+            "xls_link": "https://kf.hope.unicef.org/api/v2/assets/aWnA2d5YBBDgQ5WZXpbaRe/?format=json.xls",
+        }

tests/unit/apps/registration_datahub/test_pull_kobo_submissions.py

@@ -0,0 +1,50 @@
+import json
+from pathlib import Path
+from typing import Any
+from unittest import mock
+
+from django.conf import settings
+
+import pytest
+
+from hct_mis_api.apps.core.fixtures import create_afghanistan
+from hct_mis_api.apps.program.fixtures import ProgramFactory
+from hct_mis_api.apps.registration_data.models import KoboImportData
+from hct_mis_api.apps.registration_datahub.tasks.pull_kobo_submissions import (
+    PullKoboSubmissions,
+)
+
+pytestmark = pytest.mark.django_db(databases=(("default",)), transaction=True)
+
+
+class TestPullKoboSubmissions:
+    @pytest.fixture(autouse=True)
+    def use_kobo_master_token(self, settings: Any) -> None:
+        settings.KOBO_MASTER_API_TOKEN = "token-from-env"
+        settings.KOBO_KF_URL = "https://kf.hope.unicef.org"
+
+    def test_pull_kobo_submissions(self) -> None:
+        create_afghanistan()
+        kobo_import_data = KoboImportData(
+            status=KoboImportData.STATUS_PENDING,
+            business_area_slug="afghanistan",
+            data_type=KoboImportData.JSON,
+            kobo_asset_id="aWnA2d5YBBDgQ5WZXpbaRe",
+            only_active_submissions=False,
+        )
+        kobo_import_data.save()
+        program = ProgramFactory()
+
+        content = Path(
+            f"{settings.TESTS_ROOT}/apps/registration_datahub/test_file/kobo_submissions_collectors.json"
+        ).read_text()
+        content = json.loads(content)
+        with mock.patch(
+            "hct_mis_api.apps.registration_datahub.tasks.pull_kobo_submissions.KoboAPI.get_project_submissions",
+            return_value=content,
+        ):
+            service = PullKoboSubmissions()
+            result = service.execute(kobo_import_data, program)
+
+        kobo_import_data.refresh_from_db()
+        assert kobo_import_data.id == result["kobo_import_data_id"]