Best Python code snippet using tempest_python
test_floating_ips_rbac.py
Source:test_floating_ips_rbac.py  
...104        """105        floating_ip = self._create_floatingip()106        self.rbac_utils.switch_role(self, toggle_rbac_role=True)107        # Delete the floating IP...floatingip.py
Source:floatingip.py  
1import kopf2import pykube3from open4k import utils4from open4k import kube5from open4k import client6from open4k import settings7from open4k import hooks8LOG = utils.get_logger(__name__)9kopf_on_args = ["open4k.amadev.ru", "v1alpha1", "floatingips"]10class FloatingIP(pykube.objects.NamespacedAPIObject, kube.HelmBundleMixin):11    version = "open4k.amadev.ru/v1alpha1"12    endpoint = "floatingips"13    kind = "FloatingIP"14    api = {15        "service": "network",16        "objects": "floatingips",17        "object": "floatingip",18        "get_": "get_floatingip",19        "list": "list_floatingips",20        "create": "create_floatingip",21        "delete": "delete_floatingip",22    }23    @staticmethod24    def get_os_obj(c, obj_id, id_name=None):25        if not id_name:26            id_name = "floatingip_id"27        os_obj = getattr(getattr(c, "floatingips"), "get_floatingip")(28            **{id_name: obj_id}29        )30        if {31            "service": "network",32            "objects": "floatingips",33            "object": "floatingip",34            "get_": "get_floatingip",35            "list": "list_floatingips",36            "create": "create_floatingip",37            "delete": "delete_floatingip",38        }.get("object_envelope", True):39            os_obj = os_obj[list(os_obj)[0]]40        return os_obj41    def create_os_obj(c, body):42        os_obj = c.floatingips.create_floatingip(floatingip=body)43        if {44            "service": "network",45            "objects": "floatingips",46            "object": "floatingip",47            "get_": "get_floatingip",48            "list": "list_floatingips",49            "create": "create_floatingip",50            "delete": "delete_floatingip",51        }.get("object_envelope", True):52            os_obj = os_obj[list(os_obj)[0]]53        return os_obj54    def delete_os_obj(c, obj_id):55        getattr(getattr(c, "floatingips"), "delete_floatingip")(56            floatingip_id=obj_id57        )58@kopf.on.create(*kopf_on_args)59@kopf.on.update(*kopf_on_args)60@kopf.on.resume(*kopf_on_args)61async def floatingip_change_handler(body, name, namespace, **kwargs):62    LOG.info(f"Got FloatingIP change event {name}")63    if body["spec"].get("managed") == False:64        LOG.info(f"{name} is not managed")65        return66    c = client.get_client(67        settings.OPEN4K_NAMESPACE, body["spec"]["cloud"], "network"68    )69    obj = kube.find(FloatingIP, name, namespace=namespace)70    klass = FloatingIP71    if body.get("status", {}).get("applied") == True:72        LOG.info(f"{name} exists, updating ...")73        obj_id = body["status"]["object"].get("id")74        id_name = None75        if not obj_id:76            id_name = "uuid"77            obj_id = body["status"]["object"].get("uuid")78        os_obj = klass.get_os_obj(c, obj_id, id_name)79        obj.patch(80            {"status": {"object": os_obj}},81            subresource="status",82        )83        return84    try:85        os_obj = klass.create_os_obj(c, body["spec"]["body"])86    except Exception as e:87        obj.patch(88            {"status": {"applied": False, "error": str(e)}},89            subresource="status",90        )91        raise92    obj.patch(93        {"status": {"applied": True, "error": "", "object": os_obj}},94        subresource="status",95    )96    await hooks.call("floatingip", "post_create", c, klass, obj, os_obj)97@kopf.on.delete(*kopf_on_args)98async def floatingip_delete_handler(body, name, namespace, **kwargs):99    LOG.info(f"Got FloatingIP delete event {name}")100    if body["spec"].get("managed") == False:101        LOG.info(f"{name} is not managed")102        return103    if not body.get("status", {}).get("applied"):104        LOG.info(f"{name} was not applied successfully")105        return106    klass = FloatingIP107    os_obj_id = body["status"].get("object", {}).get("id")108    if not os_obj_id:109        LOG.info(f"Cannot get id for {name}")110        return111    c = client.get_client(112        settings.OPEN4K_NAMESPACE, body["spec"]["cloud"], "network"113    )...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
