pyceo/tests/utils.py

import ceod.utils as ceod_utils
import contextlib
import os
import shutil
import subprocess
from subprocess import DEVNULL
import tempfile
from unittest.mock import patch

import gssapi
import pytest


# map principals to GSSAPI credentials
_cache = {}


@contextlib.contextmanager
def gssapi_token_ctx(principal: str):
    """
    Temporarily set KRB5CCNAME to a ccache storing credentials
    for the specified user, and yield the GSSAPI credentials.
    """
    old_krb5ccname = os.environ['KRB5CCNAME']
    try:
        if principal not in _cache:
            f = tempfile.NamedTemporaryFile()
            os.environ['KRB5CCNAME'] = 'FILE:' + f.name
            args = ['kinit', principal]
            if principal == 'ceod/admin':
                args = ['kinit', '-k', principal]
            subprocess.run(
                args, stdout=DEVNULL, text=True, input='krb5', check=True)
            creds = gssapi.Credentials(name=gssapi.Name(principal), usage='initiate')
            # Keep the credential cache files around as long as the creds are
            # used, otherwise we get a "Invalid credential was supplied" error
            _cache[principal] = creds, f
        else:
            creds, f = _cache[principal]
            os.environ['KRB5CCNAME'] = 'FILE:' + f.name
        yield creds.export()
    finally:
        os.environ['KRB5CCNAME'] = old_krb5ccname


@pytest.fixture(scope='session', autouse=True)
def ccache_cleanup():
    """Make sure the ccache files get deleted at the end of the tests."""
    yield
    _cache.clear()


@contextlib.contextmanager
def gen_password_mock_ctx():
    with patch.object(ceod_utils, 'gen_password') as mock:
        mock.return_value = 'krb5'
        yield


@contextlib.contextmanager
def mocks_for_create_user_ctx():
    with gen_password_mock_ctx():
        yield


def set_datetime_in_app_process(app_process, value):
    pipe = app_process
    pipe.send(value)
    pipe.recv()  # wait for ACK


def restore_datetime_in_app_process(app_process):
    set_datetime_in_app_process(app_process, None)


def reset_disable_club_conf(webhosting_srv):
    shutil.copy('tests/resources/apache2/disable-club.conf', webhosting_srv.conf_available_dir)


def create_php_file_for_club(clubs_home, club_name):
    www = f'{clubs_home}/{club_name}/www'
    os.makedirs(www, exist_ok=True)
    open(www + '/index.php', 'w').close()


def create_website_config_for_club(sites_available_dir, cn, filename=None):
    if filename is None:
        filename = f'club-{cn}.conf'
    filepath = os.path.join(sites_available_dir, filename)
    with open(filepath, 'w') as fo:
        fo.write(f"""
        <VirtualHost *:80>
            ServerName {cn}.uwaterloo.internal
            ServerAdmin {cn}@{cn}.uwaterloo.internal
            DocumentRoot /users/{cn}/www/
        </VirtualHost>
        """)