CORD-3054 delete steps for k8s synchronizer
Change-Id: I95d61ae5837892e3d7783791506d9ba6d7babcb4
diff --git a/xos/synchronizer/tests/test_sync_configmap.py b/xos/synchronizer/tests/test_sync_configmap.py
index 1ebd9e8..efc1111 100644
--- a/xos/synchronizer/tests/test_sync_configmap.py
+++ b/xos/synchronizer/tests/test_sync_configmap.py
@@ -118,6 +118,24 @@
self.assertEqual(configmap.backend_handle, "1234")
+ def test_delete_record(self):
+ with patch.object(self.step_class, "init_kubernetes_client", new=fake_init_kubernetes_client):
+ data = {"foo": "bar"}
+ configmap = KubernetesConfigMap(trust_domain=self.trust_domain, name="test-configmap", data=json.dumps(data))
+
+ orig_map = MagicMock()
+ orig_map.data = {"foo": "not_bar"}
+ orig_map.metadata.self_link = "1234"
+
+ step = self.step_class()
+ step.v1core.read_namespaced_config_map.return_value = orig_map
+
+ step.delete_record(configmap)
+
+ step.v1core.delete_namespaced_config_map.assert_called_with("test-configmap", self.trust_domain.name, ANY)
+
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/xos/synchronizer/tests/test_sync_kubernetesserviceinstance.py b/xos/synchronizer/tests/test_sync_kubernetesserviceinstance.py
index 9bf285a..2a20868 100644
--- a/xos/synchronizer/tests/test_sync_kubernetesserviceinstance.py
+++ b/xos/synchronizer/tests/test_sync_kubernetesserviceinstance.py
@@ -95,5 +95,18 @@
step.v1core.create_namespaced_pod.assert_called()
self.assertEqual(xos_si.backend_handle, "1234")
+ def test_delete_record(self):
+ with patch.object(self.step_class, "init_kubernetes_client", new=fake_init_kubernetes_client):
+ xos_si = KubernetesServiceInstance(name="test-instance", slice=self.slice, image=self.image)
+ xos_si.kubernetes_config_volume_mounts = self.MockObjectList([])
+ xos_si.kubernetes_secret_volume_mounts = self.MockObjectList([])
+
+ step = self.step_class()
+ pod = MagicMock()
+ step.v1core.read_namespaced_pod.return_value = pod
+
+ step.delete_record(xos_si)
+ step.v1core.delete_namespaced_pod.assert_called_with("test-instance", self.trust_domain.name, ANY)
+
if __name__ == '__main__':
unittest.main()
diff --git a/xos/synchronizer/tests/test_sync_principal.py b/xos/synchronizer/tests/test_sync_principal.py
index 81f9b30..3a55eb4 100644
--- a/xos/synchronizer/tests/test_sync_principal.py
+++ b/xos/synchronizer/tests/test_sync_principal.py
@@ -88,5 +88,17 @@
step.v1core.create_namespaced_service_account.assert_called()
self.assertEqual(xos_principal.backend_handle, "1234")
+ def test_delete_record(self):
+ with patch.object(self.step_class, "init_kubernetes_client", new=fake_init_kubernetes_client):
+ xos_principal = Principal(name="test-principal", trust_domain=self.trust_domain)
+
+ step = self.step_class()
+ k8s_sa = MagicMock()
+ step.v1core.read_namespaced_service_account.return_value = k8s_sa
+
+ step.delete_record(xos_principal)
+
+ step.v1core.delete_namespaced_service_account.assert_called_with("test-principal", self.trust_domain.name, ANY)
+
if __name__ == '__main__':
unittest.main()
diff --git a/xos/synchronizer/tests/test_sync_secret.py b/xos/synchronizer/tests/test_sync_secret.py
index ba7b39e..632adf6 100644
--- a/xos/synchronizer/tests/test_sync_secret.py
+++ b/xos/synchronizer/tests/test_sync_secret.py
@@ -118,6 +118,22 @@
self.assertEqual(xos_secret.backend_handle, "1234")
+ def test_delete_record(self):
+ with patch.object(self.step_class, "init_kubernetes_client", new=fake_init_kubernetes_client):
+ data = {"foo": "bar"}
+ xos_secret = KubernetesSecret(trust_domain=self.trust_domain, name="test-secret", data=json.dumps(data))
+
+ orig_secret = MagicMock()
+ orig_secret.data = {"foo": "not_bar"}
+ orig_secret.metadata.self_link = "1234"
+
+ step = self.step_class()
+ step.v1core.read_namespaced_secret.return_value = orig_secret
+
+ step.delete_record(xos_secret)
+
+ step.v1core.delete_namespaced_secret.assert_called_with("test-secret", self.trust_domain.name, ANY)
+
if __name__ == '__main__':
unittest.main()
diff --git a/xos/synchronizer/tests/test_sync_service.py b/xos/synchronizer/tests/test_sync_service.py
index 550d0c2..3030193 100644
--- a/xos/synchronizer/tests/test_sync_service.py
+++ b/xos/synchronizer/tests/test_sync_service.py
@@ -60,7 +60,7 @@
service = MagicMock()
step.v1core.read_namespaced_service.return_value = service
- result = step.get_service(xos_service, self.trust_domain)
+ result = step.get_service(xos_service, self.trust_domain.name)
self.assertEqual(result, service)
step.v1core.read_namespaced_service.assert_called_with("test-service", "test-trust")
@@ -74,7 +74,7 @@
step = self.step_class()
step.v1core.read_namespaced_service.side_effect = step.ApiException(status=404)
- result = step.get_service(xos_service, self.trust_domain)
+ result = step.get_service(xos_service, self.trust_domain.name)
self.assertEqual(result, None)
@@ -100,5 +100,24 @@
step.v1core.create_namespaced_service.assert_called()
self.assertEqual(xos_service.backend_handle, "1234")
+ def test_delete_record(self):
+ with patch.object(self.step_class, "init_kubernetes_client", new=fake_init_kubernetes_client):
+ xos_service = Service(name="test-service")
+ xos_slice = Slice(service=xos_service, trust_domain=self.trust_domain)
+ xos_service.slices = self.MockObjectList([xos_slice])
+
+ xos_serviceport = ServicePort(service=xos_service, name="web", external_port=123, internal_port=345,
+ protocol="TCP")
+ xos_service.serviceports=self.MockObjectList([xos_serviceport])
+
+ step = self.step_class()
+ k8s_service = MagicMock()
+ step.v1core.read_namespaced_service.return_value = k8s_service
+ step.v1core.delete_namespaced_service.return_value = None
+
+ step.delete_record(xos_service)
+
+ step.v1core.delete_namespaced_service.assert_called_with("test-service", self.trust_domain.name, ANY)
+
if __name__ == '__main__':
unittest.main()
diff --git a/xos/synchronizer/tests/test_sync_trustdomain.py b/xos/synchronizer/tests/test_sync_trustdomain.py
index 1695b26..4b12b9a 100644
--- a/xos/synchronizer/tests/test_sync_trustdomain.py
+++ b/xos/synchronizer/tests/test_sync_trustdomain.py
@@ -87,5 +87,17 @@
step.v1core.create_namespace.assert_called()
self.assertEqual(xos_trustdomain.backend_handle, "1234")
+ def test_delete_record(self):
+ with patch.object(self.step_class, "init_kubernetes_client", new=fake_init_kubernetes_client):
+ xos_trustdomain = TrustDomain(name="test-trust")
+
+ step = self.step_class()
+ td = MagicMock()
+ step.v1core.read_namespace.return_value = td
+
+ step.delete_record(xos_trustdomain)
+
+ step.v1core.delete_namespace.assert_called_with("test-trust", ANY)
+
if __name__ == '__main__':
unittest.main()