added tests to cover on demand token refresh enabled/disable

Author: JAMES FUQIAN

cms_bluebutton/fhir_request.py

@@ -6,7 +6,6 @@
 
 def fhir_request(bb, config):
     auth_token = config["auth_token"]
-
     if bb.token_refresh_on_expire:
         auth_token = handle_expired(bb, auth_token)
 

cms_bluebutton/tests/test_fhir_request.py

@@ -26,6 +26,18 @@ def json(self):
         return self.json_data
 
 
+class MockResponseWithRaiseForStatus:
+    def __init__(self, json_data, status_code):
+        self.json_data = json_data
+        self.status_code = status_code
+
+    def json(self):
+        return self.json_data
+
+    def raise_for_status(self):
+        pass
+
+
 class MockSession:
     def __init__(self, json_data, status_code):
         self.response = MockResponse(json_data, status_code)
@@ -58,10 +70,45 @@ def get(self, *args, **kwargs):
                 return MockResponse(json.load(f), 200)
 
 
+class MockSessionTokenRefresh:
+    def __init__(self, json_data, status_code):
+        self.response = MockResponse(json_data, status_code)
+
+    def mount(self, *args, **kwargs):
+        return
+
+    def get(self, *args, **kwargs):
+        endpoint_url = kwargs['url']
+        parsed_url = urlparse(endpoint_url)
+        if parsed_url.path.endswith('Patient/'):
+            # patient
+            return self.response
+        else:
+            raise ValueError("Unexpected GET path={}".format(parsed_url.path))
+
+
 def success_fhir_patient_request_mock(*args, **kwargs):
     return MockSession({"resourceType": "Patient", "id": "-20140000010000"}, 200)
 
 
+def mocked_token_refresh_post(*args, **kwargs):
+    endpoint_url = kwargs['url']
+    parsed_url = urlparse(endpoint_url)
+    if parsed_url.path.endswith('token/'):
+        # auth token - refreshed
+        expires_at_str = str(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=36000))
+        return MockResponseWithRaiseForStatus({"access_token": "fake_access_token",
+                                               "refresh_token": "fake_refresh_token",
+                                               "expires_at": expires_at_str,
+                                               "patient": "-20140000010000"}, 200)
+    else:
+        raise ValueError("Unexpected POST path={}".format(parsed_url.path))
+
+
+def success_fhir_patient_request_refresh_token_mock(*args, **kwargs):
+    return MockSessionTokenRefresh({"resourceType": "Patient", "id": "-20140000010000"}, 200)
+
+
 def success_fhir_coverage_request_mock(*args, **kwargs):
     return MockSession({"resourceType": "Bundle", "id": "aaa-111-111-111-aaaa"}, 200)
 
@@ -101,6 +148,21 @@ def generate_mock_config():
     }
 
 
+def generate_mock_config_w_expired_access_token():
+    return {
+        "params": {},
+        "auth_token": AuthorizationToken(
+            {
+                "access_token": "fake_access_token",
+                "refresh_token": "fake_refresh_token",
+                "expires_at": datetime.datetime.now(datetime.timezone.utc)
+                - datetime.timedelta(seconds=36000),
+                "patient": "-20140000010000",
+            }
+        ),
+    }
+
+
 class TestAPI(unittest.TestCase):
     @mock.patch("requests.Session", side_effect=success_fhir_patient_request_mock)
     def test_successful_fhir_patient_request(self, get_request_mock):
@@ -151,6 +213,40 @@ def test_successful_fhir_eob_pages_request(self, get_request_mock):
         self.assertEqual(len(pages["pages"]), 6)
         self.assertEqual(get_request_mock.call_count, 6)
 
+    @mock.patch("requests.post", side_effect=mocked_token_refresh_post)
+    @mock.patch("requests.Session", side_effect=success_fhir_patient_request_refresh_token_mock)
+    def test_successful_fhir_request_token_refreshed(self, post_mock, get_mock):
+        bb = BlueButton(config=MOCK_BB_CONFIG)
+        config = generate_mock_config_w_expired_access_token()
+        response = bb.get_patient_data(config)
+        self.assertTrue(config['auth_token'].access_token_expired())
+        self.assertIsNotNone(response["auth_token"])
+        self.assertFalse(response['auth_token'].access_token_expired())
+        self.assertEqual(response["response"].status_code, 200)
+        self.assertEqual(response["response"].json()["id"], "-20140000010000")
+        self.assertEqual(response["response"].json()["resourceType"], "Patient")
+        # mock post called once for token refresh
+        self.assertEqual(post_mock.call_count, 1)
+        # mock get called once for fhir resource
+        self.assertEqual(get_mock.call_count, 1)
+
+    @mock.patch("requests.post", side_effect=mocked_token_refresh_post)
+    @mock.patch("requests.Session", side_effect=success_fhir_patient_request_refresh_token_mock)
+    def test_successful_fhir_request_token_refresh_disabled(self, get_mock, post_mock):
+        bb = BlueButton(config="./tests/test_configs/json/bluebutton-sample-config-disable-token-refresh-on-expire.json")
+        config = generate_mock_config_w_expired_access_token()
+        response = bb.get_patient_data(config)
+        self.assertTrue(config['auth_token'].access_token_expired())
+        self.assertIsNotNone(response["auth_token"])
+        self.assertTrue(response['auth_token'].access_token_expired())
+        self.assertEqual(response["response"].status_code, 200)
+        self.assertEqual(response["response"].json()["id"], "-20140000010000")
+        self.assertEqual(response["response"].json()["resourceType"], "Patient")
+        # mock post for toekn not called
+        self.assertEqual(post_mock.call_count, 0)
+        # mock get called once for fhir resource
+        self.assertEqual(get_mock.call_count, 1)
+
     @mock.patch("requests.Session", side_effect=success_fhir_profile_request_mock)
     def test_successful_fhir_profile_request(self, get_request_mock):
         bb = BlueButton(config=MOCK_BB_CONFIG)