Allow direct communication to Openshift Users API

Author: QuanMPhm

src/coldfront_plugin_cloud/openshift.py

@@ -232,12 +232,25 @@ def get_federated_user(self, username):
         logger.info(f"User ({username}) does not exist")
 
     def create_federated_user(self, unique_id):
-        url = f"{self.auth_url}/users/{unique_id}"
-        try:
-            r = self.session.put(url)
-            self.check_response(r)
-        except Conflict:
-            pass
+        user_def = {
+            "metadata": {"name": unique_id},
+            "fullName": unique_id,
+        }
+
+        identity_def = {
+            "providerName": self.id_provider,
+            "providerUserName": unique_id,
+        }
+
+        identity_mapping_def = {
+            "user": {"name": unique_id},
+            "identity": {"name": self.qualified_id_user(unique_id)},
+        }
+
+        self._openshift_create_user(user_def)
+        self._openshift_create_identity(identity_def)
+        self._openshift_create_useridentitymapping(identity_mapping_def)
+        logger.info(f"User {unique_id} successfully created")
 
     def assign_role_on_user(self, username, project_id):
         # /users/<user_name>/projects/<project>/roles/<role>
@@ -291,9 +304,9 @@ def _get_project(self, project_id):
         return self.check_response(r)
 
     def _delete_user(self, username):
-        url = f"{self.auth_url}/users/{username}"
-        r = self.session.delete(url)
-        return self.check_response(r)
+        self._openshift_delete_user(username)
+        self._openshift_delete_identity(username)
+        logger.info(f"User {username} successfully deleted")
 
     def get_users(self, project_id):
         url = f"{self.auth_url}/projects/{project_id}/users"
@@ -304,12 +317,43 @@ def _openshift_get_user(self, username):
         api = self.get_resource_api(API_USER, "User")
         return clean_openshift_metadata(api.get(name=username).to_dict())
 
+    def _openshift_create_user(self, user_def):
+        api = self.get_resource_api(API_USER, "User")
+        try:
+            return clean_openshift_metadata(api.create(body=user_def).to_dict())
+        except kexc.ConflictError:
+            pass
+
+    def _openshift_delete_user(self, username):
+        api = self.get_resource_api(API_USER, "User")
+        return clean_openshift_metadata(api.delete(name=username).to_dict())
+
     def _openshift_get_identity(self, id_user):
         api = self.get_resource_api(API_USER, "Identity")
         return clean_openshift_metadata(
             api.get(name=self.qualified_id_user(id_user)).to_dict()
         )
 
+    def _openshift_create_identity(self, identity_def):
+        api = self.get_resource_api(API_USER, "Identity")
+        try:
+            return clean_openshift_metadata(api.create(body=identity_def).to_dict())
+        except kexc.ConflictError:
+            pass
+
+    def _openshift_delete_identity(self, username):
+        api = self.get_resource_api(API_USER, "Identity")
+        return api.delete(name=self.qualified_id_user(username)).to_dict()
+
+    def _openshift_create_useridentitymapping(self, identity_mapping_def):
+        api = self.get_resource_api(API_USER, "UserIdentityMapping")
+        try:
+            return clean_openshift_metadata(
+                api.create(body=identity_mapping_def).to_dict()
+            )
+        except kexc.ConflictError:
+            pass
+
     def _openshift_user_exists(self, user_name):
         try:
             self._openshift_get_user(user_name)

src/coldfront_plugin_cloud/tests/functional/openshift/test_allocation.py

@@ -286,3 +286,33 @@ def test_project_default_labels(self):
         self.assertTrue(
             namespace_dict_labels.items() > openshift.PROJECT_DEFAULT_LABELS.items()
         )
+
+    def test_create_incomplete(self):
+        """Creating a user that only has user, but no identity or mapping should not raise an error."""
+        user = self.new_user()
+        project = self.new_project(pi=user)
+        allocation = self.new_allocation(project, self.resource, 1)
+        allocator = openshift.OpenShiftResourceAllocator(self.resource, allocation)
+        user_def = {
+            "metadata": {"name": user.username},
+            "fullName": user.username,
+        }
+
+        allocator._openshift_create_user(user_def)
+        self.assertTrue(allocator._openshift_user_exists(user.username))
+        self.assertFalse(allocator._openshift_identity_exists(user.username))
+        self.assertFalse(
+            allocator._openshift_useridentitymapping_exists(
+                user.username, user.username
+            )
+        )
+
+        # Now create identity and mapping, no errors should be raised
+        allocator.get_or_create_federated_user(user.username)
+        self.assertTrue(allocator._openshift_user_exists(user.username))
+        self.assertTrue(allocator._openshift_identity_exists(user.username))
+        self.assertTrue(
+            allocator._openshift_useridentitymapping_exists(
+                user.username, user.username
+            )
+        )

src/coldfront_plugin_cloud/tests/unit/openshift/test_user.py

@@ -37,3 +37,48 @@ def test_get_federated_user_not_exist(self):
 
         output = self.allocator.get_federated_user("fake_user_2")
         self.assertEqual(output, None)
+
+    def test_create_federated_user(self):
+        fake_client_output = mock.Mock(spec=["to_dict"])
+        fake_client_output.to_dict.return_value = {}
+        self.allocator.k8_client.resources.get.return_value.create.return_value = (
+            fake_client_output
+        )
+
+        self.allocator.create_federated_user("fake_user_name")
+
+        # Assert called to create user
+        self.allocator.k8_client.resources.get.return_value.create.assert_any_call(
+            body={"metadata": {"name": "fake_user_name"}, "fullName": "fake_user_name"}
+        )
+
+        # Assert called to add identity
+        self.allocator.k8_client.resources.get.return_value.create.assert_any_call(
+            body={
+                "providerName": "fake_idp",
+                "providerUserName": "fake_user_name",
+            }
+        )
+
+        # Assert called to add identity mapping
+        self.allocator.k8_client.resources.get.return_value.create.assert_any_call(
+            body={
+                "user": {"name": "fake_user_name"},
+                "identity": {"name": "fake_idp:fake_user_name"},
+            }
+        )
+
+    def test_delete_user(self):
+        fake_client_output = mock.Mock(spec=["to_dict"])
+        fake_client_output.to_dict.return_value = {}
+        self.allocator.k8_client.resources.get.return_value.delete.return_value = (
+            fake_client_output
+        )
+
+        self.allocator._delete_user("fake_user_name")
+        self.allocator.k8_client.resources.get.return_value.delete.assert_any_call(
+            name="fake_user_name"
+        )
+        self.allocator.k8_client.resources.get.return_value.delete.assert_any_call(
+            name="fake_idp:fake_user_name"
+        )