unit tests

Author: tomklapiscak

src/mas/devops/users.py

@@ -545,7 +545,7 @@ def create_or_get_manage_api_key_for_user(self, user_id):
             self.logger.info(f"Creating new Manage API Key for user {user_id}")
         else:
             # any other status code is unexpected
-            raise Exception(response.text)
+            raise Exception(f"{response.status_code} {response.text}")
 
         # otherwise, retrieve the apikey (either it already existed, or we just created it)
 
@@ -562,8 +562,6 @@ def create_or_get_manage_api_key_for_user(self, user_id):
             #       NOTE: the manage sanity test seems to create maxadmin apikey and not clean it up
             atexit.register(self.delete_manage_api_key, apikey)
 
-            pass
-
         return apikey
 
     def get_manage_api_key_for_user(self, user_id):
@@ -763,17 +761,16 @@ def get_mas_application_availability(self, mas_application_id):
             return response.json()
         raise Exception(f"{response.status_code} {response.text}")
 
-    def await_mas_application_availability(self, mas_application_id, timeout_secs=60 * 10):
+    def await_mas_application_availability(self, mas_application_id, timeout_secs=60 * 10, retry_interval_secs=5):
         t_end = time.time() + timeout_secs
         self.logger.info(f"Waiting for {mas_application_id} to become ready and available: {t_end - time.time():.2f} seconds remaining")
         while time.time() < t_end:
             app = self.get_mas_application_availability(mas_application_id)
             if "available" in app and "ready" in app and app["ready"] and app["available"]:
                 return
-            # TODO: error state?
             else:
-                self.logger.info(f"{mas_application_id} is not ready or available, retry in 5 seconds: {t_end - time.time():.2f} seconds remaining")
-                time.sleep(5)
+                self.logger.info(f"{mas_application_id} is not ready or available, retry in {retry_interval_secs} seconds: {t_end - time.time():.2f} seconds remaining")
+                time.sleep(retry_interval_secs)
         raise Exception(f"{mas_application_id} did not become ready and available in time, aborting")
 
     def parse_initial_users_from_aws_secret_json(self, secret_json):

test/src/test_users.py

@@ -724,3 +724,145 @@ def test_check_user_sync_appstate_persistent_error(user_utils, requests_mock):
 
     # an "update_user_display_name" should have been triggered for every 2 get calls (1 call by check_user_sync, 1 by resync)
     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(user_utils, requests_mock):
+    pass
+    # TODO
+
+
+def test_delete_manage_api_key(user_utils, requests_mock):
+    pass
+    # TODO
+
+
+def test_get_manage_group_id(user_utils, requests_mock):
+    pass
+    # TODO
+
+
+def test_is_user_in_manage_group(user_utils, requests_mock):
+    pass
+    # TODO
+
+
+def test_add_user_to_manage_group(user_utils, requests_mock):
+    pass
+    # TODO
+
+
+def test_get_mas_applications_in_workspace(user_utils, requests_mock):
+    get = requests_mock.get(
+        f"{MAS_API_URL}/workspaces/{MAS_WORKSPACE_ID}/applications",
+        request_headers={"x-access-token": TOKEN},
+        json=[{"id": "manage"}],
+        status_code=200
+    )
+    assert user_utils.get_mas_applications_in_workspace() == [{"id": "manage"}]
+    assert get.call_count == 1
+
+
+def test_get_mas_applications_in_workspace_error(user_utils, requests_mock):
+    get = requests_mock.get(
+        f"{MAS_API_URL}/workspaces/{MAS_WORKSPACE_ID}/applications",
+        request_headers={"x-access-token": TOKEN},
+        json={"error": "internal"},
+        status_code=500
+    )
+    with pytest.raises(Exception) as excinfo:
+        user_utils.get_mas_applications_in_workspace()
+    assert get.call_count == 1
+    assert str(excinfo.value) == '500 {"error": "internal"}'
+
+
+def test_get_mas_application_availability(user_utils, requests_mock):
+    application_id = "manage"
+    get = requests_mock.get(
+        f"{MAS_API_URL}/workspaces/{MAS_WORKSPACE_ID}/applications/{application_id}",
+        request_headers={"x-access-token": TOKEN},
+        json={"id": "manage"},
+        status_code=200
+    )
+    assert user_utils.get_mas_application_availability(application_id) == {"id": "manage"}
+    assert get.call_count == 1
+
+
+def test_get_mas_application_availability_error(user_utils, requests_mock):
+    application_id = "manage"
+    get = requests_mock.get(
+        f"{MAS_API_URL}/workspaces/{MAS_WORKSPACE_ID}/applications/{application_id}",
+        request_headers={"x-access-token": TOKEN},
+        json={"error": "internal"},
+        status_code=500
+    )
+    with pytest.raises(Exception) as excinfo:
+        user_utils.get_mas_application_availability(application_id)
+    assert get.call_count == 1
+    assert str(excinfo.value) == '500 {"error": "internal"}'
+
+
+def test_await_mas_application_availability(user_utils, requests_mock):
+    pass
+    # TODO
+
+
+def test_parse_initial_users_from_aws_secret_json(user_utils):
+
+    actual_initial_users = user_utils.parse_initial_users_from_aws_secret_json(
+        {
+            "user1@example.com": "primary,joe,bloggs",
+            "user2@example.com": "  primary     ,   ben     ,   bob  ",
+            "user3@example.com": "secondary     ,bill,  bibb"
+        }
+    )
+
+    expected_initial_users = {
+        "users": {
+            "primary": [
+                {
+                    "email": "user1@example.com",
+                    "given_name": "joe",
+                    "family_name": "bloggs"
+                },
+                {
+                    "email": "user2@example.com",
+                    "given_name": "ben",
+                    "family_name": "bob"
+                }
+            ],
+            "secondary": [
+                {
+                    "email": "user3@example.com",
+                    "given_name": "bill",
+                    "family_name": "bibb"
+                }
+            ]
+        }
+    }
+
+    assert actual_initial_users == expected_initial_users
+
+    with pytest.raises(Exception) as excinfo:
+        user_utils.parse_initial_users_from_aws_secret_json({
+            "user1@example.com": "primary"
+        })
+    assert "Wrong number of CSV values for user1@example.com (expected 3 but got 1)" == str(excinfo.value)
+
+    with pytest.raises(Exception) as excinfo:
+        user_utils.parse_initial_users_from_aws_secret_json({
+            "user1@example.com": "unknown,x,y"
+        })
+    assert "Unknown user type for user1@example.com: unknown" == str(excinfo.value)
+
+
+def test_create_initial_user_for_saas(user_utils, requests_mock):
+    pass
+
+
+def test_create_initial_users_for_saas(user_utils, requests_mock):
+    pass