From 8632ae1421dbc79705917694d2ff7cd17a75a14b Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Sun, 17 May 2026 20:44:45 +0700 Subject: [PATCH 1/2] fix(spp_dci_client_dr): parse DCI v1.0.0 spec envelope and record fields The sync DR path treated `search_response[*].data` as a flat record, and both sync and callback paths read flat fields that do not exist in spec-shape records (`disability_status` string, `disability_details[*].impairment_type`). A spec-conformant DR server was silently classified as "no disability". Route both paths through a shared `dr_parsing` module so any spec-shape envelope is correctly unwrapped and record fields are derived from `disability_status` and `disability_details`. Date coercion (ISO datetime / `Z` suffix) is centralized so both sync and callback hand the ORM a `date` object. Co-Authored-By: Claude Opus 4.7 --- spp_dci_client_dr/routers/callback.py | 21 +- spp_dci_client_dr/services/__init__.py | 1 + spp_dci_client_dr/services/dr_parsing.py | 151 ++++++++ spp_dci_client_dr/services/dr_service.py | 120 ++----- spp_dci_client_dr/tests/__init__.py | 2 + spp_dci_client_dr/tests/test_callback.py | 205 +++++++++++ spp_dci_client_dr/tests/test_dr_parsing.py | 294 ++++++++++++++++ spp_dci_client_dr/tests/test_dr_service.py | 379 ++++++++++++++------- 8 files changed, 948 insertions(+), 225 deletions(-) create mode 100644 spp_dci_client_dr/services/dr_parsing.py create mode 100644 spp_dci_client_dr/tests/test_callback.py create mode 100644 spp_dci_client_dr/tests/test_dr_parsing.py diff --git a/spp_dci_client_dr/routers/callback.py b/spp_dci_client_dr/routers/callback.py index cdc8e443..8d58cffe 100644 --- a/spp_dci_client_dr/routers/callback.py +++ b/spp_dci_client_dr/routers/callback.py @@ -6,6 +6,7 @@ from datetime import UTC, datetime from typing import Annotated +from odoo import fields from odoo.api import Environment from odoo.addons.fastapi.dependencies import odoo_env @@ -14,6 +15,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status from ..middleware.signature import verify_dr_signature +from ..services.dr_parsing import extract_disability_data _logger = logging.getLogger(__name__) @@ -245,11 +247,8 @@ def _update_disability_status( # Use sudo() for API access - authentication is handled by signature verification DisabilityStatus = env["spp.dci.disability.status"].sudo() # nosemgrep: odoo-sudo-without-context - # Extract disability data from record - has_disability = record.get("has_disability", False) or record.get("is_pwd", False) - disability_types = record.get("disability_types", []) - functional_scores = record.get("functional_scores", {}) - assessment_date = record.get("assessment_date") + # Extract disability data using spec-aware parsing + extracted = extract_disability_data(record) # Find existing status existing = DisabilityStatus.search( @@ -259,16 +258,14 @@ def _update_disability_status( vals = { "partner_id": partner.id, - "has_disability": has_disability, - "disability_types": json.dumps(disability_types) if isinstance(disability_types, list) else disability_types, - "functional_scores": json.dumps(functional_scores) - if isinstance(functional_scores, dict) - else functional_scores, - "assessment_date": assessment_date, + "has_disability": extracted["has_disability"], + "disability_types": json.dumps(extracted["disability_types"]), + "functional_scores": json.dumps(extracted["functional_scores"]), + "assessment_date": extracted["assessment_date"], "source_registry": source_registry, "raw_data": json.dumps(record), "state": "synced", - "last_sync_date": datetime.now(UTC), + "last_sync_date": fields.Datetime.now(), "synced_by": env.user.id, } diff --git a/spp_dci_client_dr/services/__init__.py b/spp_dci_client_dr/services/__init__.py index 79f65215..c6e1ff5c 100644 --- a/spp_dci_client_dr/services/__init__.py +++ b/spp_dci_client_dr/services/__init__.py @@ -1 +1,2 @@ +from . import dr_parsing from . import dr_service diff --git a/spp_dci_client_dr/services/dr_parsing.py b/spp_dci_client_dr/services/dr_parsing.py new file mode 100644 index 00000000..9d27c979 --- /dev/null +++ b/spp_dci_client_dr/services/dr_parsing.py @@ -0,0 +1,151 @@ +# Part of OpenSPP. See LICENSE file for full copyright and licensing details. +"""Pure parsing helpers for DCI v1.0.0 Disability Registry records. + +All functions are stateless and require no Odoo env, making them +independently testable and easy to reuse. +""" + +import logging +from datetime import date, datetime + +_logger = logging.getLogger(__name__) + +# Status values that indicate the person is registered as disabled. +# The SP DCI v1.0.0 spec does not declare an enum; these are the +# spec-aligned workflow tokens. Empty / missing / other values trigger +# the impairment-list fallback in extract_disability_data. +_APPROVED_STATUSES = {"approved", "registered"} + +# Status values that explicitly reject disability registration. +_REJECTED_STATUSES = {"rejected", "denied"} + + +def _coerce_date(value) -> date | None: + """Coerce a DCI date/datetime value into a ``date`` object. + + Accepts ISO date strings (``YYYY-MM-DD``), ISO datetime strings with an + optional trailing ``Z`` (``YYYY-MM-DDTHH:MM:SSZ``), naive/aware datetimes, + and date objects. Returns ``None`` for empty input or unparseable values + (with a WARNING logged). + + For tz-aware inputs, the local wall-clock date is returned; + no UTC normalization is applied. + """ + if not value: + return None + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + try: + return datetime.fromisoformat(str(value).removesuffix("Z")).date() + except ValueError: + _logger.warning("Could not parse date from value: %r", value) + return None + + +def unwrap_search_data(data) -> list: + """Extract the reg_records list from a DCI v1.0.0 search response data envelope. + + Args: + data: The value of ``search_response[*].data`` from the API response. + Expected to be a dict with a ``reg_records`` key per the spec. + + Returns: + list: The contents of ``data["reg_records"]``, or an empty list when + the envelope is absent, empty, or malformed. + """ + if data is None: + return [] + + if not isinstance(data, dict): + _logger.warning( + "Unexpected type for search response data envelope: %s; expected dict", + type(data).__name__, + ) + return [] + + if not data: + return [] + + records = data.get("reg_records") + if records is None: + return [] + if not isinstance(records, list): + _logger.warning( + "Unexpected type for reg_records: %s; expected list", + type(records).__name__, + ) + return [] + return records + + +def extract_disability_data(record: dict) -> dict: + """Extract structured disability information from a DCI v1.0.0 record. + + Args: + record: A single record dict from ``reg_records``. + + Returns: + dict with keys: + - ``has_disability`` (bool) + - ``disability_types`` (list[str]) + - ``functional_scores`` (dict, always ``{}``: spec has no numeric scores) + - ``assessment_date`` (``date`` | None) + - ``source_registry`` (str | None) + - ``raw_data`` (the input record, unchanged) + """ + # Extract impairment types from disability_details. + # Use `or []` so an explicit null on the wire does not crash the loop. + details = record.get("disability_details") or [] + disability_types = [d["impairment_type"] for d in details if isinstance(d, dict) and d.get("impairment_type")] + + # Resolve has_disability from the disability_status string + status_str = str(record.get("disability_status", "")).strip().lower() + + if status_str in _APPROVED_STATUSES: + has_disability = True + elif status_str in _REJECTED_STATUSES: + has_disability = False + elif status_str == "": + # No explicit status: fall back to impairment list presence + has_disability = bool(disability_types) + else: + _logger.warning( + "Unknown disability_status value: %s; falling back to impairment list", + record.get("disability_status"), + ) + has_disability = bool(disability_types) + + # Assessment date: prefer last_updated, fall back to registration_date. + # The spec uses ISO datetime strings; coerce to a date for the ORM. + assessment_date = _coerce_date(record.get("last_updated") or record.get("registration_date")) + + # Source registry: prefer source_registry, fall back to registry_name + source_registry = record.get("source_registry") or record.get("registry_name") + + return { + "has_disability": has_disability, + "disability_types": disability_types, + "functional_scores": {}, + "assessment_date": assessment_date, + "source_registry": source_registry, + "raw_data": record, + } + + +def extract_functional_scores(record: dict) -> dict: + """Return functional assessment scores from a DCI v1.0.0 record. + + The DCI v1.0.0 spec does not define numeric functional scores. + ``impairment_level`` is a free-text string, not a number. + This function always returns ``{}`` and exists as a hook for future + spec versions that may introduce numeric scoring. + + Args: + record: A single record dict from ``reg_records``. + + Returns: + dict: Always ``{}``. + """ + return {} diff --git a/spp_dci_client_dr/services/dr_service.py b/spp_dci_client_dr/services/dr_service.py index fdf09d19..79d8223d 100644 --- a/spp_dci_client_dr/services/dr_service.py +++ b/spp_dci_client_dr/services/dr_service.py @@ -9,6 +9,8 @@ from odoo.addons.spp_dci_client.services import DCIClient +from .dr_parsing import extract_disability_data, extract_functional_scores, unwrap_search_data + _logger = logging.getLogger(__name__) @@ -122,14 +124,11 @@ def get_disability_status(self, partner) -> dict | None: # Extract first result search_response = message["search_response"][0] - if "data" not in search_response or not search_response["data"]: + records = unwrap_search_data(search_response.get("data")) + if not records: return None - record_data = ( - search_response["data"][0] if isinstance(search_response["data"], list) else search_response["data"] - ) - - # Extract disability information + record_data = records[0] disability_data = self._extract_disability_data(record_data) _logger.info( @@ -207,14 +206,11 @@ def get_functional_assessment(self, identifier_type: str, identifier_value: str) # Extract first result search_response = message["search_response"][0] - if "data" not in search_response or not search_response["data"]: + records = unwrap_search_data(search_response.get("data")) + if not records: return None - record_data = ( - search_response["data"][0] if isinstance(search_response["data"], list) else search_response["data"] - ) - - # Extract functional scores + record_data = records[0] scores = self._extract_functional_scores(record_data) _logger.info( @@ -389,103 +385,29 @@ def _get_partner_identifier(self, partner): return None def _extract_disability_data(self, record_data: dict) -> dict: - """Extract disability information from DCI record data. + """Extract disability information from a DCI v1.0.0 record. + + Delegates to the stateless module-level helper in dr_parsing. Args: - record_data: DCI record data from search response + record_data: A single record dict from reg_records Returns: dict: Extracted disability data """ - disability_data = { - "has_disability": False, - "disability_types": [], - "functional_scores": {}, - "raw_data": record_data, - } - - # Check for disability flag - if "has_disability" in record_data: - disability_data["has_disability"] = bool(record_data["has_disability"]) - elif "is_pwd" in record_data: - disability_data["has_disability"] = bool(record_data["is_pwd"]) - - # Extract disability types - if "disability_types" in record_data: - types_data = record_data["disability_types"] - if isinstance(types_data, list): - disability_data["disability_types"] = types_data - elif isinstance(types_data, str): - # Handle comma-separated string - disability_data["disability_types"] = [t.strip() for t in types_data.split(",") if t.strip()] - - # Extract functional scores - disability_data["functional_scores"] = self._extract_functional_scores(record_data) - - # Extract assessment date - if "assessment_date" in record_data: - disability_data["assessment_date"] = record_data["assessment_date"] - elif "disability_assessment_date" in record_data: - disability_data["assessment_date"] = record_data["disability_assessment_date"] - - # Extract source registry - if "source_registry" in record_data: - disability_data["source_registry"] = record_data["source_registry"] - elif "registry_name" in record_data: - disability_data["source_registry"] = record_data["registry_name"] - - return disability_data + return extract_disability_data(record_data) def _extract_functional_scores(self, record_data: dict) -> dict: - """Extract functional assessment scores from record data. + """Return functional assessment scores from a DCI v1.0.0 record. + + Delegates to the stateless module-level helper in dr_parsing. + The DCI v1.0.0 spec has no numeric functional scores, so this always + returns ``{}``. Args: - record_data: DCI record data + record_data: A single record dict from reg_records Returns: - dict: Functional scores by domain - Example: {'Vision': 3, 'Hearing': 1, 'Mobility': 4, ...} + dict: Always ``{}`` """ - scores = {} - - # Try to extract from functional_scores field - if "functional_scores" in record_data: - scores_data = record_data["functional_scores"] - if isinstance(scores_data, dict): - scores = scores_data - elif isinstance(scores_data, str): - # Try to parse JSON string - try: - scores = json.loads(scores_data) - except json.JSONDecodeError: - _logger.warning("Failed to parse functional_scores JSON") - - # Try to extract individual domain scores - functional_domains = [ - "Vision", - "Hearing", - "Mobility", - "Cognition", - "SelfCare", - "Communication", - ] - - for domain in functional_domains: - # Try various field name formats - for field_name in [ - f"functional_{domain.lower()}", - f"{domain.lower()}_score", - domain.lower(), - domain, - ]: - if field_name in record_data and record_data[field_name]: - try: - scores[domain] = int(record_data[field_name]) - except (ValueError, TypeError): - _logger.warning( - "Invalid functional score for %s: %s", - domain, - record_data[field_name], - ) - - return scores + return extract_functional_scores(record_data) diff --git a/spp_dci_client_dr/tests/__init__.py b/spp_dci_client_dr/tests/__init__.py index 015e6a78..86384bbd 100644 --- a/spp_dci_client_dr/tests/__init__.py +++ b/spp_dci_client_dr/tests/__init__.py @@ -1,4 +1,6 @@ # Part of OpenSPP. See LICENSE file for full copyright and licensing details. +from . import test_callback from . import test_disability_status +from . import test_dr_parsing from . import test_dr_service diff --git a/spp_dci_client_dr/tests/test_callback.py b/spp_dci_client_dr/tests/test_callback.py new file mode 100644 index 00000000..d54338f0 --- /dev/null +++ b/spp_dci_client_dr/tests/test_callback.py @@ -0,0 +1,205 @@ +# Part of OpenSPP. See LICENSE file for full copyright and licensing details. +"""Tests for DR callback endpoint processing.""" + +import json +from unittest.mock import patch + +from odoo.tests import TransactionCase, tagged + + +@tagged("post_install", "-at_install") +class TestDRCallbackProcessing(TransactionCase): + """Tests for _process_dr_search_result and _update_disability_status.""" + + @classmethod + def setUpClass(cls): + """Set up test data.""" + super().setUpClass() + + cls.Partner = cls.env["res.partner"] + cls.DisabilityStatus = cls.env["spp.dci.disability.status"] + cls.VocabularyCode = cls.env["spp.vocabulary.code"] + + # Get or create ID type vocabulary + id_type_vocab = cls.env["spp.vocabulary"].search([("namespace_uri", "=", "urn:openspp:vocab:id-type")], limit=1) + if not id_type_vocab: + id_type_vocab = cls.env["spp.vocabulary"].create( + { + "name": "ID Type", + "namespace_uri": "urn:openspp:vocab:id-type", + } + ) + + cls.id_type_uin = cls.VocabularyCode.create( + { + "vocabulary_id": id_type_vocab.id, + "code": "UIN_DR_TEST", + "display": "Universal Identification Number", + "target_type": "individual", + "is_local": True, + } + ) + + cls.partner = cls.Partner.create( + { + "name": "CB Test Person", + "is_registrant": True, + "is_group": False, + } + ) + + cls.IdRecord = cls.env["spp.registry.id"] + cls.IdRecord.create( + { + "partner_id": cls.partner.id, + "id_type_id": cls.id_type_uin.id, + "value": "UIN-CB-001", + } + ) + + def _make_spec_result(self, disability_status="Approved", impairment_types=None): + """Build a DCI v1.0.0 spec-shaped search result item.""" + if impairment_types is None: + impairment_types = ["Vision"] + return { + "status": "succ", + "data": { + "version": "1.0.0", + "reg_record_type": "PERSON", + "reg_records": [ + { + "personal_details": {"identifier": "UIN-CB-001"}, + "identifier": [ + { + "identifier_type": "UIN_DR_TEST", + "identifier_value": "UIN-CB-001", + } + ], + "disability_status": disability_status, + "disability_details": [{"impairment_type": t} for t in impairment_types], + "registration_date": "2024-01-01T00:00:00Z", + } + ], + }, + } + + def test_approved_status_creates_disability_record(self): + """Spec-shape approved result creates spp.dci.disability.status with has_disability=True.""" + from odoo.addons.spp_dci_client_dr.routers.callback import _process_dr_search_result + + result = self._make_spec_result(disability_status="Approved", impairment_types=["Vision"]) + + with patch( + "odoo.addons.spp_dci_client_dr.routers.callback._find_partner_by_identifier", + return_value=self.partner, + ): + _process_dr_search_result(self.env, result, "test-source-registry") + + status = self.DisabilityStatus.search([("partner_id", "=", self.partner.id)]) + self.assertEqual(len(status), 1) + self.assertTrue(status.has_disability) + self.assertEqual(status.state, "synced") + + types_list = json.loads(status.disability_types) + self.assertIn("Vision", types_list) + + def test_rejected_status_creates_record_with_false(self): + """Spec-shape rejected result creates spp.dci.disability.status with has_disability=False.""" + from odoo.addons.spp_dci_client_dr.routers.callback import _process_dr_search_result + + # Use a fresh partner so there is no pre-existing disability record + rejected_partner = self.Partner.create( + { + "name": "CB Rejected Person", + "is_registrant": True, + "is_group": False, + } + ) + + result = self._make_spec_result(disability_status="Rejected", impairment_types=["Hearing"]) + + with patch( + "odoo.addons.spp_dci_client_dr.routers.callback._find_partner_by_identifier", + return_value=rejected_partner, + ): + _process_dr_search_result(self.env, result, "test-source-registry") + + status = self.DisabilityStatus.search([("partner_id", "=", rejected_partner.id)]) + self.assertEqual(len(status), 1) + self.assertFalse(status.has_disability) + self.assertEqual(status.state, "synced") + + def test_non_success_status_is_skipped(self): + """Search result with non-success status does not create a disability record.""" + from odoo.addons.spp_dci_client_dr.routers.callback import _process_dr_search_result + + skipped_partner = self.Partner.create( + { + "name": "CB Skipped Person", + "is_registrant": True, + "is_group": False, + } + ) + + result = { + "status": "rjct", + "data": { + "version": "1.0.0", + "reg_record_type": "PERSON", + "reg_records": [ + { + "identifier": [{"identifier_type": "UIN_DR_TEST", "identifier_value": "UIN-CB-SKIP"}], + "disability_status": "Approved", + } + ], + }, + } + + with patch( + "odoo.addons.spp_dci_client_dr.routers.callback._find_partner_by_identifier", + return_value=skipped_partner, + ) as mock_find: + _process_dr_search_result(self.env, result, "test-source-registry") + + # No record should be created + status = self.DisabilityStatus.search([("partner_id", "=", skipped_partner.id)]) + self.assertEqual(len(status), 0) + mock_find.assert_not_called() + + def test_update_overwrites_existing_record(self): + """Calling _process_dr_search_result twice updates the existing record.""" + from odoo.addons.spp_dci_client_dr.routers.callback import _process_dr_search_result + + update_partner = self.Partner.create( + { + "name": "CB Update Person", + "is_registrant": True, + "is_group": False, + } + ) + + # First call: approved + result_approved = self._make_spec_result(disability_status="Approved") + with patch( + "odoo.addons.spp_dci_client_dr.routers.callback._find_partner_by_identifier", + return_value=update_partner, + ): + _process_dr_search_result(self.env, result_approved, "test-source-registry") + + status = self.DisabilityStatus.search([("partner_id", "=", update_partner.id)]) + self.assertEqual(len(status), 1) + self.assertTrue(status.has_disability) + + # Second call: rejected + result_rejected = self._make_spec_result(disability_status="Rejected") + with patch( + "odoo.addons.spp_dci_client_dr.routers.callback._find_partner_by_identifier", + return_value=update_partner, + ): + _process_dr_search_result(self.env, result_rejected, "test-source-registry") + + # Should still be one record, now with has_disability=False + status.invalidate_recordset() + all_status = self.DisabilityStatus.search([("partner_id", "=", update_partner.id)]) + self.assertEqual(len(all_status), 1) + self.assertFalse(all_status.has_disability) diff --git a/spp_dci_client_dr/tests/test_dr_parsing.py b/spp_dci_client_dr/tests/test_dr_parsing.py new file mode 100644 index 00000000..efd88c2c --- /dev/null +++ b/spp_dci_client_dr/tests/test_dr_parsing.py @@ -0,0 +1,294 @@ +# Part of OpenSPP. See LICENSE file for full copyright and licensing details. +"""Pure-function tests for dr_parsing helpers. + +These tests do not need the Odoo env, but inherit from ``TransactionCase`` +so the Odoo test runner picks them up at post_install.""" + +from datetime import date + +from odoo.tests import TransactionCase, tagged + + +@tagged("post_install", "-at_install") +class TestUnwrapSearchData(TransactionCase): + """Tests for unwrap_search_data.""" + + def _fn(self, data): + from odoo.addons.spp_dci_client_dr.services.dr_parsing import unwrap_search_data + + return unwrap_search_data(data) + + def test_none_returns_empty_list(self): + self.assertEqual(self._fn(None), []) + + def test_empty_dict_returns_empty_list(self): + self.assertEqual(self._fn({}), []) + + def test_spec_envelope_returns_reg_records(self): + data = { + "version": "1.0.0", + "reg_record_type": "PERSON", + "reg_records": [{"x": 1}, {"x": 2}], + } + self.assertEqual(self._fn(data), [{"x": 1}, {"x": 2}]) + + def test_spec_envelope_empty_reg_records_returns_empty_list(self): + data = { + "version": "1.0.0", + "reg_record_type": "PERSON", + "reg_records": [], + } + self.assertEqual(self._fn(data), []) + + def test_spec_envelope_missing_reg_records_returns_empty_list(self): + data = { + "version": "1.0.0", + "reg_record_type": "PERSON", + } + self.assertEqual(self._fn(data), []) + + def test_spec_envelope_none_reg_records_returns_empty_list(self): + data = { + "version": "1.0.0", + "reg_records": None, + } + self.assertEqual(self._fn(data), []) + + def test_unexpected_list_returns_empty_and_warns(self): + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING"): + result = self._fn([{"a": 1}]) + self.assertEqual(result, []) + + def test_unexpected_int_returns_empty_and_warns(self): + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING"): + result = self._fn(42) + self.assertEqual(result, []) + + def test_unexpected_string_returns_empty_and_warns(self): + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING"): + result = self._fn("not-a-dict") + self.assertEqual(result, []) + + def test_reg_records_non_list_returns_empty_and_warns(self): + """If reg_records is present but not a list, return [] and warn rather than + passing garbage downstream where records[0].get(...) would crash.""" + data = { + "version": "1.0.0", + "reg_records": "oops", + } + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING"): + result = self._fn(data) + self.assertEqual(result, []) + + +@tagged("post_install", "-at_install") +class TestExtractDisabilityData(TransactionCase): + """Tests for extract_disability_data.""" + + def _fn(self, record): + from odoo.addons.spp_dci_client_dr.services.dr_parsing import extract_disability_data + + return extract_disability_data(record) + + def test_approved_with_impairments(self): + record = { + "disability_status": "Approved", + "disability_details": [{"impairment_type": "Physical"}], + } + result = self._fn(record) + self.assertTrue(result["has_disability"]) + self.assertEqual(result["disability_types"], ["Physical"]) + self.assertEqual(result["functional_scores"], {}) + self.assertIs(result["raw_data"], record) + + def test_approved_no_disability_details(self): + record = { + "disability_status": "Approved", + } + result = self._fn(record) + self.assertTrue(result["has_disability"]) + self.assertEqual(result["disability_types"], []) + + def test_rejected_impairments_still_false(self): + """Explicit rejection overrides impairment list.""" + record = { + "disability_status": "Rejected", + "disability_details": [{"impairment_type": "Physical"}], + } + result = self._fn(record) + self.assertFalse(result["has_disability"]) + + def test_empty_status_with_impairments_is_true(self): + """Empty status is ambiguous; fall back to impairment list signal.""" + record = { + "disability_status": "", + "disability_details": [{"impairment_type": "X"}], + } + result = self._fn(record) + self.assertTrue(result["has_disability"]) + + def test_empty_status_no_impairments_is_false(self): + """Empty status with no impairments resolves to False.""" + record = { + "disability_status": "", + } + result = self._fn(record) + self.assertFalse(result["has_disability"]) + + def test_unknown_status_with_impairments_warns_and_is_true(self): + """Unknown non-empty status emits a WARNING and falls back to impairment list.""" + record = { + "disability_status": "Pending", + "disability_details": [{"impairment_type": "X"}], + } + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING") as cm: + result = self._fn(record) + self.assertTrue(result["has_disability"]) + self.assertTrue(any("Pending" in line for line in cm.output)) + + def test_unknown_status_no_impairments_warns_and_is_false(self): + """Unknown status with no impairments resolves to False after warning.""" + record = { + "disability_status": "Pending", + } + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING") as cm: + result = self._fn(record) + self.assertFalse(result["has_disability"]) + self.assertTrue(any("Pending" in line for line in cm.output)) + + def test_case_insensitive_approved(self): + record = {"disability_status": "approved"} + result = self._fn(record) + self.assertTrue(result["has_disability"]) + + def test_case_insensitive_rejected(self): + record = {"disability_status": "REJECTED"} + result = self._fn(record) + self.assertFalse(result["has_disability"]) + + def test_case_insensitive_registered(self): + record = {"disability_status": "Registered"} + result = self._fn(record) + self.assertTrue(result["has_disability"]) + + def test_assessment_date_from_last_updated(self): + record = { + "disability_status": "Approved", + "last_updated": "2024-01-01T00:00:00Z", + } + result = self._fn(record) + self.assertEqual(result["assessment_date"], date(2024, 1, 1)) + + def test_assessment_date_fallback_to_registration_date(self): + record = { + "disability_status": "Approved", + "registration_date": "2023-06-15T00:00:00Z", + } + result = self._fn(record) + self.assertEqual(result["assessment_date"], date(2023, 6, 15)) + + def test_last_updated_takes_precedence_over_registration_date(self): + record = { + "disability_status": "Approved", + "last_updated": "2024-01-01T00:00:00Z", + "registration_date": "2023-06-15T00:00:00Z", + } + result = self._fn(record) + self.assertEqual(result["assessment_date"], date(2024, 1, 1)) + + def test_assessment_date_plain_iso_date(self): + record = { + "disability_status": "Approved", + "last_updated": "2024-03-10", + } + result = self._fn(record) + self.assertEqual(result["assessment_date"], date(2024, 3, 10)) + + def test_assessment_date_unparseable_returns_none_and_warns(self): + record = { + "disability_status": "Approved", + "last_updated": "not a date", + } + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING"): + result = self._fn(record) + self.assertIsNone(result["assessment_date"]) + + def test_assessment_date_missing_returns_none(self): + record = {"disability_status": "Approved"} + result = self._fn(record) + self.assertIsNone(result["assessment_date"]) + + def test_disability_details_none_returns_empty_list(self): + """Explicit null on the wire must not crash the parser.""" + record = { + "disability_status": "Approved", + "disability_details": None, + } + result = self._fn(record) + self.assertEqual(result["disability_types"], []) + self.assertTrue(result["has_disability"]) + + def test_legacy_truthy_token_no_longer_approved(self): + """`yes` / `true` are not spec status tokens. Treated as unknown, + which logs a WARN and falls back to the impairment list.""" + record = {"disability_status": "yes"} + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING"): + result = self._fn(record) + self.assertFalse(result["has_disability"]) + + def test_disability_details_missing_impairment_type_skipped(self): + record = { + "disability_status": "Approved", + "disability_details": [ + {"impairment_level": "Severe"}, # no impairment_type + {"impairment_type": "Vision"}, + ], + } + result = self._fn(record) + self.assertEqual(result["disability_types"], ["Vision"]) + + def test_source_registry_field(self): + record = { + "disability_status": "Approved", + "source_registry": "National DR", + } + result = self._fn(record) + self.assertEqual(result["source_registry"], "National DR") + + def test_source_registry_fallback_to_registry_name(self): + record = { + "disability_status": "Approved", + "registry_name": "Regional DR", + } + result = self._fn(record) + self.assertEqual(result["source_registry"], "Regional DR") + + +@tagged("post_install", "-at_install") +class TestExtractFunctionalScores(TransactionCase): + """Tests for extract_functional_scores.""" + + def _fn(self, record): + from odoo.addons.spp_dci_client_dr.services.dr_parsing import extract_functional_scores + + return extract_functional_scores(record) + + def test_spec_record_returns_empty_dict(self): + record = { + "disability_status": "Approved", + "disability_details": [{"impairment_type": "Physical"}], + } + self.assertEqual(self._fn(record), {}) + + def test_empty_record_returns_empty_dict(self): + self.assertEqual(self._fn({}), {}) + + def test_record_with_arbitrary_keys_returns_empty_dict(self): + """The function is intentionally a no-op until the spec adds numeric scores.""" + record = { + "functional_vision": 3, + "hearing_score": 1, + "mobility": 4, + "disability_details": [{"impairment_level": "Severe"}], + } + self.assertEqual(self._fn(record), {}) diff --git a/spp_dci_client_dr/tests/test_dr_service.py b/spp_dci_client_dr/tests/test_dr_service.py index 18162e32..70552a02 100644 --- a/spp_dci_client_dr/tests/test_dr_service.py +++ b/spp_dci_client_dr/tests/test_dr_service.py @@ -1,7 +1,6 @@ # Part of OpenSPP. See LICENSE file for full copyright and licensing details. """Tests for DRService.""" -import json from unittest.mock import MagicMock, patch from odoo.exceptions import UserError, ValidationError @@ -85,40 +84,32 @@ def setUpClass(cls): def _create_mock_response( self, - has_disability=True, - disability_types=None, - functional_scores=None, - assessment_date="2024-11-15", + disability_status="Approved", + impairment_types=None, + last_updated="2024-11-15", + source_registry="National DR", ): - """Helper to create mock DR search response.""" - if disability_types is None: - disability_types = ["Vision", "Mobility"] - if functional_scores is None: - functional_scores = { - "Vision": 3, - "Hearing": 1, - "Mobility": 4, - "Cognition": 1, - "SelfCare": 2, - "Communication": 1, - } - + """Build a DCI v1.0.0 spec-envelope mock response.""" + if impairment_types is None: + impairment_types = ["Vision", "Mobility"] + record = { + "disability_status": disability_status, + "disability_details": [{"impairment_type": t} for t in impairment_types], + "last_updated": last_updated, + "source_registry": source_registry, + } return { "message": { "search_response": [ { "reference_id": "ref-001", "status": "succ", - "data": [ - { - "has_disability": has_disability, - "is_pwd": has_disability, - "disability_types": disability_types, - "functional_scores": functional_scores, - "assessment_date": assessment_date, - "source_registry": "National DR", - } - ], + "data": { + "version": "1.0.0", + "reg_type": "DR", + "reg_record_type": "PERSON", + "reg_records": [record], + }, } ] } @@ -147,10 +138,9 @@ def test_service_init_validates_registry_type(self): @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") def test_get_disability_status_success(self, mock_client_class): - """Test successful disability status retrieval.""" + """Test successful disability status retrieval with spec envelope.""" from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - # Setup mock mock_client = MagicMock() mock_client.search_by_id.return_value = self._create_mock_response() mock_client_class.return_value = mock_client @@ -162,8 +152,7 @@ def test_get_disability_status_success(self, mock_client_class): self.assertTrue(result["has_disability"]) self.assertIn("Vision", result["disability_types"]) self.assertIn("Mobility", result["disability_types"]) - self.assertEqual(result["functional_scores"]["Vision"], 3) - self.assertEqual(result["functional_scores"]["Mobility"], 4) + self.assertEqual(result["functional_scores"], {}) @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") def test_get_disability_status_no_partner(self, mock_client_class): @@ -201,7 +190,6 @@ def test_get_disability_status_no_results(self, mock_client_class): """Test get_disability_status returns None when no DR record found.""" from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - # Mock empty response mock_client = MagicMock() mock_client.search_by_id.return_value = {"message": {"search_response": []}} mock_client_class.return_value = mock_client @@ -216,7 +204,6 @@ def test_get_disability_status_api_error(self, mock_client_class): """Test get_disability_status handles API errors.""" from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - # Mock API error mock_client = MagicMock() mock_client.search_by_id.side_effect = Exception("API connection failed") mock_client_class.return_value = mock_client @@ -230,10 +217,9 @@ def test_get_disability_status_api_error(self, mock_client_class): @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") def test_get_functional_assessment_success(self, mock_client_class): - """Test successful functional assessment retrieval.""" + """Test functional assessment returns empty dict (no numeric scores in spec).""" from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - # Setup mock mock_client = MagicMock() mock_client.search_by_id.return_value = self._create_mock_response() mock_client_class.return_value = mock_client @@ -242,9 +228,7 @@ def test_get_functional_assessment_success(self, mock_client_class): result = service.get_functional_assessment("UIN", "UIN-TEST-12345") self.assertIsNotNone(result) - self.assertEqual(result["Vision"], 3) - self.assertEqual(result["Mobility"], 4) - self.assertEqual(result["Hearing"], 1) + self.assertEqual(result, {}) @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") def test_get_functional_assessment_missing_params(self, mock_client_class): @@ -305,7 +289,7 @@ def test_is_pwd_fetches_when_no_cache(self, mock_client_class): ) mock_client = MagicMock() - mock_client.search_by_id.return_value = self._create_mock_response(has_disability=True) + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Approved") mock_client_class.return_value = mock_client service = DRService(self.env, data_source_code="dr_test") @@ -329,7 +313,7 @@ def test_is_pwd_ignores_stale_cache(self, mock_client_class): ) mock_client = MagicMock() - mock_client.search_by_id.return_value = self._create_mock_response(has_disability=True) + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Approved") mock_client_class.return_value = mock_client service = DRService(self.env, data_source_code="dr_test") @@ -360,7 +344,7 @@ def test_sync_disability_data_creates_record(self, mock_client_class): ) mock_client = MagicMock() - mock_client.search_by_id.return_value = self._create_mock_response() + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Approved") mock_client_class.return_value = mock_client service = DRService(self.env, data_source_code="dr_test") @@ -389,9 +373,7 @@ def test_sync_disability_data_updates_record(self, mock_client_class): ) mock_client = MagicMock() - mock_client.search_by_id.return_value = self._create_mock_response( - has_disability=True # New value - ) + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Approved") mock_client_class.return_value = mock_client service = DRService(self.env, data_source_code="dr_test") @@ -439,111 +421,280 @@ def test_sync_disability_data_no_result(self, mock_client_class): self.assertFalse(status.has_disability) self.assertEqual(status.state, "synced") - def test_extract_disability_data_is_pwd_field(self): - """Test _extract_disability_data handles is_pwd field.""" + def test_get_partner_identifier_priority(self): + """Test _get_partner_identifier follows priority order.""" from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - # Test with is_pwd field instead of has_disability - record_data = { - "is_pwd": True, - "disability_types": ["Hearing"], - } + # Create partner with multiple IDs + partner = self.Partner.create( + { + "name": "Multi ID Person", + "is_registrant": True, + } + ) + + # Add DRN first (lower priority) + self.IdRecord.create( + { + "partner_id": partner.id, + "id_type_id": self.id_type_drn.id, + "value": "DRN-123", + } + ) + + # Add UIN second (higher priority) + self.IdRecord.create( + { + "partner_id": partner.id, + "id_type_id": self.id_type_uin.id, + "value": "UIN-456", + } + ) with patch.object(DRService, "__init__", lambda x, y, z: None): service = DRService.__new__(DRService) - result = service._extract_disability_data(record_data) + service.env = self.env + result = service._get_partner_identifier(partner) + + # Should return UIN_DR_TEST (higher priority vocabulary code) + self.assertEqual(result[0], "UIN_DR_TEST") + self.assertEqual(result[1], "UIN-456") + + # --- additional spec-form coverage --- + + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_spec_rejected(self, mock_client_class): + """Test get_disability_status with rejected status returns has_disability=False.""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService + + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Rejected") + mock_client_class.return_value = mock_client + + service = DRService(self.env, data_source_code="dr_test") + result = service.get_disability_status(self.partner) + + self.assertIsNotNone(result) + self.assertFalse(result["has_disability"]) + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_spec_unknown_status_warns(self, mock_client_class): + """Test get_disability_status with unknown status emits WARNING and returns True (has impairments).""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService + + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Pending") + mock_client_class.return_value = mock_client + + service = DRService(self.env, data_source_code="dr_test") + with self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING") as cm: + result = service.get_disability_status(self.partner) + + self.assertIsNotNone(result) self.assertTrue(result["has_disability"]) + self.assertTrue(any("Pending" in line for line in cm.output)) + + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_extracts_impairment_types(self, mock_client_class): + """Test that impairment_type values are extracted into disability_types.""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService + + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response( + impairment_types=["Physical and movement related functions"] + ) + mock_client_class.return_value = mock_client + + service = DRService(self.env, data_source_code="dr_test") + result = service.get_disability_status(self.partner) + + self.assertIsNotNone(result) + self.assertIn("Physical and movement related functions", result["disability_types"]) + + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_sync_disability_data_creates_record_rejected(self, mock_client_class): + """Test sync with rejected status creates record with has_disability=False.""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService + + rejected_partner = self.Partner.create( + { + "name": "Rejected Person", + "is_registrant": True, + } + ) + self.IdRecord.create( + { + "partner_id": rejected_partner.id, + "id_type_id": self.id_type_uin.id, + "value": "UIN-REJECTED-123", + } + ) + + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Rejected") + mock_client_class.return_value = mock_client + + service = DRService(self.env, data_source_code="dr_test") + result = service.sync_disability_data(rejected_partner) + + self.assertTrue(result) + + status = self.DisabilityStatus.search([("partner_id", "=", rejected_partner.id)]) + self.assertEqual(len(status), 1) + self.assertFalse(status.has_disability) + self.assertEqual(status.state, "synced") - def test_extract_disability_data_string_types(self): - """Test _extract_disability_data handles comma-separated types.""" + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_functional_assessment_spec_envelope(self, mock_client_class): + """Test get_functional_assessment with spec envelope returns empty scores without raising.""" from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - record_data = { - "has_disability": True, - "disability_types": "Vision, Hearing, Mobility", + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response(disability_status="Approved") + mock_client_class.return_value = mock_client + + service = DRService(self.env, data_source_code="dr_test") + result = service.get_functional_assessment("UIN", "UIN-TEST-12345") + + # Spec has no numeric scores: must return empty dict, not raise + self.assertIsNotNone(result) + self.assertEqual(result, {}) + + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_empty_data(self, mock_client_class): + """Test get_disability_status returns None when data is empty envelope.""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService + + mock_client = MagicMock() + mock_client.search_by_id.return_value = { + "message": { + "search_response": [ + { + "reference_id": "ref-001", + "status": "succ", + "data": { + "version": "1.0.0", + "reg_record_type": "PERSON", + "reg_records": [], + }, + } + ] + } } + mock_client_class.return_value = mock_client - with patch.object(DRService, "__init__", lambda x, y, z: None): - service = DRService.__new__(DRService) - result = service._extract_disability_data(record_data) + service = DRService(self.env, data_source_code="dr_test") + result = service.get_disability_status(self.partner) - self.assertEqual(len(result["disability_types"]), 3) - self.assertIn("Vision", result["disability_types"]) - self.assertIn("Hearing", result["disability_types"]) - self.assertIn("Mobility", result["disability_types"]) + self.assertIsNone(result) + + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_missing_data_key(self, mock_client_class): + """Test get_disability_status returns None when data key is absent.""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - def test_extract_functional_scores_domain_fields(self): - """Test _extract_functional_scores handles various field formats.""" + mock_client = MagicMock() + mock_client.search_by_id.return_value = { + "message": { + "search_response": [ + { + "reference_id": "ref-001", + "status": "succ", + } + ] + } + } + mock_client_class.return_value = mock_client + + service = DRService(self.env, data_source_code="dr_test") + result = service.get_disability_status(self.partner) + + self.assertIsNone(result) + + def test_extract_disability_data_delegates_to_module(self): + """Test _extract_disability_data thin-delegates to dr_parsing module function.""" + from odoo.addons.spp_dci_client_dr.services.dr_parsing import extract_disability_data from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - # Test with individual domain fields - record_data = { - "functional_vision": 3, - "hearing_score": 1, - "mobility": 4, - "Cognition": 2, + record = { + "disability_status": "Approved", + "disability_details": [{"impairment_type": "Vision"}], } with patch.object(DRService, "__init__", lambda x, y, z: None): service = DRService.__new__(DRService) - result = service._extract_functional_scores(record_data) - self.assertEqual(result.get("Vision"), 3) - self.assertEqual(result.get("Hearing"), 1) - self.assertEqual(result.get("Mobility"), 4) - self.assertEqual(result.get("Cognition"), 2) + service_result = service._extract_disability_data(record) + module_result = extract_disability_data(record) - def test_extract_functional_scores_json_string(self): - """Test _extract_functional_scores handles JSON string.""" + self.assertEqual(service_result, module_result) + + def test_extract_functional_scores_delegates_to_module(self): + """Test _extract_functional_scores thin-delegates to dr_parsing module function.""" + from odoo.addons.spp_dci_client_dr.services.dr_parsing import extract_functional_scores from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - record_data = { - "functional_scores": json.dumps({"Vision": 3, "Hearing": 1}), + record = { + "disability_status": "Approved", + "disability_details": [{"impairment_type": "Vision"}], } with patch.object(DRService, "__init__", lambda x, y, z: None): service = DRService.__new__(DRService) - result = service._extract_functional_scores(record_data) - self.assertEqual(result["Vision"], 3) - self.assertEqual(result["Hearing"], 1) + self.assertEqual(service._extract_functional_scores(record), extract_functional_scores(record)) + self.assertEqual(service._extract_functional_scores(record), {}) - def test_get_partner_identifier_priority(self): - """Test _get_partner_identifier follows priority order.""" + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_spec_envelope_approved(self, mock_client_class): + """Test get_disability_status with DCI v1.0.0 spec envelope, approved status.""" from odoo.addons.spp_dci_client_dr.services.dr_service import DRService - # Create partner with multiple IDs - partner = self.Partner.create( - { - "name": "Multi ID Person", - "is_registrant": True, - } + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response( + disability_status="Approved", + impairment_types=["Physical and movement related functions"], ) + mock_client_class.return_value = mock_client - # Add DRN first (lower priority) - self.IdRecord.create( - { - "partner_id": partner.id, - "id_type_id": self.id_type_drn.id, - "value": "DRN-123", - } + service = DRService(self.env, data_source_code="dr_test") + result = service.get_disability_status(self.partner) + + self.assertIsNotNone(result) + self.assertTrue(result["has_disability"]) + + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_spec_envelope_rejected(self, mock_client_class): + """Test get_disability_status with DCI v1.0.0 spec envelope, rejected status.""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService + + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response( + disability_status="Rejected", + impairment_types=["Physical"], ) + mock_client_class.return_value = mock_client - # Add UIN second (higher priority) - self.IdRecord.create( - { - "partner_id": partner.id, - "id_type_id": self.id_type_uin.id, - "value": "UIN-456", - } + service = DRService(self.env, data_source_code="dr_test") + result = service.get_disability_status(self.partner) + + self.assertIsNotNone(result) + self.assertFalse(result["has_disability"]) + + @patch("odoo.addons.spp_dci_client_dr.services.dr_service.DCIClient") + def test_get_disability_status_spec_envelope_impairments(self, mock_client_class): + """Test that impairment_type values are extracted into disability_types.""" + from odoo.addons.spp_dci_client_dr.services.dr_service import DRService + + mock_client = MagicMock() + mock_client.search_by_id.return_value = self._create_mock_response( + disability_status="Approved", + impairment_types=["Physical and movement related functions"], ) + mock_client_class.return_value = mock_client - with patch.object(DRService, "__init__", lambda x, y, z: None): - service = DRService.__new__(DRService) - service.env = self.env - result = service._get_partner_identifier(partner) + service = DRService(self.env, data_source_code="dr_test") + result = service.get_disability_status(self.partner) - # Should return UIN_DR_TEST (higher priority vocabulary code) - self.assertEqual(result[0], "UIN_DR_TEST") - self.assertEqual(result[1], "UIN-456") + self.assertIsNotNone(result) + self.assertIn("Physical and movement related functions", result["disability_types"]) From 55a219d28e8835e65bcbb1a231e8907101a507bf Mon Sep 17 00:00:00 2001 From: Jeremi Joslin Date: Sun, 17 May 2026 20:53:21 +0700 Subject: [PATCH 2/2] refactor(spp_dci_client_dr): route callback envelope unwrap through dr_parsing The callback path duplicated `data.get("reg_records", [])` without the type guards and WARN logging now centralized in `unwrap_search_data`. A malformed `data` envelope was silently coerced to `{}` and skipped, hiding spec drift from operators. Route through the shared helper so both sync and callback paths get identical validation behavior. Co-Authored-By: Claude Opus 4.7 --- spp_dci_client_dr/routers/callback.py | 5 ++--- spp_dci_client_dr/tests/test_callback.py | 20 ++++++++++++++++++++ 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/spp_dci_client_dr/routers/callback.py b/spp_dci_client_dr/routers/callback.py index 8d58cffe..fa38c460 100644 --- a/spp_dci_client_dr/routers/callback.py +++ b/spp_dci_client_dr/routers/callback.py @@ -15,7 +15,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status from ..middleware.signature import verify_dr_signature -from ..services.dr_parsing import extract_disability_data +from ..services.dr_parsing import extract_disability_data, unwrap_search_data _logger = logging.getLogger(__name__) @@ -151,8 +151,7 @@ def _process_dr_search_result(env: Environment, result: dict, source_registry: s ) return - data = result.get("data", {}) - reg_records = data.get("reg_records", []) + reg_records = unwrap_search_data(result.get("data")) for record in reg_records: # Extract identifiers to find matching partner diff --git a/spp_dci_client_dr/tests/test_callback.py b/spp_dci_client_dr/tests/test_callback.py index d54338f0..cde37f73 100644 --- a/spp_dci_client_dr/tests/test_callback.py +++ b/spp_dci_client_dr/tests/test_callback.py @@ -166,6 +166,26 @@ def test_non_success_status_is_skipped(self): self.assertEqual(len(status), 0) mock_find.assert_not_called() + def test_malformed_envelope_warns_and_does_not_crash(self): + """`data` of an unexpected type is rejected by unwrap_search_data with a WARN + rather than crashing the callback or silently processing garbage.""" + from odoo.addons.spp_dci_client_dr.routers.callback import _process_dr_search_result + + result = { + "status": "succ", + "data": "not-an-envelope", + } + + with ( + patch( + "odoo.addons.spp_dci_client_dr.routers.callback._find_partner_by_identifier", + ) as mock_find, + self.assertLogs("odoo.addons.spp_dci_client_dr.services.dr_parsing", level="WARNING"), + ): + _process_dr_search_result(self.env, result, "test-source-registry") + + mock_find.assert_not_called() + def test_update_overwrites_existing_record(self): """Calling _process_dr_search_result twice updates the existing record.""" from odoo.addons.spp_dci_client_dr.routers.callback import _process_dr_search_result