Skip to content

Commit

Permalink
Extract kubernetes mockups to a common py file
Browse files Browse the repository at this point in the history
Some common mockups are used in the tests of all providers.
This PR prevents the duplication of code
  • Loading branch information
ilausuch committed Apr 5, 2023
1 parent 442cb69 commit 9749935
Show file tree
Hide file tree
Showing 3 changed files with 53 additions and 95 deletions.
51 changes: 51 additions & 0 deletions tests/kubernetes.py
Original file line number Diff line number Diff line change
@@ -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
52 changes: 1 addition & 51 deletions tests/test_eks.py
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -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):
Expand Down
45 changes: 1 addition & 44 deletions tests/test_gke.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down

0 comments on commit 9749935

Please sign in to comment.