#!/usr/bin/python DOCUMENTATION = r''' --- module: decort_account description: See L(Module Documentation,https://repository.basistech.ru/BASIS/decort-ansible/wiki/Home). # noqa: E501 ''' from typing import Any, Iterable from ansible.module_utils.basic import AnsibleModule from ansible.module_utils.decort_utils import DecortController, sdk_types class DecortAccount(DecortController): OBJ = 'account' def __init__(self): super().__init__(AnsibleModule(**self.amodule_init_args)) self.check_amodule_args() @property def amodule_init_args(self) -> dict: return self.pack_amodule_init_args( argument_spec=dict( send_access_emails=dict( type='bool', ), acl=dict( type='dict', options=dict( mode=dict( type='str', choices=[ 'match', 'revoke', 'update', ], default='update', ), users=dict( type='list', required=True, elements='dict', options=dict( rights=dict( type='str', choices=( sdk_types.AccessTypeForSet ._member_names_ ), default=( sdk_types.AccessTypeForSet.R.name ), ), id=dict( type='str', required=True, ), ), ), ), ), id=dict( type='int', ), name=dict( type='str', ), get_resource_consumption=dict( type='bool', default=False, ), quotas=dict( type='dict', options=dict( cpu_count=dict( type='int', ), storage_size_gb=dict( type='int', ), gpu_count=dict( type='int', ), ext_ip_count=dict( type='int', ), ram_size_mb=dict( type='int', ), ), ), state=dict( type='str', choices=[ 'absent', 'absent_permanently', 'confirmed', 'disabled', 'present', ], ), sep_pools=dict( type='list', elements='dict', options=dict( sep_id=dict( type='int', required=True, ), pool_names=dict( type='list', required=True, elements='str', ), ), ), description=dict( type='str', ), default_zone_id=dict( type='int', ), ), required_one_of=[ ('id', 'name') ], supports_check_mode=True, ) def check_amodule_args(self): """ Additional Ansible Module arguments validation that cannot be implemented using Ansible Argument spec. """ arg_state = self.aparams['state'] if arg_state is not None and 'absent' in arg_state: # Parameters or combinations of parameters that can # cause changing the object. changing_params = [ 'send_access_emails', 'acl', ['id', 'name'], 'quotas', ] check_error = False for elem in changing_params: if isinstance(elem, str): param = elem if self.aparams[elem] is not None: self.message( f'If the parameter "state" is set to' f' "{arg_state}", then using the parameter' f' "{param}" is not allowed.' ) check_error = True elif isinstance(elem, Iterable): params = elem params_using = map( lambda x: self.aparams[x] is not None, params ) if all(params_using): params_str = ', '.join(f'"{p}"' for p in params) self.message( f'If the parameter "state" is set to' f' "{arg_state}", then using the combination' f' of parameters {params_str} are not allowed.' ) check_error = True if check_error: self.exit(fail=True) def check_amodule_args_for_change(self): check_error = False if self.check_aparam_default_zone_id() is False: check_error = True if check_error: self.exit(fail=True) @DecortController.handle_sdk_exceptions def run(self): self.get_info() self.check_amodule_args_for_change() self.change() self.exit() def get_info(self): # If this is the first getting info if self._acc_info is None: self.acc_id, self._acc_info = self.account_find( account_name=self.aparams['name'], account_id=self.aparams['id'], ) # If this is a repeated getting info else: # If check mode is enabled, there is no needed to # request info again if not self.amodule.check_mode: self.acc_id, self._acc_info = self.account_find( account_id=self._acc_info.id, ) if self._acc_info is not None: acc_info_dict = self.acc_info.model_dump() if self.aparams['get_resource_consumption']: resource_consumption_dict = ( self.api.cloudapi.account.get_resource_consumption( account_id=self.acc_info.id, ).model_dump() ) resource_consumption_dict.pop('id') resource_consumption_dict.pop('quotas') acc_info_dict['resource_consumption'] = ( resource_consumption_dict ) self.facts = acc_info_dict def change(self): self.change_state() self.change_acl() need_account_update_api_call: bool = False sdk_param_send_access_emails = None aparam_send_access_emails: bool | None = self.aparams[ 'send_access_emails' ] if ( aparam_send_access_emails is not None and self.acc_info.send_access_emails != aparam_send_access_emails ): sdk_param_send_access_emails = aparam_send_access_emails need_account_update_api_call = True sdk_param_name = None aparam_name: str | None = self.aparams['name'] if ( self.aparams['id'] and aparam_name and self.acc_info.name != aparam_name ): sdk_param_name = aparam_name need_account_update_api_call = True sdk_param_cpu_count_quota = None sdk_param_storage_size_quota_gb = None sdk_param_gpu_count_quota = None sdk_param_ext_ip_count_quota = None sdk_param_ram_size_quota_mb = None aparam_quotas: dict[str, Any] = self.aparams['quotas'] if aparam_quotas: aparam_quotas_cpu_count: int | None = aparam_quotas['cpu_count'] if ( aparam_quotas_cpu_count is not None and self.acc_info.quotas.cpu_count != aparam_quotas_cpu_count ): sdk_param_cpu_count_quota = aparam_quotas_cpu_count need_account_update_api_call = True aparam_quotas_storage_size_gb: int | None = aparam_quotas[ 'storage_size_gb' ] if ( aparam_quotas_storage_size_gb is not None and ( self.acc_info.quotas.storage_size_gb != aparam_quotas_storage_size_gb ) ): sdk_param_storage_size_quota_gb = aparam_quotas_storage_size_gb need_account_update_api_call = True aparam_quotas_gpu_count: int | None = aparam_quotas['gpu_count'] if ( aparam_quotas_gpu_count is not None and self.acc_info.quotas.gpu_count != aparam_quotas_gpu_count ): sdk_param_gpu_count_quota = aparam_quotas_gpu_count need_account_update_api_call = True aparam_quotas_ext_ip_count: int | None = aparam_quotas[ 'ext_ip_count' ] if ( aparam_quotas_ext_ip_count is not None and ( self.acc_info.quotas.ext_ip_count != aparam_quotas_ext_ip_count ) ): sdk_param_ext_ip_count_quota = aparam_quotas_ext_ip_count need_account_update_api_call = True aparam_quotas_ram_size_mb: int | None = aparam_quotas[ 'ram_size_mb' ] if ( aparam_quotas_ram_size_mb is not None and ( self.acc_info.quotas.ram_size_mb != aparam_quotas_ram_size_mb ) ): sdk_param_ram_size_quota_mb = aparam_quotas_ram_size_mb need_account_update_api_call = True sdk_param_sep_pools = None aparam_sep_pools: list[dict[str, Any]] = self.aparams['sep_pools'] if aparam_sep_pools is not None: sep_pools: set[str] = set() for sep in aparam_sep_pools: sep_pool_names: list[str] = sep['pool_names'] for pool_name in sep_pool_names: sep_pools.add(f'{sep["sep_id"]}_{pool_name}') if set(self.acc_info.sep_pools) != sep_pools: sdk_param_sep_pools = list(sep_pools) need_account_update_api_call = True sdk_param_description = None aparam_desc: str | None = self.aparams['description'] if ( aparam_desc is not None and self.acc_info.description != aparam_desc ): sdk_param_description = aparam_desc need_account_update_api_call = True sdk_param_default_zone_id = None aparam_default_zone_id: int | None = self.aparams['default_zone_id'] if ( aparam_default_zone_id is not None and self.acc_info.default_zone_id != aparam_default_zone_id ): sdk_param_default_zone_id = aparam_default_zone_id need_account_update_api_call = True if need_account_update_api_call: OBJ = 'account' self.sdk_checkmode(self.api.ca.account.update)( account_id=self.acc_info.id, cpu_count_quota=sdk_param_cpu_count_quota, gpu_count_quota=sdk_param_gpu_count_quota, name=sdk_param_name, ext_ip_count_quota=sdk_param_ext_ip_count_quota, ram_size_quota_mb=sdk_param_ram_size_quota_mb, send_access_emails=sdk_param_send_access_emails, storage_size_quota_gb=sdk_param_storage_size_quota_gb, sep_pools=sdk_param_sep_pools, description=sdk_param_description, default_zone_id=sdk_param_default_zone_id, ) if sdk_param_send_access_emails is not None: smth = 'sending access emails' if sdk_param_send_access_emails: self.message( self.MESSAGES.obj_smth_enabled( obj=OBJ, id=self.acc_info.id, smth=smth, ) ) else: self.message( self.MESSAGES.obj_smth_disabled( obj=OBJ, id=self.acc_info.id, smth=smth, ) ) if sdk_param_name is not None: self.message( self.MESSAGES.obj_renamed( obj=OBJ, id=self.acc_info.id, new_name=sdk_param_name, ) ) quotas = { 'CPU count quota': sdk_param_cpu_count_quota, 'storage size quota GB': sdk_param_storage_size_quota_gb, 'GPU count quota': sdk_param_gpu_count_quota, 'ext IP count quota': sdk_param_ext_ip_count_quota, 'RAM size quota MB': sdk_param_ram_size_quota_mb, } for q_name, q_value in quotas.items(): if q_value is not None: self.message( self.MESSAGES.obj_smth_changed( obj=OBJ, id=self.acc_info.id, smth=q_name, new_value=q_value ) ) if sdk_param_default_zone_id is not None: self.message( self.MESSAGES.obj_smth_changed( obj=OBJ, id=self.acc_info.id, smth='default_zone_id', new_value=sdk_param_default_zone_id, ) ) self.get_info() def change_state(self): if self._acc_info is None: self.message(self.MESSAGES.obj_not_found(obj=self.OBJ)) match self.aparams: case {'state': 'absent' | 'absent_permanently'}: pass case {'state': 'confirmed' | 'disabled' | 'present'}: self.exit(fail=True) elif self._acc_info.status == sdk_types.AccountStatus.DESTROYED: match self.aparams: case {'state': 'absent' | 'absent_permanently'}: self.message( self.MESSAGES.obj_deleted( obj=self.OBJ, id=self.acc_info.id, permanently=True, already=True, ) ) case {'state': 'confirmed' | 'disabled' | 'present'}: self.message( self.MESSAGES.obj_not_restored(obj=self.OBJ, id=self.acc_info.id) ) self.exit(fail=True) elif self._acc_info.status == sdk_types.AccountStatus.DELETED: match self.aparams: case {'state': 'absent'}: self.message( self.MESSAGES.obj_deleted( obj=self.OBJ, id=self.acc_info.id, permanently=False, already=True, ) ) case {'state': 'absent_permanently'}: self.delete(permanently=True) case {'state': 'confirmed' | 'present'}: self.restore() case {'state': 'disabled'}: self.restore() self.disable() elif self._acc_info.status == sdk_types.AccountStatus.CONFIRMED: match self.aparams: case {'state': 'absent'}: self.delete() case {'state': 'absent_permanently'}: self.delete(permanently=True) case {'state': 'confirmed' | 'present'}: pass case {'state': 'disabled'}: self.disable() elif self._acc_info.status == sdk_types.AccountStatus.DISABLED: match self.aparams: case {'state': 'absent'}: self.delete() case {'state': 'absent_permanently'}: self.delete(permanently=True) case {'state': 'confirmed'}: self.enable() case {'state': 'present' | 'disabled'}: pass def delete(self, permanently=False): self.account_delete( account_id=self.acc_info.id, permanently=permanently ) self.get_info() def disable(self): self.sdk_checkmode(self.api.ca.account.disable)( account_id=self.acc_info.id ) self.get_info() def enable(self): self.sdk_checkmode(self.api.ca.account.enable)( account_id=self.acc_info.id ) self.get_info() def restore(self): self.account_restore(account_id=self.acc_info.id) self.get_info() def change_acl(self): if not self.aparams['acl']: return actual_users = { u.user_name: u.access_type for u in self.acc_info.acl } actual_users_ids = set(actual_users.keys()) aparams_acl = self.aparams['acl'] aparams_users = {u['id']: u['rights'] for u in aparams_acl['users']} aparams_users_ids = set(aparams_users.keys()) del_users_ids = None upd_users = None new_users = None match aparams_acl: case {'mode': 'revoke'}: del_users_ids = aparams_users_ids.intersection( actual_users_ids, ) case {'mode': 'update' | 'match' as mode}: new_users_ids = aparams_users_ids.difference( actual_users_ids, ) new_users = dict( u for u in aparams_users.items() if u[0] in new_users_ids ) upd_users_ids =\ aparams_users_ids.intersection(actual_users_ids) upd_users = dict() for id in upd_users_ids: if actual_users[id] == sdk_types.AccessType.CXDRAU: actual_user_rights = sdk_types.AccessTypeForSet.ARCXDU else: actual_user_rights = actual_users[id] if actual_user_rights != aparams_users[id]: upd_users[id] = aparams_users[id] if mode == 'match': del_users_ids =\ actual_users_ids.difference(aparams_users_ids) if del_users_ids or new_users or upd_users: self.account_change_acl(account_id=self.acc_info.id, del_users=del_users_ids, add_users=new_users, upd_users=upd_users) self.get_info() def check_aparam_default_zone_id(self) -> bool | None: aparam_default_zone_id = self.aparams['default_zone_id'] if aparam_default_zone_id is not None: if aparam_default_zone_id in self.acc_zone_ids: return True else: self.message( 'Check for parameter "default_zone_id" failed: ' f'zone ID {aparam_default_zone_id} not available ' f'for account ID {self.acc_id}.' ) return False def main(): DecortAccount().run() if __name__ == '__main__': main()