Incremental, adding impl, tests.

Author: jadudm

apps/fhir/bluebutton/tests/test_insurancecard.py

@@ -0,0 +1,62 @@
+
+from django.conf import settings
+from django.test.client import Client
+from django.urls import reverse
+from httmock import all_requests, HTTMock
+from http import HTTPStatus
+from oauth2_provider.models import get_access_token_model
+
+from apps.test import BaseApiTest
+
+# Get the pre-defined Conformance statement
+from waffle.testutils import override_switch
+
+AccessToken = get_access_token_model()
+
+FHIR_ID_V2 = settings.DEFAULT_SAMPLE_FHIR_ID_V2
+
+POSSIBLE_COVERAGE_SCOPES = ['patient/Coverage.read', 'patient/Coverage.rs', 'patient/Coverage.s']
+POSSIBLE_PATIENT_SCOPES = ['patient/Patient.read', 'patient/Patient.rs', 'patient/Patient.r']
+
+
+class InsuranceCardTest(BaseApiTest):
+    TEST_TABLE = []
+    # Add all of the passing combos. These should be 200s.
+    for C in POSSIBLE_COVERAGE_SCOPES:
+        for P in POSSIBLE_PATIENT_SCOPES:
+            TEST_TABLE.append({'status': HTTPStatus.OK, 'scope': [C, P]})
+    # If we only have one, those should fail with a 403.
+    for C in POSSIBLE_COVERAGE_SCOPES:
+        TEST_TABLE.append({'status': HTTPStatus.FORBIDDEN, 'scope': [C]})
+    for P in POSSIBLE_PATIENT_SCOPES:
+        TEST_TABLE.append({'status': HTTPStatus.FORBIDDEN, 'scope': [P]})
+
+    def setUp(self):
+        # create read and write capabilities
+        self.read_capability = self._create_capability('Read', [])
+        self.write_capability = self._create_capability('Write', [])
+        self.client = Client()
+
+    @override_switch('v3_endpoints', active=True)
+    def test_scope_combinations(self):
+        for tt in InsuranceCardTest.TEST_TABLE:
+            with self.subTest(tt=tt):
+                token = self.create_token('Annie', 'User', fhir_id_v2=FHIR_ID_V2)
+                ac = AccessToken.objects.get(token=token)
+                ac.scope = " ".join(tt['scope'])
+                ac.save()
+
+                @all_requests
+                def catchall(url, req):
+                    return {
+                        'status_code': 200,
+                        'content': {
+                            'doesnot': 'matter',
+                        }
+                    }
+                with HTTMock(catchall):
+                    response = self.client.get(
+                        reverse('bb_oauth_fhir_dic_read'),
+                        Authorization='Bearer %s' % (token)
+                    )
+                    self.assertEqual(response.status_code, tt['status'])

apps/fhir/bluebutton/v3/urls.py

@@ -11,7 +11,7 @@
     SearchViewExplanationOfBenefit,
 )
 from apps.fhir.bluebutton.views.patient_viewset import PatientViewSet
-from apps.fhir.bluebutton.views.insurancecard_viewset import DigitalInsuranceCardViewSet
+from apps.fhir.bluebutton.views.insurancecard import DigitalInsuranceCardView
 
 admin.autodiscover()
 
@@ -59,7 +59,7 @@
     # application, which means we're kinda breaking REST principles here.
     re_path(
         r'DigitalInsuranceCard[/]?',
-        waffle_switch('v3_endpoints')(DigitalInsuranceCardViewSet.as_view({'get': 'list'}, version=3)),
+        waffle_switch('v3_endpoints')(DigitalInsuranceCardView.as_view(version=3)),
         name='bb_oauth_fhir_dic_read',
     ),
 ]

apps/fhir/bluebutton/views/insurancecard.py

@@ -1,14 +1,16 @@
+from waffle import switch_is_active
 from apps.fhir.bluebutton.views.generic import FhirDataView
 from apps.fhir.bluebutton.permissions import (SearchCrosswalkPermission,
                                               ResourcePermission,
                                               ApplicationActivePermission)
 from apps.authorization.permissions import DataAccessGrantPermission
 from apps.capabilities.permissions import TokenHasProtectedCapability
+from django.http import JsonResponse
 
 from rest_framework import permissions  # pyright: ignore[reportMissingImports]
 
 
-def _is_not_empty(s: set):
+def _is_not_empty(s: set) -> bool:
     if len(s) > 0:
         return True
     else:
@@ -21,23 +23,41 @@ class HasDigitalInsuranceCardScope(permissions.BasePermission):
     required_patient_read_scopes = ['patient/Patient.r', 'patient/Patient.rs', 'patient/Patient.read']
 
     def has_permission(self, request, view) -> bool:  # type: ignore
+        print("HasDigitalInsuranceCardScope has_permission")
+
         # Is this an authorized request? If not, exit.
-        if request.get('auth', None) is None:
+        if not hasattr(request, 'auth'):
+            return False
+        if request.auth is None:
             return False
 
         # If we're authenticated, then we can check the scopes from the token.
-        token_scopes = request.auth.scope
+        token_scope_string = request.auth.scope
+        # This will be a space-separated string.
+        token_scopes = list(map(lambda s: s.strip(), token_scope_string.split(" ")))
+
         # Two things need to be true:
         #  1. At least one of the scopes in the token needs to be one of the above coverage scopes.
         #  2. At leaset one of the scopes in the token needs to be one of the above read scopes.
         coverage_set = set(HasDigitalInsuranceCardScope.required_coverage_search_scopes)
         patient_set = set(HasDigitalInsuranceCardScope.required_patient_read_scopes)
         token_set = set(token_scopes)
+
+        # print()
+        # print("CS", coverage_set)
+        # print("PS", patient_set)
+        # print("TS", token_set)
+
         return (_is_not_empty(coverage_set.intersection(token_set))
                 and _is_not_empty(patient_set.intersection(token_set)))
 
 
-class DigitalInsuranceCardSearchView(FhirDataView):
+class WaffleSwitchV3IsActive(permissions.BasePermission):
+    def has_permission(self, request, view):
+        return switch_is_active('v3_endpoints')
+
+
+class DigitalInsuranceCardView(FhirDataView):
     '''Digital Insurance Card view for handling BFD Endpoint'''
 
     permission_classes = [
@@ -59,7 +79,20 @@ def __init__(self, version=1):
         super().__init__(version)
         self.resource_type = 'Bundle'
 
+    def initial(self, request, *args, **kwargs):
+        return super().initial(request, self.resource_type, *args, **kwargs)
+
+    def get(self, request, *args, **kwargs):
+        # return super().get(request, self.resource_type, *args, **kwargs)
+        return JsonResponse(status=200, data={"ok": "go"})
+
+    # How do the has_permission herre and the has_permission in the permission classes
+    # play together? If they pass, can this fail? Visa-versa?
+
     def has_permission(self, request, view):
+        # TODO: Why is this not being called?
+        # A print statement where this comment is does not appear when unit tests are run.
+        # But, the permission classes run. Where/when does has_permission get called?
         required_scopes = getattr(view, 'required_scopes', None)
         if required_scopes is None:
             return False

apps/fhir/bluebutton/views/insurancecard_viewset.py

@@ -9,11 +9,36 @@
     ApplicationActivePermission,
 )
 
+from rest_framework.response import Response
+
+
+def _is_not_empty(s: set) -> bool:
+    if len(s) > 0:
+        return True
+    else:
+        return False
+
 
 class HasDigitalInsuranceCardScope(permissions.BasePermission):
+
+    required_coverage_search_scopes = ['patient/Coverage.rs', 'patient/Coverage.s', 'patient/Coverage.read']
+    required_patient_read_scopes = ['patient/Patient.r', 'patient/Patient.rs', 'patient/Patient.read']
+
     def has_permission(self, request, view) -> bool:  # type: ignore
-        # TODO - implement scope checking logic
-        return True
+        # Is this an authorized request? If not, exit.
+        if request.GET.get('auth', None) is None:
+            return False
+
+        # If we're authenticated, then we can check the scopes from the token.
+        token_scopes = request.auth.scope
+        # Two things need to be true:
+        #  1. At least one of the scopes in the token needs to be one of the above coverage scopes.
+        #  2. At leaset one of the scopes in the token needs to be one of the above read scopes.
+        coverage_set = set(HasDigitalInsuranceCardScope.required_coverage_search_scopes)
+        patient_set = set(HasDigitalInsuranceCardScope.required_patient_read_scopes)
+        token_set = set(token_scopes)
+        return (_is_not_empty(coverage_set.intersection(token_set))
+                and _is_not_empty(patient_set.intersection(token_set)))
 
 
 class DigitalInsuranceCardViewSet(ResourceViewSet):
@@ -44,6 +69,9 @@ def __init__(self, version, **kwargs):
     def initial(self, request, *args, **kwargs):
         return super().initial(request, self.resource_type, *args, **kwargs)
 
+    def list(self, request, resource_id, *args, **kwargs):
+        return Response({"ok": "go"})
+
     def build_url(self, fhir_settings, resource_type, resource_id=None, *args, **kwargs):
         if fhir_settings.fhir_url.endswith('v1/fhir/'):
             # only if called by tests

apps/test.py

@@ -74,6 +74,12 @@ def _create_user(
         """
         fhir_id_v2 = fhir_id_v2 or settings.DEFAULT_SAMPLE_FHIR_ID_V2
         fhir_id_v3 = fhir_id_v3 or settings.DEFAULT_SAMPLE_FHIR_ID_V3
+
+        # Some of our tests might create users more than once in the DB.
+        # Just return them if they exist.
+        if User.objects.filter(username=username).exists():
+            return User.objects.get(username=username)
+
         user = User.objects.create_user(username, password=password, **extra_fields)
         self._create_crosswalk(
             user=user,
@@ -490,6 +496,7 @@ def create_token(
         self, first_name, last_name, fhir_id_v2=None, fhir_id_v3=None, hicn_hash=None, mbi=None
     ):
         passwd = "123456"
+
         user = self._create_user(
             first_name,
             passwd,
@@ -499,7 +506,7 @@ def create_token(
             fhir_id_v3=fhir_id_v3,
             user_hicn_hash=hicn_hash if hicn_hash is not None else self.test_hicn_hash,
             user_mbi=mbi if mbi is not None else self.test_mbi,
-            email="%s@%s.net" % (first_name, last_name),
+            email="%s@%s.notanagency.gov" % (first_name, last_name),
         )
 
         # create a oauth2 application and add capabilities