Merge pull request #1 from CMSgov/BB2-1117_fhir_endpoints

[ BB2-1117 ] fhir endpoints

Author: Nick Bragdon

.gitignore

@@ -148,6 +148,12 @@ dmypy.json
 # Cython debug symbols
 cython_debug/
 
+# VS Code
+.vscode/
+
+# Virtual Env
+bb2_env/
+
 # PyCharm
 #  JetBrains specific template is maintainted in a separate JetBrains.gitignore that can
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore

src/constants.py

@@ -0,0 +1,3 @@
+SDK_HEADER = "python"
+SDK_HEADER_KEY = "X-BLUEBUTTON-SDK"
+REFRESH_TOKEN_ENDPOINT = "/o/token/"

src/fhir_request.py

@@ -0,0 +1,52 @@
+import datetime
+from constants import REFRESH_TOKEN_ENDPOINT, SDK_HEADER, SDK_HEADER_KEY
+import requests
+from requests.adapters import HTTPAdapter, Retry
+
+
+def fhir_request(bb, config):
+    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=5, status_forcelist=[500, 502, 503, 504]
+    )
+    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)
+
+    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"])
+
+    return None
+
+
+def refresh_access_token(bb, refresh_token):
+    full_url = bb["base_url"] + "/v" + bb["version"] + REFRESH_TOKEN_ENDPOINT
+
+    params = {
+        "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"])
+    )
+    response_json = my_response.json()
+    response_json["expires_at"] = datetime.datetime.now() + datetime.timedelta(
+        seconds=response_json["expires_in"]
+    )
+    return response_json

src/test_fhir_request.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)