update API endpoints to add isClubRep attribute

pull/27/head
Max Erenberg 1 year ago
parent 1fbc068f3b
commit 743d63efb0
  1. 6
      .drone/csc.schema
  2. 2
      ceo/utils.py
  3. 6
      ceo_common/interfaces/IUser.py
  4. 14
      ceod/api/members.py
  5. 2
      ceod/model/LDAPService.py
  6. 26
      ceod/model/User.py
  7. 3
      ceod/transactions/members/RenewMemberTransaction.py
  8. 4
      tests/ceo/cli/test_members.py
  9. 3
      tests/ceod/api/test_members.py
  10. 26
      tests/ceod/model/test_user.py

@ -20,10 +20,14 @@ attributetype ( 1.3.6.1.4.1.27934.1.1.5 NAME 'nonMemberTerm'
EQUALITY caseIgnoreIA5Match
SYNTAX 1.3.6.1.4.1.1466.115.121.1.26{5} )
attributetype ( 1.3.6.1.4.1.27934.1.1.6 NAME 'isClubRep'
EQUALITY booleanMatch
SYNTAX 1.3.6.1.4.1.1466.115.121.1.7 )
objectclass ( 1.3.6.1.4.1.27934.1.2.1 NAME 'member'
SUP top AUXILIARY
MUST ( cn $ uid )
MAY ( studentid $ program $ term $ nonMemberTerm $ description $ position ) )
MAY ( studentid $ program $ term $ nonMemberTerm $ description $ position $ isClubRep ) )
objectclass ( 1.3.6.1.4.1.27934.1.2.2 NAME 'club'
SUP top AUXILIARY

@ -105,6 +105,8 @@ def user_dict_kv(d: Dict) -> List[Tuple[str]]:
pairs.append(('home directory', d['home_directory']))
if 'is_club' in d:
pairs.append(('is a club', str(d['is_club'])))
if 'is_club_rep' in d:
pairs.append(('is a club rep', str(d['is_club_rep'])))
if 'forwarding_addresses' in d:
if len(d['forwarding_addresses']) > 0:
pairs.append(('forwarding addresses', d['forwarding_addresses'][0]))

@ -19,6 +19,7 @@ class IUser(Interface):
non_member_terms = Attribute('list of terms for which this person was '
'a club rep')
mail_local_addresses = Attribute('email aliases')
is_club_rep = Attribute('whether this user is a club rep or not')
# Non-LDAP attributes
ldap3_entry = Attribute('cached ldap3.Entry instance for this user')
@ -35,6 +36,11 @@ class IUser(Interface):
Returns False if this is the Unix user for a member.
"""
def should_be_club_rep(self) -> bool:
"""
Returns True iff this user's most recent term was a non-member term.
"""
def add_to_ldap():
"""
Add a new record to LDAP for this user.

@ -20,12 +20,17 @@ bp = Blueprint('members', __name__)
@authz_restrict_to_staff
def create_user():
body = request.get_json(force=True)
terms = body.get('terms')
non_member_terms = body.get('non_member_terms')
if (terms and non_member_terms) or not (terms or non_member_terms):
raise BadRequest('Must specify either terms or non-member terms')
txn = AddMemberTransaction(
uid=body['uid'],
cn=body['cn'],
program=body.get('program'),
terms=body.get('terms'),
non_member_terms=body.get('non_member_terms'),
terms=terms,
non_member_terms=non_member_terms,
forwarding_addresses=body.get('forwarding_addresses'),
)
return create_streaming_response(txn)
@ -67,6 +72,11 @@ def patch_user(auth_user: str, username: str):
@authz_restrict_to_staff
def renew_user(username: str):
body = request.get_json(force=True)
terms = body.get('terms')
non_member_terms = body.get('non_member_terms')
if (terms and non_member_terms) or not (terms or non_member_terms):
raise BadRequest('Must specify either terms or non-member terms')
ldap_srv = component.getUtility(ILDAPService)
user = ldap_srv.get_user(username)
if body.get('terms'):

@ -197,6 +197,8 @@ class LDAPService:
entry.position = user.positions
if user.mail_local_addresses:
entry.mailLocalAddress = user.mail_local_addresses
if user.is_club_rep:
entry.isClubRep = True
if not user.is_club():
entry.userPassword = '{SASL}%s@%s' % (user.uid, self.ldap_sasl_realm)

@ -9,6 +9,7 @@ from zope.interface import implementer
from .validators import is_valid_shell, is_valid_term
from ceo_common.interfaces import ILDAPService, IKerberosService, IFileService, \
IUser, IConfig, IMailmanService
from ceo_common.model import Term
@implementer(IUser)
@ -26,6 +27,7 @@ class User:
home_directory: Union[str, None] = None,
positions: Union[List[str], None] = None,
mail_local_addresses: Union[List[str], None] = None,
is_club_rep: Union[bool, None] = None,
is_club: bool = False,
ldap3_entry: Union[ldap3.Entry, None] = None,
):
@ -52,6 +54,10 @@ class User:
self.positions = positions or []
self.mail_local_addresses = mail_local_addresses or []
self._is_club = is_club
if is_club_rep is None:
self.is_club_rep = self.should_be_club_rep()
else:
self.is_club_rep = is_club_rep
self.ldap3_entry = ldap3_entry
self.ldap_srv = component.getUtility(ILDAPService)
@ -66,6 +72,7 @@ class User:
'login_shell': self.login_shell,
'home_directory': self.home_directory,
'is_club': self.is_club(),
'is_club_rep': self.is_club_rep,
'program': self.program or 'Unknown',
}
if self.terms:
@ -86,6 +93,19 @@ class User:
def is_club(self) -> bool:
return self._is_club
def should_be_club_rep(self) -> bool:
if self._is_club:
# not a real user
return False
if not self.terms:
# no member terms => was only ever a club rep
return True
if not self.non_member_terms:
# no non-member terms => was only ever a member
return False
# decide using the most recent term (member or non-member)
return max(map(Term, self.non_member_terms)) > max(map(Term, self.terms))
def add_to_ldap(self):
self.ldap_srv.add_user(self)
@ -131,6 +151,7 @@ class User:
home_directory=attrs['homeDirectory'][0],
positions=attrs.get('position'),
mail_local_addresses=attrs.get('mailLocalAddress'),
is_club_rep=attrs.get('isClubRep', [False])[0],
is_club=('club' in attrs['objectClass']),
ldap3_entry=entry,
)
@ -148,7 +169,10 @@ class User:
raise Exception('%s is not a valid term' % term)
with self.ldap_srv.entry_ctx_for_user(self) as entry:
entry.term.add(terms)
if entry.isClubRep.value:
entry.isClubRep.remove()
self.terms.extend(terms)
self.is_club_rep = False
def add_non_member_terms(self, terms: List[str]):
for term in terms:
@ -156,7 +180,9 @@ class User:
raise Exception('%s is not a valid term' % term)
with self.ldap_srv.entry_ctx_for_user(self) as entry:
entry.nonMemberTerm.add(terms)
entry.isClubRep = True
self.non_member_terms.extend(terms)
self.is_club_rep = True
def set_positions(self, positions: List[str]):
with self.ldap_srv.entry_ctx_for_user(self) as entry:

@ -3,7 +3,6 @@ from typing import Union, List
from zope import component
from ..AbstractTransaction import AbstractTransaction
from ceo_common.errors import BadRequest
from ceo_common.interfaces import ILDAPService
@ -23,8 +22,6 @@ class RenewMemberTransaction(AbstractTransaction):
):
super().__init__()
self.username = username
if (terms and non_member_terms) or not (terms or non_member_terms):
raise BadRequest('Must specify either terms or non-member terms')
self.terms = terms
self.non_member_terms = non_member_terms
self.ldap_srv = component.getUtility(ILDAPService)

@ -20,6 +20,7 @@ def test_members_get(cli_setup, ldap_user):
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"is a club rep: {ldap_user.is_club_rep}\n"
"forwarding addresses: \n"
f"member terms: {','.join(ldap_user.terms)}\n"
)
@ -58,7 +59,8 @@ def test_members_add(cli_setup):
"login shell: /bin/bash\n"
"home directory: [a-z0-9/_-]+/test_1\n"
"is a club: False\n"
"forwarding addresses: test_1@uwaterloo.internal\n"
"is a club rep: False\n"
"forwarding addresses: test_1@uwaterloo\\.internal\n"
"member terms: [sfw]\\d{4}\n"
"password: \\S+\n$"
), re.MULTILINE)

@ -56,6 +56,7 @@ def test_api_create_user(cfg, create_user_resp, mock_mail_server):
"login_shell": "/bin/bash",
"home_directory": "/tmp/test_users/test_1",
"is_club": False,
"is_club_rep": False,
"program": "Math",
"terms": ["s2021"],
"forwarding_addresses": ['test_1@uwaterloo.internal'],
@ -175,6 +176,7 @@ def test_api_renew_user(cfg, client, create_user_result, ldap_conn):
_, data = client.get(f'/api/members/{uid}')
assert data['terms'] == old_terms + new_terms
assert data['non_member_terms'] == old_non_member_terms + new_non_member_terms
assert data['is_club_rep']
# cleanup
base_dn = cfg.get('ldap_users_base')
@ -182,6 +184,7 @@ def test_api_renew_user(cfg, client, create_user_result, ldap_conn):
changes = {
'term': [(ldap3.MODIFY_REPLACE, old_terms)],
'nonMemberTerm': [(ldap3.MODIFY_REPLACE, old_non_member_terms)],
'isClubRep': [(ldap3.MODIFY_REPLACE, [])],
}
ldap_conn.modify(dn, changes)

@ -162,6 +162,7 @@ def test_user_to_dict(cfg):
'login_shell': '/bin/bash',
'home_directory': user.home_directory,
'is_club': False,
'is_club_rep': False,
}
assert user.to_dict() == expected
@ -172,3 +173,28 @@ def test_user_to_dict(cfg):
user.create_home_dir()
expected['forwarding_addresses'] = []
assert user.to_dict(True) == expected
def test_user_is_club_rep(ldap_user, ldap_srv):
user = User(
uid='test_jsmith',
cn='John Smith',
program='Math',
terms=['s2021'],
)
assert not user.is_club_rep
club_rep = User(
uid='test_jdoe',
cn='John Doe',
program='Math',
non_member_terms=['s2021'],
)
assert club_rep.is_club_rep
ldap_user.add_terms(['f2021'])
assert not ldap_user.is_club_rep
assert not ldap_srv.get_user(ldap_user.uid).is_club_rep
ldap_user.add_non_member_terms(['w2022'])
assert ldap_user.is_club_rep
assert ldap_srv.get_user(ldap_user.uid).is_club_rep

Loading…
Cancel
Save