Decode FHIR pseudonym tokens in LocalizationListService

Author: thomas-samoht

app/models/fhir/resources/localization_list/request.py

@@ -9,8 +9,6 @@
     DEVICE_SYSTEM,
     PSEUDONYM_SYSTEM,
 )
-from app.models.pseudonym import OprfPseudonym
-from app.services.fhir.pseudonym_token import decode_pseudonym_token
 
 logger = logging.getLogger(__name__)
 
@@ -92,9 +90,6 @@ def empty(self) -> bool:
     def is_localize_params(self) -> bool:
         return self.subject is not None and self.source is None
 
-    def get_subject_as_class_or_none(self) -> OprfPseudonym | None:
-        return decode_pseudonym_token(self.subject, SUBJECT_IDENTIFIER_PARAM) if self.subject else None
-
 
 class LocalizationListSearchParams(LocalizationListParams):
     """Search parameters for a List search, which also controls the page that is returned."""

app/models/fhir/resources/localization_list/resource.py

@@ -19,9 +19,7 @@
     URA_SYSTEM_EXTENSION,
 )
 from app.models.fhir.resources.domain_resource import DomainResource
-from app.models.pseudonym import OprfPseudonym
 from app.models.ura import UraNumber
-from app.services.fhir.pseudonym_token import decode_pseudonym_token
 
 
 class ReferenceExtension(Extension):
@@ -88,14 +86,14 @@ def get_ura(self) -> UraNumber:
 
         return UraNumber(target_ura)
 
-    def get_encoded_pseudonym(self) -> OprfPseudonym:
+    def get_encoded_pseudonym(self) -> str:
         if self.subject is None:
             raise ValueError("List.subject is required")
 
         if self.subject.identifier.system != PSEUDONYM_SYSTEM:
             raise ValueError(f"List.subject.identifier.system must be {PSEUDONYM_SYSTEM}")
 
-        return decode_pseudonym_token(self.subject.identifier.value, "List.subject.identifier.value")
+        return self.subject.identifier.value
 
     def get_device(self) -> str:
         if self.source.identifier.system != DEVICE_SYSTEM:

app/services/fhir/localization_list.py

@@ -9,19 +9,26 @@
 from app.models.fhir.resources.localization_list.request import (
     COUNT_PARAM,
     OFFSET_PARAM,
+    SUBJECT_IDENTIFIER_PARAM,
     LocalizationListParams,
     LocalizationListSearchParams,
 )
 from app.models.fhir.resources.localization_list.resource import LocalizationList
 from app.models.fhir.resources.operation_outcome.resource import OperationOutcome
+from app.models.pseudonym import OprfPseudonym
 from app.models.ura import UraNumber
 from app.services.exceptions import UnauthorizedUraError
+from app.services.fhir.pseudonym_token import decode_pseudonym_token
 from app.services.pseudonym_referral_service import PseudonymReferralService
 from app.services.referral_service import ReferralService
 
 logger = logging.getLogger(__name__)
 
 
+def _subject_pseudonym(params: LocalizationListParams) -> OprfPseudonym | None:
+    return decode_pseudonym_token(params.subject, SUBJECT_IDENTIFIER_PARAM) if params.subject else None
+
+
 class LocalizationListService:
     def __init__(
         self,
@@ -43,7 +50,6 @@ def create(
         """Returns the stored List and whether it was created by this call."""
         ura_number = data.get_ura()
         device = data.get_device()
-        pseudonym = data.get_encoded_pseudonym()
 
         if ura_number != authenticated_ura:
             gflog.emit(
@@ -54,6 +60,7 @@ def create(
             )
             raise UnauthorizedUraError("Registration not linked to the authorized URA")
 
+        pseudonym = decode_pseudonym_token(data.get_encoded_pseudonym(), "List.subject.identifier.value")
         referral, created = self.pseudonym_referral_service.register(pseudonym, ura_number=ura_number, source=device)
 
         return LocalizationList.from_referral(referral), created
@@ -72,7 +79,7 @@ def query(
 
         When `request_url` is given, the bundle links to the current, next and previous page.
         """
-        subject = params.get_subject_as_class_or_none()
+        subject = _subject_pseudonym(params)
         page_size = min(params.count or self.default_page_size, self.max_page_size)
 
         if params.is_localize_params() and subject is not None:
@@ -132,7 +139,7 @@ def delete_by_query(
         authenticated_ura: UraNumber,
         source: str,
     ) -> tuple[OperationOutcome, int]:
-        subject = params.get_subject_as_class_or_none()
+        subject = _subject_pseudonym(params)
         deleted_count = self.pseudonym_referral_service.delete_many(authenticated_ura, pseudonym=subject, source=source)
 
         if deleted_count < 1:

tests/services/fhir/test_localization_list.py

@@ -63,10 +63,11 @@ def test_stores_and_returns_the_resource(self, service: LocalizationListService)
 
         assert result.id is not None
 
+    @pytest.mark.parametrize("pseudonym", [TEST_PSEUDONYM_TOKEN, INVALID_PSEUDONYM_TOKEN], ids=["valid", "invalid"])
     def test_rejects_a_resource_for_another_ura(
-        self, service: LocalizationListService, caplog: pytest.LogCaptureFixture
+        self, service: LocalizationListService, caplog: pytest.LogCaptureFixture, pseudonym: str
     ) -> None:
-        data = make_list_resource(ura=str(OTHER_URA))
+        data = make_list_resource(ura=str(OTHER_URA), pseudonym=pseudonym)
 
         with caplog.at_level(logging.INFO), pytest.raises(UnauthorizedUraError):
             service.create(data, AUTH_URA)