Merge branch 'master' into clewellyn-nava/bb2-4271

Author: clewellyn-nava

Makefile

@@ -15,3 +15,9 @@ reqs-install:
 
 reqs-install-dev:
 	pip install -r requirements/requirements.dev.txt --no-index --find-links ./vendor/
+
+build-local:
+	cd dev-local ; make build-local ; cd ..
+
+run-local:
+	cd dev-local ; make run-local ; cd ..

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()
@@ -31,7 +32,6 @@ class Meta:
         fields = ('id', 'name', 'logo_uri', 'tos_uri', 'policy_uri', 'contacts')
 
     def get_contacts(self, obj):
-        print(obj)
         application = Application.objects.get(id=obj.id)
         return application.support_email or ""
 
@@ -68,9 +68,21 @@ 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
+
+            # V1 is treated as the same as V2 since their pathways are very similar and use the same fhir_id_v2 despite the name
+            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/core/management/commands/create_test_feature_switches.py

@@ -34,7 +34,6 @@ def handle(self, *args, **options):
         for switch in WAFFLE_FEATURE_SWITCHES:
             try:
                 Switch.objects.get(name=switch[0])
-                self._log("Feature switch already exists: %s" % (str(switch)))
             except Switch.DoesNotExist:
                 Switch.objects.create(name=switch[0], active=switch[1], note=switch[2])
                 self._log("Feature switch created: %s" % (str(switch)))
@@ -46,7 +45,6 @@ def handle(self, *args, **options):
 
             try:
                 flag_obj = Flag.objects.get(name=flag[0])
-                self._log("Feature flag already exists: %s" % (str(flag_obj)))
             except Flag.DoesNotExist:
                 flag_obj = Flag.objects.create(name=flag[0])
                 self._log("Feature flag created: %s" % (str(flag[0])))
@@ -62,7 +60,6 @@ def handle(self, *args, **options):
                                 flag_obj.save()
                                 self._log("User {} added to feature flag: {}".format(u, flag))
                             except Exception as e:
-                                print(e)
                                 self._log("Exception when adding user {} to feature flag: {}".format(u, flag))
                         except User.DoesNotExist:
                             # assuming test users exist before creating flags associated with them

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/tests/test_views.py

@@ -24,6 +24,8 @@
     SCOPES_TO_URL_BASE_PATH,
 )
 
+import os
+
 from hhs_oauth_server.settings.base import MOCK_FHIR_ENDPOINT_HOSTNAME
 
 
@@ -576,16 +578,15 @@ def test_delete_token_success(self):
 
         # This assertion is incorrectly crafted - it actually requires a local server started
         # so that the fhir fetch data is called and hence generate cert file not found error.
-        # TODO: refactor test to not depend on a server up and running.
-
-        # Post Django 2.2:  An OSError exception is expected when trying to reach the
-        #                   backend FHIR server and proves authentication worked.
-        with self.assertRaisesRegexp(
-            OSError, 'Could not find the TLS certificate file'
-        ):
-            response = self.client.get(
-                '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + anna_token.token}
-            )
+        # 20251120 This test is now gated on a variable; if the variable does not exist, or
+        # is not set, the test will run. This is the desired behavior.
+        if os.getenv("RUNNING_IN_LOCAL_STACK", None) != "true":
+            with self.assertRaisesRegexp(
+                OSError, 'Could not find the TLS certificate file'
+            ):
+                response = self.client.get(
+                    '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + anna_token.token}
+                )
 
         bob_tkn = self._create_test_token(bob, bob_application)
         self.assertTrue(
@@ -638,24 +639,26 @@ def test_delete_token_success(self):
 
         # Post Django 2.2:  An OSError exception is expected when trying to reach the
         #                   backend FHIR server and proves authentication worked.
-        with self.assertRaisesRegexp(
-            OSError, 'Could not find the TLS certificate file'
-        ):
-            response = self.client.get(
-                '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + bob_tkn.token}
-            )
+        if os.getenv("RUNNING_IN_LOCAL_STACK", None) != "true":
+            with self.assertRaisesRegexp(
+                OSError, 'Could not find the TLS certificate file'
+            ):
+                response = self.client.get(
+                    '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + bob_tkn.token}
+                )
 
         next_tkn = self._create_test_token(anna, anna_application)
 
         # Post Django 2.2:  An OSError exception is expected when trying to reach the
         #                   backend FHIR server and proves authentication worked.
-        with self.assertRaisesRegexp(
-            OSError, 'Could not find the TLS certificate file'
-        ):
-            response = self.client.get(
-                '/v1/fhir/Patient',
-                headers={'authorization': 'Bearer ' + next_tkn.token},
-            )
+        if os.getenv("RUNNING_IN_LOCAL_STACK", None) != "true":
+            with self.assertRaisesRegexp(
+                OSError, 'Could not find the TLS certificate file'
+            ):
+                response = self.client.get(
+                    '/v1/fhir/Patient',
+                    headers={'authorization': 'Bearer ' + next_tkn.token},
+                )
 
         # self.assertEqual(next_tkn.token, tkn.token)
         self.assertTrue(