drop Utils

e2e/tests/test_smoke.py

@@ -305,82 +305,5 @@ class K8s:
     def create_with_kubectl(self, path):
         subprocess.run(["kubectl", "create", "-f", path])
 
-
-class Utils:
-
-    RETRY_TIMEOUT_SEC = 5
-
-    @staticmethod
-    def get_spilo_nodes(k8s, pod_labels):
-        master_pod_node = ''
-        replica_pod_nodes = []
-        podsList = k8s.core_v1.list_namespaced_pod('default', label_selector=pod_labels)
-        for pod in podsList.items:
-            if ('spilo-role', 'master') in pod.metadata.labels.items():
-                master_pod_node = pod.spec.node_name
-            elif ('spilo-role', 'replica') in pod.metadata.labels.items():
-                replica_pod_nodes.append(pod.spec.node_name)
-
-        return master_pod_node, replica_pod_nodes
-
-    @staticmethod
-    def wait_for_pod_start(k8s, pod_labels):
-        pod_phase = 'No pod running'
-        while pod_phase != 'Running':
-            pods = k8s.core_v1.list_namespaced_pod('default', label_selector=pod_labels).items
-            if pods:
-                pod_phase = pods[0].status.phase
-            time.sleep(Utils.RETRY_TIMEOUT_SEC)
-
-    @staticmethod
-    def wait_for_pg_to_scale(k8s, number_of_instances):
-
-        body = {
-            "spec": {
-                "numberOfInstances": number_of_instances
-            }
-        }
-        _ = k8s.crd.patch_namespaced_custom_object("acid.zalan.do",
-                                                       "v1", "default", "postgresqls", "acid-minimal-cluster", body)
-
-        labels = 'version=acid-minimal-cluster'
-        while Utils.count_pods_with_label(k8s, labels) != number_of_instances:
-            time.sleep(Utils.RETRY_TIMEOUT_SEC)
-
-    @staticmethod
-    def count_pods_with_label(k8s, labels):
-        return len(k8s.core_v1.list_namespaced_pod('default', label_selector=labels).items)
-
-    @staticmethod
-    def wait_for_master_failover(k8s, expected_master_nodes):
-        pod_phase = 'Failing over'
-        new_master_node = ''
-        labels = 'spilo-role=master,version=acid-minimal-cluster'
-
-        while (pod_phase != 'Running') or (new_master_node not in expected_master_nodes):
-            pods = k8s.core_v1.list_namespaced_pod('default', label_selector=labels).items
-
-            if pods:
-                new_master_node = pods[0].spec.node_name
-                pod_phase = pods[0].status.phase
-            time.sleep(Utils.RETRY_TIMEOUT_SEC)
-
-    @staticmethod
-    def get_logical_backup_job(k8s):
-        return k8s.batch_v1_beta1.list_namespaced_cron_job("default", label_selector="application=spilo")
-
-    @staticmethod
-    def wait_for_logical_backup_job(k8s, expected_num_of_jobs):
-        while (len(Utils.get_logical_backup_job(k8s).items) != expected_num_of_jobs):
-            time.sleep(Utils.RETRY_TIMEOUT_SEC)
-
-    @staticmethod
-    def wait_for_logical_backup_job_deletion(k8s):
-        Utils.wait_for_logical_backup_job(k8s, expected_num_of_jobs = 0)
-
-    @staticmethod
-    def wait_for_logical_backup_job_creation(k8s):
-        Utils.wait_for_logical_backup_job(k8s, expected_num_of_jobs = 1)
-
 if __name__ == '__main__':
     unittest.main()