# 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