Fix some failing unit tests, add integration tests for 403s

Author: JamesDemeryNava

apps/authorization/tests/test_data_access_grant_permissions.py

@@ -13,7 +13,7 @@
 from apps.authorization.models import (
     DataAccessGrant,
 )
-from waffle.testutils import override_switch
+from waffle.testutils import override_switch, override_flag
 from apps.fhir.bluebutton.tests.test_fhir_resources_read_search_w_validation import (
     get_response_json,
 )
@@ -145,6 +145,7 @@ def setUp(self):
         self.client = Client()
 
     @override_switch('v3_endpoints', active=True)
+    @override_flag('v3_early_adopter', active=True)
     def _assert_call_all_fhir_endpoints(
         self,
         access_token=None,

apps/core/models.py

@@ -1,9 +1,12 @@
 from waffle.models import AbstractUserFlag
+from django.db import models
 
 
 class Flag(AbstractUserFlag):
     """ Custom version of waffle feature Flag model """
     """ This makes future extensions nicer """
+    # Added as part of BB2-4250 testing
+    objects = models.Manager()
 
     def is_active_for_user(self, user):
         # use app_user which is the owner of the current app

apps/dot_ext/tests/test_authorization.py

@@ -11,7 +11,8 @@
 from django.http import HttpRequest
 from django.urls import reverse
 from django.test import Client
-from waffle.testutils import override_switch
+from apps.core.models import Flag
+from waffle.testutils import override_switch, override_flag
 
 from apps.test import BaseApiTest
 from ..models import Application, ArchivedToken
@@ -1020,3 +1021,66 @@ def _execute_token_endpoint(self, token_path):
         c = Client()
         response = c.post(token_path, data=token_request_data)
         self.assertEqual(response.status_code, 200)
+
+    @override_switch('v3_endpoints', active=True)
+    @override_flag('v3_early_adopter', active=True)
+    def test_v3_token_endpoint_with_early_adopter_flag_enabled(self):
+        self._execute_token_endpoint('/v3/o/token/')
+
+    @override_switch('v3_endpoints', active=True)
+    # @override_flag('v3_early_adopter', active=False)
+    def test_v3_token_endpoint_with_early_adopter_flag_disabled(self):
+        self._execute_token_endpoint_for_flag_test('/v3/o/token/')
+
+    # @override_switch('v3_endpoints', active=True)
+    # @override_flag('v3_early_adopter', active=True)
+    # def test_v3_token_endpoint_without_trailling_slash(self):
+    #     self._execute_token_endpoint('/v3/o/token')
+
+    def _execute_token_endpoint_for_flag_test(self, token_path):
+        Flag.objects.create(name='v3_early_adopter', everyone=None)
+        redirect_uri = 'http://localhost'
+        # create a user
+        user = self._create_user('anna', '123456')
+        capability_a = self._create_capability('Capability A', [])
+        capability_b = self._create_capability('Capability B', [])
+        print("USER ID: ", user.id)
+        # create an application and add capabilities
+        application = self._create_application(
+            'an app',
+            grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            client_type=Application.CLIENT_CONFIDENTIAL,
+            redirect_uris=redirect_uri,
+            user_id=user.id)
+        application.scope.add(capability_a, capability_b)
+        # user logs in
+        request = HttpRequest()
+        self.client.login(request=request, username='anna', password='123456')
+        # post the authorization form with only one scope selected
+        payload = {
+            'client_id': application.client_id,
+            'response_type': 'code',
+            'redirect_uri': redirect_uri,
+            'scope': ['capability-a'],
+            'expires_in': 86400,
+            'allow': True,
+            "state": "0123456789abcdef",
+            'refresh_token': 'asdfj23h4q98wuafidj'
+        }
+        response = self.client.post(reverse('oauth2_provider:authorize'), data=payload)
+        self.client.logout()
+        self.assertEqual(response.status_code, 302)
+        # now extract the authorization code and use it to request an access_token
+        query_dict = parse_qs(urlparse(response['Location']).query)
+        authorization_code = query_dict.pop('code')
+        token_request_data = {
+            'grant_type': 'authorization_code',
+            'code': authorization_code,
+            'redirect_uri': redirect_uri,
+            'client_id': application.client_id,
+            'client_secret': application.client_secret_plain,
+        }
+        c = Client()
+        print("token path: ", token_path)
+        response = c.post(token_path, data=token_request_data)
+        self.assertEqual(response.status_code, 403)

apps/dot_ext/views/authorization.py

@@ -448,8 +448,7 @@ def validate_token_endpoint_request_body(self, request):
 
     def validate_v3_token_call(self, request) -> None:
         flag = get_waffle_flag_model().get('v3_early_adopter')
-        req_meta = request.META
-        url_query = parse_qs(req_meta.get('QUERY_STRING'))
+
         try:
             url_query = parse_qs(request._body.decode('utf-8'))
             refresh_token_from_request = url_query.get('refresh_token', [None])

apps/fhir/bluebutton/tests/test_wellknown_endpoints.py

@@ -3,10 +3,9 @@
 from collections import namedtuple as NT
 from django.conf import settings
 from django.test.client import Client
-from httmock import all_requests, HTTMock
 from oauth2_provider.models import get_access_token_model
 from unittest import skipIf
-from waffle.testutils import override_switch
+from waffle.testutils import override_switch, override_flag
 
 # Introduced in bb2-4184
 # Rudimentary tests to make sure endpoints exist and are returning
@@ -52,29 +51,29 @@ def setUp(self):
 
     @skipIf((not settings.RUN_ONLINE_TESTS), 'Can\'t reach external sites.')
     @override_switch('v3_endpoints', active=True)
+    def test_userinfo_returns_403(self):
+        first_access_token = self.create_token('John', 'Smith', fhir_id_v2=FHIR_ID_V2)
+        ac = AccessToken.objects.get(token=first_access_token)
+        ac.save()
+
+        response = self.client.get(
+            f'{BASEURL}/v3/connect/userinfo',
+            Authorization='Bearer %s' % (first_access_token))
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.json()['detail'], settings.APPLICATION_DOES_NOT_HAVE_V3_ENABLED_YET.format('John_Smith_test'))
+
+    @skipIf((not settings.RUN_ONLINE_TESTS), 'Can\'t reach external sites.')
+    @override_switch('v3_endpoints', active=True)
+    @override_flag('v3_early_adopter', active=True)
     def test_userinfo_returns_200(self):
         first_access_token = self.create_token('John', 'Smith', fhir_id_v2=FHIR_ID_V2)
         ac = AccessToken.objects.get(token=first_access_token)
         ac.save()
 
-        @all_requests
-        def catchall(url, req):
-            return {'status_code': 200,
-                    'content': {
-                        'sub': FHIR_ID_V2,
-                        'name': ' ',
-                        'given_name': '',
-                        'family_name': '',
-                        'email': '',
-                        'iat': '2025-10-14T18:01:01.660Z',
-                        'patient': FHIR_ID_V2
-                    }}
-
-        with HTTMock(catchall):
-            response = self.client.get(
-                f'{BASEURL}/v3/connect/userinfo',
-                Authorization='Bearer %s' % (first_access_token))
-            self.assertEqual(response.status_code, 200)
+        response = self.client.get(
+            f'{BASEURL}/v3/connect/userinfo',
+            Authorization='Bearer %s' % (first_access_token))
+        self.assertEqual(response.status_code, 200)
 
     # This makes sure URLs return 200s.
     @skipIf((not settings.RUN_ONLINE_TESTS), "Can't reach external sites.")

apps/integration_tests/integration_test_fhir_resources.py

@@ -2,10 +2,12 @@
 
 from django.conf import settings
 from django.contrib.staticfiles.testing import StaticLiveServerTestCase
+from http import HTTPStatus
 from oauth2_provider.models import AccessToken
 from rest_framework.test import APIClient
 from waffle.testutils import override_switch
 
+from apps.core.models import Flag
 from apps.test import BaseApiTest
 from apps.testclient.utils import extract_last_page_index
 from .common_utils import validate_json_schema
@@ -36,6 +38,9 @@
 FHIR_RES_TYPE_EOB = "ExplanationOfBenefit"
 FHIR_RES_TYPE_PATIENT = "Patient"
 FHIR_RES_TYPE_COVERAGE = "Coverage"
+V3_403_DETAIL = 'This application, John_Doe_test, does not yet have access to v3 endpoints. ' \
+                'If you are the app maintainer, please contact the Blue Button API team. If you are a Medicare Beneficiary ' \
+                'and need assistance, please contact the support team for the application you are trying to access.'
 
 
 def dump_content(json_str, file_name):
@@ -759,3 +764,59 @@ def _err_response_caused_by_illegalarguments(self, v2=False):
         # This now returns 400 after BB2-2063 work.
         # for both v1 and v2
         self.assertEqual(response.status_code, 400)
+
+    @override_switch('v3_endpoints', active=True)
+    def test_patient_read_endpoint_v3_403(self):
+        '''
+        test patient read and search v2
+        '''
+        self._call_v3_endpoint_to_assert_403(FHIR_RES_TYPE_PATIENT, settings.DEFAULT_SAMPLE_FHIR_ID_V3, False, None)
+
+    @override_switch('v3_endpoints', active=True)
+    def test_coverage_read_endpoint_v3_403(self):
+        '''
+        test patient read and search v2
+        '''
+        self._call_v3_endpoint_to_assert_403(FHIR_RES_TYPE_COVERAGE, 'part-a-99999999999999', False, None)
+
+    @override_switch('v3_endpoints', active=True)
+    def test_eob_read_endpoint_v3_403(self):
+        '''
+        test patient read and search v2
+        '''
+        self._call_v3_endpoint_to_assert_403(FHIR_RES_TYPE_EOB, 'outpatient--9999999999999', False, None)
+
+    @override_switch('v3_endpoints', active=True)
+    def test_patient_search_endpoint_v3_403(self):
+        '''
+        test patient read and search v2
+        '''
+        self._call_v3_endpoint_to_assert_403(FHIR_RES_TYPE_PATIENT, settings.DEFAULT_SAMPLE_FHIR_ID_V3, True, '_id=')
+
+    @override_switch('v3_endpoints', active=True)
+    def test_coverage_search_endpoint_v3_403(self):
+        '''
+        test patient read and search v2
+        '''
+        self._call_v3_endpoint_to_assert_403(FHIR_RES_TYPE_COVERAGE, settings.DEFAULT_SAMPLE_FHIR_ID_V3, True, 'beneficiary=')
+
+    @override_switch('v3_endpoints', active=True)
+    def test_eob_search_endpoint_v3_403(self):
+        '''
+        test patient read and search v2
+        '''
+        self._call_v3_endpoint_to_assert_403(FHIR_RES_TYPE_EOB, settings.DEFAULT_SAMPLE_FHIR_ID_V3, True, 'patient=')
+
+    def _call_v3_endpoint_to_assert_403(self, resource_type: str, resource_value: str, search: bool, search_param: str):
+        client = APIClient()
+
+        # Authenticate
+        self._setup_apiclient(client)
+        Flag.objects.create(name='v3_early_adopter', everyone=None)
+        if search:
+            endpoint_url = "{}/v3/fhir/{}/?{}{}".format(self.live_server_url, resource_type, search_param, resource_value)
+        else:
+            endpoint_url = "{}/v3/fhir/{}/{}".format(self.live_server_url, resource_type, resource_value)
+        response = client.get(endpoint_url)
+        self.assertEqual(response.status_code, HTTPStatus.FORBIDDEN)
+        self.assertEqual(response.json()['detail'], V3_403_DETAIL)