add test for RemoteMailmanService

pull/7/head
Max Erenberg 1 year ago
parent 490abb302c
commit 46fd926acc
  1. 2
      setup.cfg
  2. 0
      tests/ceo_common/model/__init__.py
  3. 42
      tests/ceo_common/model/test_remote_mailman.py
  4. 47
      tests/conftest.py

@ -2,4 +2,4 @@
ignore =
# line too long
E501
exclude = .git,.vscode,venv,__pycache__,__init__.py,conftest.py,build,dist
exclude = .git,.vscode,venv,__pycache__,__init__.py,build,dist

@ -0,0 +1,42 @@
from multiprocessing import Process
import socket
import sys
import time
import requests
from ceo_common.model import RemoteMailmanService
def test_remote_mailman(cfg, http_client, app, mock_mailman_server, g_syscom):
port = cfg.get('ceod_port')
hostname = socket.gethostname()
def server_start():
sys.stdout = open('/dev/null', 'w')
sys.stderr = sys.stdout
app.run(debug=False, host='0.0.0.0', port=port)
proc = Process(target=server_start)
proc.start()
for _ in range(5):
try:
http_client.get(hostname, '/ping')
except requests.exceptions.ConnectionError:
time.sleep(0.5)
continue
break
try:
mailman_srv = RemoteMailmanService()
assert mock_mailman_server.subscriptions['csc-general'] == []
# RemoteMailmanService -> app -> MailmanService -> MockMailmanServer
address = 'test_1@csclub.internal'
mailman_srv.subscribe(address, 'csc-general')
assert mock_mailman_server.subscriptions['csc-general'] == [address]
mailman_srv.unsubscribe(address, 'csc-general')
assert mock_mailman_server.subscriptions['csc-general'] == []
finally:
proc.terminate()
proc.join()

@ -5,6 +5,7 @@ import os
import pwd
import shutil
import subprocess
from subprocess import DEVNULL
import tempfile
from unittest.mock import patch
@ -16,15 +17,14 @@ 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 ceo_common.model import Config, 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
import ceod.utils as utils
from .MockSMTPServer import MockSMTPServer
from .MockMailmanServer import MockMailmanServer
from .conftest_ceod_api import client
from .conftest_ceod_api import client # noqa: F401
@pytest.fixture(scope='session')
@ -96,6 +96,40 @@ def g_admin_ctx(cfg, ceod_admin_creds, app):
return wrapper
@pytest.fixture(scope='session')
def syscom_creds():
"""
Acquire credentials for a syscom member and store them in a ccache.
Yields the name of the ccache file.
"""
with tempfile.NamedTemporaryFile() as f:
subprocess.run(
['kinit', '-c', f.name, 'ctdalek'],
check=True, text=True, input='krb5', stdout=DEVNULL,
)
yield f.name
@pytest.fixture
def g_syscom(syscom_creds, app):
"""
Store the principal for the syscom member in flask.g, and point
KRB5CCNAME to the file where the TGT is stored.
Use this fixture if you need syscom credentials for an HTTP request
to a different process.
"""
filename = syscom_creds
with app.app_context():
old_krb5ccname = os.environ['KRB5CCNAME']
os.environ['KRB5CCNAME'] = 'FILE:' + filename
try:
flask.g.sasl_user = 'ctdalek'
yield filename
finally:
os.environ['KRB5CCNAME'] = old_krb5ccname
flask.g.pop('sasl_user')
@pytest.fixture(scope='session')
def ldap_conn(cfg, ceod_admin_creds) -> ldap3.Connection:
# Assume that the same server URL is being used for the CSC
@ -119,7 +153,6 @@ def ldap_srv_session(cfg, krb_srv, ldap_conn):
for base_dn in [users_base, groups_base, sudo_base]:
delete_subtree(conn, base_dn)
ou = base_dn.split(',', 1)[0].split('=')[1]
conn.add(base_dn, 'organizationalUnit')
_ldap_srv = LDAPService()
@ -156,9 +189,9 @@ def file_srv(cfg):
@pytest.fixture(scope='session')
def http_client(cfg):
client = HTTPClient()
component.provideUtility(client, IHTTPClient)
return
_client = HTTPClient()
component.provideUtility(_client, IHTTPClient)
return _client
@pytest.fixture(scope='session')

Loading…
Cancel
Save