CORD-3048 Unit Tests for Kubernetes Synchronizer
Change-Id: I0ff9146d544a2e0a212264b7d366500d6a51ff1c
diff --git a/xos/synchronizer/steps/sync_secret.py b/xos/synchronizer/steps/sync_secret.py
index 903be4f..a020b43 100644
--- a/xos/synchronizer/steps/sync_secret.py
+++ b/xos/synchronizer/steps/sync_secret.py
@@ -26,9 +26,6 @@
from xosconfig import Config
from multistructlog import create_logger
-from kubernetes.client.rest import ApiException
-from kubernetes import client as kubernetes_client, config as kubernetes_config
-
log = create_logger(Config().get('logging'))
class SyncKubernetesSecret(SyncStep):
@@ -45,16 +42,23 @@
def __init__(self, *args, **kwargs):
super(SyncKubernetesSecret, self).__init__(*args, **kwargs)
+ self.init_kubernetes_client()
+
+ def init_kubernetes_client(self):
+ from kubernetes.client.rest import ApiException
+ from kubernetes import client as kubernetes_client, config as kubernetes_config
kubernetes_config.load_incluster_config()
- self.v1 = kubernetes_client.CoreV1Api()
+ self.kubernetes_client = kubernetes_client
+ self.v1core = kubernetes_client.CoreV1Api()
+ self.ApiException = ApiException
def get_secret(self, o):
""" Given an XOS KubernetesSecret object, read the corresponding Secret from Kubernetes.
return None if no Secret exists.
"""
try:
- secret = self.v1.read_namespaced_secret(o.name, o.trust_domain.name)
- except ApiException, e:
+ secret = self.v1core.read_namespaced_secret(o.name, o.trust_domain.name)
+ except self.ApiException, e:
if e.status == 404:
return None
raise
@@ -63,14 +67,14 @@
def sync_record(self, o):
secret = self.get_secret(o)
if not secret:
- secret = kubernetes_client.V1Secret()
+ secret = self.kubernetes_client.V1Secret()
secret.data = json.loads(o.data)
- secret.metadata = kubernetes_client.V1ObjectMeta(name=o.name)
+ secret.metadata = self.kubernetes_client.V1ObjectMeta(name=o.name)
- secret = self.v1.create_namespaced_secret(o.trust_domain.name, secret)
+ secret = self.v1core.create_namespaced_secret(o.trust_domain.name, secret)
else:
secret.data = json.loads(o.data)
- self.v1.patch_namespaced_secret(o.name, o.trust_domain.name, secret)
+ self.v1core.patch_namespaced_secret(o.name, o.trust_domain.name, secret)
if (not o.backend_handle):
o.backend_handle = secret.metadata.self_link