| 3767 | |
| 3768 | |
| 3769 | class UserReportCommand(EnterpriseCommand): |
| 3770 | def __init__(self): |
| 3771 | super(UserReportCommand, self).__init__() |
| 3772 | self.nodes = {} |
| 3773 | self.roles = {} |
| 3774 | self.teams = {} |
| 3775 | self.users = {} |
| 3776 | self.user_roles = {} |
| 3777 | self.user_teams = {} |
| 3778 | |
| 3779 | def get_parser(self): |
| 3780 | return user_report_parser |
| 3781 | |
| 3782 | def execute(self, params, **kwargs): |
| 3783 | self.nodes.clear() |
| 3784 | if 'nodes' in params.enterprise: |
| 3785 | for node in params.enterprise['nodes']: |
| 3786 | self.nodes[node['node_id']] = { |
| 3787 | 'parent_id': node.get('parent_id') or 0, |
| 3788 | 'name': node['data'].get('displayname') or '' |
| 3789 | } |
| 3790 | |
| 3791 | self.roles.clear() |
| 3792 | if 'roles' in params.enterprise: |
| 3793 | for role in params.enterprise['roles']: |
| 3794 | if 'data' in role: |
| 3795 | self.roles[role['role_id']] = role['data'].get('displayname') or '' |
| 3796 | |
| 3797 | self.teams.clear() |
| 3798 | if 'teams' in params.enterprise: |
| 3799 | for team in params.enterprise['teams']: |
| 3800 | self.teams[team['team_uid']] = team['name'] |
| 3801 | |
| 3802 | self.users.clear() |
| 3803 | kw_euid = 'enterprise_user_id' |
| 3804 | ei_cmd = EnterpriseInfoCommand() |
| 3805 | cmd_output = ei_cmd.execute(params, users=True, columns='roles,teams', format='json', quiet=True) |
| 3806 | enterprise_users = json.loads(cmd_output) |
| 3807 | |
| 3808 | def get_user_info(user): |
| 3809 | u = { |
| 3810 | 'enterprise_user_id': user.get('enterprise_user_id'), |
| 3811 | 'node_id': user.get('node_id'), |
| 3812 | 'username': user.get('username'), |
| 3813 | 'name': user.get('data', {}).get('displayname') or '', |
| 3814 | 'status': user.get('status'), |
| 3815 | 'lock': user.get('lock') |
| 3816 | } |
| 3817 | if 'account_share_expiration' in user: |
| 3818 | u['account_share_expiration'] = user['account_share_expiration'] |
| 3819 | return u |
| 3820 | |
| 3821 | self.user_teams = {u.get('user_id'): u.get('teams') for u in enterprise_users} |
| 3822 | self.user_roles = {u.get('user_id'): u.get('roles') for u in enterprise_users} |
| 3823 | euids = list(self.user_roles.keys()) |
| 3824 | users_cache_filtered = {u.get(kw_euid): u for u in params.enterprise.get('users', []) if u.get(kw_euid) in euids} |
| 3825 | self.users = {k: get_user_info(u) for k, u in users_cache_filtered.items()} |
| 3826 |
no outgoing calls
no test coverage detected