more unit tests

tomklapiscak · tomklapiscak · commit 9cf9fceb84ef · 2025-04-11T10:34:35.000+01:00
diff --git a/src/mas/devops/users.py b/src/mas/devops/users.py
@@ -172,7 +172,7 @@ def manage_internal_ca_pem_file_path(self):
     @property
     def manage_maxadmin_api_key(self):
         if self._manage_maxadmin_api_key is None:
-            self._manage_maxadmin_api_key = self.create_or_get_manage_api_key_for_user(MASUserUtils.MAXADMIN)
+            self._manage_maxadmin_api_key = self.create_or_get_manage_api_key_for_user(MASUserUtils.MAXADMIN, temporary=True)
         return self._manage_maxadmin_api_key
 
     @property
@@ -502,7 +502,11 @@ def resync_users(self, user_ids):
             user = self.get_user(user_id)
             self.update_user_display_name(user_id, user["displayName"])
 
-    def create_or_get_manage_api_key_for_user(self, user_id):
+    def create_or_get_manage_api_key_for_user(self, user_id, temporary=False):
+        '''
+        Get singleton API for user_id if it already exists, create it if not
+        if temporary is True AND we created the API key, delete it on exit
+        '''
         self.logger.debug(f"Attempting to create Manage API Key for user {user_id}")
         url = f"{self.manage_api_url_internal}/maximo/api/os/mxapiapikey"
         querystring = {
@@ -531,14 +535,19 @@ def create_or_get_manage_api_key_for_user(self, user_id):
         )
 
         if response.status_code == 400:
-            error_json = response.json()
+            # Assisted by watsonx Code Assistant
+            try:
+                error_json = response.json()
+            except ValueError:
+                raise Exception(f"{response.status_code} {response.text}")
+
             if "Error" in error_json and "reasonCode" in error_json["Error"] and error_json["Error"]["reasonCode"] == "BMXAA10051E":
                 # BMXAA10051E - Only one API key allowed per user.
                 self.logger.info(f"Reusing existing Manage API Key for user {user_id}")
                 pass
             else:
                 # any other 400 error is unexpected
-                raise Exception(response.text)
+                raise Exception(f"{response.status_code} {response.text}")
 
         elif response.status_code == 201:
             self.logger.info(f"Creating new Manage API Key for user {user_id}")
@@ -554,11 +563,7 @@ def create_or_get_manage_api_key_for_user(self, user_id):
             # so we expect the get call to find it
             raise Exception("API key was unexpectedly not found")
 
-        if response.status_code == 201:
-            # We created this api key, register a handler to delete it when we're done
-            # TODO: is there a danger of some other process adopting this singleton api key
-            #       which may then fail if we subsequently delete it?
-            #       NOTE: the manage sanity test seems to create maxadmin apikey and not clean it up
+        if temporary and response.status_code == 201:
             atexit.register(self.delete_manage_api_key, apikey)
 
         return apikey
@@ -592,7 +597,7 @@ def get_manage_api_key_for_user(self, user_id):
 
             return None
 
-        raise Exception(response.text)
+        raise Exception(f"{response.status_code} {response.text}")
 
     def delete_manage_api_key(self, manage_api_key):
         self.logger.info(f"Deleting Manage API Key for user {manage_api_key['userid']}")
diff --git a/test/src/test_users.py b/test/src/test_users.py
@@ -36,17 +36,20 @@
 
 ADMIN_DASHBOARD_PORT = 1
 COREAPI_PORT = 2
+MANAGE_API_PORT = 3
 
 MAS_ADMIN_URL = f"https://admin-dashboard.{MAS_CORE_NAMESPACE}.svc.cluster.local:{ADMIN_DASHBOARD_PORT}"
 MAS_API_URL = f'https://coreapi.{MAS_CORE_NAMESPACE}.svc.cluster.local:{COREAPI_PORT}'
+MANAGE_API_URL = f'https://{MAS_INSTANCE_ID}-{MAS_WORKSPACE_ID}.{MANAGE_NAMESPACE}.svc.cluster.local:{MANAGE_API_PORT}'
 
 PEM_PATH = "pempath"
 
 
-def additional_matcher(req, json=None, verify=PEM_PATH):
+def additional_matcher(req, json=None, verify=PEM_PATH, cert=None):
     if json is not None:
         assert req.json() == json
     assert req.verify == verify
+    assert req.cert == cert
     return True
 
 
@@ -120,7 +123,8 @@ def user_utils(mock_v1_secrets, mock_logininitial_endpoint, mock_named_temporary
         MAS_WORKSPACE_ID,
         k8s_client,
         coreapi_port=COREAPI_PORT,
-        admin_dashboard_port=ADMIN_DASHBOARD_PORT
+        admin_dashboard_port=ADMIN_DASHBOARD_PORT,
+        manage_api_port=MANAGE_API_PORT
     )
 
     yield user_utils
@@ -905,14 +909,141 @@ def test_check_user_sync_appstate_persistent_error(user_utils, requests_mock):
     assert patche.call_count == get.call_count / 2
 
 
-def test_create_or_get_manage_api_key_for_user(user_utils, requests_mock):
-    pass
-    # TODO
+def test_get_manage_api_key_for_user_exists(user_utils, requests_mock):
+    user_id = "user1"
+    apikey = {"userid": user_id, "href": f"https://{MANAGE_API_URL}/maximo/api/os/mxapikey/theapikeyid"}
 
+    get = requests_mock.get(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1&oslc.select=*&oslc.where=userid=\"{user_id}\"",
+        request_headers={"accept": "application/json"},
+        json={"member": [apikey]},
+        status_code=200,
+        additional_matcher=lambda req: additional_matcher(req, cert=PEM_PATH)
+    )
 
-def test_get_manage_api_key_for_user(user_utils, requests_mock):
-    pass
-    # TODO
+    assert user_utils.get_manage_api_key_for_user(user_id) == apikey
+    assert get.call_count == 1
+
+
+def test_get_manage_api_key_for_user_notfound(user_utils, requests_mock):
+    user_id = "user1"
+
+    get = requests_mock.get(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1&oslc.select=*&oslc.where=userid=\"{user_id}\"",
+        request_headers={"accept": "application/json"},
+        json={"member": []},
+        status_code=200,
+        additional_matcher=lambda req: additional_matcher(req, cert=PEM_PATH)
+    )
+
+    assert user_utils.get_manage_api_key_for_user(user_id) is None
+    assert get.call_count == 1
+
+
+def test_get_manage_api_key_for_user_error(user_utils, requests_mock):
+    user_id = "user1"
+
+    get = requests_mock.get(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1&oslc.select=*&oslc.where=userid=\"{user_id}\"",
+        request_headers={"accept": "application/json"},
+        text="boom",
+        status_code=500,
+        additional_matcher=lambda req: additional_matcher(req, cert=PEM_PATH)
+    )
+
+    with pytest.raises(Exception) as excinfo:
+        user_utils.get_manage_api_key_for_user(user_id)
+    assert str(excinfo.value) == "500 boom"
+    assert get.call_count == 1
+
+
+@pytest.mark.parametrize("temporary", [(True), (False)])
+def test_create_or_get_manage_api_key_for_user_new_api_key(temporary, user_utils, requests_mock, mock_atexit):
+    user_id = "user1"
+    apikey = {"userid": user_id, "href": f"https://{MANAGE_API_URL}/maximo/api/os/mxapikey/theapikeyid"}
+
+    post = requests_mock.post(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1",
+        request_headers={"content-type": "application/json"},
+        json={"id": user_id},
+        status_code=201,
+        additional_matcher=lambda req: additional_matcher(req, json={"expiration": -1, "userid": user_id}, cert=PEM_PATH)
+    )
+
+    get = requests_mock.get(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1&oslc.select=*&oslc.where=userid=\"{user_id}\"",
+        request_headers={"accept": "application/json"},
+        json={"member": [apikey]},
+        status_code=200,
+        additional_matcher=lambda req: additional_matcher(req, cert=PEM_PATH)
+    )
+
+    assert user_utils.create_or_get_manage_api_key_for_user(user_id, temporary=temporary) == apikey
+    assert post.call_count == 1
+    assert get.call_count == 1
+
+    # if temporary, check we registered the exit hook to delete the temporary Manage API Key
+    if temporary:
+        assert call(user_utils.delete_manage_api_key, apikey) in mock_atexit.mock_calls, "delete_manage_api_key exit hook not registered for temporary api key that we created"
+    else:
+        assert call(user_utils.delete_manage_api_key, apikey) not in mock_atexit.mock_calls, "delete_manage_api_key exit hook registered unexpectedly for non-temporary api key that we created"
+
+
+@pytest.mark.parametrize("temporary", [(True), (False)])
+def test_create_or_get_manage_api_key_for_user_existing_api_key(temporary, user_utils, requests_mock, mock_atexit):
+    user_id = "user1"
+    apikey = {"userid": user_id, "href": f"https://{MANAGE_API_URL}/maximo/api/os/mxapikey/theapikeyid"}
+
+    post = requests_mock.post(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1",
+        request_headers={"content-type": "application/json"},
+        json={"Error": {"reasonCode": "BMXAA10051E"}},
+        status_code=400,
+        additional_matcher=lambda req: additional_matcher(req, json={"expiration": -1, "userid": user_id}, cert=PEM_PATH)
+    )
+
+    get = requests_mock.get(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1&oslc.select=*&oslc.where=userid=\"{user_id}\"",
+        request_headers={"accept": "application/json"},
+        json={"member": [apikey]},
+        status_code=200,
+        additional_matcher=lambda req: additional_matcher(req, cert=PEM_PATH)
+    )
+
+    assert user_utils.create_or_get_manage_api_key_for_user(user_id, temporary=temporary) == apikey
+    assert post.call_count == 1
+    assert get.call_count == 1
+
+    # even if temporary is set, because we did not create the api key, we should not registered a hook to delete it
+    assert call(user_utils.delete_manage_api_key, apikey) not in mock_atexit.mock_calls, "delete_manage_api_key exit hook registered unexpectedly for existing API Key that we did not create"
+
+
+def test_create_or_get_manage_api_key_for_user_error(user_utils, requests_mock, mock_atexit):
+    user_id = "user1"
+    apikey = {"userid": user_id, "href": f"https://{MANAGE_API_URL}/maximo/api/os/mxapikey/theapikeyid"}
+
+    post = requests_mock.post(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1",
+        request_headers={"content-type": "application/json"},
+        text="boom",
+        status_code=400,
+        additional_matcher=lambda req: additional_matcher(req, json={"expiration": -1, "userid": user_id}, cert=PEM_PATH)
+    )
+
+    get = requests_mock.get(
+        f"{MANAGE_API_URL}/maximo/api/os/mxapiapikey?ccm=1&lean=1&oslc.select=*&oslc.where=userid=\"{user_id}\"",
+        request_headers={"accept": "application/json"},
+        json={"member": [apikey]},
+        status_code=200,
+        additional_matcher=lambda req: additional_matcher(req, cert=PEM_PATH)
+    )
+
+    with pytest.raises(Exception) as excinfo:
+        user_utils.create_or_get_manage_api_key_for_user(user_id, temporary=True)
+    assert str(excinfo.value) == "400 boom"
+    assert post.call_count == 1
+    assert get.call_count == 0
+    assert call(user_utils.delete_manage_api_key, apikey) not in mock_atexit.mock_calls, "delete_manage_api_key exit hook not registered even though we failed to create the api key"
 
 
 def test_delete_manage_api_key(user_utils, requests_mock):
