add updateprograms CLI

pull/13/head
Max Erenberg 1 year ago
parent 831ebf17aa
commit 51737585bd
  1. 2
      ceo/cli/entrypoint.py
  2. 2
      ceo/cli/groups.py
  3. 7
      ceo/cli/members.py
  4. 35
      ceo/cli/updateprograms.py
  5. 10
      ceod/api/members.py
  6. 19
      tests/ceo/cli/test_members.py
  7. 43
      tests/ceo/cli/test_updatemembers.py

@ -8,6 +8,7 @@ from zope import component
from ..krb_check import krb_check
from .members import members
from .groups import groups
from .updateprograms import updateprograms
from ceo_common.interfaces import IConfig, IHTTPClient
from ceo_common.model import Config, HTTPClient
@ -28,6 +29,7 @@ def cli(ctx):
cli.add_command(members)
cli.add_command(groups)
cli.add_command(updateprograms)
def register_services():

@ -15,7 +15,7 @@ from ceod.transactions.groups import (
)
@click.group()
@click.group(short_help='Perform operations on CSC groups/clubs')
def groups():
pass

@ -15,7 +15,7 @@ from ceod.transactions.members import (
)
@click.group()
@click.group(short_help='Perform operations on CSC members and club reps')
def members():
pass
@ -102,7 +102,10 @@ def print_user_lines(result: Dict):
('is a club', result['is_club']),
]
if 'forwarding_addresses' in result:
lines.append(('forwarding addresses', ','.join(result['forwarding_addresses'])))
if len(result['forwarding_addresses']) != 0:
lines.append(('forwarding addresses', result['forwarding_addresses'][0]))
for address in result['forwarding_addresses'][1:]:
lines.append(('', address))
if 'terms' in result:
lines.append(('terms', ','.join(result['terms'])))
if 'non_member_terms' in result:

@ -0,0 +1,35 @@
import click
from ..utils import http_post
from .utils import handle_sync_response, print_colon_kv
@click.command(short_help="Sync the 'program' attribute with UWLDAP")
@click.option('--dry-run', is_flag=True, default=False)
@click.option('--members', required=False)
def updateprograms(dry_run, members):
body = {}
if dry_run:
body['dry_run'] = True
if members is not None:
body['members'] = ','.split(members)
if not dry_run:
click.confirm('Are you sure that you want to sync programs with UWLDAP?', abort=True)
resp = http_post('/api/uwldap/updateprograms', json=body)
result = handle_sync_response(resp)
if len(result) == 0:
click.echo('All programs are up-to-date.')
return
if dry_run:
click.echo('Members whose program would be changed:')
else:
click.echo('Members whose program was changed:')
lines = []
for uid, csc_program, uw_program in result:
csc_program = csc_program or 'Unknown'
csc_program = click.style(csc_program, fg='yellow')
uw_program = click.style(uw_program, fg='green')
lines.append((uid, csc_program + ' -> ' + uw_program))
print_colon_kv(lines)

@ -35,10 +35,12 @@ def create_user():
@requires_authentication_no_realm
def get_user(auth_user: str, username: str):
get_forwarding_addresses = False
if auth_user == username or user_is_in_group(auth_user, 'syscom'):
# Only syscom members, or the user themselves, may see the user's
# forwarding addresses, since this requires reading a file in the
# user's home directory
if user_is_in_group(auth_user, 'syscom'):
# Only syscom members may see the user's forwarding addresses,
# since this requires reading a file in the user's home directory.
# To avoid situations where an unprivileged user symlinks their
# ~/.forward file to /etc/shadow or something, we don't allow
# non-syscom members to use this option either.
get_forwarding_addresses = True
ldap_srv = component.getUtility(ILDAPService)
user = ldap_srv.get_user(username)

@ -12,16 +12,15 @@ def test_members_get(cli_setup, ldap_user):
runner = CliRunner()
result = runner.invoke(cli, ['members', 'get', ldap_user.uid])
expected = (
f"uid: {ldap_user.uid}\n"
f"cn: {ldap_user.cn}\n"
f"program: {ldap_user.program}\n"
f"UID number: {ldap_user.uid_number}\n"
f"GID number: {ldap_user.gid_number}\n"
f"login shell: {ldap_user.login_shell}\n"
f"home directory: {ldap_user.home_directory}\n"
f"is a club: {ldap_user.is_club()}\n"
"forwarding addresses: \n"
f"terms: {','.join(ldap_user.terms)}\n"
f"uid: {ldap_user.uid}\n"
f"cn: {ldap_user.cn}\n"
f"program: {ldap_user.program}\n"
f"UID number: {ldap_user.uid_number}\n"
f"GID number: {ldap_user.gid_number}\n"
f"login shell: {ldap_user.login_shell}\n"
f"home directory: {ldap_user.home_directory}\n"
f"is a club: {ldap_user.is_club()}\n"
f"terms: {','.join(ldap_user.terms)}\n"
)
assert result.exit_code == 0
assert result.output == expected

@ -0,0 +1,43 @@
from click.testing import CliRunner
import ldap3
from ceo.cli import cli
def test_updatemembers(cli_setup, cfg, ldap_conn, ldap_user, uwldap_user):
# sanity check
assert ldap_user.uid == uwldap_user.uid
# modify the user's program in UWLDAP
conn = ldap_conn
base_dn = cfg.get('uwldap_base')
dn = f'uid={uwldap_user.uid},{base_dn}'
changes = {'ou': [(ldap3.MODIFY_REPLACE, ['New Program'])]}
conn.modify(dn, changes)
runner = CliRunner()
result = runner.invoke(cli, ['updateprograms', '--dry-run'])
expected = (
"Members whose program would be changed:\n"
f"{ldap_user.uid}: {ldap_user.program} -> New Program\n"
)
assert result.exit_code == 0
assert result.output == expected
runner = CliRunner()
result = runner.invoke(cli, ['updateprograms'], input='y\n')
expected = (
"Are you sure that you want to sync programs with UWLDAP? [y/N]: y\n"
"Members whose program was changed:\n"
f"{ldap_user.uid}: {ldap_user.program} -> New Program\n"
)
assert result.exit_code == 0
assert result.output == expected
runner = CliRunner()
result = runner.invoke(cli, ['updateprograms'], input='y\n')
expected = (
"Are you sure that you want to sync programs with UWLDAP? [y/N]: y\n"
"All programs are up-to-date.\n"
)
assert result.exit_code == 0
assert result.output == expected
Loading…
Cancel
Save