add unit tests

pull/35/head
Max Erenberg 1 year ago
parent 7425d69feb
commit 0a52b0b395
  1. 3
      .drone/common.sh
  2. 3
      .drone/mail-setup.sh
  3. 43
      ceo/cli/cloud.py
  4. 4
      ceo/utils.py
  5. 5
      ceo_common/errors.py
  6. 4
      ceo_common/model/HTTPClient.py
  7. 37
      ceod/api/cloud.py
  8. 6
      ceod/api/error_handlers.py
  9. 11
      ceod/api/utils.py
  10. 20
      ceod/model/CloudService.py
  11. 8
      ceod/model/VHostManager.py
  12. 9
      etc/ceod.ini
  13. 26
      tests/ceo/cli/test_cloud.py
  14. 100
      tests/ceod/api/test_cloud.py
  15. 28
      tests/ceod/model/test_vhosts.py
  16. 2
      tests/ceod_dev.ini
  17. 2
      tests/ceod_test_local.ini
  18. 16
      tests/conftest.py
  19. 3
      tests/conftest_ceod_api.py

@ -6,6 +6,9 @@ sed -E '/^(domain|search)[[:space:]]+csclub.uwaterloo.ca/d' /etc/resolv.conf > /
cp /tmp/resolv.conf /etc/resolv.conf
rm /tmp/resolv.conf
# mock out systemctl
ln -s /bin/true /usr/local/bin/systemctl
get_ip_addr() {
getent hosts $1 | cut -d' ' -f1
}

@ -18,5 +18,8 @@ apt update
apt install -y netcat-openbsd
auth_setup mail
# for the VHostManager
mkdir -p /run/ceod/member-vhosts
# sync with phosphoric-acid
nc -l 0.0.0.0 9000 &

@ -3,8 +3,8 @@ from zope import component
from ceo_common.interfaces import IConfig
from ..utils import http_post
from .utils import handle_sync_response
from ..utils import http_post, http_put, http_get, http_delete
from .utils import Abort, handle_sync_response, print_colon_kv
@click.group(short_help='Perform operations on the CSC cloud')
@ -44,3 +44,42 @@ def purge():
result = handle_sync_response(resp)
click.echo('Accounts to be deleted: ' + ','.join(result['accounts_to_be_deleted']))
click.echo('Accounts which were deleted: ' + ','.join(result['accounts_deleted']))
@cloud.group(short_help='Manage your virtual hosts')
def vhosts():
pass
@vhosts.command(name='add', short_help='Add a virtual host')
@click.argument('domain')
@click.argument('ip_address')
def add_vhost(domain, ip_address):
body = {'ip_address': ip_address}
if '/' in domain:
raise Abort('invalid domain name')
resp = http_put('/api/cloud/vhosts/' + domain, json=body)
handle_sync_response(resp)
click.echo('Done.')
@vhosts.command(name='delete', short_help='Delete a virtual host')
@click.argument('domain')
def delete_vhost(domain):
if '/' in domain:
raise Abort('invalid domain name')
resp = http_delete('/api/cloud/vhosts/' + domain)
handle_sync_response(resp)
click.echo('Done.')
@vhosts.command(name='list', short_help='List virtual hosts')
def list_vhosts():
resp = http_get('/api/cloud/vhosts')
result = handle_sync_response(resp)
vhosts = result['vhosts']
if not vhosts:
click.echo('No vhosts found.')
return
pairs = [(d['domain'], d['ip_address']) for d in vhosts]
print_colon_kv(pairs)

@ -42,6 +42,10 @@ def http_delete(path: str, **kwargs) -> requests.Response:
return http_request('DELETE', path, **kwargs)
def http_put(path: str, **kwargs) -> requests.Response:
return http_request('PUT', path, **kwargs)
def get_failed_operations(data: List[Dict]) -> List[str]:
"""
Get a list of the failed operations using the JSON objects

@ -78,3 +78,8 @@ class CloudStackAPIError(Exception):
class InvalidDomainError(Exception):
def __init__(self):
super().__init__('domain is invalid')
class InvalidIPError(Exception):
def __init__(self):
super().__init__('IP address is invalid')

@ -27,8 +27,8 @@ class HTTPClient:
host = host + '.' + self.base_domain
if method == 'GET':
# This is the only GET endpoint which requires auth
need_auth = path.startswith('/api/members')
need_auth = path.startswith('/api/members') or \
path.startswith('/api/cloud')
delegate = False
else:
need_auth = True

@ -1,8 +1,9 @@
from flask import Blueprint
from flask import Blueprint, request
from zope import component
from .utils import requires_authentication_no_realm, authz_restrict_to_syscom
from ceo_common.interfaces import ICloudService, ILDAPService
from .utils import requires_authentication_no_realm, authz_restrict_to_syscom, \
get_valid_member_or_throw
from ceo_common.interfaces import ICloudService
bp = Blueprint('cloud', __name__)
@ -10,9 +11,8 @@ bp = Blueprint('cloud', __name__)
@bp.route('/accounts/create', methods=['POST'])
@requires_authentication_no_realm
def create_account(auth_user: str):
user = get_valid_member_or_throw(auth_user)
cloud_srv = component.getUtility(ICloudService)
ldap_srv = component.getUtility(ILDAPService)
user = ldap_srv.get_user(auth_user)
cloud_srv.create_account(user)
return {'status': 'OK'}
@ -22,3 +22,30 @@ def create_account(auth_user: str):
def purge_accounts():
cloud_srv = component.getUtility(ICloudService)
return cloud_srv.purge_accounts()
@bp.route('/vhosts/<domain>', methods=['PUT'])
@requires_authentication_no_realm
def create_vhost(auth_user: str, domain: str):
get_valid_member_or_throw(auth_user)
cloud_srv = component.getUtility(ICloudService)
body = request.get_json(force=True)
ip_address = body['ip_address']
cloud_srv.create_vhost(auth_user, domain, ip_address)
return {'status': 'OK'}
@bp.route('/vhosts/<domain>', methods=['DELETE'])
@requires_authentication_no_realm
def delete_vhost(auth_user: str, domain: str):
cloud_srv = component.getUtility(ICloudService)
cloud_srv.delete_vhost(auth_user, domain)
return {'status': 'OK'}
@bp.route('/vhosts', methods=['GET'])
@requires_authentication_no_realm
def get_vhosts(auth_user: str):
cloud_srv = component.getUtility(ICloudService)
vhosts = cloud_srv.get_vhosts(auth_user)
return {'vhosts': vhosts}

@ -8,7 +8,7 @@ from werkzeug.exceptions import HTTPException
from ceo_common.errors import UserNotFoundError, GroupNotFoundError, \
UserAlreadyExistsError, GroupAlreadyExistsError, BadRequest, \
UserAlreadySubscribedError, InvalidMembershipError, \
CloudStackAPIError
CloudStackAPIError, InvalidDomainError, InvalidIPError
from ceo_common.logger_factory import logger_factory
__all__ = ['register_error_handlers']
@ -24,7 +24,9 @@ def generic_error_handler(err: Exception):
"""Return JSON for all errors."""
if isinstance(err, HTTPException):
status_code = err.code
elif isinstance(err, BadRequest):
elif any(isinstance(err, cls) for cls in [
BadRequest, InvalidDomainError, InvalidIPError
]):
status_code = 400
elif isinstance(err, ldap3.core.exceptions.LDAPStrongerAuthRequiredResult) \
or isinstance(err, InvalidMembershipError):

@ -7,14 +7,25 @@ import traceback
from typing import Callable, List
from flask import current_app, stream_with_context
from zope import component
from .spnego import requires_authentication
from ceo_common.errors import InvalidMembershipError
from ceo_common.interfaces import IUser, ILDAPService
from ceo_common.logger_factory import logger_factory
from ceod.transactions import AbstractTransaction
logger = logger_factory(__name__)
def get_valid_member_or_throw(username: str) -> IUser:
ldap_srv = component.getUtility(ILDAPService)
user = ldap_srv.get_user(username)
if not user.membership_is_valid():
raise InvalidMembershipError()
return user
def requires_authentication_no_realm(f: Callable) -> Callable:
"""
Like requires_authentication, but strips the realm out of the principal string.

@ -2,6 +2,7 @@ from base64 import b64encode
import datetime
import hashlib
import hmac
import ipaddress
import json
import os
import re
@ -13,8 +14,7 @@ from zope import component
from zope.interface import implementer
from .VHostManager import VHostManager
from ceo_common.errors import InvalidMembershipError, CloudStackAPIError, \
InvalidDomainError
from ceo_common.errors import CloudStackAPIError, InvalidDomainError, InvalidIPError
from ceo_common.logger_factory import logger_factory
from ceo_common.interfaces import ICloudService, IConfig, IUser, ILDAPService, \
IMailService
@ -41,6 +41,8 @@ class CloudService:
)
self.max_vhosts_per_account = cfg.get('cloud vhosts_max_vhosts_per_account')
self.vhost_domain = cfg.get('cloud vhosts_members_domain')
self.vhost_ip_min = ipaddress.ip_address(cfg.get('cloud vhosts_ip_range_min'))
self.vhost_ip_max = ipaddress.ip_address(cfg.get('cloud vhosts_ip_range_max'))
state_dir = '/run/ceod'
if not os.path.isdir(state_dir):
@ -100,8 +102,6 @@ class CloudService:
resp.raise_for_status()
def create_account(self, user: IUser):
if not user.membership_is_valid():
raise InvalidMembershipError()
domain_id = self._get_domain_id(self.members_domain)
url = self._create_url({
@ -158,7 +158,10 @@ class CloudService:
if user.membership_is_valid():
continue
account_id = username_to_account_id[username]
self._delete_account(account_id)
self.vhost_mgr.delete_all_vhosts_for_user(username)
accounts_deleted.append(username)
mail_srv.send_cloud_account_has_been_deleted_message(user)
logger.info(f'Deleted cloud account for {username}')
@ -195,12 +198,21 @@ class CloudService:
return False
return True
def _is_valid_ip_address(self, ip_address: str) -> bool:
try:
addr = ipaddress.ip_address(ip_address)
except ValueError:
return False
return self.vhost_ip_min <= addr <= self.vhost_ip_max
def create_vhost(self, username: str, domain: str, ip_address: str):
if self.vhost_mgr.get_num_vhosts(username) >= self.max_vhosts_per_account:
raise Exception(f'Only {self.max_vhosts_per_account} vhosts '
'allowed per account')
if not self._is_valid_domain(username, domain):
raise InvalidDomainError()
if not self._is_valid_ip_address(ip_address):
raise InvalidIPError()
self.vhost_mgr.create_vhost(username, domain, ip_address)
def delete_vhost(self, username: str, domain: str):

@ -4,7 +4,6 @@ import re
import subprocess
from typing import List, Dict
from flask import current_app
import jinja2
from zope.interface import implementer
@ -12,7 +11,7 @@ from ceo_common.logger_factory import logger_factory
from ceo_common.interfaces import IVHostManager
PROXY_PASS_IP_RE = re.compile(r'^\s+proxy_pass\s+http://(?P<ip_address>[\d.]+);$')
VHOST_FILENAME_RE = re.compile(r'^member_(?P<username>[0-9a-z-]+)_(?P<domain>[0-9a-z-]+)$')
VHOST_FILENAME_RE = re.compile(r'^member_(?P<username>[0-9a-z-]+)_(?P<domain>[0-9a-z.-]+)$')
logger = logger_factory(__name__)
@ -47,10 +46,7 @@ class VHostManager:
return glob.glob(os.path.join(self.vhost_dir, 'member_' + username + '_*'))
def _reload_web_server(self):
if current_app.config.get('ENV') == 'development' or \
current_app.config.get('TESTING'):
logger.info('Not reloading web server because we are in development')
return
logger.debug('Reloading nginx')
subprocess.run(['systemctl', 'reload', 'nginx'], check=True)
def create_vhost(self, username: str, domain: str, ip_address: str):

@ -79,3 +79,12 @@ host = localhost
api_key = REPLACE_ME
secret_key = REPLACE_ME
base_url = http://localhost:8080/client/api
[cloud vhosts]
config_dir = /etc/nginx/ceod-member-vhosts
ssl_cert_path = /etc/ssl/private/csclub.cloud.chain
ssl_key_path = /etc/ssl/private/csclub.cloud.key
max_vhosts_per_account = 10
members_domain = m.csclub.cloud
ip_range_min = 172.19.134.10
ip_range_max = 172.19.134.160

@ -26,3 +26,29 @@ def test_cloud_accounts_purge(cli_setup, mock_cloud_server):
runner = CliRunner()
result = runner.invoke(cli, ['cloud', 'accounts', 'purge'])
assert result.exit_code == 0
def test_cloud_vhosts(cli_setup, new_user, cfg):
members_domain = cfg.get('cloud vhosts_members_domain')
uid = new_user.uid
domain1 = uid + '.' + members_domain
ip1 = '172.19.134.11'
runner = CliRunner()
with gssapi_token_ctx(uid):
result = runner.invoke(cli, ['cloud', 'vhosts', 'add', domain1, ip1])
expected = 'Done.\n'
assert result.exit_code == 0
assert result.output == expected
with gssapi_token_ctx(uid):
result = runner.invoke(cli, ['cloud', 'vhosts', 'list'])
expected = domain1 + ': ' + ip1 + '\n'
assert result.exit_code == 0
assert result.output == expected
with gssapi_token_ctx(uid):
result = runner.invoke(cli, ['cloud', 'vhosts', 'delete', domain1])
expected = 'Done.\n'
assert result.exit_code == 0
assert result.output == expected

@ -84,3 +84,103 @@ def test_purge_accounts(
assert new_user.uid not in mock_cloud_server.users_by_username
assert len(mock_mail_server.messages) == 2
mock_mail_server.messages.clear()
def test_cloud_vhosts(cfg, client, new_user, ldap_conn):
members_domain = cfg.get('cloud vhosts_members_domain')
max_vhosts = cfg.get('cloud vhosts_max_vhosts_per_account')
uid = new_user.uid
domain1 = uid + '.' + members_domain
ip1 = '172.19.134.11'
status, _ = client.put(
f'/api/cloud/vhosts/{domain1}', json={'ip_address': ip1},
principal=uid)
assert status == 200
status, data = client.get('/api/cloud/vhosts', principal=uid)
assert status == 200
assert data == {'vhosts': [{'domain': domain1, 'ip_address': ip1}]}
# invalid domain name
domain2 = uid + 'm.cloud.' + cfg.get('base_domain')
ip2 = ip1
status, _ = client.put(
f'/api/cloud/vhosts/{domain2}', json={'ip_address': ip2},
principal=uid)
assert status == 400
# invalid IP address
domain3 = domain1
ip3 = '129.97.134.10'
status, _ = client.put(
f'/api/cloud/vhosts/{domain3}', json={'ip_address': ip3},
principal=uid)
assert status == 400
# new vhost with same domain should replace old one
domain4 = domain1
ip4 = '172.19.134.14'
status, _ = client.put(
f'/api/cloud/vhosts/{domain4}', json={'ip_address': ip4},
principal=uid)
assert status == 200
status, data = client.get('/api/cloud/vhosts', principal=uid)
assert status == 200
assert data == {'vhosts': [{'domain': domain4, 'ip_address': ip4}]}
# maximum number of vhosts
for i in range(max_vhosts):
domain = 'app' + str(i + 1) + '.' + uid + '.' + members_domain
status, _ = client.put(
f'/api/cloud/vhosts/{domain}', json={'ip_address': ip1},
principal=uid)
if i < max_vhosts - 1:
assert status == 200
else:
assert status != 200
# delete a vhost
status, _ = client.delete(f'/api/cloud/vhosts/{domain1}', principal=uid)
assert status == 200
# expired members may not create vhosts
expire_member(new_user, ldap_conn)
status, _ = client.put(
f'/api/cloud/vhosts/{domain1}', json={'ip_address': ip1},
principal=uid)
assert status == 403
def test_cloud_vhosts_purged_account(
cfg, client, mock_cloud_server, mock_mail_server, cloud_srv, new_user,
ldap_conn,
):
uid = new_user.uid
members_domain = cfg.get('cloud vhosts_members_domain')
mock_cloud_server.clear()
current_term = Term.current()
beginning_of_term = current_term.to_datetime()
domain1 = uid + '.' + members_domain
ip1 = '172.19.134.11'
client.post('/api/cloud/accounts/create', principal=uid)
client.put(
f'/api/cloud/vhosts/{domain1}', json={'ip_address': ip1},
principal=uid)
expire_member(new_user, ldap_conn)
with patch.object(ceo_common_utils, 'get_current_datetime') as now_mock:
# grace period has passed - user should be sent a warning
now_mock.return_value = beginning_of_term + datetime.timedelta(days=31)
client.post('/api/cloud/accounts/purge')
# one week has passed - the account can now be deleted
now_mock.return_value += datetime.timedelta(days=8)
client.post('/api/cloud/accounts/purge')
# vhosts should have been deleted
status, data = client.get('/api/cloud/vhosts', principal=uid)
assert status == 200
assert data == {'vhosts': []}
mock_mail_server.messages.clear()

@ -0,0 +1,28 @@
import os
def test_vhost_mgr(cloud_srv):
vhost_mgr = cloud_srv.vhost_mgr
username = 'test1'
domain = username + '.m.csclub.cloud'
filename = f'member_{username}_{domain}'
ip_address = '172.19.134.11'
vhost_mgr.create_vhost(username, domain, ip_address)
path = os.path.join(vhost_mgr.vhost_dir, filename)
assert os.path.isfile(path)
assert vhost_mgr.get_num_vhosts(username) == 1
assert vhost_mgr.get_vhosts(username) == [{
'domain': domain, 'ip_address': ip_address,
}]
domain2 = 'app.' + domain
vhost_mgr.create_vhost(username, domain2, ip_address)
assert vhost_mgr.get_num_vhosts(username) == 2
vhost_mgr.delete_vhost(username, domain)
assert vhost_mgr.get_num_vhosts(username) == 1
vhost_mgr.delete_all_vhosts_for_user(username)
assert vhost_mgr.get_num_vhosts(username) == 0

@ -80,3 +80,5 @@ ssl_cert_path = /etc/ssl/private/csclub.cloud.chain
ssl_key_path = /etc/ssl/private/csclub.cloud.key
max_vhosts_per_account = 10
members_domain = m.csclub.cloud
ip_range_min = 172.19.134.10
ip_range_max = 172.19.134.160

@ -79,3 +79,5 @@ ssl_cert_path = /etc/ssl/private/csclub.cloud.chain
ssl_key_path = /etc/ssl/private/csclub.cloud.key
max_vhosts_per_account = 10
members_domain = m.csclub.cloud
ip_range_min = 172.19.134.10
ip_range_max = 172.19.134.160

@ -59,7 +59,7 @@ def cfg(_drone_hostname_mock):
def delete_test_princs(krb_srv):
proc = subprocess.run([
'kadmin', '-k', '-p', krb_srv.admin_principal, 'listprincs', 'test_*',
'kadmin', '-k', '-p', krb_srv.admin_principal, 'listprincs', 'test*',
], text=True, capture_output=True, check=True)
princs = [line.strip() for line in proc.stdout.splitlines()]
for princ in princs:
@ -268,7 +268,17 @@ def postgresql_srv(cfg):
@pytest.fixture(scope='session')
def cloud_srv(cfg):
def vhost_dir_setup(cfg):
vhost_dir = cfg.get('cloud vhosts_config_dir')
if os.path.isdir(vhost_dir):
shutil.rmtree(vhost_dir)
os.makedirs(vhost_dir)
yield
shutil.rmtree(vhost_dir)
@pytest.fixture(scope='session')
def cloud_srv(cfg, vhost_dir_setup):
_cloud_srv = CloudService()
component.getGlobalSiteManager().registerUtility(_cloud_srv, ICloudService)
return _cloud_srv
@ -350,7 +360,7 @@ _new_user_id_counter = 10001
@pytest.fixture # noqa: E302
def new_user(client, g_admin_ctx, ldap_srv_session): # noqa: F811
global _new_user_id_counter
uid = 'test_' + str(_new_user_id_counter)
uid = 'test' + str(_new_user_id_counter)
_new_user_id_counter += 1
status, data = client.post('/api/members', json={
'uid': uid,

@ -73,3 +73,6 @@ class CeodTestClient:
def delete(self, path, principal=None, need_auth=True, delegate=True, **kwargs):
return self.request('DELETE', path, principal, need_auth, delegate, **kwargs)
def put(self, path, principal=None, need_auth=True, delegate=True, **kwargs):
return self.request('PUT', path, principal, need_auth, delegate, **kwargs)

Loading…
Cancel
Save