prmdr-327 add test for jwt token

nhsconnect · Oct 17, 2023 · 0923b82 · 0923b82
1 parent bfda67f
commit 0923b82
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 9 deletions.
diff --git a/lambdas/services/pds_api_service.py b/lambdas/services/pds_api_service.py
@@ -1,7 +1,7 @@
 import json
 import logging
 import uuid
-from time import time
+import time
 
 import jwt
 import requests
@@ -60,7 +60,7 @@ def pds_request(self, nshNumber: str, retry_on_expired: bool):
             access_token_response["issued_at"]
         )
         time_safety_margin_seconds = 10
-        if time() - access_token_expiration < time_safety_margin_seconds:
+        if time.time() - access_token_expiration < time_safety_margin_seconds:
             access_token = self.get_new_access_token()
 
         x_request_id = str(uuid.uuid4())
@@ -119,17 +119,17 @@ def get_parameters_for_pds_api_request(self):
         )
         return ssm_response[parameters[0]], json.loads(ssm_response[parameters[1]])
 
-    def create_jwt_token_for_new_access_token_request(self, access_token_ssm_parameter):
-        nhs_oauth_endpoint = access_token_ssm_parameter[SSMParameter.NHS_OAUTH_ENDPOINT]
-        kid = access_token_ssm_parameter[SSMParameter.PDS_KID]
-        nhs_key = access_token_ssm_parameter[SSMParameter.NHS_OAUTH_KEY]
-        pds_key = access_token_ssm_parameter[SSMParameter.PDS_API_KEY]
+    def create_jwt_token_for_new_access_token_request(self, access_token_ssm_parameters):
+        nhs_oauth_endpoint = access_token_ssm_parameters[SSMParameter.NHS_OAUTH_ENDPOINT]
+        kid = access_token_ssm_parameters[SSMParameter.PDS_KID]
+        nhs_key = access_token_ssm_parameters[SSMParameter.NHS_OAUTH_KEY]
+        pds_key = access_token_ssm_parameters[SSMParameter.PDS_API_KEY]
         payload = {
             "iss": nhs_key,
             "sub": nhs_key,
             "aud": nhs_oauth_endpoint,
             "jti": str(uuid.uuid4()),
-            "exp": int(time()) + 300,
+            "exp": int(time.time()) + 300,
         }
         return jwt.encode(payload, pds_key, algorithm="RS512", headers={"kid": kid})
 

diff --git a/lambdas/tests/unit/services/test_pds_api_service.py b/lambdas/tests/unit/services/test_pds_api_service.py
@@ -6,10 +6,16 @@
 
 from services.pds_api_service import PdsApiService
 
+from enums.pds_ssm_parameters import SSMParameter
+
+
 class FakeSSMService:
     def __init__(self, *arg, **kwargs):
         pass
 
+    def get_ssm_parameters(self, *arg, **kwargs):
+        pass
+
 
 pds_service = PdsApiService(FakeSSMService)
 
@@ -65,7 +71,7 @@ def test_handle_response_catch_all_raises_PdsErrorException(mocker):
         pds_service.handle_response(response, nhs_number)
 
 def test_request_new_token_is_call_with_correct_data(mocker):
-    mock_jwt_token = "fgjkstjgkld"
+    mock_jwt_token = "testtest"
     mock_endpoint = "api.endpoint/mock"
     access_token_headers = {"content-type": "application/x-www-form-urlencoded"}
     access_token_data = {
@@ -77,3 +83,23 @@ def test_request_new_token_is_call_with_correct_data(mocker):
     pds_service.request_new_access_token(mock_jwt_token, mock_endpoint)
     mock_post.assert_called_with(url=mock_endpoint, headers=access_token_headers, data=access_token_data)
 
+def test_create_jwt_for_new_access_token(mocker):
+    access_token_parameters = {SSMParameter.NHS_OAUTH_ENDPOINT : "api.endpoint/mock",
+                               SSMParameter.PDS_KID: "test_string_pds_kid",
+                                SSMParameter.NHS_OAUTH_KEY: "test_string_key_oauth",
+                               SSMParameter.PDS_API_KEY: "test_string_key_pds"
+    }
+    expected_payload = {
+        "iss": "test_string_key_oauth",
+        "sub": "test_string_key_oauth",
+        "aud": "api.endpoint/mock",
+        "jti": "123412342",
+        "exp": 1534,
+    }
+    mocker.patch("time.time", return_value=1234.1)
+    mocker.patch("uuid.uuid4", return_value="123412342")
+
+    mock_jwt_encode = mocker.patch("jwt.encode")
+    pds_service.create_jwt_token_for_new_access_token_request(access_token_parameters)
+    mock_jwt_encode.assert_called_with(expected_payload, "test_string_key_pds", algorithm="RS512", headers={"kid": "test_string_pds_kid"})
+