import sys from typing import Dict import click from zope import component from ..term_utils import get_terms_for_new_user, get_terms_for_renewal from ..utils import http_post, http_get, http_patch, http_delete, \ get_failed_operations, user_dict_lines, get_adduser_operations from .utils import handle_stream_response, handle_sync_response, print_lines, \ check_if_in_development from ceo_common.interfaces import IConfig from ceod.transactions.members import DeleteMemberTransaction @click.group(short_help='Perform operations on CSC members and club reps') def members(): pass @members.command(short_help='Add a new member or club rep') @click.argument('username') @click.option('--cn', help='Full name', prompt='Full name') @click.option('--program', required=False, help='Academic program') @click.option('--terms', 'num_terms', type=click.IntRange(1, 100), help='Number of terms to add', default=1) @click.option('--clubrep', is_flag=True, default=False, help='Add non-member terms instead of member terms') @click.option('--forwarding-address', required=False, help=('Forwarding address to set in ~/.forward. ' 'Default is UW address. ' 'Set to the empty string to disable forwarding.')) def add(username, cn, program, num_terms, clubrep, forwarding_address): cfg = component.getUtility(IConfig) uw_domain = cfg.get('uw_domain') terms = get_terms_for_new_user(num_terms) # TODO: get email address from UWLDAP if forwarding_address is None: forwarding_address = username + '@' + uw_domain body = { 'uid': username, 'cn': cn, } if program is not None: body['program'] = program if clubrep: body['non_member_terms'] = terms else: body['terms'] = terms if forwarding_address != '': body['forwarding_addresses'] = [forwarding_address] else: body['forwarding_addresses'] = [] click.echo("The following user will be created:") print_user_lines(body) click.confirm('Do you want to continue?', abort=True) operations = get_adduser_operations(body) resp = http_post('/api/members', json=body) data = handle_stream_response(resp, operations) result = data[-1]['result'] print_user_lines(result) failed_operations = get_failed_operations(data) if 'send_welcome_message' in failed_operations: click.echo(click.style( 'Warning: welcome message was not sent. You now need to manually ' 'send the user their password.', fg='yellow')) def print_user_lines(d: Dict): """Pretty-print a serialized User.""" print_lines(user_dict_lines(d)) @members.command(short_help='Get info about a user') @click.argument('username') def get(username): resp = http_get('/api/members/' + username) result = handle_sync_response(resp) print_user_lines(result) @members.command(short_help="Replace a user's login shell or forwarding addresses") @click.argument('username') @click.option('--login-shell', required=False, help='Login shell') @click.option('--forwarding-addresses', required=False, help=( 'Comma-separated list of forwarding addresses. ' 'Set to the empty string to disable forwarding.' )) def modify(username, login_shell, forwarding_addresses): if login_shell is None and forwarding_addresses is None: click.echo('Nothing to do.') sys.exit() operations = [] body = {} if login_shell is not None: body['login_shell'] = login_shell operations.append('replace_login_shell') click.echo('Login shell will be set to: ' + login_shell) if forwarding_addresses is not None: if forwarding_addresses == '': forwarding_addresses = [] else: forwarding_addresses = forwarding_addresses.split(',') body['forwarding_addresses'] = forwarding_addresses operations.append('replace_forwarding_addresses') prefix = '~/.forward will be set to: ' if len(forwarding_addresses) > 0: click.echo(prefix + forwarding_addresses[0]) for address in forwarding_addresses[1:]: click.echo((' ' * len(prefix)) + address) else: click.echo(prefix) click.confirm('Do you want to continue?', abort=True) resp = http_patch('/api/members/' + username, json=body) handle_stream_response(resp, operations) @members.command(short_help="Renew a member or club rep's membership") @click.argument('username') @click.option('--terms', 'num_terms', type=click.IntRange(1, 100), help='Number of terms to add', prompt='Number of terms') @click.option('--clubrep', is_flag=True, default=False, help='Add non-member terms instead of member terms') def renew(username, num_terms, clubrep): terms = get_terms_for_renewal(username, num_terms, clubrep) if clubrep: body = {'non_member_terms': terms} click.echo('The following non-member terms will be added: ' + ','.join(terms)) else: body = {'terms': terms} click.echo('The following member terms will be added: ' + ','.join(terms)) click.confirm('Do you want to continue?', abort=True) resp = http_post(f'/api/members/{username}/renew', json=body) handle_sync_response(resp) click.echo('Done.') @members.command(short_help="Reset a user's password") @click.argument('username') def pwreset(username): click.confirm(f"Are you sure you want to reset {username}'s password?", abort=True) resp = http_post(f'/api/members/{username}/pwreset') result = handle_sync_response(resp) click.echo('New password: ' + result['password']) @members.command(short_help="Delete a user") @click.argument('username') def delete(username): check_if_in_development() click.confirm(f"Are you sure you want to delete {username}?", abort=True) resp = http_delete(f'/api/members/{username}') handle_stream_response(resp, DeleteMemberTransaction.operations)