add CLI and unit tests

pull/42/head
Max Erenberg 1 year ago
parent a16ca8f5fd
commit e863678236
  1. 2
      ceo/cli/entrypoint.py
  2. 21
      ceo/cli/registry.py
  3. 2
      ceod/model/CloudResourceManager.py
  4. 16
      tests/MockHarborServer.py
  5. 12
      tests/ceo/cli/test_cloud.py
  6. 16
      tests/ceod/api/test_cloud.py
  7. 24
      tests/ceod/model/test_container_registry.py
  8. 22
      tests/conftest.py

@ -9,6 +9,7 @@ from .postgresql import postgresql
from .mailman import mailman
from .cloud import cloud
from .k8s import k8s
from .registry import registry
@click.group()
@ -25,3 +26,4 @@ cli.add_command(postgresql)
cli.add_command(mailman)
cli.add_command(cloud)
cli.add_command(k8s)
cli.add_command(registry)

@ -0,0 +1,21 @@
import click
from ..utils import http_post
from .utils import handle_sync_response
@click.group(short_help='Manage your container registry account')
def registry():
pass
@registry.group(short_help='Manage your container registry project')
def project():
pass
@project.command(short_help='Create a registry project')
def create():
resp = http_post('/api/cloud/registry/projects')
handle_sync_response(resp)
click.echo('Congratulations! Your registry project was successfully created.')

@ -176,7 +176,7 @@ class CloudResourceManager:
}
now = utils.get_current_datetime()
if self._in_grace_period():
if self._in_grace_period(now):
return result
# get a list of all cloud services each user is using

@ -8,6 +8,7 @@ class MockHarborServer(MockHTTPServerBase):
prefix = '/api/v2.0'
routes = [
web.get(prefix + '/users', self.users_get_handler),
web.get(prefix + '/projects', self.projects_get_handler),
web.post(prefix + '/projects', self.projects_post_handler),
web.post(prefix + '/projects/{project}/members', self.members_post_handler),
web.get(prefix + '/projects/{project}/repositories', self.repositories_get_handler),
@ -15,7 +16,7 @@ class MockHarborServer(MockHTTPServerBase):
web.delete(prefix + '/projects/{project}', self.projects_delete_handler),
# for debugging purposes
web.post('/reset', self.reset_handler),
web.post('/users/{username}', self.users_post_handler),
web.delete('/users/{username}', self.users_delete_handler),
]
super().__init__(port, routes)
@ -26,6 +27,12 @@ class MockHarborServer(MockHTTPServerBase):
'exec1': [],
}
async def projects_get_handler(self, request):
return web.json_response([
{'name': name, 'project_id': i + 1}
for i, name in enumerate(self.projects.keys())
])
async def projects_delete_handler(self, request):
project_name = request.match_info['project']
if project_name not in self.projects:
@ -78,14 +85,17 @@ class MockHarborServer(MockHTTPServerBase):
self.projects[project_name] = ['repo1', 'repo2']
return web.Response(text='', status=201)
async def users_post_handler(self, request):
async def users_delete_handler(self, request):
username = request.match_info['username']
self.users.remove(username)
return web.Response(text='OK\n', status=201)
async def reset_handler(self, request):
def reset(self):
self.users.clear()
self.projects.clear()
async def reset_handler(self, request):
self.reset()
return web.Response(text='OK\n')

@ -76,3 +76,15 @@ def test_k8s_account_activate(cli_setup, new_user):
assert result.exit_code == 0
assert result.output == expected
assert os.path.isfile(os.path.join(new_user.home_directory, '.kube', 'config'))
def test_registry_project_create(cli_setup, mock_harbor_server, new_user):
uid = new_user.uid
runner = CliRunner()
mock_harbor_server.reset()
mock_harbor_server.users.append(uid)
with gssapi_token_ctx(uid):
result = runner.invoke(cli, ['registry', 'project', 'create'])
expected = 'Congratulations! Your registry project was successfully created.\n'
assert result.exit_code == 0
assert result.output == expected

@ -35,8 +35,8 @@ def test_create_account(client, mock_cloud_server, new_user, ldap_conn):
def test_purge_accounts(
client, mock_cloud_server, cloud_mgr, mock_mail_server, new_user,
ldap_conn,
client, mock_cloud_server, cloud_mgr, mock_mail_server,
mock_harbor_server, new_user, ldap_conn,
):
uid = new_user.uid
mock_cloud_server.clear()
@ -167,12 +167,14 @@ def test_cloud_vhosts(cfg, client, new_user, ldap_conn):
assert status == 403
def test_cloud_vhosts_purged_account(
cfg, client, mock_cloud_server, mock_mail_server, new_user, ldap_conn,
def test_cloud_resources_purged_account(
cfg, client, mock_cloud_server, mock_mail_server, mock_harbor_server,
new_user, ldap_conn,
):
uid = new_user.uid
members_domain = cfg.get('cloud vhosts_members_domain')
mock_cloud_server.clear()
mock_harbor_server.reset()
current_term = Term.current()
beginning_of_term = current_term.to_datetime()
domain1 = uid + '.' + members_domain
@ -182,6 +184,9 @@ def test_cloud_vhosts_purged_account(
client.put(
f'/api/cloud/vhosts/{domain1}', json={'ip_address': ip1},
principal=uid)
mock_harbor_server.users.append(uid)
client.post('/api/cloud/registry/projects', principal=uid)
assert len(mock_harbor_server.projects) == 1
expire_member(new_user, ldap_conn)
with patch.object(ceo_common_utils, 'get_current_datetime') as now_mock:
@ -198,6 +203,9 @@ def test_cloud_vhosts_purged_account(
assert status == 200
assert data == {'vhosts': []}
# registry project should have been deleted
assert len(mock_harbor_server.projects) == 0
mock_mail_server.messages.clear()

@ -0,0 +1,24 @@
import pytest
from ceo_common.errors import UserNotFoundError
def test_registry(mock_harbor_server, registry_srv):
mock_harbor_server.reset()
username = 'test1'
with pytest.raises(UserNotFoundError):
registry_srv.create_project_for_user(username)
mock_harbor_server.users.append(username)
registry_srv.create_project_for_user(username)
assert username in mock_harbor_server.projects
# trying to create a project with the same name should have no effect
registry_srv.create_project_for_user(username)
assert registry_srv.get_accounts() == [username]
registry_srv.delete_project_for_user(username)
assert username not in mock_harbor_server.projects
# trying to delete a nonexistent project should have no effect
registry_srv.delete_project_for_user(username)

@ -26,16 +26,18 @@ from .utils import ( # noqa: F401
from ceo_common.interfaces import IConfig, IKerberosService, ILDAPService, \
IFileService, IMailmanService, IHTTPClient, IUWLDAPService, IMailService, \
IDatabaseService, ICloudStackService, IKubernetesService, IVHostManager, \
ICloudResourceManager
ICloudResourceManager, IContainerRegistryService
from ceo_common.model import Config, HTTPClient, Term
from ceod.api import create_app
from ceod.db import MySQLService, PostgreSQLService
from ceod.model import KerberosService, LDAPService, FileService, User, \
MailmanService, Group, UWLDAPService, UWLDAPRecord, MailService, \
CloudStackService, KubernetesService, VHostManager, CloudResourceManager
CloudStackService, KubernetesService, VHostManager, CloudResourceManager, \
ContainerRegistryService
from .MockSMTPServer import MockSMTPServer
from .MockMailmanServer import MockMailmanServer
from .MockCloudStackServer import MockCloudStackServer
from .MockHarborServer import MockHarborServer
from .conftest_ceod_api import client # noqa: F401
from .conftest_ceo import cli_setup # noqa: F401
@ -256,6 +258,21 @@ def mock_cloud_server():
mock_server.stop()
@pytest.fixture(scope='session')
def mock_harbor_server():
mock_server = MockHarborServer()
mock_server.start()
yield mock_server
mock_server.stop()
@pytest.fixture(scope='session')
def registry_srv(cfg):
reg_srv = ContainerRegistryService()
component.getGlobalSiteManager().registerUtility(reg_srv, IContainerRegistryService)
return reg_srv
@pytest.fixture(scope='session')
def mysql_srv(cfg):
mysql_srv = MySQLService()
@ -322,6 +339,7 @@ def app(
cloudstack_srv,
vhost_mgr,
k8s_srv,
registry_srv,
cloud_mgr,
):
app = create_app({'TESTING': True})

Loading…
Cancel
Save