How to use create_from_yaml method in avocado

Best Python code snippet using avocado_python

test_utils.py

Source:test_utils.py

...38        """39        Should be able to create an apps/v1 deployment.40        """41        k8s_client = client.api_client.ApiClient(configuration=self.config)42        utils.create_from_yaml(43            k8s_client, self.path_prefix + "apps-deployment.yaml")44        app_api = client.AppsV1Api(k8s_client)45        dep = app_api.read_namespaced_deployment(name="nginx-app",46                                                 namespace="default")47        self.assertIsNotNone(dep)48        while True:49            try:50                app_api.delete_namespaced_deployment(51                    name="nginx-app", namespace="default",52                    body={})53                break54            except ApiException:55                continue56    def test_create_apps_deployment_from_yaml_object(self):57        """58        Should be able to pass YAM objects directly to helper function.59        """60        k8s_client = client.api_client.ApiClient(configuration=self.config)61        _path = self.path_prefix + "apps-deployment.yaml"62        with open(path.abspath(_path)) as f:63            yaml_objects = yaml.safe_load_all(f)64            utils.create_from_yaml(65                k8s_client,66                yaml_objects=yaml_objects,67            )68        app_api = client.AppsV1Api(k8s_client)69        dep = app_api.read_namespaced_deployment(name="nginx-app",70                                                 namespace="default")71        self.assertIsNotNone(dep)72        while True:73            try:74                app_api.delete_namespaced_deployment(75                    name="nginx-app", namespace="default",76                    body={})77                break78            except ApiException:79                continue80    def test_create_apps_deployment_from_yaml_obj(self):81        k8s_client = client.api_client.ApiClient(configuration=self.config)82        with open(self.path_prefix + "apps-deployment.yaml") as f:83            yml_obj = yaml.safe_load(f)84        yml_obj["metadata"]["name"] = "nginx-app-3"85        utils.create_from_dict(k8s_client, yml_obj)86        app_api = client.AppsV1Api(k8s_client)87        dep = app_api.read_namespaced_deployment(name="nginx-app-3",88                                                 namespace="default")89        self.assertIsNotNone(dep)90        app_api.delete_namespaced_deployment(91            name="nginx-app-3", namespace="default",92            body={})93    def test_create_pod_from_yaml(self):94        """95        Should be able to create a pod.96        """97        k8s_client = client.api_client.ApiClient(configuration=self.config)98        utils.create_from_yaml(99            k8s_client, self.path_prefix + "core-pod.yaml")100        core_api = client.CoreV1Api(k8s_client)101        pod = core_api.read_namespaced_pod(name="myapp-pod",102                                           namespace="default")103        self.assertIsNotNone(pod)104        core_api.delete_namespaced_pod(105            name="myapp-pod", namespace="default",106            body={})107    def test_create_service_from_yaml(self):108        """109        Should be able to create a service.110        """111        k8s_client = client.api_client.ApiClient(configuration=self.config)112        utils.create_from_yaml(113            k8s_client, self.path_prefix + "core-service.yaml")114        core_api = client.CoreV1Api(k8s_client)115        svc = core_api.read_namespaced_service(name="my-service",116                                               namespace="default")117        self.assertIsNotNone(svc)118        core_api.delete_namespaced_service(119            name="my-service", namespace="default",120            body={})121    def test_create_namespace_from_yaml(self):122        """123        Should be able to create a namespace.124        """125        k8s_client = client.api_client.ApiClient(configuration=self.config)126        utils.create_from_yaml(127            k8s_client, self.path_prefix + "core-namespace.yaml")128        core_api = client.CoreV1Api(k8s_client)129        nmsp = core_api.read_namespace(name="development")130        self.assertIsNotNone(nmsp)131        core_api.delete_namespace(name="development", body={})132    def test_create_rbac_role_from_yaml(self):133        """134        Should be able to create an rbac role.135        """136        k8s_client = client.api_client.ApiClient(configuration=self.config)137        utils.create_from_yaml(138            k8s_client, self.path_prefix + "rbac-role.yaml")139        rbac_api = client.RbacAuthorizationV1Api(k8s_client)140        rbac_role = rbac_api.read_namespaced_role(141            name="pod-reader", namespace="default")142        self.assertIsNotNone(rbac_role)143        rbac_api.delete_namespaced_role(144            name="pod-reader", namespace="default", body={})145    def test_create_rbac_role_from_yaml_with_verbose_enabled(self):146        """147        Should be able to create an rbac role with verbose enabled.148        """149        k8s_client = client.api_client.ApiClient(configuration=self.config)150        utils.create_from_yaml(151            k8s_client, self.path_prefix + "rbac-role.yaml", verbose=True)152        rbac_api = client.RbacAuthorizationV1Api(k8s_client)153        rbac_role = rbac_api.read_namespaced_role(154            name="pod-reader", namespace="default")155        self.assertIsNotNone(rbac_role)156        rbac_api.delete_namespaced_role(157            name="pod-reader", namespace="default", body={})158    def test_create_deployment_non_default_namespace_from_yaml(self):159        """160        Should be able to create a namespace "dep",161        and then create a deployment in the just-created namespace.162        """163        k8s_client = client.ApiClient(configuration=self.config)164        utils.create_from_yaml(165            k8s_client, self.path_prefix + "dep-namespace.yaml")166        utils.create_from_yaml(167            k8s_client, self.path_prefix + "dep-deployment.yaml")168        core_api = client.CoreV1Api(k8s_client)169        ext_api = client.AppsV1Api(k8s_client)170        nmsp = core_api.read_namespace(name="dep")171        self.assertIsNotNone(nmsp)172        dep = ext_api.read_namespaced_deployment(name="nginx-deployment",173                                                 namespace="dep")174        self.assertIsNotNone(dep)175        ext_api.delete_namespaced_deployment(176            name="nginx-deployment", namespace="dep",177            body={})178        core_api.delete_namespace(name="dep", body={})179    def test_create_apiservice_from_yaml_with_conflict(self):180        """181        Should be able to create an API service.182        Should verify that creating the same API service should183        fail due to conflict.184        """185        k8s_client = client.api_client.ApiClient(configuration=self.config)186        utils.create_from_yaml(187            k8s_client, self.path_prefix + "api-service.yaml")188        reg_api = client.ApiregistrationV1Api(k8s_client)189        svc = reg_api.read_api_service(190            name="v1alpha1.wardle.k8s.io")191        self.assertIsNotNone(svc)192        with self.assertRaises(utils.FailToCreateError) as cm:193            utils.create_from_yaml(194                k8s_client, "kubernetes/e2e_test/test_yaml/api-service.yaml")195        exp_error = ('Error from server (Conflict): '196                     '{"kind":"Status","apiVersion":"v1","metadata":{},'197                     '"status":"Failure",'198                     '"message":"apiservices.apiregistration.k8s.io '199                     '\\"v1alpha1.wardle.k8s.io\\" already exists",'200                     '"reason":"AlreadyExists",'201                     '"details":{"name":"v1alpha1.wardle.k8s.io",'202                     '"group":"apiregistration.k8s.io","kind":"apiservices"},'203                     '"code":409}\n'204                     )205        self.assertEqual(exp_error, str(cm.exception))206        reg_api.delete_api_service(207            name="v1alpha1.wardle.k8s.io", body={})208    # Tests for creating API objects from lists209    def test_create_general_list_from_yaml(self):210        """211        Should be able to create a service and a deployment212        from a kind: List yaml file213        """214        k8s_client = client.api_client.ApiClient(configuration=self.config)215        utils.create_from_yaml(216            k8s_client, self.path_prefix + "list.yaml")217        core_api = client.CoreV1Api(k8s_client)218        ext_api = client.AppsV1Api(k8s_client)219        svc = core_api.read_namespaced_service(name="list-service-test",220                                               namespace="default")221        self.assertIsNotNone(svc)222        dep = ext_api.read_namespaced_deployment(name="list-deployment-test",223                                                 namespace="default")224        self.assertIsNotNone(dep)225        core_api.delete_namespaced_service(name="list-service-test",226                                           namespace="default", body={})227        ext_api.delete_namespaced_deployment(name="list-deployment-test",228                                             namespace="default", body={})229    def test_create_namespace_list_from_yaml(self):230        """231        Should be able to create two namespaces232        from a kind: NamespaceList yaml file233        """234        k8s_client = client.api_client.ApiClient(configuration=self.config)235        utils.create_from_yaml(236            k8s_client, self.path_prefix + "namespace-list.yaml")237        core_api = client.CoreV1Api(k8s_client)238        nmsp_1 = core_api.read_namespace(name="mock-1")239        self.assertIsNotNone(nmsp_1)240        nmsp_2 = core_api.read_namespace(name="mock-2")241        self.assertIsNotNone(nmsp_2)242        core_api.delete_namespace(name="mock-1", body={})243        core_api.delete_namespace(name="mock-2", body={})244    def test_create_implicit_service_list_from_yaml_with_conflict(self):245        """246        Should be able to create two services from a kind: ServiceList247        json file that implicitly indicates the kind of individual objects248        """249        k8s_client = client.api_client.ApiClient(configuration=self.config)250        with self.assertRaises(utils.FailToCreateError):251            utils.create_from_yaml(252                k8s_client, self.path_prefix + "implicit-svclist.json")253        core_api = client.CoreV1Api(k8s_client)254        svc_3 = core_api.read_namespaced_service(name="mock-3",255                                                 namespace="default")256        self.assertIsNotNone(svc_3)257        svc_4 = core_api.read_namespaced_service(name="mock-4",258                                                 namespace="default")259        self.assertIsNotNone(svc_4)260        core_api.delete_namespaced_service(name="mock-3",261                                           namespace="default", body={})262        core_api.delete_namespaced_service(name="mock-4",263                                           namespace="default", body={})264    # Tests for creating multi-resource from directory265    def test_create_multi_resource_from_directory(self):266        """267        Should be able to create a service and a replication controller268        from a directory269        """270        k8s_client = client.api_client.ApiClient(configuration=self.config)271        utils.create_from_directory(272            k8s_client, self.path_prefix + "multi-resource/")273        core_api = client.CoreV1Api(k8s_client)274        svc = core_api.read_namespaced_service(name="mock",275                                               namespace="default")276        self.assertIsNotNone(svc)277        ctr = core_api.read_namespaced_replication_controller(278            name="mock", namespace="default")279        self.assertIsNotNone(ctr)280        core_api.delete_namespaced_replication_controller(281            name="mock", namespace="default", propagation_policy="Background")282        core_api.delete_namespaced_service(name="mock",283                                           namespace="default", body={})284    # Tests for multi-resource yaml objects285    def test_create_from_multi_resource_yaml(self):286        """287        Should be able to create a service and a replication controller288        from a multi-resource yaml file289        """290        k8s_client = client.api_client.ApiClient(configuration=self.config)291        utils.create_from_yaml(292            k8s_client, self.path_prefix + "multi-resource.yaml")293        core_api = client.CoreV1Api(k8s_client)294        svc = core_api.read_namespaced_service(name="mock",295                                               namespace="default")296        self.assertIsNotNone(svc)297        ctr = core_api.read_namespaced_replication_controller(298            name="mock", namespace="default")299        self.assertIsNotNone(ctr)300        core_api.delete_namespaced_replication_controller(301            name="mock", namespace="default", propagation_policy="Background")302        core_api.delete_namespaced_service(name="mock",303                                           namespace="default", body={})304    def test_create_from_list_in_multi_resource_yaml(self):305        """306        Should be able to create the items in the PodList and a deployment307        specified in the multi-resource file308        """309        k8s_client = client.api_client.ApiClient(configuration=self.config)310        utils.create_from_yaml(311            k8s_client, self.path_prefix + "multi-resource-with-list.yaml")312        core_api = client.CoreV1Api(k8s_client)313        app_api = client.AppsV1Api(k8s_client)314        pod_0 = core_api.read_namespaced_pod(315            name="mock-pod-0", namespace="default")316        self.assertIsNotNone(pod_0)317        pod_1 = core_api.read_namespaced_pod(318            name="mock-pod-1", namespace="default")319        self.assertIsNotNone(pod_1)320        dep = app_api.read_namespaced_deployment(321            name="mock", namespace="default")322        self.assertIsNotNone(dep)323        core_api.delete_namespaced_pod(324            name="mock-pod-0", namespace="default", body={})325        core_api.delete_namespaced_pod(326            name="mock-pod-1", namespace="default", body={})327        app_api.delete_namespaced_deployment(328            name="mock", namespace="default", body={})329    def test_create_from_multi_resource_yaml_with_conflict(self):330        """331        Should be able to create a service from the first yaml file.332        Should fail to create the same service from the second yaml file333        and create a replication controller.334        Should raise an exception for failure to create the same service twice.335        """336        k8s_client = client.api_client.ApiClient(configuration=self.config)337        utils.create_from_yaml(338            k8s_client, self.path_prefix + "yaml-conflict-first.yaml")339        core_api = client.CoreV1Api(k8s_client)340        svc = core_api.read_namespaced_service(name="mock-2",341                                               namespace="default")342        self.assertIsNotNone(svc)343        with self.assertRaises(utils.FailToCreateError) as cm:344            utils.create_from_yaml(345                k8s_client, self.path_prefix + "yaml-conflict-multi.yaml")346        exp_error = ('Error from server (Conflict): {"kind":"Status",'347                     '"apiVersion":"v1","metadata":{},"status":"Failure",'348                     '"message":"services \\"mock-2\\" already exists",'349                     '"reason":"AlreadyExists","details":{"name":"mock-2",'350                     '"kind":"services"},"code":409}\n'351                     )352        self.assertEqual(exp_error, str(cm.exception))353        ctr = core_api.read_namespaced_replication_controller(354            name="mock-2", namespace="default")355        self.assertIsNotNone(ctr)356        core_api.delete_namespaced_replication_controller(357            name="mock-2", namespace="default", propagation_policy="Background")358        core_api.delete_namespaced_service(name="mock-2",359                                           namespace="default", body={})360    def test_create_from_multi_resource_yaml_with_multi_conflicts(self):361        """362        Should create an apps/v1 deployment363        and fail to create the same deployment twice.364        Should raise an exception that contains two error messages.365        """366        k8s_client = client.api_client.ApiClient(configuration=self.config)367        with self.assertRaises(utils.FailToCreateError) as cm:368            utils.create_from_yaml(369                k8s_client, self.path_prefix + "triple-nginx.yaml")370        exp_error = ('Error from server (Conflict): {"kind":"Status",'371                     '"apiVersion":"v1","metadata":{},"status":"Failure",'372                     '"message":"deployments.apps \\"triple-nginx\\" '373                     'already exists","reason":"AlreadyExists",'374                     '"details":{"name":"triple-nginx","group":"apps",'375                     '"kind":"deployments"},"code":409}\n'376                     )377        exp_error += exp_error378        self.assertEqual(exp_error, str(cm.exception))379        ext_api = client.AppsV1Api(k8s_client)380        dep = ext_api.read_namespaced_deployment(name="triple-nginx",381                                                 namespace="default")382        self.assertIsNotNone(dep)383        ext_api.delete_namespaced_deployment(384            name="triple-nginx", namespace="default",385            body={})386    def test_create_namespaced_apps_deployment_from_yaml(self):387        """388        Should be able to create an apps/v1beta1 deployment389                in a test namespace.390        """391        k8s_client = client.api_client.ApiClient(configuration=self.config)392        utils.create_from_yaml(393            k8s_client, self.path_prefix + "apps-deployment.yaml",394            namespace=self.test_namespace)395        app_api = client.AppsV1Api(k8s_client)396        dep = app_api.read_namespaced_deployment(name="nginx-app",397                                                 namespace=self.test_namespace)398        self.assertIsNotNone(dep)399        app_api.delete_namespaced_deployment(400            name="nginx-app", namespace=self.test_namespace,401            body={})402    def test_create_from_list_in_multi_resource_yaml_namespaced(self):403        """404        Should be able to create the items in the PodList and a deployment405        specified in the multi-resource file in a test namespace406        """407        k8s_client = client.api_client.ApiClient(configuration=self.config)408        utils.create_from_yaml(409            k8s_client, self.path_prefix + "multi-resource-with-list.yaml",410            namespace=self.test_namespace)411        core_api = client.CoreV1Api(k8s_client)412        app_api = client.AppsV1Api(k8s_client)413        pod_0 = core_api.read_namespaced_pod(414            name="mock-pod-0", namespace=self.test_namespace)415        self.assertIsNotNone(pod_0)416        pod_1 = core_api.read_namespaced_pod(417            name="mock-pod-1", namespace=self.test_namespace)418        self.assertIsNotNone(pod_1)419        dep = app_api.read_namespaced_deployment(420            name="mock", namespace=self.test_namespace)421        self.assertIsNotNone(dep)422        core_api.delete_namespaced_pod(...

kubeTestDashboard.py

Source:kubeTestDashboard.py

...42class TestDashboard(unittest.TestCase):43    def test_a_create_user_cluster_role(self):44        # create service account and cluster role bindings45        try:46            utils.create_from_yaml(47                aApiClient, filepath + "cluster/files/" + "dashboard-adminuser.yaml")48            core_api = client.CoreV1Api(aApiClient)49            user_resp = core_api.read_namespaced_service_account(name="admin-user", namespace="kube-system")50            #print("user response %s " %user_resp)51            self.assertIsNotNone(user_resp)52            rbac_body = utils.create_from_yaml(53                aApiClient, filepath + "cluster/files/" + "dashboard-adminrolebind.yaml")54            rbac_api = client.RbacAuthorizationV1Api(aApiClient)55            cluster_resp = rbac_api.read_cluster_role_binding(name="admin-user")56            #print("cluster response %s " %cluster_resp)57            self.assertIsNotNone(user_resp)58        except ApiException as e:59            print("Exception creating user role: %s\n"%e)60        # tidy up created settings61        """ try:62            core_api = client.CoreV1Api(aApiClient)63            resp = core_api.delete_namespaced_service_account(name="admin-user", namespace='kube-system')64            rbac_api = client.RbacAuthorizationV1Api(aApiClient)65            resp = rbac_api.delete_cluster_role_binding(name="admin-user")66        except ApiException as e:67            print("Exception deleting user role: %s\n"%e) """68    def test_b_create_user_duplicate(self):69        # create a dupluicate user70        #self.maxDiff = None71        try:    72            with self.assertRaises(utils.FailToCreateError) as cm:73                utils.create_from_yaml(74                    aApiClient, filepath + "cluster/files/" + "dashboard-adminuser.yaml")75            exp_error = ('Error from server (Conflict): '76                        '{"kind":"Status","apiVersion":"v1","metadata":{},'77                        '"status":"Failure",'78                        '"message":"serviceaccounts \\"admin-user\\" already exists",'79                        '"reason":"AlreadyExists",'80                        '"details":{"name":"admin-user","kind":"serviceaccounts"},'81                        '"code":409}\n'82                        )83            self.assertEqual(exp_error, str(cm.exception))84        except ApiException as e:85            print("Exception creating duplicate user role: %s\n"%e)86    def test_c_create_secret_certs(self):87        # create certificates88        try:89            utils.create_from_yaml(90                aApiClient, filepath + "cluster/files/" + "dashboard-certs.yaml")91            core_api = client.CoreV1Api(aApiClient)92            certs_resp = core_api.read_namespaced_secret(name="kubernetes-dashboard-certs", namespace="kube-system")93            #print("certs response %s " %certs_resp)94            self.assertIsNotNone(certs_resp)95            utils.create_from_yaml(96                aApiClient, filepath + "cluster/files/" + "dashboard-csrf.yaml")97            core_api = client.CoreV1Api(aApiClient)98            csrf_resp = core_api.read_namespaced_secret(name="kubernetes-dashboard-csrf", namespace="kube-system")99            #print("certs response %s " %csrf_resp)100            self.assertIsNotNone(csrf_resp)101            utils.create_from_yaml(102                aApiClient, filepath + "cluster/files/" + "dashboard-keyh.yaml")103            core_api = client.CoreV1Api(aApiClient)104            keyh_resp = core_api.read_namespaced_secret(name="kubernetes-dashboard-key-holder", namespace="kube-system")105            #print("certs response %s " %keyh_resp)106            self.assertIsNotNone(keyh_resp)107        except ApiException as e:108            print("Exception creating certs: %s\n"%e)109        # tidy up created settings110        """ try:111            certs_resp = core_api.delete_namespaced_secret(name="kubernetes-dashboard-certs", namespace="kube-system")112            print("certs response %s " %certs_resp)113            csrf_resp = core_api.delete_namespaced_secret(name="kubernetes-dashboard-csrf", namespace="kube-system")114            print("certs response %s " %csrf_resp)115            keyh_resp = core_api.delete_namespaced_secret(name="kubernetes-dashboard-key-holder", namespace="kube-system")116            print("certs response %s " %keyh_resp)117        except ApiException as e:118            print("Exception deleting certs: %s\n"%e) """119    def test_d_create_config_map(self):120        # create config map121        try:122            utils.create_from_yaml(123                aApiClient, filepath + "cluster/files/" + "dashboard-configmap.yaml")124            core_api = client.CoreV1Api(aApiClient)125            conf_resp = core_api.read_namespaced_config_map(name="kubernetes-dashboard-settings", namespace="kube-system")126            #print("config map response %s " %conf_resp)127            self.assertIsNotNone(conf_resp)128        except ApiException as e:129            print("Exception creating config map: %s\n"%e)130        # tidy up created settings131        """ try:132            core_api = client.CoreV1Api(aApiClient)133            resp = core_api.delete_namespaced_config_map(name="kubernetes-dashboard-settings", namespace='kube-system')134        except ApiException as e:135            print("Exception deleting config map: %s\n"%e) """136    def test_e_create_cluster_role(self):137        # create cluster role138        try:139            utils.create_from_yaml(140                aApiClient, filepath + "cluster/files/" + "dashboard-clusterrole.yaml")141            core_api = client.RbacAuthorizationV1Api(aApiClient)142            cluster_resp = core_api.read_cluster_role(name="kubernetes-dashboard")143            #print("cluster role response %s " %cluster_resp)144            self.assertIsNotNone(cluster_resp)145        except ApiException as e:146            print("Exception creating cluster role: %s\n"%e)147        # tidy up created settings148        """ try:149            core_api = client.RbacAuthorizationV1Api(aApiClient)150            resp = core_api.delete_namespaced_cluster_role(name="kubernetes-dashboard", namespace='kube-system')151        except ApiException as e:152            print("Exception deleting cluster role: %s\n"%e) """153    def test_f_create_role_bindings(self):154        # create cluster role155        try:156            utils.create_from_yaml(157                aApiClient, filepath + "cluster/files/" + "dashboard-rolebinding.yaml")158            core_api = client.RbacAuthorizationV1Api(aApiClient)159            role_resp = core_api.read_namespaced_role(name="kubernetes-dashboard", namespace="kube-system")160            #print("role response %s " %role_resp)161            self.assertIsNotNone(cluster_resp)162        except ApiException as e:163            print("Exception creating role: %s\n"%e)164        # tidy up created settings165        """ try:166            core_api = client.RbacAuthorizationV1Api(aApiClient)167            resp = core_api.delete_namespaced_role(name="kubernetes-dashboard", namespace='kube-system')168        except ApiException as e:169            print("Exception deleting role: %s\n"%e) """170    def test_g_create_cluster_role_bindings(self):171        # create cluster role172        try:173            utils.create_from_yaml(174                aApiClient, filepath + "cluster/files/" + "dashboard-clusterrolebinding.yaml")175            core_api = client.RbacAuthorizationV1Api(aApiClient)176            role_resp = core_api.read_cluster_role_binding(name="kubernetes-dashboard")177            #print("role response %s " %role_resp)178            self.assertIsNotNone(cluster_resp)179        except ApiException as e:180            print("Exception creating role: %s\n"%e)181        # tidy up created settings182        """ try:183            core_api = client.RbacAuthorizationV1Api(aApiClient)184            resp = core_api.delete_cluster_role_binding(name="kubernetes-dashboard", namespace='kube-system')185        except ApiException as e:186            print("Exception deleting role: %s\n"%e) """187    def test_h_create_deployment(self):188        # create deployment189        try:190            utils.create_from_yaml(191                aApiClient, filepath + "cluster/files/" + "dashboard-deployment.yaml")192            apps_api = client.AppsV1Api(aApiClient)193            deploy_resp = apps_api.read_namespaced_deployment(name="kubernetes-dashboard", namespace="kube-system")194            #print("deploy response %s " %deploy_resp)195            self.assertIsNotNone(deploy_resp)196        except ApiException as e:197            print("Exception creating app: %s\n"%e)198        # tidy up created settings199        """ try:200            apps_api = client.AppsV1Api(aApiClient)201            resp = apps_api.delete_namespaced_deployment(name="kubernetes-dashboard", namespace='kube-system')202        except ApiException as e:203            print("Exception deleting app: %s\n"%e) """204    def test_i_create_metrics_service(self):205        # create service206        try:207            utils.create_from_yaml(208                aApiClient, filepath + "cluster/files/" + "dashboard-metrics-service.yaml")209            core_api = client.CoreV1Api(aApiClient)210            deploy_resp = core_api.read_namespaced_service(name="dashboard-metrics-scraper", namespace="kube-system")211            #print("deploy response %s " %deploy_resp)212            self.assertIsNotNone(deploy_resp)213        except ApiException as e:214            print("Exception creating service: %s\n"%e)215        # tidy up created settings216        """ try:217            core_api = client.CoreV1Api(aApiClient)218            resp = core_api.delete_namespaced_service(name="dashboard-metrics-scraper", namespace='kube-system')219        except ApiException as e:220            print("Exception deleting service: %s\n"%e) """221    def test_j_create_metrics_deployment(self):222        # create deployment223        try:224            utils.create_from_yaml(225                aApiClient, filepath + "cluster/files/" + "dashboard-metrics-deploy.yaml")226            apps_api = client.AppsV1Api(aApiClient)227            deploy_resp = apps_api.read_namespaced_deployment(name="dashboard-metrics-scraper", namespace="kube-system")228            #print("deploy response %s " %deploy_resp)229            self.assertIsNotNone(deploy_resp)230        except ApiException as e:231            print("Exception creating app: %s\n"%e)232        # tidy up created settings233        """ try:234            apps_api = client.AppsV1Api(aApiClient)235            resp = apps_api.delete_namespaced_deployment(name="dashboard-metrics-scraper", namespace='kube-system')236        except ApiException as e:237            print("Exception deleting app: %s\n"%e) """238if __name__ == '__main__':...

GE_GSCH_pre_warmer.py

Source:GE_GSCH_pre_warmer.py

...35        for full_filename in self.policyYamls:36            #with open(path.join(path.dirname(__file__),filename)) as f:37            try:38                yaml_file = full_filename39                resp = utils.create_from_yaml(k8s_client, yaml_file)40                #print('resp of utils.create_from_yaml ====>',resp)41                print('create_from_yaml is completed ',yaml_file)   42            except :...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.