diff --git a/tests/kubernetes.py b/tests/kubernetes.py new file mode 100644 index 00000000..d91c07a2 --- /dev/null +++ b/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 diff --git a/tests/test_eks.py b/tests/test_eks.py index 6afd1ecf..6c48e3d6 100644 --- a/tests/test_eks.py +++ b/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): diff --git a/tests/test_gke.py b/tests/test_gke.py index e28012d0..c2652690 100644 --- a/tests/test_gke.py +++ b/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