Adds the pathways for v3 handling around fhir_ids (BB-4166)

apps/accounts/views/oauth2_profile.py

@@ -9,7 +9,7 @@
 from apps.fhir.bluebutton.models import Crosswalk
 from apps.fhir.bluebutton.permissions import ApplicationActivePermission
 
-from apps.constants import Versions
+from apps.versions import Versions
 
 
 def _get_userinfo(user, version=Versions.NOT_AN_API_VERSION):
@@ -46,8 +46,6 @@ def _get_userinfo(user, version=Versions.NOT_AN_API_VERSION):
 def _openidconnect_userinfo(request, version=Versions.NOT_AN_API_VERSION):
     # NOTE: The **kwargs are not used anywhere down the callchain, and are being ignored.
 
-    # BB2-4166-TODO: will the request have a version? do we get here from redirects or is this
-    # a straight url that we need to get the version from the url (like we do in the fhir app)
     return JsonResponse(_get_userinfo(request.resource_owner, version))
 
 

apps/authorization/permissions.py

@@ -1,5 +1,6 @@
 from django.conf import settings
 from rest_framework import (permissions, exceptions)
+from apps.versions import Versions, VersionNotMatched
 
 from .models import DataAccessGrant
 
@@ -32,8 +33,10 @@ def has_object_permission(self, request, view, obj):
         # Patient resources were taken care of above
         # Return 404 on error to avoid notifying unauthorized user the object exists
 
-        # BB2-4166-TODO: this is hardcoded to be version 2
-        return is_resource_for_patient(obj, request.crosswalk.fhir_id(2))
+        if view.version in Versions.supported_versions():
+            return is_resource_for_patient(obj, request.crosswalk.fhir_id(view.version))
+        else:
+            raise VersionNotMatched()
 
 
 def is_resource_for_patient(obj, patient_id):

apps/authorization/views.py

@@ -14,10 +14,11 @@
 from oauth2_provider.views.base import OAuthLibMixin
 from oauth2_provider.views.generic import ClientProtectedResourceView
 
+from apps.versions import VersionNotMatched, Versions
 from apps.dot_ext.authentication import SLSAuthentication
-from .models import DataAccessGrant
-from ..dot_ext.utils import get_application_from_meta
-from ..fhir.bluebutton.models import Crosswalk
+from apps.authorization.models import DataAccessGrant
+from apps.dot_ext.utils import get_application_from_meta, get_api_version_number_from_url
+from apps.fhir.bluebutton.models import Crosswalk
 
 
 Application = get_application_model()
@@ -68,9 +69,20 @@ class ExpireDataAccessGrantView(ClientProtectedResourceView, OAuthLibMixin):
     @staticmethod
     def post(request, *args, **kwargs):
         try:
+            path_info = request.__dict__.get('path_info')
+            version = get_api_version_number_from_url(path_info)
             patient_id = kwargs.pop('patient_id', None)
-            # BB2-4166-TODO: currently hardcoded for v2, might need to not be static
-            user = Crosswalk.objects.get(fhir_id_v2=patient_id).user
+
+            match version:
+                case Versions.V1:
+                    user = Crosswalk.objects.get(fhir_id_v2=patient_id).user
+                case Versions.V2:
+                    user = Crosswalk.objects.get(fhir_id_v2=patient_id).user
+                case Versions.V3:
+                    user = Crosswalk.objects.get(fhir_id_v3=patient_id).user
+                case _:
+                    raise VersionNotMatched(f"{version} is not a valid version constant")
+
             client = get_application_from_meta(request)
             DataAccessGrant.objects.get(beneficiary=user.id, application=client).delete()
         except Crosswalk.DoesNotExist:

apps/bb2_tools/admin.py

@@ -21,7 +21,7 @@
     DummyAdminObject,
     UserStats,
 )
-from apps.fhir.bluebutton.utils import get_patient_by_id
+from apps.fhir.bluebutton.utils import get_v2_patient_by_id
 
 ADMIN_PREPEND = getattr(settings, "ADMIN_PREPEND_URL", "")
 BB2_TOOLS_PATH = (
@@ -332,8 +332,7 @@ class BeneficiaryDashboardAdmin(ReadOnlyAdmin):
         "get_connected_applications",
         "date_created",
     )
-    # BB2-4166-TODO: add support for v3
-    search_fields = ('user__username', 'fhir_id_v2', '_user_id_hash', '_user_mbi')
+    search_fields = ('user__username', 'fhir_id_v2', 'fhir_id_v3', '_user_id_hash', '_user_mbi')
     readonly_fields = ('date_created',)
     raw_id_fields = ('user',)
 
@@ -361,10 +360,9 @@ def get_access_tokens(self, obj):
         ordering="MyIdentities",
     )
     def get_identities(self, obj):
-        # BB2-4166-TODO: add support for v3
         return format_html(
-            '<div><ul><li>FHIR_ID_V2:{}</li><li>HICN HASH:{}</li><li>MBI:{}</li>'.format(
-                obj.fhir_id(2), obj.user_hicn_hash, obj.user_mbi
+            '<div><ul><li>FHIR_ID_V2:{}</li><li>FHIR_ID_V3:{}</li><li>HICN HASH:{}</li><li>MBI:{}</li>'.format(
+                obj.fhir_id(2), obj.fhir_id(3), obj.user_hicn_hash, obj.user_mbi
             )
         )
 
@@ -416,10 +414,9 @@ def change_view(self, request, object_id, form_url="", extra_context=None):
         crosswalk = BeneficiaryDashboard.objects.get(pk=int(object_id))
 
         json_resp = None
-
         try:
-            # BB2-4166-TODO: this is hardcoded to be version 2
-            json_resp = get_patient_by_id(crosswalk.fhir_id(2), request)
+            # DEPRECATE_V2: If we ever deprecate v2, this function and call need to be updated
+            json_resp = get_v2_patient_by_id(crosswalk.fhir_id(2), request)
         except Exception as e:
             json_resp = {"backend_error": str(e)}
 

apps/dot_ext/oauth2_backends.py

@@ -4,6 +4,7 @@
 from ..fhir.bluebutton.models import Crosswalk
 from .loggers import (clear_session_auth_flow_trace, update_session_auth_flow_trace_from_code,
                       set_session_auth_flow_trace_value)
+from apps.dot_ext.utils import get_api_version_number_from_url
 
 
 class OAuthLibSMARTonFHIR(OAuthLibCore):
@@ -31,8 +32,8 @@ def create_token_response(self, request):
             if Crosswalk.objects.filter(user=token.user).exists():
                 fhir_body = json.loads(body)
                 cw = Crosswalk.objects.get(user=token.user)
-                # BB2-4166-TODO: this is hardcoded to be version 2
-                fhir_body["patient"] = cw.fhir_id(2)
+                version = get_api_version_number_from_url(request.path)
+                fhir_body['patient'] = cw.fhir_id(version)
                 body = json.dumps(fhir_body)
 
         return uri, headers, body, status

apps/dot_ext/tests/test_api_error_codes.py

@@ -6,7 +6,7 @@
 from apps.authorization.models import (
     DataAccessGrant,
 )
-from apps.constants import AccessType
+from apps.versions import AccessType
 from datetime import datetime
 from dateutil.relativedelta import relativedelta
 from unittest import mock

apps/dot_ext/tests/test_utils.py

@@ -0,0 +1,24 @@
+from django.test import TestCase
+
+from apps.versions import VersionNotMatched
+from ..utils import get_api_version_number_from_url
+
+SUPPORTED_VERSION_TEST_CASES = [
+    {'url_path': '/v2/fhir/Patient/', 'expected': 2},
+    # return 0 because v2 does not have a leading /
+    {'url_path': 'v2/fhir/Patient/', 'expected': 0},
+    {'url_path': '/v3/fhir/Coverage/', 'expected': 3},
+    {'url_path': '/v3/fhir/Coverage/v2/', 'expected': 3},
+]
+
+
+class TestDOTUtils(TestCase):
+    def test_get_api_version_number(self):
+        for test in SUPPORTED_VERSION_TEST_CASES:
+            result = get_api_version_number_from_url(test['url_path'])
+            assert result == test['expected']
+
+    def test_get_api_version_number_unsupported_version(self):
+        # unsupported version will raise an exception
+        with self.assertRaises(VersionNotMatched, msg='4 extracted from /v4/fhir/Coverage/'):
+            get_api_version_number_from_url('/v4/fhir/Coverage/')

apps/dot_ext/utils.py

@@ -7,6 +7,8 @@
 from oauth2_provider.models import AccessToken, RefreshToken, get_application_model
 from oauthlib.oauth2.rfc6749.errors import InvalidClientError, InvalidGrantError, InvalidRequestError
 from http import HTTPStatus
+import re
+from apps.versions import Versions, VersionNotMatched
 
 from apps.authorization.models import DataAccessGrant
 
@@ -252,3 +254,26 @@ def json_response_from_oauth2_error(error):
         ret_data['error_description'] = error.description
 
     return JsonResponse(ret_data, status=error.status_code)
+
+
+def get_api_version_number_from_url(url_path: str) -> int:
+    """Utility function to extract what version of the API a URL is
+    If there are multiple occurrences of 'v{{VERSION}} in a url path,
+    only return the first one
+    EX. /v2/o/authorize will return v2.
+
+    Args:
+        url_path (str): The url being called that we want to extract the api version
+
+    Returns:
+        Optional[str]: Returns a string of v2
+    """
+    match = re.search(r'/v(\d+)/', url_path, re.IGNORECASE)
+    if match:
+        version = int(match.group(1))
+        if version in Versions.supported_versions():
+            return version
+        else:
+            raise VersionNotMatched(f'{version} extracted from {url_path}')
+
+    return Versions.NOT_AN_API_VERSION

apps/fhir/bluebutton/models.py

@@ -12,6 +12,8 @@
 from django.core.validators import MinLengthValidator
 from apps.accounts.models import get_user_id_salt
 
+from apps.versions import Versions, VersionNotMatched
+
 
 class BBFhirBluebuttonModelException(APIException):
     # BB2-237 custom exception
@@ -164,29 +166,25 @@ class Meta:
             )
         ]
 
-    def fhir_id(self, version: int = 2) -> str:
+    def fhir_id(self, version: int = Versions.V2) -> str:
         """Helper method to return fhir_id based on BFD version, preferred over direct access"""
-        if version in (1, 2):
+        if version in [Versions.V1, Versions.V2]:
             if self.fhir_id_v2 is not None and self.fhir_id_v2 != '':
                 # TODO - This is legacy code, to be removed before migration bluebutton 0010
                 # If fhir_id is empty, try to populate it from fhir_id_v2 to support old code
                 self._fhir_id = self.fhir_id_v2
                 self.save()
-                return str(self.fhir_id_v2)
+                return self.fhir_id_v2
             # TODO - This is legacy code, to be removed before migration bluebutton 0010
             # If fhir_id_v2 is empty, try to populate it from _fhir_id to support new code
             if self._fhir_id is not None and self._fhir_id != '':
                 self.fhir_id_v2 = self._fhir_id
                 self.save()
-                return str(self._fhir_id)
+                return self._fhir_id
             return ''
-        elif version == 3:
-            # TODO BB2-4166: This will want to change. In order to make
-            # BB2-4181 work, the V3 value needed to be found in the V2 column.
-            # 4166 should flip this to _v3, and we should be able to find
-            # values there when using (say) the test client.
-            if self.fhir_id_v2 is not None and self.fhir_id_v2 != '':
-                return str(self.fhir_id_v2)
+        elif version == Versions.V3:
+            if self.fhir_id_v3 is not None and self.fhir_id_v3 != '':
+                return self.fhir_id_v3
             return ''
         else:
             raise ValidationError(f'{version} is not a valid BFD version')
@@ -203,7 +201,7 @@ def set_fhir_id(self, value, version: int = 2) -> None:
         elif version == 3:
             self.fhir_id_v3 = value
         else:
-            raise ValidationError(f'{version} is not a valid BFD version')
+            raise VersionNotMatched(f'{version} is not a valid BFD version')
 
     @property
     def user_hicn_hash(self):

apps/fhir/bluebutton/permissions.py

@@ -5,6 +5,7 @@
 from rest_framework import permissions, exceptions
 from rest_framework.exceptions import AuthenticationFailed
 from .constants import ALLOWED_RESOURCE_TYPES
+from apps.versions import Versions, VersionNotMatched
 
 import apps.logging.request_logger as bb2logging
 
@@ -30,35 +31,36 @@ def has_permission(self, request, view):
 
 class HasCrosswalk(permissions.BasePermission):
     def has_permission(self, request, view):
-        return bool(
-            # BB2-4166-TODO : this needs to use version to determine fhir_id, probably in request
-            request.user and request.user.crosswalk and request.user.crosswalk.fhir_id(2)
-        )
+        if view.version in Versions.supported_versions():
+            return request.user and request.user.crosswalk and request.user.crosswalk.fhir_id(view.version)
+        else:
+            # this should not happen where we'd get an unsupported version
+            raise VersionNotMatched("Version not matched in has_permission")
 
 
 class ReadCrosswalkPermission(HasCrosswalk):
     def has_object_permission(self, request, view, obj):
         # Now check that the user has permission to access the data
         # Patient resources were taken care of above # TODO - verify this
         # Return 404 on error to avoid notifying unauthorized user the object exists
-
+        if view.version in Versions.supported_versions():
+            fhir_id = request.crosswalk.fhir_id(view.version)
+        else:
+            raise VersionNotMatched("Version not matched in has_object_permission in ReadCrosswalkPermission")
         try:
             if request.resource_type == "Coverage":
                 reference = obj["beneficiary"]["reference"]
                 reference_id = reference.split("/")[1]
-                # BB2-4166-TODO : this needs to use version to determine fhir_id, probably in request
-                if reference_id != request.crosswalk.fhir_id(2):
+                if reference_id != fhir_id:
                     raise exceptions.NotFound()
             elif request.resource_type == "ExplanationOfBenefit":
                 reference = obj["patient"]["reference"]
                 reference_id = reference.split("/")[1]
-                # BB2-4166-TODO : this needs to use version to determine fhir_id, probably in request
-                if reference_id != request.crosswalk.fhir_id(2):
+                if reference_id != fhir_id:
                     raise exceptions.NotFound()
             else:
                 reference_id = obj["id"]
-                # BB2-4166-TODO : this needs to use version to determine fhir_id, probably in request
-                if reference_id != request.crosswalk.fhir_id(2):
+                if reference_id != fhir_id:
                     raise exceptions.NotFound()
 
         except exceptions.NotFound:
@@ -71,9 +73,10 @@ def has_object_permission(self, request, view, obj):
 
 class SearchCrosswalkPermission(HasCrosswalk):
     def has_object_permission(self, request, view, obj):
-        # BB2-4166-TODO: this is hardcoded to be version 2
-        patient_id = request.crosswalk.fhir_id(2)
-
+        if view.version in Versions.supported_versions():
+            patient_id = request.crosswalk.fhir_id(view.version)
+        else:
+            raise VersionNotMatched("Version not matched in has_object_permission in SearchCrosswalkPermission")
         if "patient" in request.GET and request.GET["patient"] != patient_id:
             return False
 

apps/fhir/bluebutton/tests/test_models.py

@@ -42,29 +42,23 @@ def test_not_require_user_mbi(self):
         cw = Crosswalk.objects.get(user=user)
         self.assertEqual(cw.user_mbi, None)
 
-    # TODO BB2-4166: This was commented out as part of
-    # BB2-4181, but after the work of 4166 is done, this test should
-    # pass. However, in 4181, we are storing _v3 values in _v2 (at the model level), and
-    # therefore the logic of this test does not work in 4181's branch.
-    # Once 4166 is fully implemented, uncomment this test, and it should pass and
-    # navigating the test client using the v3 pathway should work.
-    # def test_mutable_fhir_id(self):
-    #     user = self._create_user(
-    #         "john",
-    #         "password",
-    #         first_name="John",
-    #         last_name="Smith",
-    #         email="[email protected]",
-    #         fhir_id_v2="-20000000000001",
-    #         fhir_id_v3="-30000000000001",
-    #     )
-    # cw = Crosswalk.objects.get(user=user)
-    # self.assertEqual(cw.fhir_id(2), "-20000000000001")
-    # self.assertEqual(cw.fhir_id(3), "-30000000000001")
-    # cw.set_fhir_id("-20000000000002", 2)
-    # cw.set_fhir_id("-30000000000002", 3)
-    # self.assertEqual(cw.fhir_id(2), "-20000000000002")
-    # self.assertEqual(cw.fhir_id(3), "-30000000000002")
+    def test_mutable_fhir_id(self):
+        user = self._create_user(
+            "john",
+            "password",
+            first_name="John",
+            last_name="Smith",
+            email="[email protected]",
+            fhir_id_v2="-20000000000001",
+            fhir_id_v3="-30000000000001",
+        )
+        cw = Crosswalk.objects.get(user=user)
+        self.assertEqual(cw.fhir_id(2), "-20000000000001")
+        self.assertEqual(cw.fhir_id(3), "-30000000000001")
+        cw.set_fhir_id("-20000000000002", 2)
+        cw.set_fhir_id("-30000000000002", 3)
+        self.assertEqual(cw.fhir_id(2), "-20000000000002")
+        self.assertEqual(cw.fhir_id(3), "-30000000000002")
 
     def test_mutable_user_hicn_hash(self):
         user = self._create_user(