#!/usr/bin/python DOCUMENTATION = r''' --- module: decort_user_info description: See L(Module Documentation,https://repository.basistech.ru/BASIS/decort-ansible/wiki/Home). # noqa: E501 ''' from ansible.module_utils.basic import AnsibleModule from ansible.module_utils.decort_utils import DecortController class DecortUserInfo(DecortController): 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( accounts=dict( type='dict', options=dict( deleted=dict( type='bool', default=False, ), filter=dict( type='dict', options=dict( rights=dict( type='str', choices=[ e.value for e in self.AccountUserRights ], ), id=dict( type='int', ), name=dict( type='str', ), status=dict( type='str', choices=[ e.value for e in self.AccountStatus ], ), ), ), pagination=dict( type='dict', options=dict( number=dict( type='int', default=1, ), size=dict( type='int', required=True, ), ), ), resource_consumption=dict( type='bool', default=False, ), sorting=dict( type='dict', options=dict( asc=dict( type='bool', default=True, ), field=dict( type='str', choices=[ e.value for e in self.AccountSortableField ], required=True, ), ), ), ), ), api_methods=dict( type='bool', default=False, ), audits=dict( type='dict', options=dict( filter=dict( type='dict', options=dict( api_method=dict( type='str', ), status_code=dict( type='dict', options=dict( min=dict( type='int', ), max=dict( type='int', ), ), ), time=dict( type='dict', options=dict( start=dict( type='dict', options=dict( timestamp=dict( type='int', ), datetime=dict( type='str', ), ), mutually_exclusive=[ ('timestamp', 'datetime'), ], ), end=dict( type='dict', options=dict( timestamp=dict( type='int', ), datetime=dict( type='str', ), ), mutually_exclusive=[ ('timestamp', 'datetime'), ], ), ), ), ), ), pagination=dict( type='dict', apply_defaults=True, options=dict( number=dict( type='int', default=1, ), size=dict( type='int', default=50, ), ), ), sorting=dict( type='dict', options=dict( asc=dict( type='bool', default=True, ), field=dict( type='str', choices=[ e.value for e in self.AuditsSortableField ], required=True, ), ), ), ), ), objects_search=dict( type='str', ), resource_consumption=dict( type='bool', default=False, ), zones=dict( type='dict', options=dict( filter=dict( type='dict', options=dict( deletable=dict( type='bool', ), description=dict( type='str', ), grid_id=dict( type='int', ), id=dict( type='int', ), name=dict( type='str', ), node_id=dict( type='int', ), status=dict( type='str', choices=[ e.value for e in self.ZoneStatus ], ), ), ), pagination=dict( type='dict', apply_defaults=True, options=dict( number=dict( type='int', default=1, ), size=dict( type='int', default=50, ), ), ), sorting=dict( type='dict', options=dict( asc=dict( type='bool', default=True, ), field=dict( type='str', choices=self.ZoneField._member_names_, required=True, ), ), ), ), ), trunks=dict( type='dict', options=dict( filter=dict( type='dict', options=dict( ids=dict( type='list', ), account_ids=dict( type='list', ), status=dict( type='str', choices=[ e.value for e in self.TrunkStatus ], ), vlan_ids=dict( type='list', ), ), ), pagination=dict( type='dict', apply_defaults=True, options=dict( number=dict( type='int', default=1, ), size=dict( type='int', default=50, ), ), ), sorting=dict( type='dict', options=dict( asc=dict( type='bool', default=True, ), field=dict( type='str', choices=[ e.value for e in self.TrunksSortableField ], required=True, ), ), ), ), ), ), supports_check_mode=True, ) def check_amodule_args(self): """ Additional validation of Ansible Module arguments. This validation cannot be implemented using Ansible Argument spec. """ check_error = False match self.aparams['audits']: case { 'filter': {'time': {'start': {'datetime': str() as dt_str}}} }: if self.dt_str_to_sec(dt_str=dt_str) is None: self.message(self.MESSAGES.str_not_parsed(string=dt_str)) check_error = True match self.aparams['audits']: case { 'filter': {'time': {'end': {'datetime': str() as dt_str}}} }: if self.dt_str_to_sec(dt_str=dt_str) is None: self.message(self.MESSAGES.str_not_parsed(string=dt_str)) check_error = True aparam_trunks = self.aparams['trunks'] if ( aparam_trunks is not None and aparam_trunks['filter'] is not None and aparam_trunks['filter']['vlan_ids'] is not None ): for vlan_id in aparam_trunks['filter']['vlan_ids']: if not ( self.TRUNK_VLAN_ID_MIN_VALUE <= vlan_id <= self.TRUNK_VLAN_ID_MAX_VALUE ): check_error = True self.message( 'Check for parameter "trunks.filter.vlan_ids" failed: ' f'VLAN ID {vlan_id} must be in range 1-4095.' ) if check_error: self.exit(fail=True) @property def mapped_accounts_args(self) -> None | dict: """ Map the module argument `accounts` to arguments dictionary for the method `DecortController.user_accounts`. """ input_args = self.aparams['accounts'] if not input_args: return input_args mapped_args = {} mapped_args['deleted'] = input_args['deleted'] mapped_args['resource_consumption'] = ( input_args['resource_consumption'] ) input_args_filter = input_args['filter'] if input_args_filter: input_args_filter_rights = input_args_filter['rights'] if input_args_filter_rights: mapped_args['account_user_rights'] = ( self.AccountUserRights(input_args_filter_rights) ) mapped_args['account_id'] = input_args_filter['id'] mapped_args['account_name'] = input_args_filter['name'] input_args_filter_status = input_args_filter['status'] if input_args_filter_status: mapped_args['account_status'] = ( self.AccountStatus(input_args_filter_status) ) input_args_pagination = input_args['pagination'] if input_args_pagination: mapped_args['page_number'] = input_args_pagination['number'] mapped_args['page_size'] = input_args_pagination['size'] input_args_sorting = input_args['sorting'] if input_args_sorting: mapped_args['sort_by_asc'] = input_args_sorting['asc'] input_args_sorting_field = input_args_sorting['field'] if input_args_sorting_field: mapped_args['sort_by_field'] = ( self.AccountSortableField(input_args_sorting_field) ) return mapped_args @property def mapped_audits_args(self): """ Map the module argument `audits` to arguments dictionary for the method `DecortController.user_audits`. """ input_args = self.aparams['audits'] if not input_args: return input_args mapped_args = {} input_args_filter = input_args['filter'] if input_args_filter: mapped_args['api_method'] = input_args_filter['api_method'] match input_args_filter['status_code']: case {'min': int() as min_status_code}: mapped_args['min_status_code'] = min_status_code match input_args_filter['status_code']: case {'max': int() as max_status_code}: mapped_args['max_status_code'] = max_status_code match input_args_filter['time']: case {'start': {'timestamp': int() as start_unix_time}}: mapped_args['start_unix_time'] = start_unix_time case {'start': {'datetime': str() as start_dt_str}}: mapped_args['start_unix_time'] = self.dt_str_to_sec( dt_str=start_dt_str ) match input_args_filter['time']: case {'end': {'timestamp': int() as end_unix_time}}: mapped_args['end_unix_time'] = end_unix_time case {'end': {'datetime': str() as end_dt_str}}: mapped_args['end_unix_time'] = self.dt_str_to_sec( dt_str=end_dt_str ) input_args_pagination = input_args['pagination'] if input_args_pagination: mapped_args['page_number'] = input_args_pagination['number'] mapped_args['page_size'] = input_args_pagination['size'] input_args_sorting = input_args['sorting'] if input_args_sorting: mapped_args['sort_by_asc'] = input_args_sorting['asc'] input_args_sorting_field = input_args_sorting['field'] if input_args_sorting_field: mapped_args['sort_by_field'] = ( self.AuditsSortableField(input_args_sorting_field) ) return mapped_args @property def mapped_zones_args(self): """ Map the module argument `zones` to arguments dictionary for the method `DecortController.user_zones`. """ input_args = self.aparams['zones'] if not input_args: return input_args mapped_args = {} input_args_filter = input_args['filter'] if input_args_filter: mapped_args.update(input_args_filter) input_args_filter_status = input_args_filter['status'] if input_args_filter_status: mapped_args['status'] = ( self.ZoneStatus(input_args_filter_status) ) input_args_pagination = input_args['pagination'] if input_args_pagination: mapped_args['page_number'] = input_args_pagination['number'] mapped_args['page_size'] = input_args_pagination['size'] input_args_sorting = input_args['sorting'] if input_args_sorting: mapped_args['sort_by_asc'] = input_args_sorting['asc'] input_args_sorting_field = input_args_sorting['field'] if input_args_sorting_field: mapped_args['sort_by_field'] = ( self.ZoneField._member_map_[input_args_sorting_field] ) return mapped_args @property def mapped_trunks_args(self): """ Map the module argument `trunks` to arguments dictionary for the method `DecortController.user_trunks`. """ input_args = self.aparams['trunks'] if not input_args: return input_args mapped_args = {} input_args_filter = input_args['filter'] if input_args_filter: mapped_args.update(input_args_filter) input_args_filter_status = input_args_filter['status'] if input_args_filter_status: mapped_args['status'] = ( self.TrunkStatus(input_args_filter_status) ) input_args_filter_vlan_ids = input_args_filter['vlan_ids'] if input_args_filter_vlan_ids is not None: mapped_args['vlan_ids'] = ', '.join( map(str, input_args_filter_vlan_ids) ) input_args_pagination = input_args['pagination'] if input_args_pagination: mapped_args['page_number'] = input_args_pagination['number'] mapped_args['page_size'] = input_args_pagination['size'] input_args_sorting = input_args['sorting'] if input_args_sorting: mapped_args['sort_by_asc'] = input_args_sorting['asc'] input_args_sorting_field = input_args_sorting['field'] if input_args_sorting_field: mapped_args['sort_by_field'] = ( self.TrunksSortableField(input_args_sorting_field) ) return mapped_args def run(self): self.get_info() self.exit() def get_info(self): self.facts = self.user_whoami() self.id = self.facts['name'] user_get = self.user_get(id=self.id) for key in ['emailaddresses', 'data']: self.facts[key] = user_get[key] if self.aparams['accounts']: self.facts['accounts'] = self.user_accounts( **self.mapped_accounts_args, ) if self.aparams['resource_consumption']: self.facts.update(self.user_resource_consumption()) if self.aparams['audits']: self.facts['audits'] = self.user_audits(**self.mapped_audits_args) if self.aparams['api_methods']: self.facts['api_methods'] = self.user_api_methods(id=self.id) search_string = self.aparams['objects_search'] if search_string: self.facts['objects_search'] = self.user_objects_search( search_string=search_string, ) if self.aparams['zones']: self.facts['zones'] = self.user_zones(**self.mapped_zones_args) if self.aparams['trunks']: self.facts['trunks'] = self.user_trunks(**self.mapped_trunks_args) for trunk_facts in self.facts['trunks']: trunk_facts['account_ids'] = trunk_facts.pop('accountIds') trunk_facts['created_timestamp'] = trunk_facts.pop( 'created_at' ) trunk_facts['deleted_timestamp'] = trunk_facts.pop( 'deleted_at' ) trunk_facts['updated_timestamp'] = trunk_facts.pop( 'updated_at' ) trunk_facts['native_vlan_id'] = trunk_facts.pop( 'nativeVlanId' ) trunk_facts['ovs_bridge'] = trunk_facts.pop('ovsBridge') trunk_facts['vlan_ids'] = trunk_facts.pop('trunkTags') def main(): DecortUserInfo().run() if __name__ == '__main__': main()