getting properly failing test on refreshs

Author: bwang-icf

apps/dot_ext/tests/test_authorization.py

@@ -234,6 +234,91 @@ def test_post_with_invalid_non_standard_scheme_granttype_authcode_clienttype_con
         response = self.client.post(reverse('oauth2_provider:authorize'), data=payload)
         self.assertEqual(response.status_code, 400)
 
+    # FIXME: This should be merged somehow with test_refresh_token and also include version checking.
+    # Currently, this is expected to fail because the fhir_ids somehow aren't being updated on refresh.
+    def test_refresh_token_fhir_id_storing(self):
+        redirect_uri = 'http://localhost'
+        # create a user
+        self._create_user('anna', '123456')
+        capability_a = self._create_capability('Capability A', [])
+        capability_b = self._create_capability('Capability B', [])
+        # 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)
+        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",
+        }
+        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()
+        if switch_is_active('v3_endpoints'):
+            response = c.post('/v3/o/token/', data=token_request_data)
+        else:
+            response = c.post('/v2/o/token/', data=token_request_data)
+        self.assertEqual(response.status_code, 200)
+        # Now we have a token and refresh token
+        tkn = response.json()['access_token']
+        refresh_tkn = response.json()['refresh_token']
+        refresh_request_data = {
+            'grant_type': 'refresh_token',
+            'refresh_token': refresh_tkn,
+            'redirect_uri': redirect_uri,
+            'client_id': application.client_id,
+            'client_secret': application.client_secret_plain,
+        }
+        response = self.client.post(reverse('oauth2_provider:token'), data=refresh_request_data)
+        self.assertEqual(response.status_code, 200)
+        self.assertNotEqual(response.json()['access_token'], tkn)
+        # Capture rotated refresh token (server may rotate refresh tokens)
+        new_refresh = response.json().get('refresh_token')
+        if new_refresh:
+            refresh_request_data['refresh_token'] = new_refresh
+        user = User.objects.get(username='anna')
+        crosswalk = Crosswalk.objects.get(user=user)
+        print(f'what is in crosswalk {crosswalk.fhir_id_v3}')
+        # Verify both fhir_id_v2 and fhir_id_v3 are populated
+        self.assertIsNotNone(crosswalk.fhir_id_v2)
+        self.assertIsNotNone(crosswalk.fhir_id_v3)
+        self.assertTrue(len(crosswalk.fhir_id_v2) > 0)
+        self.assertTrue(len(crosswalk.fhir_id_v3) > 0)
+        # Changing the fhir ids to test that they get updated on refresh
+        crosswalk.fhir_id_v2 = 'old_fhir_id_v2'
+        crosswalk.fhir_id_v3 = 'old_fhir_id_v3'
+        crosswalk.save()
+        response = self.client.post(reverse('oauth2_provider:token'), data=refresh_request_data)
+        print(f'Refresh response: {response.json()}')
+        self.assertEqual(response.status_code, 200)
+        self.assertNotEqual(response.json()['access_token'], tkn)
+        crosswalk.refresh_from_db()
+        # Verify both fhir_id_v2 and fhir_id_v3 are updated
+        self.assertNotEqual(crosswalk.fhir_id_v2, 'old_fhir_id_v2')
+        self.assertNotEqual(crosswalk.fhir_id_v3, 'old_fhir_id_v3')
+
     def test_refresh_token(self):
         redirect_uri = 'http://localhost'
         # create a user