Merge pull request #239 from ilausuch/common_kubernetes_mockups

Extract kubernetes mockups to a common py file

tests/kubernetes.py

@@ -0,0 +1,51 @@
+from datetime import datetime, timezone, timedelta
+
+
+class MockedKubernetesConfig():
+    def load_kube_config(self, *args, **kwargs):
+        return True
+
+
+class MockedKubernetesClient():
+    def __init__(self, jobs=[]):
+        self.jobs = jobs
+        self.deleted_jobs = []
+
+    # pylint: disable=C0103
+    def BatchV1Api(self):
+        return self
+
+    def list_job_for_all_namespaces(self, *args, **kwargs):
+        return MockedKubernetesResult(self.jobs)
+
+    def delete_namespaced_job(self, name, namespace):
+        self.deleted_jobs.append(name)
+
+
+class MockedKubernetesResult():
+    def __init__(self, items):
+        self.items = items
+
+
+class MockedKubernetesJobStatus():
+    def __init__(self, age):
+        self.start_time = datetime.now(timezone.utc) - timedelta(days=age)
+
+
+class MockedKubernetesJobMetadata():
+    def __init__(self, name):
+        self.name = name
+        self.namespace = "default"
+
+
+class MockedKubernetesJob():
+    def __init__(self, name, age):
+        self.status = MockedKubernetesJobStatus(age)
+        self.metadata = MockedKubernetesJobMetadata(name)
+
+
+class MockedSubprocessReturn():
+    def __init__(self, returncode=0, stdout="", stderr=""):
+        self.returncode = returncode
+        self.stderr = stderr
+        self.stdout = stdout

tests/test_eks.py

@@ -1,11 +1,11 @@
 import os
-from datetime import datetime, timezone, timedelta
 import pytest
 import kubernetes
 from ocw.lib.provider import Provider
 from ocw.lib.eks import EKS
 from webui.PCWConfig import PCWConfig
 from tests.generators import mock_get_feature_property
+from tests.kubernetes import MockedSubprocessReturn, MockedKubernetesClient, MockedKubernetesConfig, MockedKubernetesJob
 
 
 def test_list_clusters(eks_patch, monkeypatch):
@@ -50,56 +50,6 @@ def describe_cluster(self, name=None):
         return None
 
 
-class MockedKubernetesConfig():
-    def load_kube_config(self, *args, **kwargs):
-        return True
-
-
-class MockedKubernetesClient():
-    def __init__(self, jobs=[]):
-        self.jobs = jobs
-        self.deleted_jobs = []
-
-    # pylint: disable=C0103
-    def BatchV1Api(self):
-        return self
-
-    def list_job_for_all_namespaces(self, *args, **kwargs):
-        return MockedKubernetesResult(self.jobs)
-
-    def delete_namespaced_job(self, name, namespace):
-        self.deleted_jobs.append(name)
-
-
-class MockedKubernetesResult():
-    def __init__(self, items):
-        self.items = items
-
-
-class MockedKubernetesJobStatus():
-    def __init__(self, age):
-        self.start_time = datetime.now(timezone.utc) - timedelta(days=age)
-
-
-class MockedKubernetesJobMetadata():
-    def __init__(self, name):
-        self.name = name
-        self.namespace = "default"
-
-
-class MockedKubernetesJob():
-    def __init__(self, name, age):
-        self.status = MockedKubernetesJobStatus(age)
-        self.metadata = MockedKubernetesJobMetadata(name)
-
-
-class MockedSubprocessReturn():
-    def __init__(self, returncode=0, stdout="", stderr=""):
-        self.returncode = returncode
-        self.stderr = stderr
-        self.stdout = stdout
-
-
 @pytest.fixture
 def eks_patch(monkeypatch):
     def mocked_cmd_exec(self, cmd):

tests/test_gke.py

@@ -1,52 +1,9 @@
-from datetime import datetime, timezone, timedelta
 from ocw.lib.gke import GKE
 from ocw.lib.gce import GCE
 from ocw.lib.provider import Provider
 import pytest
 import kubernetes
-
-
-class MockedKubernetesConfig():
-    def load_kube_config(self, *args, **kwargs):
-        return True
-
-
-class MockedKubernetesClient():
-    def __init__(self, jobs=[]):
-        self.jobs = jobs
-        self.deleted_jobs = []
-
-    # pylint: disable=C0103
-    def BatchV1Api(self):
-        return self
-
-    def list_job_for_all_namespaces(self, *args, **kwargs):
-        return MockedKubernetesResult(self.jobs)
-
-    def delete_namespaced_job(self, name, namespace):
-        self.deleted_jobs.append(name)
-
-
-class MockedKubernetesResult():
-    def __init__(self, items):
-        self.items = items
-
-
-class MockedKubernetesJobStatus():
-    def __init__(self, age):
-        self.start_time = datetime.now(timezone.utc) - timedelta(days=age)
-
-
-class MockedKubernetesJobMetadata():
-    def __init__(self, name):
-        self.name = name
-        self.namespace = "default"
-
-
-class MockedKubernetesJob():
-    def __init__(self, name, age):
-        self.status = MockedKubernetesJobStatus(age)
-        self.metadata = MockedKubernetesJobMetadata(name)
+from tests.kubernetes import MockedKubernetesClient, MockedKubernetesConfig, MockedKubernetesJob
 
 
 @pytest.fixture