2075 lines
81 KiB
Python
2075 lines
81 KiB
Python
import copy
|
|
import datetime
|
|
import io
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import tarfile
|
|
|
|
import httmock
|
|
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils.timezone import now
|
|
from httmock import HTTMock
|
|
from pyquery import PyQuery
|
|
from webtest import Upload
|
|
|
|
from hobo.applications.models import (
|
|
Application,
|
|
AsyncJob,
|
|
DeploymentError,
|
|
Element,
|
|
Parameter,
|
|
Relation,
|
|
ScanError,
|
|
Version,
|
|
)
|
|
from hobo.environment.models import Authentic, Variable, Wcs
|
|
|
|
from .test_manager import login
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
TESTS_DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
|
|
|
|
|
|
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': 'comment-templates-categories',
|
|
'text': 'Categories (comment templates)',
|
|
'singular': 'Category (comment templates)',
|
|
'minor': True,
|
|
'urls': {'list': 'https://wcs.example.invalid/api/export-import/comment-templates-categories/'},
|
|
},
|
|
{
|
|
'id': 'wscalls',
|
|
'text': 'Webservice Calls',
|
|
'singular': 'Webservice Call',
|
|
'minor': True,
|
|
'urls': {'list': 'https://wcs.example.invalid/api/export-import/wscalls/'},
|
|
},
|
|
{
|
|
'id': 'roles',
|
|
'text': 'Roles',
|
|
'singular': 'Role',
|
|
'minor': True,
|
|
'urls': {'list': 'https://wcs.example.invalid/api/export-import/roles/'},
|
|
},
|
|
]
|
|
}
|
|
|
|
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/',
|
|
'redirect': 'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/',
|
|
},
|
|
},
|
|
{
|
|
'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/',
|
|
},
|
|
},
|
|
{
|
|
'id': 'foo2-form',
|
|
'text': 'Foo2 Test Form',
|
|
'type': 'forms',
|
|
'category': 'Foo',
|
|
'urls': {
|
|
'export': 'https://wcs.example.invalid/api/export-import/forms/foo2-form/',
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/forms/foo2-form/dependencies/',
|
|
},
|
|
},
|
|
{
|
|
'id': 'foo-form',
|
|
'text': 'Foo Test Form',
|
|
'type': 'forms',
|
|
'category': 'Foo',
|
|
'urls': {
|
|
'export': 'https://wcs.example.invalid/api/export-import/forms/foo-form/',
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/forms/foo-form/dependencies/',
|
|
},
|
|
},
|
|
{
|
|
'id': 'bar-form',
|
|
'text': 'Bar Test Form',
|
|
'type': 'forms',
|
|
'category': 'Bar',
|
|
'urls': {
|
|
'export': 'https://wcs.example.invalid/api/export-import/forms/bar-form/',
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/forms/bar-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/',
|
|
},
|
|
},
|
|
{
|
|
'id': 'test-role',
|
|
'text': 'Test Role',
|
|
'type': 'roles',
|
|
'urls': {}, # no urls in cache
|
|
},
|
|
]
|
|
}
|
|
|
|
|
|
class StatefulHTTMock(HTTMock):
|
|
def __enter__(self):
|
|
self.orig_wcs_available_forms = copy.deepcopy(WCS_AVAILABLE_FORMS)
|
|
return super().__enter__()
|
|
|
|
def __exit__(self, *args, **kwargs):
|
|
WCS_AVAILABLE_FORMS['data'] = self.orig_wcs_available_forms['data']
|
|
return super().__exit__(*args, **kwargs)
|
|
|
|
|
|
def mocked_http(url, request):
|
|
assert url.query.count('&signature=') == 1
|
|
|
|
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/':
|
|
# alter WCS_AVAILABLE_FORMS response with newly installed forms
|
|
with tarfile.open(mode='r', fileobj=io.BytesIO(request.original.files[0][1])) as tar:
|
|
manifest_json = json.load(tar.extractfile('manifest.json'))
|
|
for element in manifest_json.get('elements'):
|
|
if element['type'] not in [x['id'] for x in WCS_AVAILABLE_OBJECTS['data']]:
|
|
continue
|
|
if element['type'] == 'forms' and element['slug'] not in [
|
|
x['id'] for x in WCS_AVAILABLE_FORMS['data']
|
|
]:
|
|
WCS_AVAILABLE_FORMS['data'].append(
|
|
{'id': element['slug'], 'text': element['name'], 'type': 'forms'}
|
|
)
|
|
return {
|
|
'content': json.dumps({'err': 0, 'url': 'https://wcs.example.invalid/api/jobs/job-uuid/'}),
|
|
'status_code': 200,
|
|
}
|
|
|
|
if url.path == '/api/export-import/bundle-check/':
|
|
return {
|
|
'content': json.dumps(
|
|
{
|
|
'data': {
|
|
'differences': [],
|
|
'unknown_elements': [],
|
|
'no_history_elements': [{'type': 'forms', 'slug': 'legacy'}],
|
|
'legacy_elements': [],
|
|
}
|
|
}
|
|
),
|
|
'status_code': 200,
|
|
}
|
|
|
|
return {'content': json.dumps({'data': []}), 'status_code': 200}
|
|
|
|
|
|
@pytest.mark.parametrize('analyze', [True, False])
|
|
def test_create_application(app, admin_user, settings, analyze):
|
|
Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
'wcs': {
|
|
'blah': {
|
|
# simulate an instance from another collectivity
|
|
'title': 'Unknown',
|
|
'url': 'https://unknown.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
},
|
|
'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 StatefulHTTMock(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.form['documentation_url'] = 'http://foo.bar'
|
|
resp.form['visible'] = False
|
|
resp = resp.form.submit().follow()
|
|
application = Application.objects.get(slug='test')
|
|
assert application.icon.name == ''
|
|
assert application.documentation_url == 'http://foo.bar'
|
|
assert application.visible is False
|
|
|
|
# add forms
|
|
assert '/add/forms/' in resp
|
|
resp = resp.click('Forms')
|
|
assert len(resp.pyquery('.application-elements div')) == 3
|
|
assert resp.pyquery('.application-elements div:nth-child(1) h4').text() == 'Bar'
|
|
assert PyQuery(resp.pyquery('.application-elements div:nth-child(1) input')[0]).val() == 'bar-form'
|
|
assert resp.pyquery('.application-elements div:nth-child(2) h4').text() == 'Foo'
|
|
assert PyQuery(resp.pyquery('.application-elements div:nth-child(2) input')[0]).val() == 'foo-form'
|
|
assert PyQuery(resp.pyquery('.application-elements div:nth-child(2) input')[1]).val() == 'foo2-form'
|
|
assert resp.pyquery('.application-elements div:nth-child(3) h4').text() == 'Uncategorized'
|
|
assert PyQuery(resp.pyquery('.application-elements div:nth-child(3) input')[0]).val() == 'test2-form'
|
|
assert PyQuery(resp.pyquery('.application-elements div:nth-child(3) input')[1]).val() == 'test-form'
|
|
resp.form.fields['elements'][4].checked = True
|
|
resp = resp.form.submit().follow()
|
|
assert application.elements.count() == 1
|
|
element = application.elements.all()[0]
|
|
assert element.slug == 'test-form'
|
|
assert 'Test Card' not in resp.text
|
|
|
|
if analyze:
|
|
resp = resp.click('Scan dependencies').follow()
|
|
assert Application.objects.get(slug='test').elements.count() == 3
|
|
|
|
resp = resp.click('Generate application bundle')
|
|
resp.form['number'] = '1.0'
|
|
resp.form['notes'] = 'Foo bar blah.'
|
|
resp = resp.form.submit().follow()
|
|
assert 'Test Card' in resp.text
|
|
version = Version.objects.latest('pk')
|
|
assert version.number == '1.0.%s.0' % now().strftime('%Y%m%d')
|
|
assert version.notes == 'Foo bar blah.'
|
|
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
|
|
assert b'"icon": null' in resp.content
|
|
assert b'"documentation_url": "http://foo.bar"' in resp.content
|
|
assert b'"version_number": "1.0.%s.0"' % now().strftime('%Y%m%d').encode() in resp.content
|
|
assert b'"version_notes": "Foo bar blah."' in resp.content
|
|
assert b'"visible": false' in resp.content
|
|
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
versions = [e.text() for e in resp.pyquery('h3').items()]
|
|
assert versions.count('1.0.%s.0' % now().strftime('%Y%m%d')) == 1
|
|
assert resp.text.count('Creating application bundle') == 1
|
|
assert 'Compare' not in resp
|
|
resp = resp.click(href='/applications/manifest/test/download/%s/' % version.pk)
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert resp.headers[
|
|
'Content-Disposition'
|
|
] == 'attachment; filename="test-1.0.%s.0.tar"' % now().strftime('%Y%m%d')
|
|
assert b'"version_number": "1.0.%s.0"' % now().strftime('%Y%m%d').encode() in resp.content
|
|
|
|
# add an icon
|
|
resp = app.get('/applications/manifest/test/metadata/')
|
|
with open(os.path.join(TESTS_DATA_DIR, 'black.jpeg'), mode='rb') as icon_fd:
|
|
resp.form['icon'] = Upload('foo.jpeg', icon_fd.read(), 'image/jpeg')
|
|
resp.form['documentation_url'] = '' # and reset documentation_url
|
|
resp.form['visible'] = True
|
|
resp = resp.form.submit().follow()
|
|
application.refresh_from_db()
|
|
assert re.match(r'applications/icons/foo(_\w+)?.jpeg', application.icon.name)
|
|
assert application.documentation_url == ''
|
|
assert application.visible is True
|
|
|
|
# try an icon in an invalid format
|
|
resp = app.get('/applications/manifest/test/metadata/')
|
|
resp.form['icon'] = Upload('test.txt', b'hello', 'text/plain')
|
|
resp = resp.form.submit()
|
|
assert 'The icon must be in JPEG or PNG format' in resp
|
|
application.refresh_from_db()
|
|
assert re.match(r'applications/icons/foo(_\w+)?.jpeg', application.icon.name)
|
|
|
|
resp = app.get('/applications/manifest/test/')
|
|
resp = resp.click('Generate application bundle')
|
|
resp.form['number'] = '2.0'
|
|
resp.form['notes'] = 'Foo bar blah. But with an icon.'
|
|
resp = resp.form.submit().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
|
|
assert b'"icon": "foo' in resp.content
|
|
assert b'"documentation_url": ""' in resp.content
|
|
assert b'"version_number": "2.0.%s.0"' % now().strftime('%Y%m%d').encode() in resp.content
|
|
assert b'"version_notes": "Foo bar blah. But with an icon."' in resp.content
|
|
assert b'"visible": true' in resp.content
|
|
version = Version.objects.latest('pk')
|
|
assert version.number == '2.0.%s.0' % now().strftime('%Y%m%d')
|
|
assert version.notes == 'Foo bar blah. But with an icon.'
|
|
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
versions = [e.text() for e in resp.pyquery('h3').items()]
|
|
assert versions.count('1.0.%s.0' % now().strftime('%Y%m%d')) == 1
|
|
assert versions.count('2.0.%s.0' % now().strftime('%Y%m%d')) == 1
|
|
assert resp.text.count('Compare') == 2
|
|
assert resp.text.count('Creating application bundle') == 2
|
|
resp = resp.click(
|
|
href='/applications/manifest/test/download/%s/' % Version.objects.get(number__startswith='1.0').pk
|
|
)
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert resp.headers[
|
|
'Content-Disposition'
|
|
] == 'attachment; filename="test-1.0.%s.0.tar"' % now().strftime('%Y%m%d')
|
|
assert b'"version_number": "1.0.%s.0"' % now().strftime('%Y%m%d').encode() in resp.content
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
resp = resp.click(
|
|
href='/applications/manifest/test/download/%s/' % Version.objects.get(number__startswith='2.0').pk
|
|
)
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert resp.headers[
|
|
'Content-Disposition'
|
|
] == 'attachment; filename="test-2.0.%s.0.tar"' % now().strftime('%Y%m%d')
|
|
assert b'"version_number": "2.0.%s.0"' % now().strftime('%Y%m%d').encode() in resp.content
|
|
|
|
resp = app.get('/applications/manifest/test/')
|
|
resp = resp.click('Generate application bundle')
|
|
resp.form['number'] = '2.0'
|
|
resp.form['notes'] = 'Foo bar blah. But with an icon.'
|
|
resp = resp.form.submit().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
|
|
assert b'"icon": "foo' in resp.content
|
|
assert b'"documentation_url": ""' in resp.content
|
|
assert b'"version_number": "2.0.%s.1"' % now().strftime('%Y%m%d').encode() in resp.content
|
|
assert b'"version_notes": "Foo bar blah. But with an icon."' in resp.content
|
|
assert b'"visible": true' in resp.content
|
|
version = Version.objects.latest('pk')
|
|
assert version.number == '2.0.%s.1' % now().strftime('%Y%m%d')
|
|
assert version.notes == 'Foo bar blah. But with an icon.'
|
|
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
versions = [e.text() for e in resp.pyquery('h3').items()]
|
|
assert resp.text.count('Creating application bundle') == 3
|
|
assert resp.text.count('Compare') == 3
|
|
resp = resp.click('Compare', index=0)
|
|
assert 'Compare version 2.0.%s.1 to:' % now().strftime('%Y%m%d') in resp
|
|
version1 = Version.objects.get(number__startswith='1.0')
|
|
version2 = Version.objects.filter(number__startswith='2.0').latest('-pk')
|
|
version3 = Version.objects.filter(number__startswith='2.0').latest('pk')
|
|
assert resp.form['version'].options == [
|
|
(str(version2.pk), False, version2.number),
|
|
(str(version1.pk), False, version1.number),
|
|
]
|
|
resp = resp.form.submit()
|
|
assert resp.location.endswith(
|
|
'/applications/manifest/test/version/compare/?version1=%s&version2=%s'
|
|
% (version3.pk, version2.pk)
|
|
)
|
|
resp = resp.follow()
|
|
assert 'Version %s' % version3.number in resp
|
|
assert 'Version %s' % version2.number in resp
|
|
resp = resp.click('Compare elements definitions')
|
|
assert 'Version %s' % version3.number in resp
|
|
assert 'Version %s' % version2.number in resp
|
|
assert (
|
|
'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/?application=test&version1=%s&version2=%s&compare'
|
|
% (version2.number, version3.number)
|
|
in resp
|
|
)
|
|
assert (
|
|
'https://wcs.example.invalid/api/export-import/cards/test-card/redirect/?application=test&version1=%s&version2=%s&compare'
|
|
% (version2.number, version3.number)
|
|
in resp
|
|
)
|
|
assert 'Test Form <span class="extra-info">- Form</span>' in resp
|
|
assert 'Test Card <span class="extra-info">- Card Model</span>' in resp
|
|
assert 'Element not found' not in resp
|
|
|
|
Element.objects.filter(type='cards').delete()
|
|
resp = app.get(
|
|
'/applications/manifest/test/version/compare/?version1=%s&version2=%s&mode=elements'
|
|
% (version3.pk, version2.pk)
|
|
)
|
|
assert 'Version %s' % version3.number in resp
|
|
assert 'Version %s' % version2.number in resp
|
|
assert (
|
|
'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/?application=test&version1=%s&version2=%s&compare'
|
|
% (version2.number, version3.number)
|
|
in resp
|
|
)
|
|
assert (
|
|
'https://wcs.example.invalid/api/export-import/cards/test-card/redirect/?application=test&version1=%s&version2=%s&compare'
|
|
% (version2.number, version3.number)
|
|
not in resp
|
|
)
|
|
assert 'Test Form <span class="extra-info">- Form</span>' in resp
|
|
assert (
|
|
'<span class="tag tag-error">Element not found</span> Test Card <span class="extra-info">- Card Model</span>'
|
|
in resp
|
|
)
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/export-import/bundle-declare/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
with StatefulHTTMock(response_content):
|
|
with pytest.raises(DeploymentError) as e:
|
|
resp.form.submit()
|
|
assert str(e.value) == 'Failed to declare elements for module wcs (500)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to declare elements for module wcs (500)'
|
|
|
|
# non editable app
|
|
application.editable = False
|
|
application.save()
|
|
app.get('/applications/manifest/test/metadata/', status=404)
|
|
app.get('/applications/manifest/test/scandeps/', status=404)
|
|
app.get('/applications/manifest/test/generate/', status=404)
|
|
app.get('/applications/manifest/test/add/forms/', status=404)
|
|
app.get('/applications/manifest/test/delete/%s/' % application.relation_set.first().pk, status=404)
|
|
|
|
|
|
def test_application_version_number(app, admin_user, settings):
|
|
Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
'wcs': {
|
|
'blah': {
|
|
# simulate an instance from another collectivity
|
|
'title': 'Unknown',
|
|
'url': 'https://unknown.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
},
|
|
'foobar': {
|
|
'title': 'Foobar',
|
|
'url': 'https://wcs.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
},
|
|
}
|
|
}
|
|
|
|
login(app)
|
|
|
|
application = Application.objects.create(name='Test', slug='test')
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
resp.form['number'] = 'X'
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {'number': ['Enter a valid value.']}
|
|
resp.form['number'] = 'X.0'
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {'number': ['Enter a valid value.']}
|
|
resp.form['number'] = '1.1.1'
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {'number': ['Enter a valid value.']}
|
|
resp.form['number'] = '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.0' % now().strftime('%Y%m%d')
|
|
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.1' % now().strftime('%Y%m%d')
|
|
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.2' % now().strftime('%Y%m%d')
|
|
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
resp.form['number'] = '1.0'
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {
|
|
'number': ['The version number must be equal to or greater than the previous one.']
|
|
}
|
|
|
|
last_version.number = '1.1.%s.2' % (now() - datetime.timedelta(days=1)).strftime('%Y%m%d')
|
|
last_version.save()
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.0' % now().strftime('%Y%m%d')
|
|
|
|
last_version.number = 'garbage'
|
|
last_version.save()
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == ''
|
|
resp.form['number'] = '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.0' % now().strftime('%Y%m%d')
|
|
|
|
last_version.number = '1'
|
|
last_version.save()
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '1'
|
|
resp.form['number'] = '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.0' % now().strftime('%Y%m%d')
|
|
|
|
last_version.number = '1.1'
|
|
last_version.save()
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.0' % now().strftime('%Y%m%d')
|
|
|
|
last_version.number = '1.1.%s.1.1' % now().strftime('%Y%m%d')
|
|
last_version.save()
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '1.1'
|
|
with StatefulHTTMock(mocked_http):
|
|
resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert last_version.application == application
|
|
assert last_version.number == '1.1.%s.2' % now().strftime('%Y%m%d')
|
|
|
|
|
|
def test_manifest_ordering(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',
|
|
}
|
|
}
|
|
}
|
|
|
|
application = Application.objects.create(name='Test', slug='test')
|
|
objects = [
|
|
# type, slug, auto_dependency
|
|
('forms', 'bar', True),
|
|
('forms', 'foo', True),
|
|
('cards', 'bar', True),
|
|
('cards', 'foo', True),
|
|
('workflows', 'bar', True),
|
|
('workflows', 'foo', True),
|
|
('forms', 'baaaar', False),
|
|
('forms', 'foooo', False),
|
|
('cards', 'baaaar', False),
|
|
('cards', 'foooo', False),
|
|
('workflows', 'baaaar', False),
|
|
('workflows', 'foooo', False),
|
|
('unknown', 'foo', False),
|
|
]
|
|
random.shuffle(objects)
|
|
for _type, slug, auto_dependency in objects:
|
|
element = Element.objects.create(
|
|
type=_type,
|
|
slug=slug,
|
|
name=slug.title(),
|
|
cache={},
|
|
)
|
|
Relation.objects.create(application=application, element=element, auto_dependency=auto_dependency)
|
|
|
|
login(app)
|
|
with StatefulHTTMock(mocked_http):
|
|
resp = app.get('/applications/manifest/test/')
|
|
assert resp.pyquery('.application-content li a').text() == (
|
|
'Baaaar - Form remove '
|
|
'Foooo - Form remove '
|
|
'Baaaar - Card Model remove '
|
|
'Foooo - Card Model remove '
|
|
'Baaaar - Workflow remove '
|
|
'Foooo - Workflow remove '
|
|
'Foo - Unknown (unknown) remove '
|
|
'Bar - Form '
|
|
'Foo - Form '
|
|
'Bar - Card Model '
|
|
'Foo - Card Model '
|
|
'Bar - Workflow '
|
|
'Foo - Workflow'
|
|
)
|
|
|
|
|
|
def test_scandeps_on_unknown_element(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',
|
|
}
|
|
}
|
|
}
|
|
|
|
application = Application.objects.create(name='Test', slug='test')
|
|
element = Element.objects.create(
|
|
type='forms',
|
|
slug='unknown',
|
|
name='Unknown',
|
|
cache={
|
|
'urls': {
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/forms/unknown/dependencies/',
|
|
}
|
|
},
|
|
)
|
|
relation = Relation.objects.create(application=application, element=element)
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/export-import/forms/unknown/dependencies/':
|
|
return {'status_code': 404}
|
|
return mocked_http(url, request)
|
|
|
|
login(app)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
with pytest.raises(ScanError) as e:
|
|
app.get('/applications/manifest/test/scandeps/').follow()
|
|
assert str(e.value) == 'Failed to scan "Unknown" (type forms, slug unknown) dependencies (404)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to scan "Unknown" (type forms, slug unknown) dependencies (404)'
|
|
relation.refresh_from_db()
|
|
assert relation.error is True
|
|
assert relation.error_status == 'notfound'
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/forms/unknown/dependencies/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
with pytest.raises(ScanError) as e:
|
|
app.get('/applications/manifest/test/scandeps/').follow()
|
|
assert str(e.value) == 'Failed to scan "Unknown" (type forms, slug unknown) dependencies (500)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to scan "Unknown" (type forms, slug unknown) dependencies (500)'
|
|
relation.refresh_from_db()
|
|
assert relation.error is True
|
|
assert relation.error_status == 'error'
|
|
|
|
with StatefulHTTMock(mocked_http):
|
|
app.get('/applications/manifest/test/scandeps/').follow()
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'completed'
|
|
relation.refresh_from_db()
|
|
assert relation.error is False
|
|
assert relation.error_status is None
|
|
|
|
|
|
def test_scandeps_on_renamed_element(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',
|
|
}
|
|
}
|
|
}
|
|
|
|
application = Application.objects.create(name='Test', slug='test')
|
|
element = Element.objects.create(
|
|
type='forms',
|
|
slug='test-form',
|
|
name='Test 1 form',
|
|
cache={
|
|
'urls': {'redirect': 'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/'}
|
|
},
|
|
)
|
|
Relation.objects.create(application=application, element=element)
|
|
element2 = Element.objects.create(
|
|
type='cards',
|
|
slug='test-card',
|
|
name='Test Card',
|
|
cache={
|
|
'urls': {
|
|
'export': 'https://wcs.example.invalid/api/export-import/cards/test-card/',
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/',
|
|
}
|
|
},
|
|
)
|
|
Relation.objects.create(application=application, element=element2, auto_dependency=True)
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/export-import/forms/':
|
|
return {
|
|
'content': {
|
|
'data': [
|
|
{
|
|
'id': 'test-form',
|
|
'text': 'Test Form (renamed)',
|
|
'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/',
|
|
'redirect': 'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/',
|
|
},
|
|
},
|
|
]
|
|
},
|
|
'status_code': 200,
|
|
}
|
|
if url.path == '/api/export-import/forms/test-form/dependencies/':
|
|
return {
|
|
'content': {
|
|
'data': [
|
|
{
|
|
'id': 'test-card',
|
|
'text': 'Test Card (renamed)',
|
|
'type': 'cards',
|
|
'urls': {
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/',
|
|
},
|
|
}
|
|
]
|
|
},
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
login(app)
|
|
with StatefulHTTMock(response_content):
|
|
app.get('/applications/manifest/test/scandeps/').follow()
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'completed'
|
|
element.refresh_from_db()
|
|
assert element.name == 'Test Form (renamed)'
|
|
assert element.cache == {
|
|
'id': 'test-form',
|
|
'text': 'Test Form (renamed)',
|
|
'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/',
|
|
'redirect': 'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/',
|
|
},
|
|
}
|
|
element2 = Element.objects.get(relation__auto_dependency=True)
|
|
assert element2.name == 'Test Card (renamed)'
|
|
assert element2.cache == {
|
|
'id': 'test-card',
|
|
'text': 'Test Card (renamed)',
|
|
'type': 'cards',
|
|
'urls': {
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/cards/test-card/dependencies/',
|
|
},
|
|
}
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'status_code': 404}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
with pytest.raises(ScanError) as e:
|
|
app.get('/applications/manifest/test/scandeps/').follow()
|
|
assert str(e.value) == 'Failed to get elements of type forms (404)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to get elements of type forms (404)'
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
with pytest.raises(ScanError) as e:
|
|
app.get('/applications/manifest/test/scandeps/').follow()
|
|
assert str(e.value) == 'Failed to get elements of type forms (500)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to get elements of type forms (500)'
|
|
|
|
|
|
@pytest.mark.parametrize('editable', [True, False])
|
|
def test_redirect_application_element(app, admin_user, settings, editable):
|
|
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',
|
|
}
|
|
}
|
|
}
|
|
|
|
application = Application.objects.create(name='Test', slug='test', editable=editable)
|
|
element = Element.objects.create(
|
|
type='forms',
|
|
slug='test-form',
|
|
name='Test 1 form',
|
|
cache={
|
|
'urls': {'redirect': 'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/'}
|
|
},
|
|
)
|
|
Relation.objects.create(application=application, element=element)
|
|
element = Element.objects.create(
|
|
type='forms',
|
|
slug='test2-form',
|
|
name='Test 2 form',
|
|
cache={'urls': {'export': 'https://wcs.example.invalid/api/export-import/forms/test2-form/'}},
|
|
)
|
|
Relation.objects.create(application=application, element=element)
|
|
element = Element.objects.create(
|
|
type='forms',
|
|
slug='test3-form',
|
|
name='Test 3 form',
|
|
cache={},
|
|
)
|
|
Relation.objects.create(application=application, element=element)
|
|
element = Element.objects.create(
|
|
type='roles',
|
|
slug='test',
|
|
name='Test',
|
|
cache={'urls': {'redirect': 'https://wcs.example.invalid/api/export-import/roles/test/redirect/'}},
|
|
)
|
|
Relation.objects.create(application=application, element=element)
|
|
|
|
login(app)
|
|
|
|
with StatefulHTTMock(mocked_http):
|
|
resp = app.get('/applications/manifest/test/')
|
|
assert 'https://wcs.example.invalid/api/export-import/forms/test-form/redirect/' in resp
|
|
assert (
|
|
'https://wcs.example.invalid/api/export-import/forms/test2-form/redirect/' in resp
|
|
) # no redirect url, but it's ok from export url
|
|
assert (
|
|
'https://wcs.example.invalid/api/export-import/forms/test3-form/redirect/' not in resp
|
|
) # no urls
|
|
assert (
|
|
'https://wcs.example.invalid/api/export-import/roles/test/redirect/' not in resp
|
|
) # not for roles
|
|
|
|
|
|
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
|
|
|
|
with StatefulHTTMock(mocked_http):
|
|
resp = app.get('/applications/manifest/app_to_delete/')
|
|
resp = resp.click(re.compile('^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'
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/export-import/unlink/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
resp = app.get('/applications/manifest/other_app/')
|
|
resp = resp.click(re.compile('^Delete$'))
|
|
resp = resp.forms[0].submit()
|
|
resp = resp.follow()
|
|
assert 'Failed to unlink application in module wcs (500)' in resp
|
|
assert Application.objects.count() == 1
|
|
assert Application.objects.first().name == 'OtherApp'
|
|
|
|
|
|
def test_404_unknown_app(app, admin_user, settings):
|
|
login(app)
|
|
app.get('/applications/manifest/xxx/', status=404)
|
|
|
|
|
|
@pytest.fixture
|
|
def app_bundle():
|
|
return get_bundle()
|
|
|
|
|
|
@pytest.fixture
|
|
def app_bundle_with_icon():
|
|
return get_bundle(with_icon=True)
|
|
|
|
|
|
def get_bundle(with_icon=False):
|
|
tar_io = io.BytesIO()
|
|
with tarfile.open(mode='w', fileobj=tar_io) as tar:
|
|
manifest_json = {
|
|
'application': 'Test',
|
|
'slug': 'test',
|
|
'icon': 'foo.jpeg' if with_icon else None,
|
|
'description': '',
|
|
'documentation_url': 'http://foo.bar',
|
|
'version_number': '43.0' if with_icon else '42.0',
|
|
'version_notes': 'foo bar blah',
|
|
'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},
|
|
{
|
|
'type': 'comment-templates-categories',
|
|
'slug': 'test',
|
|
'name': 'test',
|
|
'auto-dependency': True,
|
|
},
|
|
],
|
|
'parameters': [
|
|
{'name': 'app_param1', 'label': 'Parameter 1', 'default_value': 'plop'},
|
|
{'name': 'app_param2', 'label': 'Parameter 2', 'default_value': None},
|
|
],
|
|
}
|
|
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)
|
|
if with_icon:
|
|
with open(os.path.join(TESTS_DATA_DIR, 'black.jpeg'), mode='rb') as icon_fd:
|
|
tarinfo = tarfile.TarInfo(manifest_json['icon'])
|
|
tarinfo.size = 558
|
|
tar.addfile(tarinfo, fileobj=icon_fd)
|
|
|
|
return tar_io.getvalue()
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['Install', 'Update'])
|
|
def test_deploy_application(app, admin_user, settings, app_bundle, app_bundle_with_icon, action):
|
|
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)
|
|
|
|
def mocked_http2(url, request):
|
|
if url.path == '/api/export-import/bundle-check/':
|
|
return {
|
|
'content': json.dumps(
|
|
{
|
|
'data': {
|
|
'differences': [],
|
|
'unknown_elements': [],
|
|
'no_history_elements': [{'type': 'forms', 'slug': 'legacy'}],
|
|
'legacy_elements': [
|
|
{'type': 'cards', 'slug': 'test', 'text': 'Test', 'url': 'http://foobar'}
|
|
],
|
|
}
|
|
}
|
|
),
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
if action == 'Update':
|
|
Application.objects.create(name='Test', slug='test', editable=False)
|
|
|
|
def install(resp, bundle):
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', bundle, 'application/x-tar')
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.form.submit()
|
|
|
|
assert Application.objects.count() == 1
|
|
application = Application.objects.get(slug='test')
|
|
assert application.name == 'Test'
|
|
if bundle == app_bundle_with_icon:
|
|
assert re.match(r'applications/icons/foo(_\w+)?.jpeg', application.icon.name)
|
|
else:
|
|
assert application.icon.name == ''
|
|
assert application.documentation_url == 'http://foo.bar'
|
|
version = application.version_set.latest('pk')
|
|
if bundle == app_bundle_with_icon:
|
|
assert version.number == '43.0'
|
|
else:
|
|
assert version.number == '42.0'
|
|
assert version.notes == 'foo bar blah'
|
|
if action == 'Install':
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert resp.location.endswith('/applications/manifest/test/confirm/%s/' % version.pk)
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.follow()
|
|
assert 'Local changes:' not in resp
|
|
assert 'Not found components (removed ?):' not in resp
|
|
assert 'Impossible to check local changes for these components' not in resp
|
|
assert 'These components alreay exist outside the application' in resp
|
|
assert 'Card Model "Test"' in resp
|
|
assert '(<a href="http://foobar">see component</a>)' in resp
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.form.submit()
|
|
assert resp.location.endswith('/applications/manifest/test/')
|
|
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = app.get('/applications/manifest/test/job/%s/check-diffs/' % job.pk)
|
|
assert 'Local changes:' not in resp
|
|
assert 'Not found components (removed ?):' not in resp
|
|
assert 'Impossible to check local changes for these components' not in resp
|
|
assert 'These components alreay exist outside the application' in resp
|
|
assert 'Card Model "Test"' in resp
|
|
assert '(<a href="http://foobar">see component</a>)' in resp
|
|
else:
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.follow()
|
|
if resp.location:
|
|
resp.follow()
|
|
else:
|
|
resp.form.submit()
|
|
|
|
assert application.elements.count() == 4
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'completed'
|
|
assert job.progression_urls == {'wcs': {'Foobar': 'https://wcs.example.invalid/api/jobs/job-uuid/'}}
|
|
assert Parameter.objects.filter(application=application).count() == 2
|
|
assert Variable.objects.filter(name__startswith='app_').count() == 2
|
|
|
|
# check forms are not marked as error (the other object types are not altered in the mocked responses
|
|
# and would be marked as errors)
|
|
assert [x.error for x in application.relation_set.filter(element__type='forms')] == [False]
|
|
return version
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.click(href='/manifest/test/')
|
|
version1 = install(resp, app_bundle)
|
|
if action == 'Update':
|
|
# update, check-install job then deploy job
|
|
assert list(AsyncJob.objects.values_list('action', flat=True).order_by('pk')) == [
|
|
'check-install',
|
|
'deploy',
|
|
]
|
|
else:
|
|
# first install, just deploy job
|
|
assert list(AsyncJob.objects.values_list('action', flat=True).order_by('pk')) == [
|
|
'check-first-install',
|
|
'deploy',
|
|
]
|
|
assert version1.application.version_set.count() == 1
|
|
version2 = install(resp, app_bundle)
|
|
assert version2.application.version_set.count() == 1
|
|
assert version1.pk == version2.pk
|
|
assert version1.creation_timestamp == version2.creation_timestamp
|
|
assert version1.last_update_timestamp < version2.last_update_timestamp
|
|
version3 = install(resp, app_bundle_with_icon)
|
|
assert version2.application.version_set.count() == 2
|
|
assert version2.pk != version3.pk
|
|
assert version2.creation_timestamp < version3.creation_timestamp
|
|
assert version2.last_update_timestamp < version3.last_update_timestamp
|
|
version4 = install(resp, app_bundle)
|
|
assert version4.application.version_set.count() == 3
|
|
assert version3.pk != version4.pk
|
|
assert version3.creation_timestamp < version4.creation_timestamp
|
|
assert version3.last_update_timestamp < version4.last_update_timestamp
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
versions = [e.text() for e in resp.pyquery('h3').items()]
|
|
assert versions.count('42.0') == 2
|
|
assert versions.count('43.0') == 1
|
|
assert resp.text.count('Deploying application bundle') == 4
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/export-import/bundle-import/':
|
|
return {'status_code': 500}
|
|
return mocked_http2(url, request)
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
resp = resp.form.submit().follow()
|
|
with pytest.raises(DeploymentError) as e:
|
|
resp.form.submit().follow()
|
|
assert str(e.value) == 'Failed to deploy module wcs (500)'
|
|
application = Application.objects.get(slug='test')
|
|
# check elements are marked as not installed if there was a failure
|
|
assert [
|
|
(x.error, x.error_status) for x in application.relation_set.filter(element__type='forms')
|
|
] == [(True, 'not-installed')]
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to deploy module wcs (500)'
|
|
|
|
form_def = {
|
|
'id': 'test',
|
|
'text': 'Test',
|
|
'type': 'forms',
|
|
'urls': {
|
|
'export': 'https://wcs.example.invalid/api/export-import/forms/test/',
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/forms/test/dependencies/',
|
|
'redirect': 'https://wcs.example.invalid/api/export-import/forms/test/redirect/',
|
|
},
|
|
}
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/forms/':
|
|
return {
|
|
'content': {'data': [form_def]},
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http2(url, request)
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
resp = resp.form.submit().follow()
|
|
resp.form.submit().follow()
|
|
application = Application.objects.get(slug='test')
|
|
elements = application.elements.all().order_by('type')
|
|
assert len(elements) == 4
|
|
assert elements[0].cache == {}
|
|
assert elements[1].cache == {}
|
|
assert elements[2].cache == form_def
|
|
assert elements[3].cache == {}
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'status_code': 500}
|
|
return mocked_http2(url, request)
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
resp = resp.form.submit().follow()
|
|
with pytest.raises(ScanError) as e:
|
|
resp.form.submit().follow()
|
|
assert str(e.value) == 'Failed to get elements of type forms (500)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to get elements of type forms (500)'
|
|
|
|
# bad file format
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', b'garbage', 'application/x-tar')
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {'bundle': ['Invalid tar file.']}
|
|
|
|
# missing manifest
|
|
tar_io = io.BytesIO()
|
|
with tarfile.open(mode='w', fileobj=tar_io) as tar:
|
|
foo_fd = io.BytesIO(json.dumps({'foo': 'bar'}, indent=2).encode())
|
|
tarinfo = tarfile.TarInfo('foo.json')
|
|
tarinfo.size = len(foo_fd.getvalue())
|
|
tar.addfile(tarinfo, fileobj=foo_fd)
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', tar_io.getvalue(), 'application/x-tar')
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {'bundle': ['Invalid tar file, missing manifest.']}
|
|
|
|
# bad icon file
|
|
tar_io = io.BytesIO()
|
|
with tarfile.open(mode='w', fileobj=tar_io) as tar:
|
|
manifest_json = {
|
|
'application': 'Test',
|
|
'slug': 'test',
|
|
'icon': 'foo.png',
|
|
'description': '',
|
|
'documentation_url': 'http://foo.bar',
|
|
'version_number': '42.0',
|
|
'version_notes': 'foo bar blah',
|
|
'elements': [],
|
|
}
|
|
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)
|
|
icon_fd = io.BytesIO(b'garbage')
|
|
tarinfo = tarfile.TarInfo('foo.png')
|
|
tarinfo.size = len(icon_fd.getvalue())
|
|
tar.addfile(tarinfo, fileobj=icon_fd)
|
|
|
|
bundle = tar_io.getvalue()
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with StatefulHTTMock(mocked_http2):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', bundle, 'application/x-tar')
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {'bundle': ['Invalid icon file.']}
|
|
|
|
|
|
def test_update_application(app, admin_user, settings, app_bundle):
|
|
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 = Application.objects.create(name='Test', slug='test', editable=True)
|
|
|
|
app.get('/applications/manifest/test/update/', status=404)
|
|
|
|
application.editable = False
|
|
application.slug = 'wrong'
|
|
application.save()
|
|
|
|
resp = app.get('/applications/manifest/wrong/update/')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
resp = resp.form.submit()
|
|
assert resp.context['form'].errors == {'bundle': ['Can not update this application, wrong slug (test).']}
|
|
|
|
AsyncJob.objects.all().delete()
|
|
application.slug = 'test'
|
|
application.save()
|
|
resp = app.get('/applications/manifest/test/update/')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with StatefulHTTMock(mocked_http):
|
|
resp = resp.form.submit()
|
|
# only legacy elements, do the update and redirect to manifest
|
|
last_version = Version.objects.latest('pk')
|
|
assert resp.location.endswith('/applications/manifest/test/confirm/%s/' % last_version.pk)
|
|
resp = resp.follow()
|
|
assert resp.location.endswith('/applications/manifest/test/')
|
|
assert list(
|
|
AsyncJob.objects.filter(version__number='42.0').values_list('action', flat=True).order_by('pk')
|
|
) == [
|
|
'check-install',
|
|
'deploy',
|
|
]
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/export-import/bundle-check/':
|
|
return {
|
|
'content': json.dumps(
|
|
{
|
|
'data': {
|
|
'differences': [{'type': 'forms', 'slug': 'test', 'url': 'http://foobar'}],
|
|
'unknown_elements': [{'type': 'workflows', 'slug': 'test'}],
|
|
'no_history_elements': [{'type': 'blocks', 'slug': 'test'}],
|
|
'legacy_elements': [
|
|
{'type': 'cards', 'slug': 'test', 'text': 'Test', 'url': 'http://foobar'}
|
|
],
|
|
}
|
|
}
|
|
),
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
Version.objects.filter(pk=last_version.pk).update(number='1.0')
|
|
resp = app.get('/applications/manifest/test/update/')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
resp = resp.form.submit()
|
|
last_version = Version.objects.latest('pk')
|
|
assert resp.location.endswith('/applications/manifest/test/confirm/%s/' % last_version.pk)
|
|
resp = resp.follow()
|
|
assert list(
|
|
AsyncJob.objects.filter(version__number='42.0').values_list('action', flat=True).order_by('pk')
|
|
) == ['check-install']
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert 'Local changes:' in resp
|
|
assert 'Form "test"' in resp
|
|
assert '(<a href="http://foobar">see changes</a>)' in resp
|
|
assert 'Not found components (removed ?):' in resp
|
|
assert 'Workflow "test"' in resp
|
|
assert (
|
|
'Impossible to check local changes for these components (no history found for this application version):'
|
|
in resp
|
|
)
|
|
assert 'Block of fields "test"' in resp
|
|
assert 'These components alreay exist outside the application' in resp
|
|
assert 'Card Model "Test"' in resp
|
|
assert '(<a href="http://foobar">see component</a>)' in resp
|
|
resp = resp.form.submit()
|
|
assert resp.location.endswith('/applications/manifest/test/')
|
|
assert list(
|
|
AsyncJob.objects.filter(version__number='42.0').values_list('action', flat=True).order_by('pk')
|
|
) == [
|
|
'check-install',
|
|
'deploy',
|
|
]
|
|
|
|
resp = app.get('/applications/manifest/test/job/%s/check-diffs/' % job.pk)
|
|
assert 'Local changes:' in resp
|
|
assert 'Form "test"' in resp
|
|
assert '(<a href="http://foobar">see changes</a>)' in resp
|
|
assert 'Not found components (removed ?):' in resp
|
|
assert 'Workflow "test"' in resp
|
|
assert (
|
|
'Impossible to check local changes for these components (no history found for this application version):'
|
|
in resp
|
|
)
|
|
assert 'Block of fields "test"' in resp
|
|
assert 'These components alreay exist outside the application' in resp
|
|
assert 'Card Model "Test"' in resp
|
|
assert '(<a href="http://foobar">see component</a>)' in resp
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/bundle-check/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
Version.objects.filter(pk=last_version.pk).update(number='2.0')
|
|
resp = app.get('/applications/manifest/test/update/')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
resp = resp.form.submit()
|
|
assert resp.location.endswith(
|
|
'/applications/manifest/test/confirm/%s/' % Version.objects.latest('pk').pk
|
|
)
|
|
resp = resp.follow()
|
|
assert list(
|
|
AsyncJob.objects.filter(version__number='42.0').values_list('action', flat=True).order_by('pk')
|
|
) == ['check-install']
|
|
assert 'Error checking local changes, update can not be run.' in resp
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/bundle-check/':
|
|
return {
|
|
'content': json.dumps({'data': {}}),
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
resp = app.get('/applications/manifest/test/update/')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
resp = resp.form.submit()
|
|
assert resp.location.endswith(
|
|
'/applications/manifest/test/confirm/%s/' % Version.objects.latest('pk').pk
|
|
)
|
|
resp = resp.follow()
|
|
assert 'No local changes found.' in resp
|
|
|
|
|
|
def test_get_version_to_check(app):
|
|
application = Application.objects.create(name='Test', slug='test')
|
|
version1 = Version.objects.create(application=application, number='1')
|
|
assert version1.get_version_to_check() is None
|
|
|
|
version2 = Version.objects.create(application=application, number='2')
|
|
assert version2.get_version_to_check() == version1
|
|
|
|
version3 = Version.objects.create(application=application, number='3')
|
|
assert version3.get_version_to_check() == version2
|
|
|
|
AsyncJob.objects.create(
|
|
label='foo',
|
|
application=application,
|
|
action='deploy',
|
|
status='completed',
|
|
# no version linked
|
|
)
|
|
assert version3.get_version_to_check() == version2
|
|
|
|
AsyncJob.objects.create(
|
|
label='foo',
|
|
application=application,
|
|
version=version2, # wrong version
|
|
action='deploy',
|
|
status='completed',
|
|
)
|
|
assert version3.get_version_to_check() == version2
|
|
|
|
AsyncJob.objects.create(
|
|
label='foo',
|
|
application=application,
|
|
version=version3,
|
|
action='deploy',
|
|
status='failed', # status
|
|
)
|
|
assert version3.get_version_to_check() == version2
|
|
|
|
AsyncJob.objects.create(
|
|
label='foo',
|
|
application=application,
|
|
version=version3,
|
|
action='foo', # wrong action
|
|
status='completed',
|
|
)
|
|
assert version3.get_version_to_check() == version2
|
|
|
|
AsyncJob.objects.create(
|
|
label='foo',
|
|
application=application,
|
|
version=version3,
|
|
action='deploy',
|
|
status='completed',
|
|
)
|
|
assert version3.get_version_to_check() == version3
|
|
|
|
|
|
def test_job_diffs(app, admin_user):
|
|
application = Application.objects.create(name='Test', slug='test')
|
|
version = Version.objects.create(application=application, number='1')
|
|
job = AsyncJob.objects.create(
|
|
label='foo',
|
|
application=application,
|
|
action='check-install',
|
|
status='failed',
|
|
version=version,
|
|
)
|
|
|
|
login(app)
|
|
|
|
app.get('/applications/manifest/test/job/%s/check-diffs/' % job.pk, status=404)
|
|
|
|
job.status = 'completed'
|
|
job.action = 'foo'
|
|
job.save()
|
|
app.get('/applications/manifest/test/job/%s/check-diffs/' % job.pk, status=404)
|
|
|
|
job.action = 'check-install'
|
|
job.save()
|
|
resp = app.get('/applications/manifest/test/job/%s/check-diffs/' % job.pk)
|
|
assert 'No local changes found.' in resp
|
|
assert 'Not found elements (removed ?):' not in resp
|
|
assert (
|
|
'Impossible to check local changes for these elements (no history found for this application version):'
|
|
not in resp
|
|
)
|
|
|
|
|
|
def test_refresh_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 = Application.objects.create(name='Test', slug='test')
|
|
element = Element.objects.create(type='forms', slug='test', name='Test', cache={})
|
|
Relation.objects.create(application=application, element=element)
|
|
|
|
form_def = {
|
|
'id': 'test',
|
|
'text': 'Test',
|
|
'type': 'forms',
|
|
'urls': {
|
|
'export': 'https://wcs.example.invalid/api/export-import/forms/test/',
|
|
'dependencies': 'https://wcs.example.invalid/api/export-import/forms/test/dependencies/',
|
|
'redirect': 'https://wcs.example.invalid/api/export-import/forms/test/redirect/',
|
|
},
|
|
}
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/export-import/forms/':
|
|
return {
|
|
'content': {'data': [form_def]},
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
resp = app.get('/applications/manifest/test/refresh/')
|
|
application = Application.objects.get(slug='test')
|
|
element.refresh_from_db()
|
|
assert element.cache == form_def
|
|
assert resp.location.endswith('/applications/manifest/test/')
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
element.cache = {'foo': 'bar'}
|
|
element.save()
|
|
with StatefulHTTMock(response_content):
|
|
resp = app.get('/applications/manifest/test/refresh/')
|
|
application = Application.objects.get(slug='test')
|
|
element.refresh_from_db()
|
|
assert element.cache == {'foo': 'bar'}
|
|
assert resp.location.endswith('/applications/manifest/test/')
|
|
|
|
|
|
@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):
|
|
Authentic.objects.create(base_url='https://idp.example.invalid', slug='idp', title='Foobar')
|
|
wcs = 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)
|
|
|
|
for i in range(2):
|
|
Application.objects.all().delete()
|
|
resp = app.get('/applications/')
|
|
resp = resp.click('Install')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
|
|
with StatefulHTTMock(httmock.remember_called(mocked_http)):
|
|
resp = resp.form.submit().follow()
|
|
assert mocked_http.call['requests'][0].url.startswith(
|
|
'https://wcs.example.invalid/api/export-import/bundle-check/?'
|
|
)
|
|
# roles
|
|
assert mocked_http.call['requests'][1].url.startswith(
|
|
'https://idp.example.invalid/api/roles/?update'
|
|
)
|
|
assert mocked_http.call['requests'][2].url.startswith(
|
|
'https://idp.example.invalid/api/provision/'
|
|
)
|
|
assert mocked_http.call['requests'][3].url.startswith(
|
|
'https://idp.example.invalid/api/roles/?update'
|
|
)
|
|
assert mocked_http.call['requests'][4].url.startswith(
|
|
'https://idp.example.invalid/api/provision/'
|
|
)
|
|
# then form
|
|
assert mocked_http.call['requests'][5].url.startswith(
|
|
'https://wcs.example.invalid/api/export-import/bundle-import/?'
|
|
)
|
|
assert mocked_http.call['requests'][6].url.startswith(
|
|
'https://wcs.example.invalid/api/export-import/?'
|
|
)
|
|
# then element refresh
|
|
available_objects = [o for o in WCS_AVAILABLE_OBJECTS['data'] if o['id'] in ['forms', 'roles']]
|
|
assert mocked_http.call['count'] == 6 + 1 + len(available_objects)
|
|
for i, object_type in enumerate(available_objects):
|
|
assert mocked_http.call['requests'][7 + i].url.startswith(
|
|
'https://wcs.example.invalid/api/export-import/%s/?' % object_type['id']
|
|
)
|
|
|
|
resp = resp.follow()
|
|
assert resp.pyquery('h2.application-title').text() == 'Test'
|
|
|
|
def response_content(url, request):
|
|
if url.path == '/api/roles/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
resp = app.get('/applications/install/')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
with pytest.raises(DeploymentError) as e:
|
|
resp.form.submit().follow()
|
|
assert str(e.value) == 'Failed to create role test-role (500)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to create role test-role (500)'
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path == '/api/export-import/bundle-check/':
|
|
return {
|
|
'content': json.dumps(
|
|
{
|
|
'data': {
|
|
'differences': [],
|
|
'unknown_elements': [],
|
|
'no_history_elements': [],
|
|
'legacy_elements': [],
|
|
}
|
|
}
|
|
),
|
|
'status_code': 200,
|
|
}
|
|
if url.path == '/api/provision/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
resp = app.get('/applications/install/')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
|
|
with StatefulHTTMock(response_content):
|
|
resp = resp.form.submit().follow()
|
|
with pytest.raises(DeploymentError) as e:
|
|
resp.form.submit().follow()
|
|
assert str(e.value) == 'Failed to provision role test-role (500)'
|
|
job = AsyncJob.objects.latest('pk')
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to provision role test-role (500)'
|
|
|
|
# test import on a specific service OU
|
|
v = Variable.objects.create(name='ou-slug', label='OU', value='service-ou')
|
|
Application.objects.all().delete()
|
|
resp = app.get('/applications/')
|
|
resp = resp.click('Install')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
|
|
with StatefulHTTMock(httmock.remember_called(mocked_http)):
|
|
resp.form.submit().follow().follow()
|
|
# ou must be specified when calling authentic API
|
|
assert mocked_http.call['requests'][1].url.startswith(
|
|
'https://idp.example.invalid/api/roles/?update_or_create=slug&update_or_create=ou'
|
|
)
|
|
assert b'"ou": "service-ou"' in mocked_http.call['requests'][1].body
|
|
|
|
# test import on a default OU
|
|
Application.objects.all().delete()
|
|
resp = app.get('/applications/')
|
|
resp = resp.click('Install')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle_roles, 'application/x-tar')
|
|
v.service_type = ContentType.objects.get_for_model(Wcs)
|
|
v.service_pk = wcs.id
|
|
v.save()
|
|
with StatefulHTTMock(httmock.remember_called(mocked_http)):
|
|
resp.form.submit().follow().follow()
|
|
assert b'"ou": "default"' in mocked_http.call['requests'][1].body
|
|
|
|
|
|
def test_job_status_page(app, admin_user, settings):
|
|
Wcs.objects.create(base_url='https://wcs.example.invalid', slug='foobar', title='Foobar')
|
|
application = Application.objects.create(name='Test', slug='test')
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
'wcs': {
|
|
'foobar': {
|
|
'title': 'Foobar',
|
|
'url': 'https://wcs.example.invalid/',
|
|
'orig': 'example.org',
|
|
'secret': 'xxx',
|
|
}
|
|
}
|
|
}
|
|
|
|
login(app)
|
|
|
|
for action in ['scandeps', 'create_bundle']:
|
|
job = AsyncJob.objects.create(
|
|
label=action,
|
|
application=application,
|
|
action=action,
|
|
status='running',
|
|
)
|
|
resp = app.get('/applications/manifest/test/job/%s/' % job.pk)
|
|
assert 'Please wait…' in resp
|
|
assert 'window.location.reload' in resp
|
|
assert 'window.location = "/applications/manifest/test/"' not in resp
|
|
|
|
job.status = 'failed'
|
|
job.exception = 'foo bar exception'
|
|
job.save()
|
|
resp = app.get('/applications/manifest/test/job/%s/' % job.pk)
|
|
assert 'Error running the job.' in resp
|
|
assert 'Please wait…' not in resp
|
|
assert 'window.location.reload' not in resp
|
|
assert 'window.location = "/applications/manifest/test/"' not in resp
|
|
|
|
job = AsyncJob.objects.create(
|
|
label='deploy',
|
|
application=application,
|
|
action='deploy',
|
|
progression_urls={'wcs': {'Foobar': 'https://wcs.example.invalid/api/jobs/job-uuid/'}},
|
|
status='completed', # completed, with async job on wcs part
|
|
)
|
|
|
|
def response_content(url, request):
|
|
if url.path.startswith('/api/jobs/'):
|
|
return {
|
|
'content': json.dumps({'err': 0, 'data': {'status': 'running', 'completion_status': '42%'}}),
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
resp = app.get('/applications/manifest/test/job/%s/' % job.pk)
|
|
assert 'Please wait…' in resp
|
|
assert 'Running: 42%' in resp
|
|
assert 'window.location.reload' in resp
|
|
assert 'window.location = "/applications/manifest/test/"' not in resp
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path.startswith('/api/jobs/'):
|
|
return {
|
|
'content': json.dumps(
|
|
{'err': 0, 'data': {'status': 'completed', 'completion_status': '100%'}}
|
|
),
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
resp = app.get('/applications/manifest/test/job/%s/' % job.pk)
|
|
assert 'Please wait…' in resp
|
|
assert 'Completed: 100%' in resp
|
|
assert 'window.location.reload' not in resp
|
|
assert 'window.location = "/applications/manifest/test/refresh/"' in resp
|
|
|
|
job.status = 'failed'
|
|
job.exception = 'foo bar exception'
|
|
job.save()
|
|
resp = app.get('/applications/manifest/test/job/%s/' % job.pk)
|
|
assert 'Error running the job.' in resp
|
|
assert 'Please wait…' not in resp
|
|
assert 'window.location.reload' not in resp
|
|
assert 'window.location = "/applications/manifest/test/"' not in resp
|
|
|
|
job.status = 'running'
|
|
job.exception = ''
|
|
job.save()
|
|
|
|
def response_content(url, request): # noqa pylint: disable=function-redefined
|
|
if url.path.startswith('/api/jobs/'):
|
|
return {
|
|
'content': json.dumps({'err': 0, 'data': {'status': 'failed', 'completion_status': '42%'}}),
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
with StatefulHTTMock(response_content):
|
|
resp = app.get('/applications/manifest/test/job/%s/' % job.pk)
|
|
assert 'Error running the job.' in resp
|
|
assert 'Please wait…' not in resp
|
|
assert 'window.location.reload' not in resp
|
|
assert 'window.location = "/applications/manifest/test/"' not in resp
|
|
job.refresh_from_db()
|
|
assert job.status == 'failed'
|
|
assert job.exception == 'Failed to deploy module wcs'
|
|
|
|
|
|
def test_create_application_parameters(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 StatefulHTTMock(mocked_http):
|
|
resp = resp.follow()
|
|
|
|
# add a form (an element is required to have the generate app button)
|
|
resp = resp.click('Forms')
|
|
resp.form.fields['elements'][4].checked = True
|
|
resp = resp.form.submit().follow()
|
|
|
|
assert resp.pyquery('.app-parameters--section h3 a')
|
|
resp = resp.click('add')
|
|
resp.form['label'] = 'Foo'
|
|
resp.form['name'] = 'app_foo'
|
|
resp.form['default_value'] = 'xxx'
|
|
resp = resp.form.submit().follow()
|
|
assert Parameter.objects.filter(name='app_foo').exists()
|
|
assert resp.pyquery('.app-parameters--list li')
|
|
resp = resp.click(href=resp.pyquery('.app-parameters--list li a:not(.delete)').attr.href)
|
|
assert resp.form['label'].value == 'Foo'
|
|
assert resp.form['name'].value == 'app_foo'
|
|
assert resp.form['default_value'].value == 'xxx'
|
|
resp = resp.click('Cancel')
|
|
|
|
resp = resp.click(href=resp.pyquery('.app-parameters--values a').attr.href)
|
|
assert resp.form['value'].value == 'xxx'
|
|
resp.form['value'] = 'changed'
|
|
resp = resp.form.submit().follow()
|
|
assert Variable.objects.get(name='app_foo').value == 'changed'
|
|
|
|
resp = resp.click('Generate application bundle')
|
|
resp.form['number'] = '1.0'
|
|
resp.form['notes'] = 'Foo bar blah.'
|
|
resp = resp.form.submit().follow()
|
|
resp_download = resp.click('Download')
|
|
assert resp_download.content_type == 'application/x-tar'
|
|
# uncompressed tar, primitive check of contents
|
|
assert b'app_foo' in resp_download.content
|
|
|
|
resp = resp.click(href=resp.pyquery('.app-parameters--list li a.delete').attr.href)
|
|
resp = resp.form.submit().follow()
|
|
assert not Parameter.objects.filter(name='app_foo').exists()
|
|
|
|
# check name is slugified
|
|
resp = resp.click('add')
|
|
resp.form['label'] = 'Foo'
|
|
resp.form['name'] = 'app-foo bar'
|
|
resp.form['default_value'] = 'xxx'
|
|
resp = resp.form.submit().follow()
|
|
assert Parameter.objects.all().count() == 1
|
|
assert Parameter.objects.filter(name='app_foo_bar').exists()
|
|
|
|
resp = resp.click(href=resp.pyquery('.app-parameters--list li a:not(.delete)').attr.href)
|
|
resp.form['name'] = 'app-foo bar'
|
|
resp = resp.form.submit().follow()
|
|
assert Parameter.objects.all().count() == 1
|
|
assert Parameter.objects.filter(name='app_foo_bar').exists()
|
|
|
|
|
|
@pytest.fixture
|
|
def app_bundle_parameters():
|
|
tar_io = io.BytesIO()
|
|
with tarfile.open(mode='w', fileobj=tar_io) as tar:
|
|
manifest_json = {
|
|
'application': 'Test',
|
|
'slug': 'test',
|
|
'description': '',
|
|
'elements': [],
|
|
'parameters': [{'label': 'Foo', 'name': 'app_foo', 'default_value': 'xxx'}],
|
|
}
|
|
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_parameters(app, admin_user, settings, app_bundle_parameters):
|
|
Parameter.objects.all().delete()
|
|
Variable.objects.all().delete()
|
|
Application.objects.all().delete()
|
|
|
|
login(app)
|
|
|
|
resp = app.get('/applications/')
|
|
resp = resp.click('Install')
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle_parameters, 'application/x-tar')
|
|
resp = resp.form.submit().follow().follow()
|
|
assert Variable.objects.get(name='app_foo').value == 'xxx'
|
|
assert resp.pyquery('h2.application-title').text() == 'Test'
|
|
assert resp.pyquery('.app-parameters--section h3')
|
|
assert not resp.pyquery('.app-parameters--section h3 a')
|
|
|
|
resp = resp.click(href=resp.pyquery('.app-parameters--values a').attr.href)
|
|
assert resp.form['value'].value == 'xxx'
|
|
resp.form['value'] = 'changed'
|
|
resp = resp.form.submit().follow()
|
|
assert Variable.objects.get(name='app_foo').value == 'changed'
|
|
|
|
# check parameter cannot be edited or deleted
|
|
parameter = Parameter.objects.all().first()
|
|
resp = app.get(f'/applications/manifest/test/edit-parameter/{parameter.id}/', status=404)
|
|
resp = app.get(f'/applications/manifest/test/delete-parameter/{parameter.id}/', status=404)
|
|
|
|
|
|
def test_non_editable_application_metadata(app, admin_user):
|
|
Application.objects.create(
|
|
name='Test',
|
|
slug='test',
|
|
editable=False,
|
|
documentation_url='https://example.net',
|
|
license='agplv3+',
|
|
authors='author1',
|
|
)
|
|
|
|
login(app)
|
|
resp = app.get('/applications/manifest/test/')
|
|
assert resp.pyquery('.meta .documentation a').attr.href == 'https://example.net'
|
|
assert resp.pyquery('.meta .authors').text() == 'Created by: author1'
|
|
assert (
|
|
resp.pyquery('.meta .license').text()
|
|
== 'License: GNU Affero General Public License v3 or later (AGPLv3+)'
|
|
)
|