pyceo/tests/conftest.py

import importlib.resources
import os
import shutil

import ldap
import pytest
import socket
from zope import component

from ceo_common.interfaces import IConfig, IKerberosService, ILDAPService, \
    IFileService, IMailmanService, IHTTPClient, IUWLDAPService, IMailService
from ceo_common.model import Config, RemoteMailmanService, HTTPClient
from ceod.api import create_app
from ceod.model import KerberosService, LDAPService, FileService, User, \
    MailmanService, Group, UWLDAPService, UWLDAPRecord, MailService
from ceod.model.utils import strings_to_bytes
from .MockSMTPServer import MockSMTPServer
from .MockMailmanServer import MockMailmanServer
from .conftest_ceod_api import *


@pytest.fixture(scope='session')
def cfg():
    with importlib.resources.path('tests', 'ceod_test_local.ini') as p:
        config_file = p.__fspath__()
    _cfg = Config(config_file)
    component.provideUtility(_cfg, IConfig)
    return _cfg


@pytest.fixture(scope='session')
def krb_srv(cfg):
    # we need to be root to read the keytab
    assert os.geteuid() == 0
    # this dance again... ugh
    if socket.gethostname() == cfg.get('ceod_admin_host'):
        principal = 'ceod/admin'
    else:
        principal = 'ceod/' + socket.getfqdn()
    cache_dir = '/tmp/ceod_test/krb5_cache'
    shutil.rmtree(cache_dir, ignore_errors=True)
    krb = KerberosService(principal, cache_dir)
    component.provideUtility(krb, IKerberosService)
    yield krb
    shutil.rmtree(cache_dir)


def recursively_delete_subtree(conn: ldap.ldapobject.LDAPObject, base_dn: str):
    try:
        records = conn.search_s(base_dn, ldap.SCOPE_ONELEVEL, attrlist=[''])
        for dn, _ in records:
            conn.delete_s(dn)
        conn.delete_s(base_dn)
    except ldap.NO_SUCH_OBJECT:
        pass


@pytest.fixture(scope='session')
def ldap_srv(cfg, krb_srv):
    conn = ldap.initialize(cfg.get('ldap_server_url'))
    conn.sasl_gssapi_bind_s()
    users_base = cfg.get('ldap_users_base')
    groups_base = cfg.get('ldap_groups_base')

    recursively_delete_subtree(conn, users_base)
    recursively_delete_subtree(conn, groups_base)

    for base_dn in [users_base, groups_base]:
        ou = base_dn.split(',', 1)[0].split('=')[1]
        conn.add_s(base_dn, ldap.modlist.addModlist({
            'objectClass': [b'organizationalUnit'],
            'ou': [ou.encode()]
        }))
    _ldap_srv = LDAPService()
    component.provideUtility(_ldap_srv, ILDAPService)
    yield _ldap_srv

    recursively_delete_subtree(conn, users_base)
    recursively_delete_subtree(conn, groups_base)


@pytest.fixture(scope='session')
def file_srv(cfg):
    _file_srv = FileService()
    component.provideUtility(_file_srv, IFileService)
    members_home = cfg.get('members_home')
    clubs_home = cfg.get('clubs_home')

    shutil.rmtree(members_home, ignore_errors=True)
    shutil.rmtree(clubs_home, ignore_errors=True)
    yield _file_srv
    shutil.rmtree(members_home, ignore_errors=True)
    shutil.rmtree(clubs_home, ignore_errors=True)


@pytest.fixture(scope='session')
def http_client(cfg):
    client = HTTPClient()
    component.provideUtility(client, IHTTPClient)
    return


@pytest.fixture(scope='session')
def mock_mailman_server():
    server = MockMailmanServer()
    server.start()
    yield server
    server.stop()


@pytest.fixture(scope='session')
def mailman_srv(mock_mailman_server, cfg, http_client):
    # TODO: test the RemoteMailmanService as well
    mailman = MailmanService()
    component.provideUtility(mailman, IMailmanService)
    return mailman


@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(scope='session')
def mock_mail_server():
    mock_server = MockSMTPServer()
    mock_server.start()
    yield mock_server
    mock_server.stop()


@pytest.fixture(scope='session')
def mail_srv(cfg, mock_mail_server):
    _mail_srv = MailService()
    component.provideUtility(_mail_srv, IMailService)
    return _mail_srv


@pytest.fixture(autouse=True, scope='session')
def app(
    cfg,
    krb_srv,
    ldap_srv,
    file_srv,
    mailman_srv,
    uwldap_srv,
    mail_srv,
):
    # need to be root to read keytab
    assert os.geteuid() == 0
    app = create_app({
        'TESTING': True,
    })
    return app


@pytest.fixture
def simple_user():
    return User(
        uid='test_jdoe',
        cn='John Doe',
        program='Math',
        terms=['s2021'],
    )


@pytest.fixture
def simple_club():
    return User(
        uid='test_club1',
        cn='Club One',
        is_club=True,
    )


@pytest.fixture
def ldap_user(simple_user):
    simple_user.add_to_ldap()
    yield simple_user
    simple_user.remove_from_ldap()


@pytest.fixture
def krb_user(simple_user):
    simple_user.add_to_kerberos('krb5')
    yield simple_user
    simple_user.remove_from_kerberos()


@pytest.fixture
def simple_group():
    return Group(
        cn='group1',
        gid_number=21000,
    )


@pytest.fixture
def ldap_group(simple_group):
    simple_group.add_to_ldap()
    yield simple_group
    simple_group.remove_from_ldap()


@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)