1244 lines
48 KiB
Python
1244 lines
48 KiB
Python
import base64
|
|
import io
|
|
import json
|
|
import random
|
|
import re
|
|
import tarfile
|
|
|
|
import httmock
|
|
import pytest
|
|
from httmock import HTTMock
|
|
from pyquery import PyQuery
|
|
from test_manager import login
|
|
from webtest import Upload
|
|
|
|
from hobo.applications.models import (
|
|
Application,
|
|
AsyncJob,
|
|
DeploymentError,
|
|
Element,
|
|
Relation,
|
|
ScanError,
|
|
Version,
|
|
)
|
|
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": "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/",
|
|
},
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
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/':
|
|
return {
|
|
'content': json.dumps({'err': 0, 'url': 'https://wcs.example.invalid/api/jobs/job-uuid/'}),
|
|
'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 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.form['documentation_url'] = 'http://foo.bar'
|
|
resp = resp.form.submit().follow()
|
|
application = Application.objects.get(slug='test')
|
|
assert application.icon.name == ''
|
|
assert application.documentation_url == 'http://foo.bar'
|
|
|
|
# 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() == 2
|
|
|
|
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'
|
|
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"' in resp.content
|
|
assert b'"version_notes": "Foo bar blah."' 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') == 1
|
|
assert resp.text.count('Creating application bundle') == 1
|
|
resp = resp.click(href='/applications/manifest/test/download/%s/' % version.pk)
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert b'"version_number": "1.0"' in resp.content
|
|
|
|
# generate again without changing version number
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '1.0' # last one
|
|
assert resp.form['notes'].value == 'Foo bar blah.' # last one
|
|
resp.form['notes'] = 'Foo bar blahha.'
|
|
resp = resp.form.submit().follow()
|
|
same_version = Version.objects.latest('pk')
|
|
assert same_version.number == '1.0'
|
|
assert same_version.notes == 'Foo bar blahha.'
|
|
assert same_version.pk == version.pk
|
|
resp = resp.click('Download')
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert b'"version_number": "1.0"' in resp.content
|
|
assert b'"version_notes": "Foo bar blahha."' 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') == 1
|
|
assert resp.text.count('Creating application bundle') == 2
|
|
resp = resp.click(href='/applications/manifest/test/download/%s/' % same_version.pk)
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert b'"version_number": "1.0"' in resp.content
|
|
|
|
# add an icon
|
|
resp = app.get('/applications/manifest/test/metadata/')
|
|
resp.form['icon'] = Upload(
|
|
'foo.png',
|
|
base64.decodebytes(
|
|
b'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABAQAAAAA3bvkkAAAACklEQVQI12NoAAAAggCB3UNq9AAAAABJRU5ErkJggg=='
|
|
),
|
|
'image/png',
|
|
)
|
|
resp.form['documentation_url'] = '' # and reset documentation_url
|
|
resp = resp.form.submit().follow()
|
|
application.refresh_from_db()
|
|
assert re.match(r'applications/icons/foo(_\w+)?.png', application.icon.name)
|
|
assert application.documentation_url == ''
|
|
|
|
# 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+)?.png', 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"' in resp.content
|
|
assert b'"version_notes": "Foo bar blah. But with an icon."' in resp.content
|
|
version = Version.objects.latest('pk')
|
|
assert version.number == '2.0'
|
|
assert version.notes == 'Foo bar blah. But with an icon.'
|
|
assert version.pk != same_version.pk
|
|
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
versions = [e.text() for e in resp.pyquery('h3').items()]
|
|
assert versions.count('1.0') == 1
|
|
assert versions.count('2.0') == 1
|
|
assert resp.text.count('Creating application bundle') == 3
|
|
resp = resp.click(href='/applications/manifest/test/download/%s/' % same_version.pk)
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert b'"version_number": "1.0"' in resp.content
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
resp = resp.click(href='/applications/manifest/test/download/%s/' % version.pk)
|
|
assert resp.content_type == 'application/x-tar'
|
|
assert b'"version_number": "2.0"' in resp.content
|
|
|
|
resp = app.get('/applications/manifest/test/generate/')
|
|
assert resp.form['number'].value == '2.0' # last one
|
|
assert resp.form['notes'].value == 'Foo bar blah. But with an icon.' # last one
|
|
resp.form['number'] = '1.0' # old number
|
|
resp = resp.form.submit().follow()
|
|
new_version = Version.objects.latest('pk')
|
|
assert new_version.number == '1.0'
|
|
assert new_version.notes == 'Foo bar blah. But with an icon.'
|
|
assert new_version.pk != version.pk # new version created
|
|
resp = app.get('/applications/manifest/test/versions/')
|
|
versions = [e.text() for e in resp.pyquery('h3').items()]
|
|
assert versions.count('1.0') == 2
|
|
assert versions.count('2.0') == 1
|
|
assert resp.text.count('Creating application bundle') == 4
|
|
|
|
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 HTTMock(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_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),
|
|
]
|
|
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 HTTMock(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 '
|
|
'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 HTTMock(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):
|
|
if url.path == '/api/export-import/forms/unknown/dependencies/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
with HTTMock(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 HTTMock(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 HTTMock(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):
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'status_code': 404}
|
|
return mocked_http(url, request)
|
|
|
|
with HTTMock(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):
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
with HTTMock(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 HTTMock(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 HTTMock(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 HTTMock(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.png' 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,
|
|
},
|
|
],
|
|
}
|
|
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:
|
|
icon_fd = io.BytesIO(
|
|
b'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABAQAAAAA3bvkkAAAACklEQVQI12NoAAAAggCB3UNq9AAAAABJRU5ErkJggg=='
|
|
)
|
|
tarinfo = tarfile.TarInfo('foo.png')
|
|
tarinfo.size = len(icon_fd.getvalue())
|
|
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)
|
|
|
|
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 HTTMock(mocked_http):
|
|
resp = resp.form.submit().follow()
|
|
|
|
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+)?.png', 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'
|
|
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/'}}
|
|
return version
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with HTTMock(mocked_http):
|
|
resp = resp.click(href='/manifest/test/')
|
|
version1 = install(resp, app_bundle)
|
|
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_http(url, request)
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with HTTMock(mocked_http):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with HTTMock(response_content):
|
|
with pytest.raises(DeploymentError) as e:
|
|
resp.form.submit()
|
|
assert str(e.value) == 'Failed to deploy module wcs (500)'
|
|
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):
|
|
if url.path == '/api/export-import/forms/':
|
|
return {
|
|
'content': {"data": [form_def]},
|
|
'status_code': 200,
|
|
}
|
|
return mocked_http(url, request)
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with HTTMock(mocked_http):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with HTTMock(response_content):
|
|
resp.form.submit()
|
|
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):
|
|
if url.path == '/api/export-import/forms/':
|
|
return {'status_code': 500}
|
|
return mocked_http(url, request)
|
|
|
|
resp = app.get('/applications/')
|
|
if action == 'Update':
|
|
with HTTMock(mocked_http):
|
|
resp = resp.click(href='manifest/test/')
|
|
resp = resp.click(action)
|
|
resp.form['bundle'] = Upload('app.tar', app_bundle, 'application/x-tar')
|
|
with HTTMock(response_content):
|
|
with pytest.raises(ScanError) as e:
|
|
resp.form.submit()
|
|
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)'
|
|
|
|
|
|
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).']}
|
|
|
|
|
|
@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['requests'][5].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'] == 5 + 1 + len(available_objects)
|
|
for i, object_type in enumerate(available_objects):
|
|
assert mocked_http.call['requests'][6 + i].url.startswith(
|
|
'https://wcs.example.invalid/api/export-import/%s/?' % object_type['id']
|
|
)
|
|
|
|
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 HTTMock(response_content):
|
|
with pytest.raises(DeploymentError) as e:
|
|
resp.form.submit()
|
|
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):
|
|
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 HTTMock(response_content):
|
|
with pytest.raises(DeploymentError) as e:
|
|
resp.form.submit()
|
|
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)'
|
|
|
|
|
|
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 HTTMock(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):
|
|
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 HTTMock(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/"' 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):
|
|
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 HTTMock(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'
|