Allow direct communication to Openshift Users API

Author: QuanMPhm

src/coldfront_plugin_cloud/openshift.py

@@ -228,12 +228,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>
@@ -287,9 +300,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"
@@ -300,12 +313,38 @@ 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")
+        return clean_openshift_metadata(api.create(body=user_def).to_dict())
+
+    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")
+        return clean_openshift_metadata(api.create(body=identity_def).to_dict())
+
+    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")
+        return clean_openshift_metadata(api.create(body=identity_mapping_def).to_dict())
+
+    def _openshift_create_rolebinding(self, project_name, role_binding_def):
+        api = self.get_resource_api(API_RBAC, "RoleBinding")
+        return clean_openshift_metadata(
+            api.create(body=role_binding_def, namespace=project_name).to_dict()
+        )
+
     def _openshift_user_exists(self, user_name):
         try:
             self._openshift_get_user(user_name)

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"
+        )