diff --git a/kubernetes/e2e_test/test_utils.py b/kubernetes/e2e_test/test_utils.py index ab752dff79..8e153100b3 100644 --- a/kubernetes/e2e_test/test_utils.py +++ b/kubernetes/e2e_test/test_utils.py @@ -408,3 +408,442 @@ def test_create_from_list_in_multi_resource_yaml_namespaced(self): name="mock-pod-1", namespace=self.test_namespace, body={}) app_api.delete_namespaced_deployment( name="mock", namespace=self.test_namespace, body={}) + + def test_create_apps_deployment_from_dict(self): + """ + Should be able to create a deployment. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + deployment_dict = { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": "nginx-app-4", + "labels": {"app": "nginx"} + }, + "spec": { + "replicas": 3, + "selector": {"matchLabels": {"app": "nginx"}}, + "template": { + "metadata": {"labels": {"app": "nginx"}}, + "spec": { + "containers": [ + { + "name": "nginx", + "image": "nginx:1.15.4", + "ports": [{"containerPort": 80}] + } + ] + } + } + } + } + utils.create_from_dict(k8s_client, deployment_dict) + + app_api = client.AppsV1Api(k8s_client) + dep = app_api.read_namespaced_deployment(name="nginx-app-4", + namespace="default") + self.assertIsNotNone(dep) + app_api.delete_namespaced_deployment( + name="nginx-app-4", namespace="default", + body={}) + + def test_create_pod_from_dict(self): + """ + Should be able to create a pod. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + core_api = client.CoreV1Api(k8s_client) + pod_dict = { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + "name": "nginx-app-pod", + "labels": {"app": "nginx"} + }, + "spec": { + "containers": [ + { + "name": "nginx", + "image": "nginx:1.15.4", + "ports": [{"containerPort": 80}] + } + ] + } + } + utils.create_from_dict(k8s_client, pod_dict) + + pod = core_api.read_namespaced_pod(name="nginx-app-pod", + namespace="default") + self.assertIsNotNone(pod) + core_api.delete_namespaced_pod( + name="nginx-app-pod", namespace="default", + body={}) + + def test_create_service_from_dict(self): + """ + Should be able to create a service. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + core_api = client.CoreV1Api(k8s_client) + service_dict = { + "kind": "Service", + "apiVersion": "v1", + "metadata": { + "name": "test-service" + }, + "spec": { + "selector": { + "app": "TestApp" + }, + "ports": [ + { + "protocol": "TCP", + "port": 80, + "targetPort": 9376 + } + ] + } + } + utils.create_from_dict(k8s_client, service_dict) + svc = core_api.read_namespaced_service(name="test-service", + namespace="default") + self.assertIsNotNone(svc) + core_api.delete_namespaced_service( + name="test-service", namespace="default", + body={}) + + def test_create_namespace_from_dict(self): + """ + Should be able to create a namespace. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + core_api = client.CoreV1Api(k8s_client) + namespace_dict = { + "apiVersion": "v1", + "kind": "Namespace", + "metadata": { + "name": "namespace-development", + "labels": {"name": "namespace-development"} + } + } + utils.create_from_dict(k8s_client, namespace_dict) + nmsp = core_api.read_namespace(name="namespace-development") + self.assertIsNotNone(nmsp) + core_api.delete_namespace(name="namespace-development", body={}) + + def test_create_rbac_role_from_dict(self): + """ + Should be able to create an rbac role. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + rbac_api = client.RbacAuthorizationV1Api(k8s_client) + rbac_role_dict = { + "apiVersion": "rbac.authorization.k8s.io/v1", + "kind": "Role", + "metadata": { + "name": "pod-reader-1", + }, + "rules": [ + { + "apiGroups": [""], + "resources": ["pods"], + "verbs": ["get", "watch", "list"], + } + ] + } + utils.create_from_dict(k8s_client, rbac_role_dict) + + rbac_role = rbac_api.read_namespaced_role( + name="pod-reader-1", namespace="default") + self.assertIsNotNone(rbac_role) + rbac_api.delete_namespaced_role( + name="pod-reader-1", namespace="default", body={}) + + def test_create_rbac_role_from_dict_with_verbose_enabled(self): + """ + Should be able to create an rbac role with verbose enabled. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + rbac_api = client.RbacAuthorizationV1Api(k8s_client) + rbac_role_dict = { + "apiVersion": "rbac.authorization.k8s.io/v1", + "kind": "Role", + "metadata": { + "name": "pod-reader-2", + }, + "rules": [ + { + "apiGroups": [""], + "resources": ["pods"], + "verbs": ["get", "watch", "list"], + } + ] + } + utils.create_from_dict(k8s_client, rbac_role_dict, verbose=True) + rbac_role = rbac_api.read_namespaced_role( + name="pod-reader-2", namespace="default") + self.assertIsNotNone(rbac_role) + rbac_api.delete_namespaced_role( + name="pod-reader-2", namespace="default", body={}) + + def test_create_deployment_non_default_namespace_from_dict(self): + """ + Should be able to create a namespace "dep", + and then create a deployment in the just-created namespace. + """ + k8s_client = client.ApiClient(configuration=self.config) + core_api = client.CoreV1Api(k8s_client) + ext_api = client.AppsV1Api(k8s_client) + namespace_dict = { + "apiVersion": "v1", + "kind": "Namespace", + "metadata": { + "name": "dep-nmsp", + "labels": {"name": "dep-nmsp"} + } + } + utils.create_from_dict(k8s_client, namespace_dict) + nmsp = core_api.read_namespace(name="dep-nmsp") + + self.assertIsNotNone(nmsp) + + deployment_dict = { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": "nginx-app-5", + "labels": {"app": "nginx"} + }, + "spec": { + "replicas": 3, + "selector": {"matchLabels": {"app": "nginx"}}, + "template": { + "metadata": {"labels": {"app": "nginx"}}, + "spec": { + "containers": [ + { + "name": "nginx", + "image": "nginx:1.15.4", + "ports": [{"containerPort": 80}] + } + ] + } + } + } + } + utils.create_from_dict(k8s_client, deployment_dict, + namespace="dep-nmsp") + + dep = ext_api.read_namespaced_deployment(name="nginx-app-5", + namespace="dep-nmsp") + self.assertIsNotNone(dep) + + ext_api.delete_namespaced_deployment( + name="nginx-app-5", namespace="dep-nmsp", + body={}) + core_api.delete_namespace(name="dep-nmsp", body={}) + + def test_create_apiservice_from_dict_with_conflict(self): + """ + Should be able to create an API service. + Should verify that creating the same API service should + fail due to conflict. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + reg_api = client.ApiregistrationV1beta1Api(k8s_client) + apiservice_dict = { + "apiVersion": "apiregistration.k8s.io/v1beta1", + "kind": "APIService", + "metadata": { + "name": "v1alpha1.test.k8s.io", + }, + "spec": { + "insecureSkipTLSVerify": True, + "group": "test.k8s.io", + "groupPriorityMinimum": 1000, + "versionPriority": 15, + "service": { + "name": "api", + "namespace": "test", + }, + "version": "v1alpha1" + } + } + utils.create_from_dict(k8s_client, apiservice_dict) + svc = reg_api.read_api_service( + name="v1alpha1.test.k8s.io") + self.assertIsNotNone(svc) + with self.assertRaises(utils.FailToCreateError) as cm: + utils.create_from_dict(k8s_client, apiservice_dict) + exp_error = ('Error from server (Conflict): ' + '{"kind":"Status","apiVersion":"v1","metadata":{},' + '"status":"Failure",' + '"message":"apiservices.apiregistration.k8s.io ' + '\\"v1alpha1.test.k8s.io\\" already exists",' + '"reason":"AlreadyExists",' + '"details":{"name":"v1alpha1.test.k8s.io",' + '"group":"apiregistration.k8s.io","kind":"apiservices"},' + '"code":409}\n' + ) + self.assertEqual(exp_error, str(cm.exception)) + reg_api.delete_api_service( + name="v1alpha1.test.k8s.io", body={}) + + # Tests for creating API objects from lists + + def test_create_general_list_from_dict(self): + """ + Should be able to create a service and a deployment + from a kind: List dict file + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + core_api = client.CoreV1Api(k8s_client) + ext_api = client.AppsV1Api(k8s_client) + + service_dict = { + "kind": "Service", + "apiVersion": "v1", + "metadata": { + "name": "list-service-test-2" + }, + "spec": { + "selector": { + "app": "TestApp" + }, + "ports": [ + { + "protocol": "TCP", + "port": 80, + "targetPort": 9376 + } + ] + } + } + + deploy_dict = { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": { + "name": "list-deployment-test-2", + "labels": {"app": "nginx"} + }, + "spec": { + "replicas": 1, + "selector": {"matchLabels": {"app": "nginx"}}, + "template": { + "metadata": {"labels": {"app": "nginx"}}, + "spec": { + "containers": [ + { + "name": "nginx", + "image": "nginx:1.15.4", + "ports": [{"containerPort": 80}] + } + ] + } + } + } + } + list_dict = { + "apiVersion": "v1", + "kind": "List", + "items": [ + service_dict, + deploy_dict, + ] + } + utils.create_from_dict(k8s_client, list_dict) + + svc = core_api.read_namespaced_service(name="list-service-test-2", + namespace="default") + self.assertIsNotNone(svc) + dep = ext_api.read_namespaced_deployment(name="list-deployment-test-2", + namespace="default") + self.assertIsNotNone(dep) + core_api.delete_namespaced_service(name="list-service-test-2", + namespace="default", body={}) + ext_api.delete_namespaced_deployment(name="list-deployment-test-2", + namespace="default", body={}) + + def test_create_namespace_list_from_dict(self): + """ + Should be able to create two namespaces + from a kind: NamespaceList dict file + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + core_api = client.CoreV1Api(k8s_client) + + nmsp_dict_1 = { + "apiVersion": "v1", + "kind": "Namespace", + "metadata": { + "name": "nmsp-1", + "labels": {"name": "nmsp-1"} + } + } + nmsp_dict_2 = { + "apiVersion": "v1", + "kind": "Namespace", + "metadata": { + "name": "nmsp-2", + "labels": {"name": "nmsp-2"} + } + } + nmsp_list_dict = { + "apiVersion": "v1", + "kind": "NamespaceList", + "items": [ + nmsp_dict_1, + nmsp_dict_2, + ] + } + + utils.create_from_dict(k8s_client, nmsp_list_dict) + nmsp_1 = core_api.read_namespace(name="nmsp-1") + self.assertIsNotNone(nmsp_1) + nmsp_2 = core_api.read_namespace(name="nmsp-2") + self.assertIsNotNone(nmsp_2) + core_api.delete_namespace(name="nmsp-1", body={}) + core_api.delete_namespace(name="nmsp-2", body={}) + + def test_create_namespace_list_from_dict_with_conflict(self): + """ + Should be able to create single namespace + Should verify that creating the same namespace should + fail due to conflict. + """ + k8s_client = client.api_client.ApiClient(configuration=self.config) + core_api = client.CoreV1Api(k8s_client) + nmsp_dict = { + "apiVersion": "v1", + "kind": "Namespace", + "metadata": { + "name": "nmsp-3", + "labels": {"name": "nmsp-3"} + } + } + nmsp_list_dict = { + "apiVersion": "v1", + "kind": "NamespaceList", + "items": [ + nmsp_dict, + nmsp_dict, + ] + } + with self.assertRaises(utils.FailToCreateError) as cm: + utils.create_from_dict(k8s_client, nmsp_list_dict) + exp_error = ('Error from server (Conflict): ' + '{"kind":"Status","apiVersion":"v1","metadata":{},' + '"status":"Failure",' + '"message":"namespaces ' + '\\"nmsp-3\\" already exists",' + '"reason":"AlreadyExists",' + '"details":{"name":"nmsp-3",' + '"kind":"namespaces"},' + '"code":409}\n' + ) + self.assertEqual(exp_error, str(cm.exception)) + nmsp = core_api.read_namespace(name="nmsp-3") + self.assertIsNotNone(nmsp) + core_api.delete_namespace(name="nmsp-3", body={}) diff --git a/kubernetes/utils/create_from_yaml.py b/kubernetes/utils/create_from_yaml.py index 80efa085f9..896f68026e 100644 --- a/kubernetes/utils/create_from_yaml.py +++ b/kubernetes/utils/create_from_yaml.py @@ -102,7 +102,7 @@ def create_from_dict(k8s_client, data, verbose=False, namespace='default', for yml_object in data["items"]: # Mitigate cases when server returns a xxxList object # See kubernetes-client/python#586 - if kind is not "": + if kind != "": yml_object["apiVersion"] = data["apiVersion"] yml_object["kind"] = kind try: