Adding tests

Author: Nick Bragdon

src/fhirRequest.py

@@ -5,47 +5,45 @@
 
 
 def fhirRequest(bb, config):
-    auth_token = config.auth_token
-    new_auth_token = handle_expired(bb, auth_token, config.refresh_token)
+    auth_token = config["auth_token"]
+    new_auth_token = handle_expired(bb, auth_token)
 
     if new_auth_token is not None:
         auth_token = new_auth_token
 
     retry_config = Retry(
         total=3, backoff_factor=0.2, status_forcelist=[500, 502, 503, 504]
     )
-    full_url = bb.base_url + "/v" + bb.version + config.url
-    headers = {
-        "Authorization": "Bearer " + auth_token.access_token,
-        [SDK_HEADER_KEY]: SDK_HEADER,
-    }
+    full_url = bb["base_url"] + "/v" + bb["version"] + config["url"]
+    headers = {"Authorization": "Bearer " + auth_token["access_token"]}
+    headers[SDK_HEADER_KEY] = SDK_HEADER
     adapter = HTTPAdapter(max_retries=retry_config)
     sesh = requests.Session()
     sesh.mount("https://", adapter)
     sesh.mount("http://", adapter)
-    response = sesh.get(url=full_url, params=config.params, headers=headers)
+    response = sesh.get(url=full_url, params=config["params"], headers=headers)
 
     return {"auth_token": new_auth_token, "response": response}
 
 
 def handle_expired(bb, auth_token):
-    if datetime.datetime.now() > auth_token.expires_at:
-        return refresh_access_token(bb, auth_token.refresh_token)
+    if datetime.datetime.now() > auth_token["expires_at"]:
+        return refresh_access_token(bb, auth_token["refresh_token"])
 
     return None
 
 
 def refresh_access_token(bb, refresh_token):
-    full_url = bb.base_url + "/v" + bb.version + REFRESH_TOKEN_ENDPOINT
+    full_url = bb["base_url"] + "/v" + bb["version"] + REFRESH_TOKEN_ENDPOINT
 
     params = {
-        "client_id": bb.client_id,
+        "client_id": bb["client_id"],
         "grant_type": "refresh_token",
         "refresh_token": refresh_token,
     }
 
     my_response = requests.post(
-        url=full_url, params=params, auth=(bb.client_id, bb.client_secret)
+        url=full_url, params=params, auth=(bb["client_id"], bb["client_secret"])
     )
     response_json = my_response.json()
     response_json["expires_at"] = datetime.datetime.now() + datetime.timedelta(

src/test_fhirRequest.py

@@ -0,0 +1,88 @@
+import datetime
+import unittest
+from unittest import mock
+from fhirRequest import fhirRequest
+
+
+class MockResponse:
+    def __init__(self, json_data, status_code):
+        self.json_data = json_data
+        self.status_code = status_code
+
+    def json(self):
+        return self.json_data
+
+
+class MockSession:
+    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):
+        return self.response
+
+
+def success_fhir_request_mock(*args, **kwargs):
+    return MockSession({"resourceType": "Patient", "id": "-20140000010000"}, 200)
+
+
+def error_fhir_request_mock(*args, **kwargs):
+    return MockSession(None, 500)
+
+
+def not_found_fhir_request_mock(*args, **kwargs):
+    return MockSession(None, 404)
+
+
+def generate_mock_bb():
+    return {
+        "base_url": "https://sandbox.bluebutton.cms.gov",
+        "client_id": "fake_client_id",
+        "client_secret": "fake_client_secret",
+        "version": "2",
+    }
+
+
+def generate_mock_config():
+    return {
+        "url": "/fhir/Patient/-20140000010000",
+        "params": {},
+        "auth_token": {
+            "access_token": "fake_access_token",
+            "refresh_token": "fake_refresh_token",
+            "expires_at": datetime.datetime.now() + datetime.timedelta(seconds=36000),
+            "patient": "-20140000010000",
+        },
+    }
+
+
+class TestAPI(unittest.TestCase):
+    @mock.patch("requests.Session", side_effect=success_fhir_request_mock)
+    def test_successful_fhir_request(self, get_request_mock):
+        bb = generate_mock_bb()
+        config = generate_mock_config()
+        response = fhirRequest(bb, config)
+        self.assertEqual(response["auth_token"], None)
+        self.assertEqual(response["response"].status_code, 200)
+        self.assertEqual(response["response"].json()["id"], "-20140000010000")
+        self.assertEqual(get_request_mock.call_count, 1)
+
+    @mock.patch("requests.Session", side_effect=error_fhir_request_mock)
+    def test_500_error_fhir_request(self, get_request_mock):
+        bb = generate_mock_bb()
+        config = generate_mock_config()
+        response = fhirRequest(bb, config)
+        self.assertEqual(response["auth_token"], None)
+        self.assertEqual(response["response"].status_code, 500)
+        self.assertEqual(get_request_mock.call_count, 1)
+
+    @mock.patch("requests.Session", side_effect=not_found_fhir_request_mock)
+    def test_not_found_fhir_request(self, get_request_mock):
+        bb = generate_mock_bb()
+        config = generate_mock_config()
+        response = fhirRequest(bb, config)
+        self.assertEqual(response["auth_token"], None)
+        self.assertEqual(response["response"].status_code, 404)
+        self.assertEqual(get_request_mock.call_count, 1)