chg ! enable parsing of API IDs in UUID or Base64 format (#104)

Author: vitali-yanushchyk-valor

src/country_workspace/contrib/hope/geo.py

@@ -7,6 +7,7 @@
 
 from country_workspace.cache.manager import cache_manager
 from country_workspace.state import state
+from country_workspace.utils.fields import extract_uuid
 
 from ...exceptions import RemoteError
 
@@ -69,7 +70,7 @@ def __init__(self, **kwargs: Any) -> None:
 
     def validate_with_parent(self, parent_value: Any, value: Any) -> None:
         choices = self.get_choices_for_parent_value(parent_value, only_codes=True)
-        if parent_value and value not in choices:
+        if parent_value and self.prepare_value(value) not in choices:
             raise ValidationError("Not valid child for selected parent")
 
     def get_choices_for_parent_value(
@@ -83,13 +84,12 @@ def get_choices_for_parent_value(
         ):
             return [] if only_codes else [("", "")]
 
-        self.code_to_id = {r["p_code"]: str(r["id"]) for r in filtered}
+        self.code_to_id = {r["p_code"]: str(extract_uuid(str(r["id"]), "Area:")) for r in filtered}
         self.id_to_code = {v: k for k, v in self.code_to_id.items()}
 
-        return {
-            True: list(self.id_to_code),
-            False: [("", ""), *[(r["p_code"], f"{r['p_code']} - {r['name']}") for r in filtered]],
-        }[only_codes]
+        if only_codes:
+            return list(self.code_to_id)
+        return [("", ""), *[(r["p_code"], f"{r['p_code']} - {r['name']}") for r in filtered]]
 
     def prepare_value(self, value: Any) -> str | None:
         val = super().prepare_value(value)

src/country_workspace/contrib/hope/push.py

@@ -138,7 +138,7 @@ def process_batch_response(self, response: dict | None, batch_ids: list[int]) ->
 
         Args:
             response (dict | None): API response.
-            batch_ids (list[int]): List of household IDs in the batch.
+            batch_ids (list[int]): List of IDs for the batch that was pushed.
 
         Returns:
             list[int]: List of successfully processed IDs.

src/country_workspace/utils/fields.py

@@ -1,6 +1,9 @@
+import binascii
 from collections.abc import Callable, Mapping
 from functools import reduce
 from typing import Any
+from base64 import b64decode
+from uuid import UUID
 
 from django.utils import timezone
 
@@ -68,3 +71,32 @@ def map_fields(fields: dict[str, str]) -> dict[str, str]:
 
     """
     return {TO_MAP_FIELDS.get(k, k): v for k, v in fields.items()}
+
+
+def extract_uuid(value: str, prefix: str | None = None) -> UUID:
+    """Extract a UUID from the given string.
+
+    - If `value` is already a UUID, returns it unchanged.
+    - Otherwise attempts Base64-decoding and stripping an optional `prefix`.
+
+    """
+    if not isinstance(value, str):
+        raise TypeError("value must be a str")
+    if prefix is not None and not isinstance(prefix, str):
+        raise TypeError("prefix must be a str or None")
+
+    try:
+        return UUID(value)
+    except ValueError:
+        pass
+
+    try:
+        decoded = b64decode(value, validate=True).decode()
+    except (binascii.Error, UnicodeDecodeError):
+        raise ValueError(f"value is neither a valid UUID nor valid Base64: {value!r}")
+
+    raw = decoded.removeprefix(prefix or "")
+    try:
+        return UUID(raw)
+    except ValueError:
+        raise ValueError(f"decoded data is not a valid UUID: {raw!r}")

tests/utils/test_utils_fields.py

@@ -3,13 +3,16 @@
 import pytest
 from django.core.files.uploadedfile import SimpleUploadedFile
 from pytest_mock import MockerFixture
+from base64 import b64encode
+from uuid import uuid4, UUID
 
 from country_workspace.contrib.kobo.api.data.helpers import VALUE_FORMAT
 from country_workspace.utils.fields import (
     clean_field_name,
     TO_REMOVE_VALUES,
     clean_field_names,
     map_fields,
+    extract_uuid,
 )
 from country_workspace.utils.flex_fields import Base64ImageInput, Base64ImageField
 
@@ -72,3 +75,38 @@ def test_base64_image_field_content_is_encoded(mocker: MockerFixture) -> None:
     instance = Mock(spec=Base64ImageField)
 
     assert Base64ImageField.clean(instance, file) == VALUE_FORMAT.format(mimetype=content_type, content=data)
+
+
+FAKE_UUID = uuid4()
+FAKE_PREFIX = "Area:"
+ENC_B64_PREF = b64encode(f"{FAKE_PREFIX}{FAKE_UUID}".encode()).decode()
+ENC_B64 = b64encode(str(FAKE_UUID).encode()).decode()
+ENC_B64_BAD = b64encode("hello-world".encode()).decode()
+
+
+@pytest.mark.parametrize(
+    ("value", "prefix", "expected"),
+    [
+        (str(FAKE_UUID), None, FAKE_UUID),
+        (ENC_B64_PREF, FAKE_PREFIX, FAKE_UUID),
+        (ENC_B64, None, FAKE_UUID),
+    ],
+    ids=["raw-uuid", "b64-with-prefix", "b64-no-prefix"],
+)
+def test_extract_uuid_success(value: str, prefix: str | None, expected: UUID) -> None:
+    assert extract_uuid(value, prefix) == expected
+
+
+@pytest.mark.parametrize(
+    ("value", "prefix", "exc_type"),
+    [
+        ("not-a-uuid-or-base64", None, ValueError),
+        (ENC_B64_BAD, None, ValueError),
+        (123, None, TypeError),
+        (str(FAKE_UUID), 123, TypeError),
+    ],
+    ids=["invalid-string", "b64-not-uuid", "value-not-str", "prefix-not-str"],
+)
+def test_extract_uuid_errors(value: str | int, prefix: str | int | None, exc_type: type[Exception]) -> None:
+    with pytest.raises(exc_type):
+        extract_uuid(value, prefix)