Skip to content

fix: DCI envelope support for disability registry search (#217) #303

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Tarekchehahde wants to merge 1 commit into
base: 19.0
Choose a base branch
from
Open

fix: DCI envelope support for disability registry search (#217) #303

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
172 changes: 88 additions & 84 deletions spp_dci_server/routers/registry_aliases.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,37 +11,100 @@
"""

import logging
from datetime import datetime
import uuid
from datetime import UTC, datetime
from typing import Annotated

from odoo.api import Environment

from odoo.addons.fastapi.dependencies import odoo_env
from odoo.addons.spp_dci.schemas import DCICallbackHeader, DCIEnvelope
from odoo.addons.spp_dci.schemas.constants import MsgHeaderStatusReasonCode
from odoo.addons.spp_dci.schemas.search import (
SearchRequest,
SearchResponse,
SearchResponseItem,
)
from odoo.addons.spp_dci.services import get_sender_id

from fastapi import APIRouter, Depends, status
from fastapi import APIRouter, Depends, HTTPException, status

from ..middleware.signature import verify_bearer_token

_logger = logging.getLogger(__name__)


def _build_stub_search_envelope(
request_envelope: DCIEnvelope,
env: Environment,
not_implemented_message: str,
) -> DCIEnvelope:
"""Return a DCI on-search envelope with per-item rjct stubs."""
try:
search_request = SearchRequest.model_validate(request_envelope.message)
except Exception as e:
_logger.error("Invalid SearchRequest message: %s", str(e))
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid search request message: {str(e)}",
) from e

response_items = []
for req_item in search_request.search_request:
response_items.append(
SearchResponseItem(
reference_id=req_item.reference_id,
timestamp=datetime.now(UTC),
status="rjct",
status_reason_code=MsgHeaderStatusReasonCode.ACTION_NOT_SUPPORTED.value,
status_reason_message=not_implemented_message,
data=None,
pagination=None,
locale=req_item.locale,
)
)

search_response = SearchResponse(
transaction_id=search_request.transaction_id,
correlation_id=str(uuid.uuid4()),
search_response=response_items,
)

our_sender_id = get_sender_id(env) or "openspp"
total_count = len(response_items)

callback_header = DCICallbackHeader(
version=request_envelope.header.version,
message_id=str(uuid.uuid4()),
message_ts=datetime.now(UTC),
action=f"on-{request_envelope.header.action}",
sender_id=our_sender_id,
receiver_id=request_envelope.header.sender_id,
total_count=total_count,
status="rjct",
status_reason_code=MsgHeaderStatusReasonCode.ACTION_NOT_SUPPORTED.value,
status_reason_message=not_implemented_message,
completed_count=0,
)

return DCIEnvelope(
signature="",
header=callback_header,
message=search_response.model_dump(mode="json", exclude_none=True),
)
Comment on lines +73 to +94

@gemini-code-assist gemini-code-assist Bot Jul 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Instead of manually constructing the DCICallbackHeader and returning an unsigned DCIEnvelope (with signature=""), you can leverage the existing build_signed_envelope helper from spp_dci.services.response_helpers. This reduces boilerplate, ensures consistency, and automatically signs the envelope if a signing key is configured in the environment.

    from odoo.addons.spp_dci.services.response_helpers import build_signed_envelope

    return build_signed_envelope(
        env=env,
        original_header=request_envelope.header,
        response_message=search_response.model_dump(mode="json", exclude_none=True),
        status_code="rjct",
        status_reason_code=MsgHeaderStatusReasonCode.ACTION_NOT_SUPPORTED.value,
        status_reason_message=not_implemented_message,
        completed_count=0,
    )



# Disability Registry router
disability_router = APIRouter(tags=["Disability Registry"], prefix="/disability/registry")


@disability_router.post(
"/sync/search",
response_model=SearchResponse,
response_model=DCIEnvelope,
response_model_exclude_none=True,
status_code=status.HTTP_501_NOT_IMPLEMENTED,
)
async def disability_sync_search(
request: SearchRequest,
request_envelope: DCIEnvelope,
env: Annotated[Environment, Depends(odoo_env)],
_bearer_token: Annotated[str, Depends(verify_bearer_token)],
):
Expand All @@ -54,29 +117,10 @@ async def disability_sync_search(
For now, returns "not implemented" response for SPDCI compliance testing.
"""
_logger.warning("Disability Registry search endpoint called but not yet implemented")

# Build response items with "not implemented" status
response_items = []
for req_item in request.search_request:
response_items.append(
SearchResponseItem(
reference_id=req_item.reference_id,
timestamp=datetime.utcnow(),
status="rjct",
status_reason_code=MsgHeaderStatusReasonCode.ACTION_NOT_SUPPORTED.value,
status_reason_message=(
"Disability Registry not yet implemented. Will be available in spp_dci_server_disability module."
),
data=None,
pagination=None,
locale=req_item.locale,
)
)

return SearchResponse(
transaction_id=request.transaction_id,
correlation_id=None,
search_response=response_items,
return _build_stub_search_envelope(
request_envelope,
env,
"Disability Registry not yet implemented. Will be available in spp_dci_server_disability module.",
)


Expand All @@ -91,15 +135,15 @@ async def disability_sync_notify(
"""
SPDCI-compliant Disability Registry notification endpoint.

Path: /disability/registry/sync/notify
Path: /disability/registry/sync/search

@gemini-code-assist gemini-code-assist Bot Jul 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The path in the docstring was incorrectly changed to /disability/registry/sync/search. Since this is the notification endpoint (disability_sync_notify), it should remain /disability/registry/sync/notify.

Suggested change
Path: /disability/registry/sync/search
Path: /disability/registry/sync/notify


This endpoint will be implemented when the Disability Registry module is added.
"""
_logger.warning("Disability Registry notify endpoint called but not yet implemented")
return {
"status": "rjct",
"message": "Disability Registry not yet implemented",
"timestamp": datetime.utcnow().isoformat(),
"timestamp": datetime.now(UTC).isoformat(),
}


Expand All @@ -109,12 +153,11 @@ async def disability_sync_notify(

@crvs_router.post(
"/sync/search",
response_model=SearchResponse,
response_model=DCIEnvelope,
response_model_exclude_none=True,
status_code=status.HTTP_501_NOT_IMPLEMENTED,
)
async def crvs_sync_search(
request: SearchRequest,
request_envelope: DCIEnvelope,
env: Annotated[Environment, Depends(odoo_env)],
_bearer_token: Annotated[str, Depends(verify_bearer_token)],
):
Expand All @@ -127,29 +170,10 @@ async def crvs_sync_search(
For now, returns "not implemented" response for SPDCI compliance testing.
"""
_logger.warning("Civil Registry search endpoint called but not yet implemented")

# Build response items with "not implemented" status
response_items = []
for req_item in request.search_request:
response_items.append(
SearchResponseItem(
reference_id=req_item.reference_id,
timestamp=datetime.utcnow(),
status="rjct",
status_reason_code=MsgHeaderStatusReasonCode.ACTION_NOT_SUPPORTED.value,
status_reason_message=(
"Civil Registry not yet implemented. Will be available in spp_dci_server_crvs module."
),
data=None,
pagination=None,
locale=req_item.locale,
)
)

return SearchResponse(
transaction_id=request.transaction_id,
correlation_id=None,
search_response=response_items,
return _build_stub_search_envelope(
request_envelope,
env,
"Civil Registry not yet implemented. Will be available in spp_dci_server_crvs module.",
)


Expand All @@ -172,7 +196,7 @@ async def crvs_sync_notify(
return {
"status": "rjct",
"message": "Civil Registry not yet implemented",
"timestamp": datetime.utcnow().isoformat(),
"timestamp": datetime.now(UTC).isoformat(),
}


Expand All @@ -182,12 +206,11 @@ async def crvs_sync_notify(

@farmer_router.post(
"/sync/search",
response_model=SearchResponse,
response_model=DCIEnvelope,
response_model_exclude_none=True,
status_code=status.HTTP_501_NOT_IMPLEMENTED,
)
async def farmer_sync_search(
request: SearchRequest,
request_envelope: DCIEnvelope,
env: Annotated[Environment, Depends(odoo_env)],
_bearer_token: Annotated[str, Depends(verify_bearer_token)],
):
Expand All @@ -200,29 +223,10 @@ async def farmer_sync_search(
For now, returns "not implemented" response for SPDCI compliance testing.
"""
_logger.warning("Farmer Registry search endpoint called but not yet implemented")

# Build response items with "not implemented" status
response_items = []
for req_item in request.search_request:
response_items.append(
SearchResponseItem(
reference_id=req_item.reference_id,
timestamp=datetime.utcnow(),
status="rjct",
status_reason_code=MsgHeaderStatusReasonCode.ACTION_NOT_SUPPORTED.value,
status_reason_message=(
"Farmer Registry not yet implemented. Will be available in spp_dci_server_farmer module."
),
data=None,
pagination=None,
locale=req_item.locale,
)
)

return SearchResponse(
transaction_id=request.transaction_id,
correlation_id=None,
search_response=response_items,
return _build_stub_search_envelope(
request_envelope,
env,
"Farmer Registry not yet implemented. Will be available in spp_dci_server_farmer module.",
)


Expand All @@ -245,5 +249,5 @@ async def farmer_sync_notify(
return {
"status": "rjct",
"message": "Farmer Registry not yet implemented",
"timestamp": datetime.utcnow().isoformat(),
"timestamp": datetime.now(UTC).isoformat(),
}
29 changes: 25 additions & 4 deletions spp_dci_server/tests/test_callback_routers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,14 @@
from odoo.tests import tagged

from odoo.addons.spp_dci.schemas.constants import MsgHeaderStatusReasonCode
from odoo.addons.spp_dci.schemas.envelope import DCIMessageHeader
from odoo.addons.spp_dci.schemas.search import (
SearchCriteria,
SearchRequest,
SearchRequestItem,
SearchResponse,
)
from odoo.addons.spp_dci.schemas import DCIEnvelope

from .common import DCIServerCommon

Expand Down Expand Up @@ -211,8 +214,24 @@ def _build_search_request(self, n_items=1):
)
return SearchRequest(transaction_id="txn-alias", search_request=items)

def _build_search_envelope(self, n_items=1):
search_request = self._build_search_request(n_items=n_items)
header = DCIMessageHeader(
message_id="msg-alias-test",
message_ts=datetime.now(UTC),
action="search",
sender_id="registry-witness",
receiver_id="openspp",
total_count=n_items,
)
return DCIEnvelope(
signature="",
header=header,
message=search_request.model_dump(mode="json"),
)

def test_search_stubs_return_per_item_rjct(self):
request = self._build_search_request(n_items=2)
request = self._build_search_envelope(n_items=2)
# Each alias mentions its corresponding spp_dci_server_* module in
# the rejection message so operators can find the missing addon.
expected_module = {
Expand All @@ -223,9 +242,11 @@ def test_search_stubs_return_per_item_rjct(self):
for registry, endpoint in self.search_endpoints.items():
with self.subTest(registry=registry):
response = _run(endpoint(request, self.env, _bearer_token="t"))
self.assertEqual(response.transaction_id, "txn-alias")
self.assertEqual(len(response.search_response), 2)
for item in response.search_response:
self.assertIsInstance(response, DCIEnvelope)
search_response = SearchResponse.model_validate(response.message)
self.assertEqual(search_response.transaction_id, "txn-alias")
self.assertEqual(len(search_response.search_response), 2)
for item in search_response.search_response:
self.assertEqual(item.status, "rjct")
self.assertEqual(
item.status_reason_code,
Expand Down