add tests

Makefile

@@ -15,7 +15,7 @@ lint:
 	python -m flake8
 
 test:
-	python -m pytest -x
+	python -m pytest
 
 build: clean
 	python -m build; twine check --strict dist/*

kfk/commands/clusters.py

@@ -15,12 +15,20 @@
 )
 from kfk.config import STRIMZI_PATH, STRIMZI_VERSION
 from kfk.kubectl_command_builder import Kubectl
+from kfk.kubernetes_commons import create_using_yaml, delete_using_yaml
 from kfk.messages import Messages
 from kfk.option_extensions import NotRequiredIf
 
 
 @click.option("-y", "--yes", "is_yes", help='"Yes" confirmation', is_flag=True)
-@click.option("-n", "--namespace", help="Namespace to use")
+@click.option(
+    "-n",
+    "--namespace",
+    help="Namespace to use",
+    required=True,
+    cls=NotRequiredIf,
+    options=["is_list"],
+)
 @click.option(
     "--delete-config",
     help="A cluster configuration override to be removed for an existing cluster",
@@ -121,7 +129,6 @@ def create(cluster, replicas, zk_replicas, config, namespace, is_yes):
         _add_config_if_provided(config, cluster_dict)
 
         cluster_yaml = yaml.dump(cluster_dict)
-
         cluster_temp_file = create_temp_file(cluster_yaml)
 
         if is_yes:
@@ -130,14 +137,8 @@ def create(cluster, replicas, zk_replicas, config, namespace, is_yes):
             open_file_in_system_editor(cluster_temp_file.name)
             is_confirmed = click.confirm(Messages.CLUSTER_CREATE_CONFIRMATION)
         if is_confirmed:
-            os.system(
-                Kubectl()
-                .create()
-                .from_file("{cluster_temp_file_path}")
-                .namespace(namespace)
-                .build()
-                .format(cluster_temp_file_path=cluster_temp_file.name)
-            )
+            create_using_yaml(cluster_temp_file.name, namespace)
+
         cluster_temp_file.close()
 
 
@@ -156,7 +157,21 @@ def delete(cluster, namespace, is_yes):
     else:
         is_confirmed = click.confirm(Messages.CLUSTER_DELETE_CONFIRMATION)
     if is_confirmed:
-        os.system(Kubectl().delete().kafkas(cluster).namespace(namespace).build())
+        with open(
+            "{strimzi_path}/examples/kafka/kafka-ephemeral.yaml".format(
+                strimzi_path=STRIMZI_PATH
+            ).format(version=STRIMZI_VERSION)
+        ) as file:
+            stream = file.read()
+
+            cluster_dict = yaml.full_load(stream)
+
+            cluster_dict["metadata"]["name"] = cluster
+
+            cluster_yaml = yaml.dump(cluster_dict)
+            cluster_temp_file = create_temp_file(cluster_yaml)
+
+            delete_using_yaml(cluster_temp_file.name, namespace)
 
 
 def alter(cluster, replicas, zk_replicas, config, delete_config, namespace):

kfk/commands/connect/clusters.py

@@ -26,6 +26,7 @@
     SpecialTexts,
 )
 from kfk.kubectl_command_builder import Kubectl
+from kfk.kubernetes_commons import create_registry_secret, create_using_yaml
 from kfk.messages import Errors, Messages
 from kfk.utils import is_valid_url
 
@@ -223,8 +224,6 @@ def create(
             is_confirmed = click.confirm(Messages.CLUSTER_CREATE_CONFIRMATION)
 
         if is_confirmed:
-            return_code = 0
-
             if connect_properties.get(SpecialTexts.CONNECT_PLUGIN_URL) is not None:
                 username = (
                     registry_username
@@ -239,37 +238,17 @@ def create(
                     else click.prompt(Messages.IMAGE_REGISTRY_PASSWORD, hide_input=True)
                 )
 
-                return_code = os.system(
-                    Kubectl()
-                    .create()
-                    .secret(
-                        "docker-registry",
-                        f"{cluster}-push-secret",
-                        "--docker-username={username}",
-                        "--docker-password={password}",
-                        "--docker-server={server}",
-                    )
-                    .namespace(namespace)
-                    .build()
-                    .format(
-                        username=username,
-                        password=password,
-                        server=connect_properties.get(SpecialTexts.CONNECT_IMAGE).data,
-                    )
+                create_registry_secret(
+                    f"{cluster}-push-secret",
+                    connect_properties.get(SpecialTexts.CONNECT_IMAGE).data,
+                    username,
+                    password,
                 )
 
-            if return_code == 0:
-                return_code = os.system(
-                    Kubectl()
-                    .create()
-                    .from_file("{cluster_temp_file_path}")
-                    .namespace(namespace)
-                    .build()
-                    .format(cluster_temp_file_path=cluster_temp_file.name)
-                )
-                if return_code == 0:
-                    for connector_config_file in connector_config_files:
-                        connectors.create(connector_config_file, cluster, namespace)
+            create_using_yaml(cluster_temp_file.name, namespace)
+
+            for connector_config_file in connector_config_files:
+                connectors.create(connector_config_file, cluster, namespace)
 
         cluster_temp_file.close()
 

kfk/commands/connect/connectors.py

@@ -16,6 +16,7 @@
 from kfk.config import STRIMZI_PATH, STRIMZI_VERSION
 from kfk.constants import SpecialTexts
 from kfk.kubectl_command_builder import Kubectl
+from kfk.kubernetes_commons import create_using_yaml, delete_using_yaml
 
 CONNECTOR_SKIPPED_PROPERTIES = (
     SpecialTexts.CONNECTOR_NAME,
@@ -78,7 +79,7 @@ def connectors(
     elif is_describe:
         describe(connector, output, namespace)
     elif is_delete:
-        delete(connector, namespace)
+        delete(connector, cluster, namespace)
     elif is_alter:
         alter(config_file, cluster, namespace)
     else:
@@ -129,14 +130,7 @@ def create(config_file, cluster, namespace):
         connector_yaml = yaml.dump(connector_dict)
         connector_temp_file = create_temp_file(connector_yaml)
 
-        os.system(
-            Kubectl()
-            .create()
-            .from_file("{connector_temp_file_path}")
-            .namespace(namespace)
-            .build()
-            .format(connector_temp_file_path=connector_temp_file.name)
-        )
+        create_using_yaml(connector_temp_file.name, namespace)
 
         connector_temp_file.close()
 
@@ -157,10 +151,21 @@ def describe(connector, output, namespace):
         )
 
 
-def delete(connector, namespace):
-    os.system(
-        Kubectl().delete().kafkaconnectors(connector).namespace(namespace).build()
-    )
+def delete(connector, cluster, namespace):
+    with open(
+        "{strimzi_path}/examples/connect/source-connector.yaml".format(
+            strimzi_path=STRIMZI_PATH
+        ).format(version=STRIMZI_VERSION)
+    ) as file:
+        connector_dict = yaml.full_load(file)
+
+        connector_dict["metadata"]["name"] = connector
+        connector_dict["metadata"]["labels"]["strimzi.io/cluster"] = cluster
+
+        connector_yaml = yaml.dump(connector_dict)
+        connector_temp_file = create_temp_file(connector_yaml)
+
+        delete_using_yaml(connector_temp_file.name, namespace)
 
 
 def alter(config_file, cluster, namespace):

kfk/commands/users.py

@@ -14,6 +14,7 @@
 )
 from kfk.config import STRIMZI_PATH, STRIMZI_VERSION
 from kfk.kubectl_command_builder import Kubectl
+from kfk.kubernetes_commons import create_using_yaml, delete_using_yaml
 from kfk.option_extensions import NotRequiredIf, RequiredIf
 from kfk.utils import snake_to_camel_case
 
@@ -197,14 +198,7 @@ def create(user, authentication_type, quota_tuple, cluster, namespace):
         user_yaml = yaml.dump(user_dict)
         user_temp_file = create_temp_file(user_yaml)
 
-        os.system(
-            Kubectl()
-            .create()
-            .from_file("{user_temp_file_path}")
-            .namespace(namespace)
-            .build()
-            .format(user_temp_file_path=user_temp_file.name)
-        )
+        create_using_yaml(user_temp_file.name, namespace)
 
         user_temp_file.close()
 
@@ -219,7 +213,20 @@ def describe(user, output, cluster, namespace):
 
 
 def delete(cluster, namespace, user):
-    os.system(Kubectl().delete().kafkausers(user).namespace(namespace).build())
+    with open(
+        "{strimzi_path}/examples/user/kafka-user.yaml".format(
+            strimzi_path=STRIMZI_PATH
+        ).format(version=STRIMZI_VERSION)
+    ) as file:
+        user_dict = yaml.full_load(file)
+
+        user_dict["metadata"]["name"] = user
+        user_dict["metadata"]["labels"]["strimzi.io/cluster"] = cluster
+
+        user_yaml = yaml.dump(user_dict)
+        user_temp_file = create_temp_file(user_yaml)
+
+        delete_using_yaml(user_temp_file.name, namespace)
 
 
 def alter(

kfk/kubectl_command_builder.py

@@ -16,10 +16,6 @@ def get(self):
         self.cmd_str = self.cmd_str + SPACE + "get"
         return self
 
-    def create(self):
-        self.cmd_str = self.cmd_str + SPACE + "create"
-        return self
-
     def apply(self):
         self.cmd_str = self.cmd_str + SPACE + "apply"
         return self
@@ -32,10 +28,6 @@ def describe(self):
         self.cmd_str = self.cmd_str + SPACE + "describe"
         return self
 
-    def delete(self):
-        self.cmd_str = self.cmd_str + SPACE + "delete"
-        return self
-
     def edit(self):
         self.cmd_str = self.cmd_str + SPACE + "edit"
         return self

kfk/kubernetes_commons.py

@@ -1,3 +1,5 @@
+import base64
+import json
 import re
 import sys
 from os import path
@@ -6,7 +8,7 @@
 from kubernetes import client, config
 
 config.load_kube_config()
-k8s_client = client.ApiClient()
+api_client = client.ApiClient()
 
 
 def yaml_object_argument_filter(func):
@@ -20,9 +22,46 @@ def inner(k8s_api, yml_object, kind, **kwargs):
     return inner
 
 
+def create_registry_secret(
+    name: str,
+    registry: str,
+    username: str,
+    password: str,
+):
+    core_api = client.CoreV1Api(api_client)
+
+    auth = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("utf-8")
+
+    docker_config_dict = {
+        "auths": {
+            registry: {
+                "username": username,
+                "password": password,
+                "email": "",
+                "auth": auth,
+            }
+        }
+    }
+
+    docker_config = base64.b64encode(
+        json.dumps(docker_config_dict).encode("utf-8")
+    ).decode("utf-8")
+
+    core_api.create_namespaced_secret(
+        namespace="default",
+        body=client.V1Secret(
+            metadata=client.V1ObjectMeta(
+                name=name,
+            ),
+            type="kubernetes.io/dockerconfigjson",
+            data={".dockerconfigjson": docker_config},
+        ),
+    )
+
+
 def create_using_yaml(file_path, namespace):
     _operate_using_yaml(
-        k8s_client,
+        api_client,
         file_path,
         "create",
         yaml_objects=None,
@@ -33,7 +72,7 @@ def create_using_yaml(file_path, namespace):
 
 def delete_using_yaml(file_path, namespace):
     _operate_using_yaml(
-        k8s_client,
+        api_client,
         file_path,
         "delete",
         yaml_objects=None,