362 lines
12 KiB
Python
362 lines
12 KiB
Python
import io
|
|
import json
|
|
import tarfile
|
|
|
|
import httmock
|
|
import pytest
|
|
from httmock import HTTMock
|
|
from test_manager import login
|
|
from webtest import Upload
|
|
|
|
from hobo.applications.models import Application
|
|
from hobo.environment.models import Authentic, Wcs
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
WCS_AVAILABLE_OBJECTS = {
|
|
"data": [
|
|
{
|
|
"id": "forms",
|
|
"text": "Forms",
|
|
"singular": "Form",
|
|
"urls": {"list": "https://wcs.example.invalid/api/export-import/forms/"},
|
|
},
|
|
{
|
|
"id": "cards",
|
|
"text": "Card Models",
|
|
"singular": "Card Model",
|
|
"urls": {"list": "https://wcs.example.invalid/api/export-import/cards/"},
|
|
},
|
|
{
|
|
"id": "workflows",
|
|
"text": "Workflows",
|
|
"singular": "Workflow",
|
|
"urls": {"list": "https://wcs.example.invalid/api/export-import/workflows/"},
|
|
},
|
|
{
|
|
"id": "blocks",
|
|
"text": "Blocks",
|
|
"singular": "Block of fields",
|
|
"minor": True,
|
|
"urls": {"list": "https://wcs.example.invalid/api/export-import/blocks/"},
|
|
},
|
|
{
|
|
"id": "data-sources",
|
|
"text": "Data Sources",
|
|
"singular": "Data Source",
|
|
"minor": True,
|
|
"urls": {"list": "https://wcs.example.invalid/api/export-import/data-sources/"},
|
|
},
|
|
{
|
|
"id": "mail-templates",
|
|
"text": "Mail Templates",
|
|
"singular": "Mail Template",
|
|
"minor": True,
|
|
"urls": {"list": "https://wcs.example.invalid/api/export-import/mail-templates/"},
|
|
},
|
|
{
|
|
"id": "wscalls",
|
|
"text": "Webservice Calls",
|
|
"singular": "Webservice Call",
|
|
"minor": True,
|
|
"urls": {"list": "https://wcs.example.invalid/api/export-import/wscalls/"},
|
|
},
|
|
]
|
|
}
|
|
|
|
WCS_AVAILABLE_FORMS = {
|
|
"data": [
|
|
{
|
|
"id": "test-form",
|
|
"text": "Test Form",
|
|
"type": "forms",
|
|
"urls": {
|
|
"export": "https://wcs.example.invalid/api/export-import/forms/test-form/",
|
|
"dependencies": "https://wcs.example.invalid/api/export-import/forms/test-form/dependencies/",
|
|
},
|
|
},
|
|
{
|
|
"id": "test2-form",
|
|
"text": "Second Test Form",
|
|
"type": "forms",
|
|
"urls": {
|
|
"export": "https://wcs.example.invalid/api/export-import/forms/test2-form/",
|
|
"dependencies": "https://wcs.example.invalid/api/export-import/forms/test2-form/dependencies/",
|
|
},
|
|
},
|
|
]
|
|
}
|
|
|
|
WCS_FORM_DEPENDENCIES = {
|
|
"data": [
|
|
{
|
|
"id": "test-card",
|
|
"text": "Test Card",
|
|
"type": "cards",
|
|
"urls": {
|
|
"export": "https://wcs.example.invalid/api/export-import/cards/test-card/",
|
|
"dependencies": "https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/",
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def mocked_http(url, request):
|
|
assert '&signature=' in url.query
|
|
|
|
if url.netloc == 'idp.example.invalid':
|
|
if url.path == '/api/roles/':
|
|
return {
|
|
'content': json.dumps({'name': 'test', 'uuid': '123', 'slug': 'test'}),
|
|
'status_code': 200,
|
|
}
|
|
if url.path == '/api/provision/':
|
|
return {
|
|
'content': json.dumps({'name': 'test', 'uuid': '123', 'slug': 'test'}),
|
|
'status_code': 200,
|
|
}
|
|
return {'content': '{}', 'status_code': 500}
|
|
|
|
if url.netloc == 'wcs.example.invalid' and url.path == '/api/export-import/':
|
|
return {'content': json.dumps(WCS_AVAILABLE_OBJECTS), 'status_code': 200}
|
|
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'content': json.dumps(WCS_AVAILABLE_FORMS), 'status_code': 200}
|
|
|
|
if url.path == '/api/export-import/forms/test-form/dependencies/':
|
|
return {'content': json.dumps(WCS_FORM_DEPENDENCIES), 'status_code': 200}
|
|
|
|
if url.path.endswith('/dependencies/'):
|
|
return {'content': json.dumps({'data': []}), 'status_code': 200}
|
|
|
|
if url.path == '/api/export-import/forms/test-form/':
|
|
return {'content': '<formdef/>', 'status_code': 200, 'headers': {'content-length': '10'}}
|
|
|
|
if url.path == '/api/export-import/cards/test-card/':
|
|
return {'content': '<carddef/>', 'status_code': 200, 'headers': {'content-length': '10'}}
|
|
|
|
if url.path == '/api/export-import/bundle-import/':
|
|
return {'content': '{}', 'status_code': 200}
|
|
|
|
|
|
def test_create_application(app, admin_user, settings):
|
|
Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
'wcs': {
|
|
'foobar': {
|
|
'title': 'Foobar',
|
|
'url': 'https://wcs.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
}
|
|
}
|
|
}
|
|
|
|
login(app)
|
|
|
|
resp = app.get('/applications/')
|
|
resp = resp.click('Create')
|
|
resp.form['name'] = 'Test'
|
|
resp = resp.form.submit()
|
|
|
|
with HTTMock(mocked_http):
|
|
resp = resp.follow()
|
|
assert 'You should now assemble the different parts of your application.' in resp.text
|
|
|
|
# edit metadata
|
|
resp = resp.click('Metadata')
|
|
resp.form['description'] = 'Lorem ipsum'
|
|
resp = resp.form.submit().follow()
|
|
|
|
# add forms
|
|
assert '/add/forms/' in resp
|
|
resp = resp.click('Forms')
|
|
assert resp.form.fields['elements'][0]._value == 'test-form'
|
|
assert resp.form.fields['elements'][1]._value == 'test2-form'
|
|
resp.form.fields['elements'][0].checked = True
|
|
resp = resp.form.submit().follow()
|
|
assert Application.objects.get(slug='test').elements.count() == 1
|
|
element = Application.objects.get(slug='test').elements.all()[0]
|
|
assert element.slug == 'test-form'
|
|
|
|
resp = resp.click('Scan dependencies').follow()
|
|
assert Application.objects.get(slug='test').elements.count() == 2
|
|
assert 'Test Card' in resp.text
|
|
|
|
resp = resp.click('Generate application bundle').follow()
|
|
resp = resp.click('Download')
|
|
assert resp.content_type == 'application/x-tar'
|
|
# uncompressed tar, primitive check of contents
|
|
assert b'<formdef/>' in resp.content
|
|
assert b'<carddef/>' in resp.content
|
|
|
|
|
|
def test_delete_application(app, admin_user, settings):
|
|
Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
'wcs': {
|
|
'foobar': {
|
|
'title': 'Foobar',
|
|
'url': 'https://wcs.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
}
|
|
}
|
|
}
|
|
|
|
login(app)
|
|
|
|
Application.objects.create(name='AppToDelete', slug='app_to_delete')
|
|
Application.objects.create(name='OtherApp', slug='other_app')
|
|
|
|
assert Application.objects.count() == 2
|
|
|
|
resp = app.get('/applications/manifest/app_to_delete/delete/')
|
|
|
|
resp = resp.forms[0].submit()
|
|
resp = resp.follow()
|
|
|
|
assert '/applications/' in resp
|
|
assert 'AppToDelete' not in resp.text
|
|
|
|
assert Application.objects.count() == 1
|
|
assert Application.objects.first().name == 'OtherApp'
|
|
|
|
|
|
@pytest.fixture
|
|
def app_bundle():
|
|
tar_io = io.BytesIO()
|
|
with tarfile.open(mode='w', fileobj=tar_io) as tar:
|
|
manifest_json = {
|
|
'application': 'Test',
|
|
'slug': 'test',
|
|
'description': '',
|
|
'elements': [
|
|
{'type': 'forms', 'slug': 'test', 'name': 'test', 'auto-dependency': False},
|
|
{'type': 'blocks', 'slug': 'test', 'name': 'test', 'auto-dependency': True},
|
|
{'type': 'workflows', 'slug': 'test', 'name': 'test', 'auto-dependency': True},
|
|
],
|
|
}
|
|
manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
|
|
tarinfo = tarfile.TarInfo('manifest.json')
|
|
tarinfo.size = len(manifest_fd.getvalue())
|
|
tar.addfile(tarinfo, fileobj=manifest_fd)
|
|
|
|
return tar_io.getvalue()
|
|
|
|
|
|
def test_deploy_application(app, admin_user, settings, app_bundle):
|
|
Application.objects.all().delete()
|
|
Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
'wcs': {
|
|
'foobar': {
|
|
'title': 'Foobar',
|
|
'url': 'https://wcs.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
}
|
|
}
|
|
}
|
|
|
|
login(app)
|
|
|
|
resp = app.get('/applications/')
|
|
for i in range(2):
|
|
resp = resp.click('Install')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with HTTMock(mocked_http):
|
|
resp = resp.form.submit().follow()
|
|
|
|
assert Application.objects.count() == 1
|
|
assert Application.objects.get(slug='test').name == 'Test'
|
|
assert Application.objects.get(slug='test').elements.count() == 3
|
|
|
|
|
|
@pytest.fixture
|
|
def app_bundle_roles():
|
|
tar_io = io.BytesIO()
|
|
with tarfile.open(mode='w', fileobj=tar_io) as tar:
|
|
manifest_json = {
|
|
'application': 'Test',
|
|
'slug': 'test',
|
|
'description': '',
|
|
'elements': [
|
|
{'type': 'forms', 'slug': 'test', 'name': 'test', 'auto-dependency': False},
|
|
{'type': 'roles', 'slug': 'test-role', 'name': 'test', 'auto-dependency': True},
|
|
{'type': 'roles', 'slug': 'test-role2', 'name': 'test2', 'auto-dependency': True},
|
|
],
|
|
}
|
|
manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
|
|
tarinfo = tarfile.TarInfo('manifest.json')
|
|
tarinfo.size = len(manifest_fd.getvalue())
|
|
tar.addfile(tarinfo, fileobj=manifest_fd)
|
|
|
|
role_json = {'name': 'Test', 'slug': 'test-role', 'uuid': '061e5de7023946c79a2f7f1273afc5a2'}
|
|
role_json_fd = io.BytesIO(json.dumps(role_json, indent=2).encode())
|
|
tarinfo = tarfile.TarInfo('roles/test-role')
|
|
tarinfo.size = len(role_json_fd.getvalue())
|
|
tar.addfile(tarinfo, fileobj=role_json_fd)
|
|
|
|
role_json = {'name': 'Test', 'slug': 'test-role2', 'uuid': '061e5de7023946c79a2f7f1273afc5a3'}
|
|
role_json_fd = io.BytesIO(json.dumps(role_json, indent=2).encode())
|
|
tarinfo = tarfile.TarInfo('roles/test-role2')
|
|
tarinfo.size = len(role_json_fd.getvalue())
|
|
tar.addfile(tarinfo, fileobj=role_json_fd)
|
|
|
|
return tar_io.getvalue()
|
|
|
|
|
|
def test_deploy_application_roles(app, admin_user, settings, app_bundle_roles):
|
|
Application.objects.all().delete()
|
|
Authentic.objects.create(base_url='https://idp.example.invalid', slug='idp', title='Foobar')
|
|
Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
'authentic': {
|
|
'idp': {
|
|
'title': 'Foobar',
|
|
'url': 'https://idp.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
}
|
|
},
|
|
'wcs': {
|
|
'foobar': {
|
|
'title': 'Foobar',
|
|
'url': 'https://wcs.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
}
|
|
},
|
|
}
|
|
|
|
login(app)
|
|
|
|
resp = app.get('/applications/')
|
|
for i in range(2):
|
|
resp = resp.click('Install')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
|
|
with HTTMock(httmock.remember_called(mocked_http)):
|
|
resp = resp.form.submit().follow()
|
|
# roles
|
|
assert mocked_http.call['requests'][0].url.startswith(
|
|
'https://idp.example.invalid/api/roles/?update'
|
|
)
|
|
assert mocked_http.call['requests'][1].url.startswith(
|
|
'https://idp.example.invalid/api/provision/'
|
|
)
|
|
assert mocked_http.call['requests'][2].url.startswith(
|
|
'https://idp.example.invalid/api/roles/?update'
|
|
)
|
|
assert mocked_http.call['requests'][3].url.startswith(
|
|
'https://idp.example.invalid/api/provision/'
|
|
)
|
|
# then form
|
|
assert 'wcs.example.invalid' in mocked_http.call['requests'][4].url
|
|
assert mocked_http.call['count'] == 5
|