forked from public/pyceo
add tests for uwldap
This commit is contained in:
parent
9e4d564a33
commit
cbf4aa43f8
|
@ -0,0 +1,15 @@
|
|||
import pytest
|
||||
|
||||
from ceo_common.errors import UserAlreadySubscribedError, UserNotSubscribedError
|
||||
|
||||
|
||||
def test_user_mailing_lists(ldap_user):
|
||||
user = ldap_user
|
||||
|
||||
user.subscribe_to_mailing_list('csc-general')
|
||||
with pytest.raises(UserAlreadySubscribedError):
|
||||
user.subscribe_to_mailing_list('csc-general')
|
||||
|
||||
user.unsubscribe_from_mailing_list('csc-general')
|
||||
with pytest.raises(UserNotSubscribedError):
|
||||
user.unsubscribe_from_mailing_list('csc-general')
|
|
@ -3,8 +3,7 @@ import subprocess
|
|||
|
||||
import pytest
|
||||
|
||||
from ceo_common.errors import UserNotFoundError, UserAlreadyExistsError, \
|
||||
UserAlreadySubscribedError, UserNotSubscribedError
|
||||
from ceo_common.errors import UserNotFoundError, UserAlreadyExistsError
|
||||
from ceod.model import User
|
||||
|
||||
|
||||
|
@ -117,18 +116,6 @@ def test_login_shell(ldap_user, ldap_srv):
|
|||
assert ldap_srv.get_user(user.uid).login_shell == user.login_shell
|
||||
|
||||
|
||||
def test_user_mailing_lists(ldap_user):
|
||||
user = ldap_user
|
||||
|
||||
user.subscribe_to_mailing_list('csc-general')
|
||||
with pytest.raises(UserAlreadySubscribedError):
|
||||
user.subscribe_to_mailing_list('csc-general')
|
||||
|
||||
user.unsubscribe_from_mailing_list('csc-general')
|
||||
with pytest.raises(UserNotSubscribedError):
|
||||
user.unsubscribe_from_mailing_list('csc-general')
|
||||
|
||||
|
||||
def test_user_to_dict(cfg):
|
||||
user = User(
|
||||
uid='test_jsmith',
|
||||
|
|
|
@ -0,0 +1,32 @@
|
|||
import ldap
|
||||
|
||||
|
||||
def test_uwldap_get(uwldap_srv, uwldap_user):
|
||||
retrieved_user = uwldap_srv.get_user(uwldap_user.uid)
|
||||
assert retrieved_user is not None
|
||||
assert retrieved_user.uid == uwldap_user.uid
|
||||
|
||||
assert uwldap_srv.get_user('no_such_user') is None
|
||||
|
||||
|
||||
def test_ldap_updateprograms(cfg, ldap_srv, uwldap_srv, ldap_user, uwldap_user):
|
||||
# sanity check
|
||||
assert ldap_user.uid == uwldap_user.uid
|
||||
# modify the user's program in UWLDAP
|
||||
conn = ldap.initialize(cfg.get('uwldap_server_url'))
|
||||
conn.sasl_gssapi_bind_s()
|
||||
base_dn = cfg.get('uwldap_base')
|
||||
dn = f'uid={uwldap_user.uid},{base_dn}'
|
||||
conn.modify_s(dn, ldap.modlist.modifyModlist(
|
||||
{'ou': [uwldap_user.program.encode()]},
|
||||
{'ou': [b'New Program']},
|
||||
))
|
||||
|
||||
assert ldap_srv.update_programs(dry_run=True) == [
|
||||
(uwldap_user.uid, uwldap_user.program, 'New Program'),
|
||||
]
|
||||
assert ldap_srv.get_user(uwldap_user.uid).program == uwldap_user.program
|
||||
assert ldap_srv.update_programs(members=['no_such_user']) == []
|
||||
|
||||
ldap_srv.update_programs()
|
||||
assert uwldap_srv.get_user(uwldap_user.uid).program == 'New Program'
|
|
@ -17,8 +17,8 @@ groups_base = ou=TestGroup,dc=csclub,dc=internal
|
|||
sudo_base = ou=TestSUDOers,dc=csclub,dc=internal
|
||||
|
||||
[uwldap]
|
||||
server_url = ldap://uwldap.uwaterloo.ca
|
||||
base = dc=uwaterloo,dc=ca
|
||||
server_url = ldap://ldap-master.csclub.internal
|
||||
base = ou=TestUWLDAP,dc=csclub,dc=internal
|
||||
|
||||
[members]
|
||||
min_id = 20001
|
||||
|
|
|
@ -8,10 +8,11 @@ import socket
|
|||
from zope import component
|
||||
|
||||
from ceo_common.interfaces import IConfig, IKerberosService, ILDAPService, \
|
||||
IFileService, IMailmanService, IHTTPClient
|
||||
IFileService, IMailmanService, IHTTPClient, IUWLDAPService
|
||||
from ceo_common.model import Config, RemoteMailmanService, HTTPClient
|
||||
from ceod.model import KerberosService, LDAPService, FileService, User, \
|
||||
MailmanService, Group
|
||||
MailmanService, Group, UWLDAPService, UWLDAPRecord
|
||||
from ceod.model.utils import strings_to_bytes
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope='session')
|
||||
|
@ -153,3 +154,56 @@ def ldap_group(simple_group):
|
|||
simple_group.add_to_ldap()
|
||||
yield simple_group
|
||||
simple_group.remove_from_ldap()
|
||||
|
||||
|
||||
@pytest.fixture(scope='session')
|
||||
def uwldap_srv(cfg, ldap_srv):
|
||||
conn = ldap.initialize(cfg.get('uwldap_server_url'))
|
||||
conn.sasl_gssapi_bind_s()
|
||||
base_dn = cfg.get('uwldap_base')
|
||||
ou = base_dn.split(',', 1)[0].split('=')[1]
|
||||
|
||||
recursively_delete_subtree(conn, base_dn)
|
||||
|
||||
conn.add_s(base_dn, ldap.modlist.addModlist({
|
||||
'objectClass': [b'organizationalUnit'],
|
||||
'ou': [ou.encode()]
|
||||
}))
|
||||
_uwldap_srv = UWLDAPService()
|
||||
component.provideUtility(_uwldap_srv, IUWLDAPService)
|
||||
yield _uwldap_srv
|
||||
|
||||
recursively_delete_subtree(conn, base_dn)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def uwldap_user(cfg, uwldap_srv):
|
||||
conn = ldap.initialize(cfg.get('uwldap_server_url'))
|
||||
conn.sasl_gssapi_bind_s()
|
||||
base_dn = cfg.get('uwldap_base')
|
||||
user = UWLDAPRecord(
|
||||
uid='test_jdoe',
|
||||
mail_local_addresses=['test_jdoe@uwaterloo.internal'],
|
||||
program='Math',
|
||||
cn='John Doe',
|
||||
sn='Doe',
|
||||
given_name='John',
|
||||
)
|
||||
dn = f'uid={user.uid},{base_dn}'
|
||||
conn.add_s(dn, ldap.modlist.addModlist(strings_to_bytes({
|
||||
'uid': [user.uid],
|
||||
'mailLocalAddress': user.mail_local_addresses,
|
||||
'ou': [user.program],
|
||||
'cn': [user.cn],
|
||||
'sn': [user.sn],
|
||||
'givenName': [user.given_name],
|
||||
'objectClass': [
|
||||
'inetLocalMailRecipient',
|
||||
'inetOrgPerson',
|
||||
'organizationalPerson',
|
||||
'person',
|
||||
'top',
|
||||
],
|
||||
})))
|
||||
yield user
|
||||
conn.delete_s(dn)
|
||||
|
|
Loading…
Reference in New Issue