Source code for idem_vra.states.vra.virtual.securitygroupstate

# Auto-generated virtuale state module

__contracts__ = ["resource"]

TREQ = {
    "present": {"require": ["vra.iaas.cloudaccount.present"]},
    "absent": {"require": []},
}


# ====================================
# State implementation
# ====================================
from idem_vra.helpers.models import StateReturn
from idem_vra.states.vra.iaas.securitygroup import SecuritygroupState


[docs]async def present( hub, ctx, name: str, groupId: str, tags: list[dict[str, str]], **kwargs ): try: # get existing security group existing = await hub.exec.vra.iaas.securitygroup.get_security_group( ctx, groupId ) existing_tags = existing.ret.tags or [] # normalize the passed tags for tag in tags: if "key" not in tag: tag["key"] = "" if "value" not in tag: tag["value"] = "" # add tags that are passed but are not in the existing tags tags_to_add = [tag for tag in tags if tag not in existing_tags] if len(tags_to_add) > 0: # build full list of tags for the update full_tags_definition = existing_tags + tags_to_add # issue the update update_result = await hub.exec.vra.iaas.securitygroup.update_security_group( ctx, groupId, tags=full_tags_definition ) old_resource = dict(groupId=existing.ret.id, tags=existing_tags) new_resource = dict( groupId=update_result.ret.id, tags=update_result.ret.tags ) return StateReturn( result=True, comment=f"Resource state updated.", old=old_resource, new=new_resource, ) else: resource = dict(groupId=existing.ret.id, tags=existing_tags) return StateReturn( result=True, comment=f"Resource state is already up-to-date.", old=resource, new=resource, ) except Exception as error: hub.log.error("Error during enforcing present state: securitygroupstate") hub.log.error(str(error)) raise error
[docs]async def absent( hub, ctx, name: str, groupId: str, tags: list[dict[str, str]], **kwargs ): try: # get existing security group existing = await hub.exec.vra.iaas.securitygroup.get_security_group( ctx, groupId ) existing_tags = existing.ret.tags or [] # normalize the passed tags for tag in tags: if "key" not in tag: tag["key"] = "" if "value" not in tag: tag["value"] = "" # remove tags that are not passed but are in the existing tags tags_to_remove = [tag for tag in existing_tags if tag in tags] if len(tags_to_remove) > 0: # build full list of tags for the update full_tags_definition = [ tag for tag in existing_tags if tag not in tags_to_remove ] # issue the update update_result = await hub.exec.vra.iaas.securitygroup.update_security_group( ctx, groupId, tags=full_tags_definition ) old_resource = dict(groupId=existing.ret.id, tags=existing_tags) new_resource = dict( groupId=update_result.ret.id, tags=update_result.ret.tags ) return StateReturn( result=True, comment=f"Resource state updated.", old=old_resource, new=new_resource, ) else: resource = dict(groupId=existing.ret.id, tags=existing_tags) return StateReturn( result=True, comment=f"Resource state is already up-to-date.", old=resource, new=resource, ) except Exception as error: hub.log.error("Error during enforcing absent state: securitygroupstate") hub.log.error(str(error)) raise error
[docs]async def describe(hub, ctx): try: result = {} # Leverage generated pagination to retrieve all available security groups state = SecuritygroupState(hub, ctx) res = await state.paginate_find(hub, ctx) for obj in res.get("ret", {}).get("content", []): # Keep track of name and id properties as they may get remapped obj_name = obj.get("name", "unknown") obj_id = obj.get("id", "unknown") # Build result result[f"virtual-{obj_name}-{obj_id.split('-')[-1]}"] = { "vra.virtual.securitygroupstate.present": [ {"name": obj_name}, {"groupId": obj_id}, {"tags": obj.get("tags", []) or []}, ] } return result except Exception as error: hub.log.error("Error during describe: securitygroupstate") hub.log.error(str(error)) raise error
[docs]def is_pending(hub, ret: dict, state: str = None, **pending_kwargs): return False