KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use reraise method in Slash
Best Python code snippet using slash
manager.py
Source:manager.py 
1#!/usr/bin/python2#3# Copyright 2018-2021 Polyaxon, Inc.4#5# Licensed under the Apache License, Version 2.0 (the "License");6# you may not use this file except in compliance with the License.7# You may obtain a copy of the License at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS,13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14# See the License for the specific language governing permissions and15# limitations under the License.=16from kubernetes import client, config17from kubernetes.client.rest import ApiException18from polyaxon.exceptions import PolyaxonK8SError19from polyaxon.k8s import constants20from polyaxon.k8s.monitor import is_pod_running21from polyaxon.logger import logger22class K8SManager:23 def __init__(self, k8s_config=None, namespace="default", in_cluster=False):24 if not k8s_config:25 if in_cluster:26 config.load_incluster_config()27 else:28 config.load_kube_config()29 self.api_client = None30 else:31 self.api_client = client.api_client.ApiClient(configuration=k8s_config)32 self._k8s_api = None33 self._k8s_batch_api = None34 self._k8s_apps_api = None35 self._k8s_beta_api = None36 self._networking_v1_beta1_api = None37 self._k8s_custom_object_api = None38 self._k8s_version_api = None39 self.namespace = namespace40 self.in_cluster = in_cluster41 @property42 def k8s_api(self):43 if not self._k8s_api:44 self._k8s_api = client.CoreV1Api(self.api_client)45 return self._k8s_api46 @property47 def k8s_batch_api(self):48 if not self._k8s_batch_api:49 self._k8s_batch_api = client.BatchV1Api(self.api_client)50 return self._k8s_batch_api51 @property52 def k8s_apps_api(self):53 if not self._k8s_apps_api:54 self._k8s_apps_api = client.AppsV1Api(self.api_client)55 return self._k8s_apps_api56 @property57 def k8s_beta_api(self):58 if not self._k8s_beta_api:59 self._k8s_beta_api = client.ExtensionsV1beta1Api(self.api_client)60 return self._k8s_beta_api61 @property62 def networking_v1_beta1_api(self):63 if not self._networking_v1_beta1_api:64 self._networking_v1_beta1_api = client.NetworkingV1beta1Api(self.api_client)65 return self._networking_v1_beta1_api66 @property67 def k8s_custom_object_api(self):68 if not self._k8s_custom_object_api:69 self._k8s_custom_object_api = client.CustomObjectsApi(self.api_client)70 return self._k8s_custom_object_api71 @property72 def k8s_version_api(self):73 if not self._k8s_version_api:74 self._k8s_version_api = client.VersionApi(self.api_client)75 return self._k8s_version_api76 def set_namespace(self, namespace):77 self.namespace = namespace78 def get_version(self, reraise=False):79 try:80 return self.k8s_version_api.get_code().to_dict()81 except ApiException as e:82 logger.error("K8S error: {}".format(e))83 if reraise:84 raise PolyaxonK8SError("Connection error: %s" % e) from e85 def _list_namespace_resource(self, resource_api, reraise=False, **kwargs):86 try:87 res = resource_api(namespace=self.namespace, **kwargs)88 if isinstance(res, dict):89 items = res["items"]90 else:91 items = res.items92 return [p for p in items]93 except ApiException as e:94 logger.error("K8S error: {}".format(e))95 if reraise:96 raise PolyaxonK8SError("Connection error: %s" % e) from e97 return []98 def list_nodes(self, reraise=False):99 try:100 res = self.k8s_api.list_node()101 return [p for p in res.items]102 except ApiException as e:103 logger.error("K8S error: {}".format(e))104 if reraise:105 raise PolyaxonK8SError("Connection error: %s" % e) from e106 return []107 def list_pods(self, include_uninitialized=True, reraise=False, **kwargs):108 return self._list_namespace_resource(109 resource_api=self.k8s_api.list_namespaced_pod,110 reraise=reraise,111 **kwargs,112 )113 def list_jobs(self, include_uninitialized=True, reraise=False, **kwargs):114 return self._list_namespace_resource(115 resource_api=self.k8s_batch_api.list_namespaced_job,116 reraise=reraise,117 **kwargs,118 )119 def list_custom_objects(self, group, version, plural, reraise=False, **kwargs):120 return self._list_namespace_resource(121 resource_api=self.k8s_custom_object_api.list_namespaced_custom_object,122 reraise=reraise,123 group=group,124 version=version,125 plural=plural,126 **kwargs,127 )128 def list_services(self, reraise=False, **kwargs):129 return self._list_namespace_resource(130 resource_api=self.k8s_api.list_namespaced_service,131 reraise=reraise,132 **kwargs,133 )134 def list_deployments(self, reraise=False, **kwargs):135 return self._list_namespace_resource(136 resource_api=self.k8s_apps_api.list_namespaced_deployment,137 reraise=reraise,138 **kwargs,139 )140 def list_ingresses(self, reraise=False, **kwargs):141 return self._list_namespace_resource(142 resource_api=self.networking_v1_beta1_api.list_namespaced_ingress,143 reraise=reraise,144 **kwargs,145 )146 def update_node_labels(self, node, labels, reraise=False):147 body = {"metadata": {"labels": labels}, "namespace": self.namespace}148 try:149 return self.k8s_api.patch_node(name=node, body=body)150 except ApiException as e:151 logger.error("K8S error: {}".format(e))152 if reraise:153 raise PolyaxonK8SError("Connection error: %s" % e) from e154 def create_config_map(self, name, body):155 resp = self.k8s_api.create_namespaced_config_map(156 namespace=self.namespace, body=body157 )158 logger.debug("Config map `{}` was created".format(name))159 return resp160 def update_config_map(self, name, body):161 resp = self.k8s_api.patch_namespaced_config_map(162 name=name, namespace=self.namespace, body=body163 )164 logger.debug("Config map `{}` was patched".format(name))165 return resp166 def create_or_update_config_map(self, name, body, reraise=False):167 try:168 return self.create_config_map(name=name, body=body)169 except ApiException:170 try:171 return self.update_config_map(name=name, body=body)172 except ApiException as e:173 if reraise:174 raise PolyaxonK8SError("Connection error: %s" % e) from e175 else:176 logger.error("K8S error: {}".format(e))177 def create_secret(self, name, body):178 resp = self.k8s_api.create_namespaced_secret(179 namespace=self.namespace, body=body180 )181 logger.debug("Secret `{}` was created".format(name))182 return resp183 def update_secret(self, name, body):184 resp = self.k8s_api.patch_namespaced_secret(185 name=name, namespace=self.namespace, body=body186 )187 logger.debug("Secret `{}` was patched".format(name))188 return resp189 def create_or_update_secret(self, name, body, reraise=False):190 try:191 return self.create_secret(name=name, body=body), True192 except ApiException:193 try:194 return self.update_secret(name=name, body=body), False195 except ApiException as e:196 if reraise:197 raise PolyaxonK8SError("Connection error: %s" % e) from e198 else:199 logger.error("K8S error: {}".format(e))200 def create_service(self, name, body):201 resp = self.k8s_api.create_namespaced_service(202 namespace=self.namespace, body=body203 )204 logger.debug("Service `{}` was created".format(name))205 return resp206 def update_service(self, name, body):207 resp = self.k8s_api.patch_namespaced_service(208 name=name, namespace=self.namespace, body=body209 )210 logger.debug("Service `{}` was patched".format(name))211 return resp212 def create_or_update_service(self, name, body, reraise=False):213 try:214 return self.create_service(name=name, body=body), True215 except ApiException:216 try:217 return self.update_service(name=name, body=body), False218 except ApiException as e:219 if reraise:220 raise PolyaxonK8SError("Connection error: %s" % e) from e221 else:222 logger.error("K8S error: {}".format(e))223 def create_pod(self, name, body):224 resp = self.k8s_api.create_namespaced_pod(namespace=self.namespace, body=body)225 logger.debug("Pod `{}` was created".format(name))226 return resp227 def update_pod(self, name, body):228 resp = self.k8s_api.patch_namespaced_pod(229 name=name, namespace=self.namespace, body=body230 )231 logger.debug("Pod `{}` was patched".format(name))232 return resp233 def create_or_update_pod(self, name, body, reraise=False):234 try:235 return self.create_pod(name=name, body=body), True236 except ApiException:237 try:238 return self.update_pod(name=name, body=body), False239 except ApiException as e:240 if reraise:241 raise PolyaxonK8SError("Connection error: %s" % e) from e242 else:243 logger.error("K8S error: {}".format(e))244 def create_job(self, name, body):245 resp = self.k8s_batch_api.create_namespaced_job(246 namespace=self.namespace, body=body247 )248 logger.debug("Job `{}` was created".format(name))249 return resp250 def update_job(self, name, body):251 resp = self.k8s_batch_api.patch_namespaced_job(252 name=name, namespace=self.namespace, body=body253 )254 logger.debug("Job `{}` was patched".format(name))255 return resp256 def create_or_update_job(self, name, body, reraise=False):257 try:258 return self.create_job(name=name, body=body), True259 except ApiException:260 try:261 return self.update_job(name=name, body=body), False262 except ApiException as e:263 if reraise:264 raise PolyaxonK8SError("Connection error: %s" % e) from e265 else:266 logger.error("K8S error: {}".format(e))267 def create_custom_object(self, name, group, version, plural, body):268 resp = self.k8s_custom_object_api.create_namespaced_custom_object(269 group=group,270 version=version,271 plural=plural,272 namespace=self.namespace,273 body=body,274 )275 logger.debug("Custom object `{}` was created".format(name))276 return resp277 def update_custom_object(self, name, group, version, plural, body):278 resp = self.k8s_custom_object_api.patch_namespaced_custom_object(279 name=name,280 group=group,281 version=version,282 plural=plural,283 namespace=self.namespace,284 body=body,285 )286 logger.debug("Custom object `{}` was patched".format(name))287 return resp288 def create_or_update_custom_object(289 self, name, group, version, plural, body, reraise=False290 ):291 try:292 return (293 self.create_custom_object(294 name=name, group=group, version=version, plural=plural, body=body295 ),296 True,297 )298 except ApiException as e_create:299 try:300 return (301 self.update_custom_object(302 name=name,303 group=group,304 version=version,305 plural=plural,306 body=body,307 ),308 False,309 )310 except ApiException as e:311 if reraise:312 raise PolyaxonK8SError(313 "Connection error: creation %s - update %s" % (e_create, e)314 ) from e315 else:316 logger.error("K8S error: {}".format(e))317 def create_deployment(self, name, body):318 resp = self.k8s_apps_api.create_namespaced_deployment(319 namespace=self.namespace, body=body320 )321 logger.debug("Deployment `{}` was created".format(name))322 return resp323 def update_deployment(self, name, body):324 resp = self.k8s_apps_api.patch_namespaced_deployment(325 name=name, namespace=self.namespace, body=body326 )327 logger.debug("Deployment `{}` was patched".format(name))328 return resp329 def create_or_update_deployment(self, name, body, reraise=False):330 try:331 return self.create_deployment(name=name, body=body), True332 except ApiException:333 try:334 return self.update_deployment(name=name, body=body), False335 except ApiException as e:336 if reraise:337 raise PolyaxonK8SError("Connection error: %s" % e) from e338 else:339 logger.error("K8S error: {}".format(e))340 def create_volume(self, name, body):341 resp = self.k8s_api.create_persistent_volume(body=body)342 logger.debug("Persistent volume `{}` was created".format(name))343 return resp344 def update_volume(self, name, body):345 resp = self.k8s_api.patch_persistent_volume(name=name, body=body)346 logger.debug("Persistent volume `{}` was patched".format(name))347 return resp348 def create_or_update_volume(self, name, body, reraise=False):349 try:350 return self.create_volume(name=name, body=body), True351 except ApiException:352 try:353 return self.update_service(name=name, body=body), False354 except ApiException as e:355 if reraise:356 raise PolyaxonK8SError("Connection error: %s" % e) from e357 else:358 logger.error("K8S error: {}".format(e))359 def create_volume_claim(self, name, body):360 resp = self.k8s_api.create_namespaced_persistent_volume_claim(361 namespace=self.namespace, body=body362 )363 logger.debug("Volume claim `{}` was created".format(name))364 return resp365 def update_volume_claim(self, name, body):366 resp = self.k8s_api.patch_namespaced_persistent_volume_claim(367 name=name, namespace=self.namespace, body=body368 )369 logger.debug("Volume claim `{}` was patched".format(name))370 return resp371 def create_or_update_volume_claim(self, name, body, reraise=False):372 try:373 return self.create_volume_claim(name=name, body=body), True374 except ApiException:375 try:376 return self.update_volume_claim(name=name, body=body), False377 except ApiException as e:378 if reraise:379 raise PolyaxonK8SError("Connection error: %s" % e) from e380 else:381 logger.error("K8S error: {}".format(e))382 def create_ingress(self, name, body):383 resp = self.networking_v1_beta1_api.create_namespaced_ingress(384 namespace=self.namespace, body=body385 )386 logger.debug("ingress `{}` was created".format(name))387 return resp388 def update_ingress(self, name, body):389 resp = self.networking_v1_beta1_api.patch_namespaced_ingress(390 name=name, namespace=self.namespace, body=body391 )392 logger.debug("Ingress `{}` was patched".format(name))393 return resp394 def create_or_update_ingress(self, name, body, reraise=False):395 try:396 return self.create_ingress(name=name, body=body), True397 except ApiException:398 try:399 return self.update_ingress(name=name, body=body), False400 except ApiException as e:401 if reraise:402 raise PolyaxonK8SError("Connection error: %s" % e) from e403 else:404 logger.error("K8S error: {}".format(e))405 def get_config_map(self, name, reraise=False):406 try:407 return self.k8s_api.read_namespaced_config_map(408 name=name, namespace=self.namespace409 )410 except ApiException as e:411 if reraise:412 raise PolyaxonK8SError("Connection error: %s" % e) from e413 return None414 def get_secret(self, name, reraise=False):415 try:416 return self.k8s_api.read_namespaced_secret(417 name=name, namespace=self.namespace418 )419 except ApiException as e:420 if reraise:421 raise PolyaxonK8SError("Connection error: %s" % e) from e422 return None423 def get_service(self, name, reraise=False):424 try:425 return self.k8s_api.read_namespaced_service(426 name=name, namespace=self.namespace427 )428 except ApiException as e:429 if reraise:430 raise PolyaxonK8SError("Connection error: %s" % e) from e431 return None432 def get_pod(self, name, reraise=False):433 try:434 return self.k8s_api.read_namespaced_pod(name=name, namespace=self.namespace)435 except ApiException as e:436 if reraise:437 raise PolyaxonK8SError("Connection error: %s" % e) from e438 return None439 def get_job(self, name, reraise=False):440 try:441 return self.k8s_batch_api.read_namespaced_job(442 name=name, namespace=self.namespace443 )444 except ApiException as e:445 if reraise:446 raise PolyaxonK8SError("Connection error: %s" % e) from e447 return None448 def get_custom_object(self, name, group, version, plural):449 return self.k8s_custom_object_api.get_namespaced_custom_object(450 name=name,451 group=group,452 version=version,453 plural=plural,454 namespace=self.namespace,455 )456 def get_deployment(self, name, reraise=False):457 try:458 return self.k8s_apps_api.read_namespaced_deployment(459 name=name, namespace=self.namespace460 )461 except ApiException as e:462 if reraise:463 raise PolyaxonK8SError("Connection error: %s" % e) from e464 return None465 def get_volume(self, name, reraise=False):466 try:467 return self.k8s_api.read_persistent_volume(name=name)468 except ApiException as e:469 if reraise:470 raise PolyaxonK8SError("Connection error: %s" % e) from e471 return None472 def get_volume_claim(self, name, reraise=False):473 try:474 return self.k8s_api.read_namespaced_persistent_volume_claim(475 name=name, namespace=self.namespace476 )477 except ApiException as e:478 if reraise:479 raise PolyaxonK8SError("Connection error: %s" % e) from e480 return None481 def get_ingress(self, name, reraise=False):482 try:483 return self.networking_v1_beta1_api.read_namespaced_ingress(484 name=name, namespace=self.namespace485 )486 except ApiException as e:487 if reraise:488 raise PolyaxonK8SError("Connection error: %s" % e) from e489 return None490 def delete_config_map(self, name, reraise=False):491 try:492 self.k8s_api.delete_namespaced_config_map(493 name=name,494 namespace=self.namespace,495 body=client.V1DeleteOptions(api_version=constants.K8S_API_VERSION_V1),496 )497 logger.debug("Config map `{}` Deleted".format(name))498 except ApiException as e:499 if reraise:500 raise PolyaxonK8SError("Connection error: %s" % e) from e501 else:502 logger.debug("Config map `{}` was not found".format(name))503 def delete_secret(self, name, reraise=False):504 try:505 self.k8s_api.delete_namespaced_secret(506 name=name,507 namespace=self.namespace,508 body=client.V1DeleteOptions(api_version=constants.K8S_API_VERSION_V1),509 )510 logger.debug("secret `{}` Deleted".format(name))511 except ApiException as e:512 if reraise:513 raise PolyaxonK8SError("Connection error: %s" % e) from e514 else:515 logger.debug("secret `{}` was not found".format(name))516 def delete_service(self, name, reraise=False):517 try:518 self.k8s_api.delete_namespaced_service(519 name=name,520 namespace=self.namespace,521 body=client.V1DeleteOptions(api_version=constants.K8S_API_VERSION_V1),522 )523 logger.debug("Service `{}` deleted".format(name))524 except ApiException as e:525 if reraise:526 raise PolyaxonK8SError("Connection error: %s" % e) from e527 else:528 logger.debug("Service `{}` was not found".format(name))529 def delete_pod(self, name, reraise=False):530 try:531 self.k8s_api.delete_namespaced_pod(532 name=name,533 namespace=self.namespace,534 body=client.V1DeleteOptions(api_version=constants.K8S_API_VERSION_V1),535 )536 logger.debug("Pod `{}` deleted".format(name))537 except ApiException as e:538 if reraise:539 raise PolyaxonK8SError("Connection error: %s" % e) from e540 else:541 logger.debug("Pod `{}` was not found".format(name))542 def delete_job(self, name, reraise=False):543 try:544 self.k8s_batch_api.delete_namespaced_job(545 name=name,546 namespace=self.namespace,547 body=client.V1DeleteOptions(api_version=constants.K8S_API_VERSION_V1),548 )549 logger.debug("Pod `{}` deleted".format(name))550 except ApiException as e:551 if reraise:552 raise PolyaxonK8SError("Connection error: %s" % e) from e553 else:554 logger.debug("Pod `{}` was not found".format(name))555 def delete_custom_object(self, name, group, version, plural):556 self.k8s_custom_object_api.delete_namespaced_custom_object(557 name=name,558 group=group,559 version=version,560 plural=plural,561 namespace=self.namespace,562 body=client.V1DeleteOptions(),563 )564 logger.debug("Custom object `{}` deleted".format(name))565 def delete_deployment(self, name, reraise=False):566 try:567 self.k8s_apps_api.delete_namespaced_deployment(568 name=name,569 namespace=self.namespace,570 body=client.V1DeleteOptions(571 api_version=constants.K8S_API_VERSION_APPS_V1,572 propagation_policy="Foreground",573 ),574 )575 logger.debug("Deployment `{}` deleted".format(name))576 except ApiException as e:577 if reraise:578 raise PolyaxonK8SError("Connection error: %s" % e) from e579 else:580 logger.debug("Deployment `{}` was not found".format(name))581 def delete_volume(self, name, reraise=False):582 try:583 self.k8s_api.delete_persistent_volume(584 name=name,585 body=client.V1DeleteOptions(api_version=constants.K8S_API_VERSION_V1),586 )587 logger.debug("Volume `{}` Deleted".format(name))588 except ApiException as e:589 if reraise:590 raise PolyaxonK8SError("Connection error: %s" % e) from e591 else:592 logger.debug("Volume `{}` was not found".format(name))593 def delete_volume_claim(self, name, reraise=False):594 try:595 self.k8s_api.delete_namespaced_persistent_volume_claim(596 name=name,597 namespace=self.namespace,598 body=client.V1DeleteOptions(api_version=constants.K8S_API_VERSION_V1),599 )600 logger.debug("Volume claim `{}` Deleted".format(name))601 except ApiException as e:602 if reraise:603 raise PolyaxonK8SError("Connection error: %s" % e) from e604 else:605 logger.debug("Volume claim `{}` was not found".format(name))606 def delete_ingress(self, name, reraise=False):607 try:608 self.networking_v1_beta1_api.delete_namespaced_ingress(609 name=name,610 namespace=self.namespace,611 body=client.V1DeleteOptions(612 api_version=constants.K8S_API_VERSION_NETWORKING_V1_BETA1,613 propagation_policy="Foreground",614 ),615 )616 logger.debug("Ingress `{}` deleted".format(name))617 except ApiException as e:618 if reraise:619 raise PolyaxonK8SError("Connection error: %s" % e) from e620 else:621 logger.debug("Ingress `{}` was not found".format(name))622 def delete_pods(self, include_uninitialized=True, reraise=False, **kwargs):623 objs = self.list_pods(624 include_uninitialized=include_uninitialized, reraise=reraise, **kwargs625 )626 for obj in objs:627 self.delete_pod(name=obj.metadata.name, reraise=reraise)628 def delete_jobs(self, include_uninitialized=True, reraise=False, **kwargs):629 objs = self.list_jobs(630 include_uninitialized=include_uninitialized, reraise=reraise, **kwargs631 )632 for obj in objs:633 self.delete_job(name=obj.metadata.name, reraise=reraise)634 def delete_services(self, reraise=False, **kwargs):635 objs = self.list_services(reraise=reraise, **kwargs)636 for obj in objs:637 self.delete_service(name=obj.metadata.name, reraise=reraise)638 def delete_deployments(self, reraise=False, **kwargs):639 objs = self.list_deployments(reraise=reraise, **kwargs)640 for obj in objs:641 self.delete_deployment(name=obj.metadata.name, reraise=reraise)642 def delete_ingresses(self, reraise=False, **kwargs):643 objs = self.list_services(reraise=reraise, **kwargs)644 for obj in objs:645 self.delete_ingress(name=obj.metadata.name, reraise=reraise)646 def is_pod_running(self, pod_id: str, container_id: str):647 event = self.k8s_api.read_namespaced_pod_status(pod_id, self.namespace)...
test_raise.py
Source:test_raise.py
...16 return self17 def __exit__(self, exc_type, exc_value, exc_tb):18 return True19class TestRaise(unittest.TestCase):20 def test_invalid_reraise(self):21 try:22 raise23 except RuntimeError as e:24 self.assertIn("No active exception", str(e))25 else:26 self.fail("No exception raised")27 def test_reraise(self):28 try:29 try:30 raise IndexError()31 except IndexError as e:32 exc1 = e33 raise34 except IndexError as exc2:35 self.assertTrue(exc1 is exc2)36 else:37 self.fail("No exception raised")38 def test_except_reraise(self):39 def reraise():40 try:41 raise TypeError("foo")42 except:43 try:44 raise KeyError("caught")45 except KeyError:46 pass47 raise48 self.assertRaises(TypeError, reraise)49 def test_finally_reraise(self):50 def reraise():51 try:52 raise TypeError("foo")53 except:54 try:55 raise KeyError("caught")56 finally:57 raise58 self.assertRaises(KeyError, reraise)59 def test_nested_reraise(self):60 def nested_reraise():61 raise62 def reraise():63 try:64 raise TypeError("foo")65 except:66 nested_reraise()67 self.assertRaises(TypeError, reraise)68 def test_raise_from_None(self):69 try:70 try:71 raise TypeError("foo")72 except:73 raise ValueError() from None74 except ValueError as e:75 self.assertTrue(isinstance(e.__context__, TypeError))76 self.assertIsNone(e.__cause__)77 def test_with_reraise1(self):78 def reraise():79 try:80 raise TypeError("foo")81 except:82 with Context():83 pass84 raise85 self.assertRaises(TypeError, reraise)86 def test_with_reraise2(self):87 def reraise():88 try:89 raise TypeError("foo")90 except:91 with Context():92 raise KeyError("caught")93 raise94 self.assertRaises(TypeError, reraise)95 def test_yield_reraise(self):96 def reraise():97 try:98 raise TypeError("foo")99 except:100 yield 1101 raise102 g = reraise()103 next(g)104 self.assertRaises(TypeError, lambda: next(g))105 self.assertRaises(StopIteration, lambda: next(g))106 def test_erroneous_exception(self):107 class MyException(Exception):108 def __init__(self):109 raise RuntimeError()110 try:111 raise MyException112 except RuntimeError:113 pass114 else:115 self.fail("No exception raised")116 def test_new_returns_invalid_instance(self):...
async_manager.py
Source:async_manager.py
1#!/usr/bin/python2#3# Copyright 2018-2021 Polyaxon, Inc.4#5# Licensed under the Apache License, Version 2.0 (the "License");6# you may not use this file except in compliance with the License.7# You may obtain a copy of the License at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS,13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14# See the License for the specific language governing permissions and15# limitations under the License.=16from kubernetes_asyncio import client, config17from kubernetes_asyncio.client import Configuration18from kubernetes_asyncio.client.rest import ApiException19from polyaxon.exceptions import PolyaxonK8SError20from polyaxon.k8s.monitor import is_pod_running21from polyaxon.logger import logger22class AsyncK8SManager:23 def __init__(self, namespace="default", in_cluster=False):24 self.namespace = namespace25 self.in_cluster = in_cluster26 self.api_client = None27 self.k8s_api = None28 self.k8s_batch_api = None29 self.k8s_beta_api = None30 self.k8s_custom_object_api = None31 self.k8s_version_api = None32 @staticmethod33 def get_managed_by_polyaxon(instance: str):34 return "app.kubernetes.io/instance={},app.kubernetes.io/managed-by=polyaxon".format(35 instance36 )37 @staticmethod38 async def load_config(in_cluster=False, k8s_config=None):39 if not k8s_config:40 if in_cluster:41 config.load_incluster_config()42 else:43 await config.load_kube_config()44 return Configuration.get_default_copy()45 @classmethod46 def get_config_auth(cls, k8s_config=None):47 additional_headers = k8s_config.api_key or {}48 return additional_headers.get("authorization", "").strip("bearer").strip()49 async def setup(self, k8s_config=None):50 if not k8s_config:51 if self.in_cluster:52 config.load_incluster_config()53 else:54 await config.load_kube_config()55 self.api_client = client.api_client.ApiClient()56 else:57 self.api_client = client.api_client.ApiClient(configuration=k8s_config)58 self.k8s_api = client.CoreV1Api(self.api_client)59 self.k8s_batch_api = client.BatchV1Api(self.api_client)60 self.k8s_beta_api = client.ExtensionsV1beta1Api(self.api_client)61 self.k8s_custom_object_api = client.CustomObjectsApi(self.api_client)62 self.k8s_version_api = client.VersionApi(self.api_client)63 async def close(self):64 if self.api_client:65 await self.api_client.close()66 async def set_namespace(self, namespace):67 self.namespace = namespace68 async def get_pod(self, name, reraise=False):69 try:70 return await self.k8s_api.read_namespaced_pod(71 name=name, namespace=self.namespace72 )73 except ApiException as e:74 if reraise:75 raise PolyaxonK8SError("Connection error: %s" % e) from e76 return None77 async def is_pod_running(self, pod_id: str, container_id: str):78 event = await self.k8s_api.read_namespaced_pod_status(pod_id, self.namespace)79 return is_pod_running(event, container_id)80 async def _list_namespace_resource(self, resource_api, reraise=False, **kwargs):81 try:82 res = await resource_api(namespace=self.namespace, **kwargs)83 return [p for p in res.items]84 except ApiException as e:85 logger.error("K8S error: {}".format(e))86 if reraise:87 raise PolyaxonK8SError("Connection error: %s" % e) from e88 return []89 async def list_pods(self, reraise=False, **kwargs):90 return await self._list_namespace_resource(91 resource_api=self.k8s_api.list_namespaced_pod,92 reraise=reraise,93 **kwargs,94 )95 async def list_jobs(self, reraise=False, **kwargs):96 return await self._list_namespace_resource(97 resource_api=self.k8s_batch_api.list_namespaced_job,98 reraise=reraise,99 **kwargs,100 )101 async def list_custom_objects(102 self, group, version, plural, reraise=False, **kwargs103 ):104 return await self._list_namespace_resource(105 resource_api=self.k8s_custom_object_api.list_namespaced_custom_object,106 reraise=reraise,107 group=group,108 version=version,109 plural=plural,110 **kwargs,111 )112 async def list_services(self, reraise=False, **kwargs):113 return await self._list_namespace_resource(114 resource_api=self.k8s_api.list_namespaced_service,115 reraise=reraise,116 **kwargs,117 )118 async def list_deployments(self, reraise=False, **kwargs):119 return await self._list_namespace_resource(120 resource_api=self.k8s_beta_api.list_namespaced_deployment,121 reraise=reraise,122 **kwargs,123 )124 async def create_custom_object(self, name, group, version, plural, body):125 resp = await self.k8s_custom_object_api.create_namespaced_custom_object(126 group=group,127 version=version,128 plural=plural,129 namespace=self.namespace,130 body=body,131 )132 logger.debug("Custom object `{}` was created".format(name))133 return resp134 async def update_custom_object(self, name, group, version, plural, body):135 resp = await self.k8s_custom_object_api.patch_namespaced_custom_object(136 name=name,137 group=group,138 version=version,139 plural=plural,140 namespace=self.namespace,141 body=body,142 )143 logger.debug("Custom object `{}` was patched".format(name))144 return resp145 async def create_or_update_custom_object(146 self, name, group, version, plural, body, reraise=False147 ):148 try:149 create = await self.create_custom_object(150 name=name, group=group, version=version, plural=plural, body=body151 )152 return create, True153 except ApiException as e_create:154 try:155 update = await self.update_custom_object(156 name=name, group=group, version=version, plural=plural, body=body157 )158 return update, False159 except ApiException as e:160 if reraise:161 raise PolyaxonK8SError(162 "Connection error: creation %s - update %s" % (e_create, e)163 ) from e164 else:165 logger.error("K8S error: {}".format(e))166 async def get_custom_object(self, name, group, version, plural, reraise=False):167 try:168 return await self.k8s_custom_object_api.get_namespaced_custom_object(169 name=name,170 group=group,171 version=version,172 plural=plural,173 namespace=self.namespace,174 )175 except ApiException as e:176 if reraise:177 raise PolyaxonK8SError("Connection error: %s" % e) from e178 return None179 async def delete_custom_object(self, name, group, version, plural, reraise=False):180 try:181 await self.k8s_custom_object_api.delete_namespaced_custom_object(182 name=name,183 group=group,184 version=version,185 plural=plural,186 namespace=self.namespace,187 body=client.V1DeleteOptions(),188 )189 logger.debug("Custom object `{}` deleted".format(name))190 except ApiException as e:191 if reraise:192 raise PolyaxonK8SError("Connection error: %s" % e) from e193 else:...
excutils.py
Source:excutils.py
...35 for those circumstances this context provides a reraise flag that36 can be used to suppress the exception. For example::37 except Exception:38 with save_and_reraise_exception() as ctxt:39 decide_if_need_reraise()40 if not should_be_reraised:41 ctxt.reraise = False42 If another exception occurs and reraise flag is False,43 the saved exception will not be logged.44 If the caller wants to raise new exception during exception handling45 he/she sets reraise to False initially with an ability to set it back to46 True if needed::47 except Exception:48 with save_and_reraise_exception(reraise=False) as ctxt:49 [if statements to determine whether to raise a new exception]50 # Not raising a new exception, so reraise51 ctxt.reraise = True52 """53 def __init__(self, reraise=True):54 self.reraise = reraise55 def __enter__(self):56 self.type_, self.value, self.tb, = sys.exc_info()57 return self58 def __exit__(self, exc_type, exc_val, exc_tb):59 if exc_type is not None:60 if self.reraise:61 logging.error(_LE('Original exception being dropped: %s'),62 traceback.format_exception(self.type_,63 self.value,64 self.tb))65 return False66 if self.reraise:67 six.reraise(self.type_, self.value, self.tb)68def forever_retry_uncaught_exceptions(infunc):69 def inner_func(*args, **kwargs):70 last_log_time = 071 last_exc_message = None72 exc_count = 073 while True:74 try:75 return infunc(*args, **kwargs)76 except Exception as exc:77 this_exc_message = six.u(str(exc))78 if this_exc_message == last_exc_message:79 exc_count += 180 else:81 exc_count = 1...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Slash automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
