wcs/tests/api/test_export_import.py

1970 lines
73 KiB
Python

import io
import json
import os
import tarfile
import uuid
import xml.etree.ElementTree as ET
import pytest
from wcs.api_export_import import klass_to_slug
from wcs.applications import Application, ApplicationElement
from wcs.blocks import BlockDef
from wcs.carddef import CardDef
from wcs.categories import (
BlockCategory,
CardDefCategory,
Category,
CommentTemplateCategory,
DataSourceCategory,
MailTemplateCategory,
WorkflowCategory,
)
from wcs.comment_templates import CommentTemplate
from wcs.data_sources import NamedDataSource
from wcs.fields import BlockField, CommentField, ComputedField, PageField, StringField
from wcs.formdef import FormDef
from wcs.mail_templates import MailTemplate
from wcs.sql import Equal
from wcs.wf.form import WorkflowFormFieldsFormDef
from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef, WorkflowVariablesFieldsFormDef
from wcs.wscalls import NamedWsCall
from ..utilities import clean_temporary_pub, create_temporary_pub, get_app
from .utils import sign_uri
@pytest.fixture
def pub():
pub = create_temporary_pub()
with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd:
fd.write(
'''\
[api-secrets]
coucou = 1234
'''
)
Application.wipe()
ApplicationElement.wipe()
Category.wipe()
FormDef.wipe()
CardDefCategory.wipe()
CardDef.wipe()
BlockCategory.wipe()
BlockDef.wipe()
WorkflowCategory.wipe()
Workflow.wipe()
MailTemplateCategory.wipe()
MailTemplate.wipe()
CommentTemplateCategory.wipe()
CommentTemplate.wipe()
DataSourceCategory.wipe()
NamedDataSource.wipe()
NamedWsCall.wipe()
pub.custom_view_class.wipe()
return pub
def teardown_module(module):
clean_temporary_pub()
def test_export_import_index(pub):
get_app(pub).get('/api/export-import/', status=403)
resp = get_app(pub).get(sign_uri('/api/export-import/'))
assert resp.json['data'][0]['id'] == 'forms'
assert resp.json['data'][0]['text'] == 'Forms'
assert resp.json['data'][0]['urls']['list'] == 'http://example.net/api/export-import/forms/'
def test_export_import_list_forms(pub):
resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
assert not resp.json['data']
formdef = FormDef()
formdef.name = 'Test'
formdef.store()
resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
assert resp.json['data'][0]['id'] == 'test'
assert resp.json['data'][0]['text'] == 'Test'
assert resp.json['data'][0]['category'] is None
category = Category(name='Test')
category.store()
formdef.category = category
formdef.store()
resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
assert resp.json['data'][0]['id'] == 'test'
assert resp.json['data'][0]['text'] == 'Test'
assert resp.json['data'][0]['category'] == 'Test'
def test_export_import_list_404(pub):
get_app(pub).get(sign_uri('/api/export-import/xxx/'), status=404)
def test_export_import_form(pub):
formdef = FormDef()
formdef.name = 'Test'
formdef.store()
resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['export']))
assert resp.text.startswith('<formdef ')
def test_export_import_form_404(pub):
get_app(pub).get(sign_uri('/api/export-import/xxx/plop/'), status=404)
get_app(pub).get(sign_uri('/api/export-import/forms/plop/'), status=404)
def test_export_import_form_dependencies_404(pub):
get_app(pub).get(sign_uri('/api/export-import/forms/plop/dependencies/'), status=404)
def test_export_import_dependencies(pub):
role = pub.role_class(name='Test role')
role.store()
role2 = pub.role_class(name='Second role')
role2.store()
role3 = pub.role_class(name='Third role')
role3.store()
role4 = pub.role_class(name='Fourth role')
role4.uuid = str(uuid.uuid4())
role4.store()
wscall = NamedWsCall(name='Test')
wscall.store()
wscall = NamedWsCall(name='Test bis')
wscall.store()
wscall = NamedWsCall(name='Test ter')
wscall.store()
wscall = NamedWsCall(name='Test quater')
wscall.store()
wscall = NamedWsCall(name='Test quinquies')
wscall.store()
wscall = NamedWsCall(name='Test sexies')
wscall.store()
wscall = NamedWsCall(name='Test in computed field')
wscall.store()
wscall = NamedWsCall(name='Test in lateral template')
wscall.store()
wscall = NamedWsCall(name='Test in loop items template')
wscall.store()
carddef = CardDef()
carddef.name = 'Test'
carddef.store()
carddef = CardDef()
carddef.name = 'Test bis'
carddef.store()
carddef = CardDef()
carddef.name = 'Test ter'
carddef.store()
carddef = CardDef()
carddef.name = 'Test quater'
carddef.store()
carddef = CardDef()
carddef.name = 'Test in loop items template'
carddef.store()
formdef = FormDef()
formdef.name = 'Test bis'
formdef.store()
formdef = FormDef()
formdef.name = 'Test ter'
formdef.store()
formdef = FormDef()
formdef.name = 'Test quater'
formdef.store()
formdef = FormDef()
formdef.name = 'Test quinquies'
formdef.store()
formdef = FormDef()
formdef.name = 'Test'
formdef.store()
resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
assert not resp.json['data']
custom_view = pub.custom_view_class()
custom_view.title = 'shared formdef custom view'
custom_view.formdef = formdef
custom_view.columns = {'list': [{'id': '1'}]}
custom_view.filters = {}
custom_view.visibility = 'role'
custom_view.role_id = role4.id
custom_view.store()
formdef.roles = ['logged-users']
formdef.backoffice_submission_roles = [role2.id]
formdef.workflow_roles = {'_receiver': role3.id}
formdef.store()
block = BlockDef(name='test')
block.store()
workflow = Workflow(name='test')
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
workflow.backoffice_fields_formdef.fields = [
BlockField(id='bo1', label='test', block_slug='test'),
]
workflow.variables_formdef = WorkflowVariablesFieldsFormDef(workflow=workflow)
workflow.variables_formdef.fields = [StringField(label='Test', id='1')]
status = workflow.add_status('New')
status.loop_items_template = '{{ webservice.test_in_loop_items_template }}'
action = status.add_action('form')
action.by = [role.id]
status = workflow.add_status('Next')
status.loop_items_template = '{{ cards|objects:"test-in-loop-items-template" }}'
data_source = NamedDataSource(name='foobar')
data_source.store()
data_source2 = NamedDataSource(name='foobaz')
data_source2.store()
display_form = status.add_action('form', id='_x')
display_form.formdef = WorkflowFormFieldsFormDef(item=display_form)
display_form.formdef.fields.append(
StringField(
label='Test',
data_source={'type': 'foobar'},
prefill={'type': 'string', 'value': '{{ webservice.test_bis.plop }}'},
)
)
display_form.formdef.fields.append(
StringField(
label='Test',
data_source={'type': 'foobar'},
prefill={'type': 'string', 'value': '{{ webservice.unknown }}'},
)
)
display_form.formdef.fields.append(
StringField(
label='Test',
data_source={'type': 'foobar'},
prefill={'type': 'string', 'value': '{{ cards|objects:"unknown" }}'},
)
)
display_form.formdef.fields.append(
StringField(
label='Test',
data_source={'type': 'foobar'},
prefill={'type': 'string', 'value': '{{ cards|objects:"test-bis" }}'},
)
)
display_form.formdef.fields.append(
StringField(
label='Test',
data_source={'type': 'foobar'},
prefill={
'type': 'string',
'value': '{{ data_source.foobaz.plop }} {{ forms|objects:"test-bis" }}',
},
)
)
display_form.formdef.fields.append(
StringField(
label='Test',
data_source={'type': 'foobar'},
prefill={'type': 'string', 'value': '{{ forms|objects:"unknown" }}'},
)
)
send_mail = status.add_action('sendmail')
send_mail.to = [role.id]
send_mail.subject = '{{ webservice.test }}'
send_mail.body = '{{ cards|objects:"test" }} {{ forms|objects:"test-ter" }}'
send_mail.condition = {
'type': 'django',
'value': '{{ cards|objects:"test-ter" }} {{ webservice.test_ter }}',
}
register_comment = status.add_action('register-comment')
register_comment.to = [role.id]
register_comment.comment = (
'{{ cards|objects:"test-quater" }} {{ forms|objects:"test-quinquies" }} {{ webservice.test_sexies }}'
)
dispatch_auto = status.add_action('dispatch')
dispatch_auto.rules = [{'role_id': role.id, 'value': 'xxx'}]
status.add_action('dispatch') # unconfigured/manual dispatch
workflow.store()
formdef.fields = [
PageField(
id='0',
label='Page',
post_conditions=[
{
'condition': {
'type': 'django',
'value': '{{ cards|objects:"test-bis" }} {{ forms|objects:"test-quater" }} {{ webservice.test_quater }}',
},
'error_message': 'You shall not pass.',
}
],
),
BlockField(
id='1',
label='test',
block_slug='test',
condition={
'type': 'django',
'value': '{{ forms|objects:"test-bis" }} {{ webservice.test_quinquies }}',
},
),
CommentField(
id='2',
label='X {{ webservice.test }} X {{ cards|objects:"test" }} X {{ forms|objects:"test-ter" }} X',
),
ComputedField(
id='3',
label='computed field',
varname='computed_field',
value_template='{{ webservice.test_in_computed_field.xxx }}',
),
]
formdef.workflow = workflow
formdef.lateral_template = 'x{{ webservice.test_in_lateral_template.blah }}y'
formdef.store()
resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
form_data = [d for d in resp.json['data'] if d['id'] == 'test']
resp = get_app(pub).get(sign_uri(form_data[0]['urls']['dependencies']))
assert {(x['id'], x['type']) for x in resp.json['data']} == {
('test', 'workflows'),
('test', 'blocks'),
('test', 'wscalls'),
('test_quater', 'wscalls'),
('test_quinquies', 'wscalls'),
('test_in_computed_field', 'wscalls'),
('test_in_lateral_template', 'wscalls'),
('test', 'cards'),
('test-bis', 'cards'),
('test-bis', 'forms'),
('test-ter', 'forms'),
('test-quater', 'forms'),
('second-role', 'roles'),
('third-role', 'roles'),
('fourth-role', 'roles'),
}
for dependency in resp.json['data']:
if dependency['type'] == 'roles':
assert dependency['urls'] == {}
continue
get_app(pub).get(sign_uri(dependency['urls']['export']))
roles = {x['id']: x for x in resp.json['data'] if x['type'] == 'roles'}
assert roles['second-role']['uuid'] is None
assert roles['third-role']['uuid'] is None
assert roles['fourth-role']['uuid'] == role4.uuid
resp = get_app(pub).get(sign_uri('/api/export-import/workflows/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
assert {(x['id'], x['type']) for x in resp.json['data']} == {
('foobar', 'data-sources'),
('foobaz', 'data-sources'),
('test', 'wscalls'),
('test_bis', 'wscalls'),
('test_ter', 'wscalls'),
('test_sexies', 'wscalls'),
('test_in_loop_items_template', 'wscalls'),
('test', 'cards'),
('test-bis', 'cards'),
('test-ter', 'cards'),
('test-quater', 'cards'),
('test-in-loop-items-template', 'cards'),
('test-ter', 'forms'),
('test-bis', 'forms'),
('test-quinquies', 'forms'),
('test', 'blocks'),
('test-role', 'roles'),
}
for dependency in resp.json['data']:
if dependency['type'] == 'roles':
continue
resp = get_app(pub).get(sign_uri(dependency['urls']['export']))
if 'test-role' in dependency['urls']['export']:
assert resp.json == {'name': 'Test role', 'slug': 'test-role', 'uuid': None}
assert resp.content_type == 'application/json'
else:
assert resp.content_type == 'text/xml'
mail_template = MailTemplate(name='test mail template')
mail_template.store()
send_mail.mail_template = mail_template.slug
comment_template = CommentTemplate(name='test comment template')
comment_template.store()
register_comment.comment_template = comment_template.slug
workflow.store()
resp = get_app(pub).get(sign_uri('/api/export-import/workflows/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
assert {(x['id'], x['type']) for x in resp.json['data']} == {
('foobar', 'data-sources'),
('foobaz', 'data-sources'),
('test_bis', 'wscalls'),
('test_ter', 'wscalls'),
('test_in_loop_items_template', 'wscalls'),
('test-bis', 'cards'),
('test-ter', 'cards'),
('test-in-loop-items-template', 'cards'),
('test-bis', 'forms'),
('test', 'blocks'),
('test-mail-template', 'mail-templates'),
('test-comment-template', 'comment-templates'),
('test-role', 'roles'),
}
for dependency in resp.json['data']:
if dependency['type'] == 'roles':
continue
get_app(pub).get(sign_uri(dependency['urls']['export']))
resp = get_app(pub).get(sign_uri(resp.json['data'][-3]['urls']['dependencies']))
assert resp.json['data'] == []
cat = MailTemplateCategory(name='Cat')
cat.store()
mail_template.category_id = cat.id
mail_template.subject = '{{ webservice.test }}'
mail_template.body = '{{ cards|objects:"test" }} {{ forms|objects:"test-ter" }}'
mail_template.store()
resp = get_app(pub).get(sign_uri('/api/export-import/workflows/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
mail_template_entry = [x for x in resp.json['data'] if x['type'] == 'mail-templates'][0]
resp = get_app(pub).get(sign_uri(mail_template_entry['urls']['dependencies']))
assert {(x['id'], x['type']) for x in resp.json['data']} == {
('cat', 'mail-templates-categories'),
('test', 'cards'),
('test', 'wscalls'),
('test-ter', 'forms'),
}
for dependency in resp.json['data']:
get_app(pub).get(sign_uri(dependency['urls']['export']))
cat = CommentTemplateCategory(name='Cat')
cat.store()
comment_template.category_id = cat.id
comment_template.comment = (
'{{ cards|objects:"test-quater" }} {{ forms|objects:"test-quinquies" }} {{ webservice.test_sexies }}'
)
comment_template.store()
resp = get_app(pub).get(sign_uri('/api/export-import/workflows/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
comment_template_entry = [x for x in resp.json['data'] if x['type'] == 'comment-templates'][0]
resp = get_app(pub).get(sign_uri(comment_template_entry['urls']['dependencies']))
assert {(x['id'], x['type']) for x in resp.json['data']} == {
('cat', 'comment-templates-categories'),
('test-quater', 'cards'),
('test-quinquies', 'forms'),
('test_sexies', 'wscalls'),
}
for dependency in resp.json['data']:
get_app(pub).get(sign_uri(dependency['urls']['export']))
resp = get_app(pub).get(sign_uri('/api/export-import/workflows/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
resp = get_app(pub).get(sign_uri(resp.json['data'][2]['urls']['dependencies']))
assert resp.json['data'] == []
cat = DataSourceCategory(name='Cat')
cat.store()
data_source.category_id = cat.id
data_source.store()
resp = get_app(pub).get(sign_uri('/api/export-import/workflows/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
data_sources_entry = [
x for x in resp.json['data'] if x['type'] == 'data-sources' and x['id'] == 'foobar'
][0]
resp = get_app(pub).get(sign_uri(data_sources_entry['urls']['dependencies']))
assert {(x['id'], x['type']) for x in resp.json['data']} == {('cat', 'data-sources-categories')}
def test_export_import_dependencies_default_workflow(pub):
formdef = FormDef()
formdef.name = 'Test'
formdef.workflow_id = '_default'
formdef.store()
carddef = CardDef()
carddef.name = 'Test'
carddef.workflow_id = '_carddef_default'
carddef.store()
resp = get_app(pub).get(sign_uri('/api/export-import/forms/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
assert resp.json['data'] == []
resp = get_app(pub).get(sign_uri('/api/export-import/cards/'))
resp = get_app(pub).get(sign_uri(resp.json['data'][0]['urls']['dependencies']))
assert resp.json['data'] == []
def test_export_import_redirect_url(pub):
workflow = Workflow(name='test')
workflow.store()
block = BlockDef(name='test')
block.store()
formdef = FormDef()
formdef.name = 'Test'
formdef.store()
carddef = CardDef()
carddef.name = 'Test'
carddef.store()
category = Category(name='Test')
category.store()
data_source = NamedDataSource(name='Test')
data_source.store()
ds_category = DataSourceCategory(name='Test')
ds_category.store()
mail_template = MailTemplate(name='Test')
mail_template.store()
mail_template_category = MailTemplateCategory(name='Test')
mail_template_category.store()
comment_template = CommentTemplate(name='Test')
comment_template.store()
comment_template_category = CommentTemplateCategory(name='Test')
comment_template_category.store()
elements = [
('forms', '/backoffice/forms/%s/' % formdef.id),
('cards', '/backoffice/cards/%s/' % carddef.id),
('blocks', '/backoffice/forms/blocks/%s/' % block.id),
('workflows', '/backoffice/workflows/%s/' % workflow.id),
('forms-categories', '/backoffice/forms/categories/%s/' % category.id),
('data-sources', '/backoffice/settings/data-sources/%s/' % data_source.id),
(
'data-sources-categories',
'/backoffice/forms/data-sources/categories/%s/' % ds_category.id,
),
('mail-templates', '/backoffice/workflows/mail-templates/%s/' % mail_template.id),
(
'mail-templates-categories',
'/backoffice/workflows/mail-templates/categories/%s/' % mail_template_category.id,
),
('comment-templates', '/backoffice/workflows/comment-templates/%s/' % comment_template.id),
(
'comment-templates-categories',
'/backoffice/workflows/comment-templates/categories/%s/' % comment_template_category.id,
),
]
for object_type, obj_url in elements:
resp = get_app(pub).get(sign_uri('/api/export-import/%s/' % object_type))
redirect_url = resp.json['data'][0]['urls']['redirect']
assert redirect_url == 'http://example.net/api/export-import/%s/test/redirect/' % object_type
resp = get_app(pub).get(redirect_url, status=302)
assert resp.location == 'http://example.net%s' % obj_url
get_app(pub).get('/api/export-import/%s/unknown/redirect/' % object_type, status=404)
resp = get_app(pub).get(redirect_url + '?compare', status=302)
assert resp.location == 'http://example.net%s' % obj_url
resp = get_app(pub).get(
redirect_url + '?compare&version1=bar&version2=bar&application=foo', status=302
)
assert (
resp.location
== 'http://example.net%shistory/compare?version1=bar&version2=bar&application=foo' % obj_url
)
role = pub.role_class(name='test')
role.store()
resp = get_app(pub).get(sign_uri('/api/export-import/roles/'))
assert resp.json['data'][0]['urls'].get('redirect') is None
get_app(pub).get('/api/export-import/roles/test/redirect/', status=404)
def create_bundle(elements, *args, **kwargs):
visible = kwargs.get('visible', True)
version_number = kwargs.get('version_number', '42.0')
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
manifest_json = {
'application': 'Test',
'slug': 'test',
'icon': 'foo.png',
'description': 'Foo Bar',
'documentation_url': 'http://foo.bar',
'visible': visible,
'version_number': version_number,
'version_notes': 'foo bar blah',
'elements': 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'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABAQAAAAA3bvkkAAAACklEQVQI12NoAAAAggCB3UNq9AAAAABJRU5ErkJggg=='
)
tarinfo = tarfile.TarInfo('foo.png')
tarinfo.size = len(icon_fd.getvalue())
tar.addfile(tarinfo, fileobj=icon_fd)
for path, obj in args:
tarinfo = tarfile.TarInfo(path)
if hasattr(obj, 'export_for_application'):
export, _ = obj.export_for_application()
export = export.encode()
else:
export = ET.tostring(obj.export_to_xml(include_id=True))
tarinfo.size = len(export)
tar.addfile(tarinfo, fileobj=io.BytesIO(export))
return tar_io.getvalue()
def test_export_import_bundle_import(pub):
workflow_category = WorkflowCategory(name='test')
workflow_category.store()
workflow = Workflow(name='test')
workflow.roles = {'_receiver': 'Receiver'}
workflow.category = workflow_category
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
workflow.backoffice_fields_formdef.fields = [
StringField(id='bo1', label='test', type='string'),
]
workflow.store()
block_category = BlockCategory(name='test')
block_category.store()
block = BlockDef(name='test')
block.category = block_category
block.store()
role = pub.role_class(name='test')
role.store()
category = Category(name='Test')
category.store()
formdef = FormDef()
formdef.name = 'Test'
formdef.fields = [
BlockField(id='1', label='test', block_slug='test'),
]
formdef.workflow = workflow
formdef.workflow_roles = {'_receiver': role.id}
formdef.disabled = False
formdef.category = category
formdef.store()
custom_view = pub.custom_view_class()
custom_view.title = 'shared formdef custom view'
custom_view.formdef = formdef
custom_view.columns = {'list': [{'id': '1'}]}
custom_view.filters = {}
custom_view.visibility = 'any'
custom_view.store()
card_category = CardDefCategory(name='Test')
card_category.store()
carddef = CardDef()
carddef.name = 'Test'
carddef.category = card_category
carddef.store()
ds_category = DataSourceCategory(name='Test')
ds_category.store()
data_source = NamedDataSource(name='Test')
data_source.category = ds_category
data_source.store()
mail_template_category = MailTemplateCategory(name='Test')
mail_template_category.store()
mail_template = MailTemplate(name='Test')
mail_template.category = mail_template_category
mail_template.store()
comment_template_category = CommentTemplateCategory(name='Test')
comment_template_category.store()
comment_template = CommentTemplate(name='Test')
comment_template.category = comment_template_category
comment_template.store()
wscall = NamedWsCall(name='Test')
wscall.store()
bundles = []
for version_number in ['42.0', '42.1']:
bundles.append(
create_bundle(
[
{'type': 'forms-categories', 'slug': 'test', 'name': 'test'},
{'type': 'forms', 'slug': 'test', 'name': 'test'},
{'type': 'cards-categories', 'slug': 'test', 'name': 'test'},
{'type': 'cards', 'slug': 'test', 'name': 'test'},
{'type': 'blocks-categories', 'slug': 'test', 'name': 'test'},
{'type': 'blocks', 'slug': 'test', 'name': 'test'},
{'type': 'roles', 'slug': 'test', 'name': 'test'},
{'type': 'workflows-categories', 'slug': 'test', 'name': 'test'},
{'type': 'workflows', 'slug': 'test', 'name': 'test'},
{'type': 'mail-templates-categories', 'slug': 'test', 'name': 'test'},
{'type': 'mail-templates', 'slug': 'test', 'name': 'test'},
{'type': 'comment-templates-categories', 'slug': 'test', 'name': 'test'},
{'type': 'comment-templates', 'slug': 'test', 'name': 'test'},
{'type': 'data-sources-categories', 'slug': 'test', 'name': 'test'},
{'type': 'data-sources', 'slug': 'test', 'name': 'test'},
{'type': 'wscalls', 'slug': 'test', 'name': 'test'},
{'type': 'foobar', 'slug': 'test', 'name': 'test'},
],
('forms-categories/test', category),
('forms/test', formdef),
('cards-categories/test', card_category),
('cards/test', carddef),
('blocks-categories/test', block_category),
('blocks/test', block),
('workflows-categories/test', workflow_category),
('workflows/test', workflow),
('data-sources-categories/test', ds_category),
('data-sources/test', data_source),
('mail-templates-categories/test', mail_template_category),
('mail-templates/test', mail_template),
('comment-templates-categories/test', comment_template_category),
('comment-templates/test', comment_template),
('roles/test', role),
('wscalls/test', wscall),
version_number=version_number,
)
)
object_classes = [
Category,
FormDef,
CardDefCategory,
CardDef,
BlockCategory,
BlockDef,
WorkflowCategory,
Workflow,
MailTemplateCategory,
MailTemplate,
CommentTemplateCategory,
CommentTemplate,
DataSourceCategory,
NamedDataSource,
pub.custom_view_class,
pub.role_class,
NamedWsCall,
]
for object_class in object_classes:
object_class.wipe()
# roles will be created beforehand with authentic provisionning
extra_role = pub.role_class(name='not this one')
extra_role.store()
role = pub.role_class(name='test')
role.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundles[0])
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert resp.json['data']['completion_status'] == '34/34 (100%)'
assert Category.count() == 1
assert FormDef.count() == 1
assert FormDef.select()[0].fields[0].key == 'block'
assert FormDef.select()[0].fields[0].block_slug == 'test'
assert FormDef.select()[0].workflow_roles == {'_receiver': role.id}
assert FormDef.select()[0].category_id == Category.select()[0].id
assert CardDefCategory.count() == 1
assert CardDef.count() == 1
assert CardDef.select()[0].category_id == CardDefCategory.select()[0].id
assert BlockCategory.count() == 1
assert BlockDef.count() == 1
assert BlockDef.select()[0].category_id == BlockCategory.select()[0].id
assert WorkflowCategory.count() == 1
assert Workflow.count() == 1
assert Workflow.select()[0].category_id == WorkflowCategory.select()[0].id
assert MailTemplateCategory.count() == 1
assert MailTemplate.count() == 1
assert MailTemplate.select()[0].category_id == MailTemplateCategory.select()[0].id
assert CommentTemplateCategory.count() == 1
assert CommentTemplate.count() == 1
assert CommentTemplate.select()[0].category_id == CommentTemplateCategory.select()[0].id
assert DataSourceCategory.count() == 1
assert NamedDataSource.count() == 1
assert NamedDataSource.select()[0].category_id == DataSourceCategory.select()[0].id
assert NamedWsCall.count() == 1
assert pub.custom_view_class().count() == 1
assert Application.count() == 1
application = Application.select()[0]
assert application.slug == 'test'
assert application.name == 'Test'
assert application.description == 'Foo Bar'
assert application.documentation_url == 'http://foo.bar'
assert application.version_number == '42.0'
assert application.version_notes == 'foo bar blah'
assert application.icon.base_filename == 'foo.png'
assert application.editable is False
assert application.visible is True
assert ApplicationElement.count() == 15
# check backoffice field have been added to table
# (a sql error would happen if it was missing)
formdef = FormDef.select()[0]
formdef.data_class().select()
# check also that workflow's label in last snapshot does not starts with '[pre-import]'
last_snapshot = pub.snapshot_class.get_latest(formdef.xml_root_node, formdef.id)
assert last_snapshot.comment == 'Application (Test) finalize initial installation'
assert '<workflow slug="test" workflow_id="1">test</workflow>' in last_snapshot.get_serialization()
for object_class in object_classes:
if object_class in [pub.custom_view_class, pub.role_class]:
# no snapshot for this objects
continue
for obj in object_class.select():
last_snapshot = pub.snapshot_class.select_object_history(obj)[0]
if last_snapshot.comment == 'Application (Test) finalize initial installation':
previous_snapshot = pub.snapshot_class.select_object_history(obj)[1]
assert previous_snapshot.comment == 'Application (Test) complete initial installation'
assert previous_snapshot.application_slug == 'test'
assert previous_snapshot.application_version == '42.0'
previous_snapshot = pub.snapshot_class.select_object_history(obj)[2]
assert previous_snapshot.comment == 'Application (Test) initial installation'
assert previous_snapshot.application_slug is None
assert previous_snapshot.application_version is None
elif last_snapshot.comment == 'Application (Test) complete initial installation':
previous_snapshot = pub.snapshot_class.select_object_history(obj)[1]
assert previous_snapshot.comment == 'Application (Test) initial installation'
assert previous_snapshot.application_slug is None
assert previous_snapshot.application_version is None
else:
assert last_snapshot.comment == 'Application (Test)'
assert last_snapshot.application_slug == 'test'
assert last_snapshot.application_version == '42.0'
# check editable flag is kept on install
application.editable = False
application.store()
# create some links to elements not present in manifest: they should be unlinked
element1 = ApplicationElement()
element1.application_id = application.id
element1.object_type = 'foobar'
element1.object_id = '42'
element1.store()
element2 = ApplicationElement()
element2.application_id = application.id
element2.object_type = 'foobarblah'
element2.object_id = '35'
element2.store()
# run new import to check it doesn't duplicate objects
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundles[1])
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert Category.count() == 1
assert FormDef.count() == 1
assert CardDefCategory.count() == 1
assert CardDef.count() == 1
assert BlockCategory.count() == 1
assert BlockDef.count() == 1
assert WorkflowCategory.count() == 1
assert Workflow.count() == 1
assert MailTemplateCategory.count() == 1
assert MailTemplate.count() == 1
assert CommentTemplateCategory.count() == 1
assert CommentTemplate.count() == 1
assert DataSourceCategory.count() == 1
assert NamedDataSource.count() == 1
assert pub.custom_view_class().count() == 1
assert NamedWsCall.count() == 1
assert Application.count() == 1
assert ApplicationElement.count() == 15
assert (
ApplicationElement.select(
[
Equal('application_id', application.id),
Equal('object_type', element1.object_type),
Equal('object_id', element1.object_id),
]
)
== []
)
assert (
ApplicationElement.select(
[
Equal('application_id', application.id),
Equal('object_type', element2.object_type),
Equal('object_id', element2.object_id),
]
)
== []
)
application = Application.select()[0]
assert application.editable is False
for object_class in object_classes:
if object_class in [pub.custom_view_class, pub.role_class]:
# no snapshot for this objects
continue
for obj in object_class.select():
last_snapshot = pub.snapshot_class.select_object_history(obj)[0]
assert last_snapshot.comment == 'Application (Test) update'
assert last_snapshot.application_slug == 'test'
assert last_snapshot.application_version == '42.1'
# change immutable attributes and check they are not reset
formdef = FormDef.select()[0]
formdef.workflow_roles = {'_receiver': extra_role.id}
formdef.disabled = True
formdef.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundles[1])
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
formdef = FormDef.select()[0]
assert formdef.disabled is True
assert formdef.workflow_roles == {'_receiver': extra_role.id}
@pytest.mark.parametrize(
'category_class',
[
Category,
CardDefCategory,
BlockCategory,
WorkflowCategory,
MailTemplateCategory,
CommentTemplateCategory,
DataSourceCategory,
],
)
def test_export_import_bundle_import_categories_ordering(pub, category_class):
category_class.wipe()
category = category_class(name='cat 1')
category.position = 1
category.store()
category = category_class(name='cat 2')
category.position = 2
category.store()
category = category_class(name='cat 3')
category.position = 3
category.store()
bundle = create_bundle(
[
{'type': klass_to_slug[category_class], 'slug': 'cat-1', 'name': 'cat 1'},
{'type': klass_to_slug[category_class], 'slug': 'cat-2', 'name': 'cat 2'},
{'type': klass_to_slug[category_class], 'slug': 'cat-3', 'name': 'cat 3'},
],
('%s/cat-1' % klass_to_slug[category_class], category_class.get(1)),
('%s/cat-2' % klass_to_slug[category_class], category_class.get(2)),
('%s/cat-3' % klass_to_slug[category_class], category_class.get(3)),
)
# delete categories
category_class.wipe()
# and recreate only cat 4 and 5 in first positions
category = category_class(name='cat 4')
category.position = 1
category.store()
category = category_class(name='cat 5')
category.position = 2
category.store()
# import bundle
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
# cat 1, 2, 3 are placed at the end
assert category_class.get_by_slug('cat-4').position == 1
assert category_class.get_by_slug('cat-5').position == 2
assert category_class.get_by_slug('cat-1').position == 3
assert category_class.get_by_slug('cat-2').position == 4
assert category_class.get_by_slug('cat-3').position == 5
# delete categories
category_class.wipe()
# recreate only cat 2, cat 4, cat 5 in this order
category = category_class(name='cat 2')
category.position = 1
category.store()
category = category_class(name='cat 4')
category.position = 2
category.store()
category = category_class(name='cat 5')
category.position = 3
category.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
# cat 1, 2, 3 are placed after cat 4
assert category_class.get_by_slug('cat-1').position == 1
assert category_class.get_by_slug('cat-2').position == 2
assert category_class.get_by_slug('cat-3').position == 3
assert category_class.get_by_slug('cat-4').position == 4
assert category_class.get_by_slug('cat-5').position == 5
# delete categories
category_class.wipe()
# recreate only cat 4, cat 2, cat 5 in this order
category = category_class(name='cat 4')
category.position = 1
category.store()
category = category_class(name='cat 2')
category.position = 2
category.store()
category = category_class(name='cat 5')
category.position = 3
category.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
# cat 1, 2, 3 are placed after cat 4
assert category_class.get_by_slug('cat-4').position == 1
assert category_class.get_by_slug('cat-1').position == 2
assert category_class.get_by_slug('cat-2').position == 3
assert category_class.get_by_slug('cat-3').position == 4
assert category_class.get_by_slug('cat-5').position == 5
# delete categories
category_class.wipe()
# recreate only cat 4, cat 5, cat 2 in this order
category = category_class(name='cat 4')
category.position = 1
category.store()
category = category_class(name='cat 5')
category.position = 2
category.store()
category = category_class(name='cat 2')
category.position = 3
category.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
# cat 1, 2, 3 are placed after cat 4
assert category_class.get_by_slug('cat-4').position == 1
assert category_class.get_by_slug('cat-5').position == 2
assert category_class.get_by_slug('cat-1').position == 3
assert category_class.get_by_slug('cat-2').position == 4
assert category_class.get_by_slug('cat-3').position == 5
# delete categories
category_class.wipe()
# recreate only cat 4, cat 2, cat1 cat 5 in this order but with weird positions
category = category_class(name='cat 4')
category.position = 4
category.store()
category = category_class(name='cat 2')
category.position = 12
category.store()
category = category_class(name='cat 1')
category.position = 13
category.store()
category = category_class(name='cat 5')
category.position = 20
category.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
# cat 1, 2, 3 are placed after cat 4
assert category_class.get_by_slug('cat-4').position == 1
assert category_class.get_by_slug('cat-1').position == 2
assert category_class.get_by_slug('cat-2').position == 3
assert category_class.get_by_slug('cat-3').position == 4
assert category_class.get_by_slug('cat-5').position == 5
# delete categories
category_class.wipe()
# recreate only cat 4, cat 2, cat1 cat 5 in this order but with weird positions
category = category_class(name='cat 4')
category.position = 1
category.store()
category = category_class(name='cat 2')
category.position = 2
category.store()
category = category_class(name='cat 1')
category.position = 2
category.store()
category = category_class(name='cat 5')
category.position = 2
category.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
# cat 1, 2, 3 are placed after cat 4
assert category_class.get_by_slug('cat-4').position == 1
assert category_class.get_by_slug('cat-1').position == 2
assert category_class.get_by_slug('cat-2').position == 3
assert category_class.get_by_slug('cat-3').position == 4
assert category_class.get_by_slug('cat-5').position == 5
def test_export_import_formdef_do_not_overwrite_table_name(pub):
formdef = FormDef()
formdef.name = 'Test2'
formdef.fields = []
formdef.disabled = False
formdef.store()
assert formdef.table_name == 'formdata_%s_test2' % formdef.id
formdata = formdef.data_class()()
formdata.just_created()
formdata.store()
# change formdef url name, internal table name won't be changed
formdef.url_name = 'test'
formdef.store()
assert formdef.table_name == 'formdata_%s_test2' % formdef.id
assert formdef.data_class().count() == 1
bundle = create_bundle([{'type': 'forms', 'slug': 'test', 'name': 'test'}], ('forms/test', formdef))
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
# check table name is not overwritten
formdef = FormDef.select()[0]
assert formdef.table_name == 'formdata_%s_test2' % formdef.id
assert formdef.data_class().count() == 1
def test_export_import_bundle_declare(pub):
workflow_category = WorkflowCategory(name='test')
workflow_category.store()
workflow = Workflow(name='test')
workflow.store()
block_category = BlockCategory(name='test')
block_category.store()
block = BlockDef(name='test')
block.store()
role = pub.role_class(name='test')
role.store()
category = Category(name='Test')
category.store()
formdef = FormDef()
formdef.name = 'Test'
formdef.store()
card_category = CardDefCategory(name='Test')
card_category.store()
carddef = CardDef()
carddef.name = 'Test'
carddef.store()
ds_category = DataSourceCategory(name='Test')
ds_category.store()
data_source = NamedDataSource(name='Test')
data_source.store()
mail_template_category = MailTemplateCategory(name='Test')
mail_template_category.store()
mail_template = MailTemplate(name='Test')
mail_template.store()
comment_template_category = CommentTemplateCategory(name='Test')
comment_template_category.store()
comment_template = CommentTemplate(name='Test')
comment_template.store()
wscall = NamedWsCall(name='Test')
wscall.store()
bundle = create_bundle(
[
{'type': 'forms-categories', 'slug': 'test', 'name': 'test'},
{'type': 'forms', 'slug': 'test', 'name': 'test'},
{'type': 'cards-categories', 'slug': 'test', 'name': 'test'},
{'type': 'cards', 'slug': 'test', 'name': 'test'},
{'type': 'blocks-categories', 'slug': 'test', 'name': 'test'},
{'type': 'blocks', 'slug': 'test', 'name': 'test'},
{'type': 'roles', 'slug': 'test', 'name': 'test'},
{'type': 'workflows-categories', 'slug': 'test', 'name': 'test'},
{'type': 'workflows', 'slug': 'test', 'name': 'test'},
{'type': 'mail-templates-categories', 'slug': 'test', 'name': 'test'},
{'type': 'mail-templates', 'slug': 'test', 'name': 'test'},
{'type': 'comment-templates-categories', 'slug': 'test', 'name': 'test'},
{'type': 'comment-templates', 'slug': 'test', 'name': 'test'},
{'type': 'data-sources-categories', 'slug': 'test', 'name': 'test'},
{'type': 'data-sources', 'slug': 'test', 'name': 'test'},
{'type': 'data-sources', 'slug': 'unknown', 'name': 'unknown'},
{'type': 'wscalls', 'slug': 'test', 'name': 'test'},
{'type': 'foobar', 'slug': 'test', 'name': 'test'},
],
('forms-categories/test', category),
('forms/test', formdef),
('cards-categories/test', card_category),
('cards/test', carddef),
('blocks-categories/test', block_category),
('blocks/test', block),
('workflows-categories/test', workflow_category),
('workflows/test', workflow),
('data-sources-categories/test', ds_category),
('data-sources/test', data_source),
('data-sources/unknown', data_source),
('mail-templates-categories/test', mail_template_category),
('mail-templates/test', mail_template),
('comment-templates-categories/test', comment_template_category),
('comment-templates/test', comment_template),
('roles/test', role),
('wscalls/test', wscall),
visible=False,
)
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-declare/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert resp.json['data']['completion_status'] == '16/16 (100%)'
assert Application.count() == 1
application = Application.select()[0]
assert application.slug == 'test'
assert application.name == 'Test'
assert application.description == 'Foo Bar'
assert application.documentation_url == 'http://foo.bar'
assert application.version_number == '42.0'
assert application.version_notes == 'foo bar blah'
assert application.icon.base_filename == 'foo.png'
assert application.editable is True
assert application.visible is False
assert ApplicationElement.count() == 15
# create some links to elements not present in manifest: they should be unlinked
element1 = ApplicationElement()
element1.application_id = application.id
element1.object_type = 'foobar'
element1.object_id = '42'
element1.store()
element2 = ApplicationElement()
element2.application_id = application.id
element2.object_type = 'foobarblah'
element2.object_id = '35'
element2.store()
# and remove an object to have an unkown reference in manifest
MailTemplate.wipe()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-declare/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert Application.count() == 1
assert ApplicationElement.count() == 14
assert (
ApplicationElement.select(
[
Equal('application_id', application.id),
Equal('object_type', element1.object_type),
Equal('object_id', element1.object_id),
]
)
== []
)
assert (
ApplicationElement.select(
[
Equal('application_id', application.id),
Equal('object_type', element2.object_type),
Equal('object_id', element2.object_id),
]
)
== []
)
assert (
ApplicationElement.select(
[
Equal('application_id', application.id),
Equal('object_type', MailTemplate.xml_root_node),
]
)
== []
)
def test_export_import_bundle_unlink(pub):
application = Application()
application.slug = 'test'
application.name = 'Test'
application.version_number = 'foo'
application.store()
other_application = Application()
other_application.slug = 'other-test'
other_application.name = 'Other Test'
other_application.version_number = 'foo'
other_application.store()
workflow_category = WorkflowCategory(name='test')
workflow_category.store()
ApplicationElement.update_or_create_for_object(application, workflow_category)
workflow = Workflow(name='test')
workflow.store()
ApplicationElement.update_or_create_for_object(application, workflow)
block_category = BlockCategory(name='test')
block_category.store()
ApplicationElement.update_or_create_for_object(application, block_category)
block = BlockDef(name='test')
block.store()
ApplicationElement.update_or_create_for_object(application, block)
category = Category(name='Test')
category.store()
ApplicationElement.update_or_create_for_object(application, category)
formdef = FormDef()
formdef.name = 'Test'
formdef.store()
ApplicationElement.update_or_create_for_object(application, formdef)
card_category = CardDefCategory(name='Test')
card_category.store()
ApplicationElement.update_or_create_for_object(application, card_category)
carddef = CardDef()
carddef.name = 'Test'
carddef.store()
ApplicationElement.update_or_create_for_object(application, carddef)
ds_category = DataSourceCategory(name='Test')
ds_category.store()
ApplicationElement.update_or_create_for_object(application, ds_category)
data_source = NamedDataSource(name='Test')
data_source.store()
ApplicationElement.update_or_create_for_object(application, data_source)
mail_template_category = MailTemplateCategory(name='Test')
mail_template_category.store()
ApplicationElement.update_or_create_for_object(application, mail_template_category)
mail_template = MailTemplate(name='Test')
mail_template.store()
ApplicationElement.update_or_create_for_object(application, mail_template)
comment_template_category = CommentTemplateCategory(name='Test')
comment_template_category.store()
ApplicationElement.update_or_create_for_object(application, comment_template_category)
comment_template = CommentTemplate(name='Test')
comment_template.store()
ApplicationElement.update_or_create_for_object(application, comment_template)
wscall = NamedWsCall(name='Test')
wscall.store()
ApplicationElement.update_or_create_for_object(application, wscall)
element = ApplicationElement()
element.application_id = application.id
element.object_type = 'foobar'
element.object_id = '42'
element.store()
other_element = ApplicationElement()
other_element.application_id = other_application.id
other_element.object_type = 'foobar'
other_element.object_id = '42'
other_element.store()
assert Application.count() == 2
assert ApplicationElement.count() == 17
get_app(pub).post(sign_uri('/api/export-import/unlink/'), {'application': 'test'})
assert Application.count() == 1
assert ApplicationElement.count() == 1
assert (
Application.count(
[
Equal('id', other_application.id),
]
)
== 1
)
assert (
ApplicationElement.count(
[
Equal('application_id', other_application.id),
]
)
== 1
)
# again
get_app(pub).post(sign_uri('/api/export-import/unlink/'), {'application': 'test'})
assert Application.count() == 1
assert ApplicationElement.count() == 1
def test_export_import_bundle_check(pub):
workflow_category = WorkflowCategory(name='test')
workflow_category.store()
workflow = Workflow(name='test')
workflow.store()
block_category = BlockCategory(name='test')
block_category.store()
block = BlockDef(name='test')
block.store()
role = pub.role_class(name='test')
role.store()
category = Category(name='Test')
category.store()
formdef = FormDef()
formdef.name = 'Test'
formdef.store()
card_category = CardDefCategory(name='Test')
card_category.store()
carddef = CardDef()
carddef.name = 'Test'
carddef.store()
ds_category = DataSourceCategory(name='Test')
ds_category.store()
data_source = NamedDataSource(name='Test')
data_source.store()
mail_template_category = MailTemplateCategory(name='Test')
mail_template_category.store()
mail_template = MailTemplate(name='Test')
mail_template.store()
comment_template_category = CommentTemplateCategory(name='Test')
comment_template_category.store()
comment_template = CommentTemplate(name='Test')
comment_template.store()
wscall = NamedWsCall(name='Test')
wscall.store()
bundles = []
for version in ['1.0', '2.0']:
bundles.append(
create_bundle(
[
{'type': 'forms-categories', 'slug': 'test', 'name': 'test'},
{'type': 'forms', 'slug': 'test', 'name': 'test'},
{'type': 'cards-categories', 'slug': 'test', 'name': 'test'},
{'type': 'cards', 'slug': 'test', 'name': 'test'},
{'type': 'blocks-categories', 'slug': 'test', 'name': 'test'},
{'type': 'blocks', 'slug': 'test', 'name': 'test'},
{'type': 'roles', 'slug': 'test', 'name': 'test'},
{'type': 'workflows-categories', 'slug': 'test', 'name': 'test'},
{'type': 'workflows', 'slug': 'test', 'name': 'test'},
{'type': 'mail-templates-categories', 'slug': 'test', 'name': 'test'},
{'type': 'mail-templates', 'slug': 'test', 'name': 'test'},
{'type': 'comment-templates-categories', 'slug': 'test', 'name': 'test'},
{'type': 'comment-templates', 'slug': 'test', 'name': 'test'},
{'type': 'data-sources-categories', 'slug': 'test', 'name': 'test'},
{'type': 'data-sources', 'slug': 'test', 'name': 'test'},
{'type': 'wscalls', 'slug': 'test', 'name': 'test'},
{'type': 'foobar', 'slug': 'test', 'name': 'test'},
],
('forms-categories/test', category),
('forms/test', formdef),
('cards-categories/test', card_category),
('cards/test', carddef),
('blocks-categories/test', block_category),
('blocks/test', block),
('workflows-categories/test', workflow_category),
('workflows/test', workflow),
('data-sources-categories/test', ds_category),
('data-sources/test', data_source),
('mail-templates-categories/test', mail_template_category),
('mail-templates/test', mail_template),
('comment-templates-categories/test', comment_template_category),
('comment-templates/test', comment_template),
('roles/test', role),
('wscalls/test', wscall),
visible=False,
version=version,
)
)
object_classes = [
Category,
FormDef,
CardDefCategory,
CardDef,
BlockCategory,
BlockDef,
WorkflowCategory,
Workflow,
MailTemplateCategory,
MailTemplate,
CommentTemplateCategory,
CommentTemplate,
DataSourceCategory,
NamedDataSource,
pub.custom_view_class,
pub.role_class,
NamedWsCall,
]
for object_class in object_classes:
object_class.wipe()
pub.snapshot_class.wipe()
incomplete_bundles = []
for manifest_json in [{'slug': 'test'}, {'version_number': '1.0'}]:
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
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)
incomplete_bundles.append(tar_io.getvalue())
# incorrect bundles, missing information
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), incomplete_bundles[0])
assert resp.json == {'data': {}}
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), incomplete_bundles[1])
assert resp.json == {'data': {}}
# not yet imported
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), bundles[0])
assert resp.json == {
'data': {
'differences': [],
'no_history_elements': [],
'unknown_elements': [
{'slug': 'test', 'type': 'forms-categories'},
{'slug': 'test', 'type': 'forms'},
{'slug': 'test', 'type': 'cards-categories'},
{'slug': 'test', 'type': 'cards'},
{'slug': 'test', 'type': 'blocks-categories'},
{'slug': 'test', 'type': 'blocks'},
{'slug': 'test', 'type': 'workflows-categories'},
{'slug': 'test', 'type': 'workflows'},
{'slug': 'test', 'type': 'mail-templates-categories'},
{'slug': 'test', 'type': 'mail-templates'},
{'slug': 'test', 'type': 'comment-templates-categories'},
{'slug': 'test', 'type': 'comment-templates'},
{'slug': 'test', 'type': 'data-sources-categories'},
{'slug': 'test', 'type': 'data-sources'},
{'slug': 'test', 'type': 'wscalls'},
],
'legacy_elements': [],
},
}
# import bundle
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundles[0])
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert resp.json['data']['completion_status'] == '34/34 (100%)'
assert Application.count() == 1
assert ApplicationElement.count() == 15
# remove application links
Application.wipe()
ApplicationElement.wipe()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), bundles[0])
assert resp.json == {
'data': {
'differences': [],
'unknown_elements': [],
'no_history_elements': [],
'legacy_elements': [
{
'slug': 'test',
'text': 'Test',
'type': 'forms-categories',
'url': 'http://example.net/api/export-import/forms-categories/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'forms',
'url': 'http://example.net/api/export-import/forms/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'cards-categories',
'url': 'http://example.net/api/export-import/cards-categories/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'cards',
'url': 'http://example.net/api/export-import/cards/test/redirect/',
},
{
'slug': 'test',
'text': 'test',
'type': 'blocks-categories',
'url': 'http://example.net/api/export-import/blocks-categories/test/redirect/',
},
{
'slug': 'test',
'text': 'test',
'type': 'blocks',
'url': 'http://example.net/api/export-import/blocks/test/redirect/',
},
{
'slug': 'test',
'text': 'test',
'type': 'workflows-categories',
'url': 'http://example.net/api/export-import/workflows-categories/test/redirect/',
},
{
'slug': 'test',
'text': 'test',
'type': 'workflows',
'url': 'http://example.net/api/export-import/workflows/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'mail-templates-categories',
'url': 'http://example.net/api/export-import/mail-templates-categories/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'mail-templates',
'url': 'http://example.net/api/export-import/mail-templates/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'comment-templates-categories',
'url': 'http://example.net/api/export-import/comment-templates-categories/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'comment-templates',
'url': 'http://example.net/api/export-import/comment-templates/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'data-sources-categories',
'url': 'http://example.net/api/export-import/data-sources-categories/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'data-sources',
'url': 'http://example.net/api/export-import/data-sources/test/redirect/',
},
{
'slug': 'test',
'text': 'Test',
'type': 'wscalls',
'url': 'http://example.net/api/export-import/wscalls/test/redirect/',
},
],
}
}
# import bundle again, recreate links
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundles[0])
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert resp.json['data']['completion_status'] == '34/34 (100%)'
assert Application.count() == 1
assert ApplicationElement.count() == 15
# no changes since last import
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), bundles[0])
assert resp.json == {
'data': {
'differences': [],
'unknown_elements': [],
'no_history_elements': [],
'legacy_elements': [],
}
}
# add local changes
snapshots = {}
for object_class in object_classes:
if object_class in [pub.custom_view_class, pub.role_class]:
# no snapshot for this objects
continue
for obj in object_class.select():
old_snapshots = pub.snapshot_class.select_object_history(obj)
obj.name = 'new name !'
obj.store(comment='local change !')
new_snapshots = pub.snapshot_class.select_object_history(obj)
snapshots['%s:%s' % (object_class.xml_root_node, obj.slug)] = (
old_snapshots[0].id,
new_snapshots[0].id,
)
assert len(new_snapshots) > len(old_snapshots)
# and check
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), bundles[0])
assert resp.json == {
'data': {
'differences': [
{
'slug': 'test',
'type': 'forms-categories',
'url': 'http://example.net/backoffice/forms/categories/1/history/compare?version1=%s&version2=%s'
% snapshots['category:test'],
},
{
'slug': 'test',
'type': 'forms',
'url': 'http://example.net/backoffice/forms/1/history/compare?version1=%s&version2=%s'
% snapshots['formdef:test'],
},
{
'slug': 'test',
'type': 'cards-categories',
'url': 'http://example.net/backoffice/cards/categories/1/history/compare?version1=%s&version2=%s'
% snapshots['carddef_category:test'],
},
{
'slug': 'test',
'type': 'cards',
'url': 'http://example.net/backoffice/cards/1/history/compare?version1=%s&version2=%s'
% snapshots['carddef:test'],
},
{
'slug': 'test',
'type': 'blocks-categories',
'url': 'http://example.net/backoffice/forms/blocks/categories/1/history/compare?version1=%s&version2=%s'
% snapshots['block_category:test'],
},
{
'slug': 'test',
'type': 'blocks',
'url': 'http://example.net/backoffice/forms/blocks/1/history/compare?version1=%s&version2=%s'
% snapshots['block:test'],
},
{
'slug': 'test',
'type': 'workflows-categories',
'url': 'http://example.net/backoffice/workflows/categories/1/history/compare?version1=%s&version2=%s'
% snapshots['workflow_category:test'],
},
{
'slug': 'test',
'type': 'workflows',
'url': 'http://example.net/backoffice/workflows/1/history/compare?version1=%s&version2=%s'
% snapshots['workflow:test'],
},
{
'slug': 'test',
'type': 'mail-templates-categories',
'url': 'http://example.net/backoffice/workflows/mail-templates/categories/1/history/compare?version1=%s&version2=%s'
% snapshots['mail_template_category:test'],
},
{
'slug': 'test',
'type': 'mail-templates',
'url': 'http://example.net/backoffice/workflows/mail-templates/1/history/compare?version1=%s&version2=%s'
% snapshots['mail-template:test'],
},
{
'slug': 'test',
'type': 'comment-templates-categories',
'url': 'http://example.net/backoffice/workflows/comment-templates/categories/1/history/compare?version1=%s&version2=%s'
% snapshots['comment_template_category:test'],
},
{
'slug': 'test',
'type': 'comment-templates',
'url': 'http://example.net/backoffice/workflows/comment-templates/1/history/compare?version1=%s&version2=%s'
% snapshots['comment-template:test'],
},
{
'slug': 'test',
'type': 'data-sources-categories',
'url': 'http://example.net/backoffice/forms/data-sources/categories/1/history/compare?version1=%s&version2=%s'
% snapshots['data_source_category:test'],
},
{
'slug': 'test',
'type': 'data-sources',
'url': 'http://example.net/backoffice/settings/data-sources/1/history/compare?version1=%s&version2=%s'
% snapshots['datasource:test'],
},
{
'slug': 'test',
'type': 'wscalls',
'url': 'http://example.net/backoffice/settings/wscalls/test/history/compare?version1=%s&version2=%s'
% snapshots['wscall:test'],
},
],
'unknown_elements': [],
'no_history_elements': [],
'legacy_elements': [],
}
}
# update bundle
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundles[1])
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert resp.json['data']['completion_status'] == '34/34 (100%)'
# and check
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), bundles[1])
assert resp.json == {
'data': {
'differences': [],
'unknown_elements': [],
'no_history_elements': [],
'legacy_elements': [],
}
}
# snapshots without application info (legacy)
for snapshot in pub.snapshot_class.select():
snapshot.application_slug = None
snapshot.application_version = None
snapshot.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-check/'), bundles[1])
assert resp.json == {
'data': {
'differences': [],
'no_history_elements': [
{'slug': 'test', 'type': 'forms-categories'},
{'slug': 'test', 'type': 'forms'},
{'slug': 'test', 'type': 'cards-categories'},
{'slug': 'test', 'type': 'cards'},
{'slug': 'test', 'type': 'blocks-categories'},
{'slug': 'test', 'type': 'blocks'},
{'slug': 'test', 'type': 'workflows-categories'},
{'slug': 'test', 'type': 'workflows'},
{'slug': 'test', 'type': 'mail-templates-categories'},
{'slug': 'test', 'type': 'mail-templates'},
{'slug': 'test', 'type': 'comment-templates-categories'},
{'slug': 'test', 'type': 'comment-templates'},
{'slug': 'test', 'type': 'data-sources-categories'},
{'slug': 'test', 'type': 'data-sources'},
{'slug': 'test', 'type': 'wscalls'},
],
'unknown_elements': [],
'legacy_elements': [],
},
}
def test_export_import_workflow_options(pub):
FormDef.wipe()
Workflow.wipe()
workflow = Workflow(name='variables')
workflow.variables_formdef = WorkflowVariablesFieldsFormDef(workflow=workflow)
workflow.variables_formdef.fields = [
StringField(id='1', label='Test', varname='foo'),
]
workflow.store()
formdef = FormDef()
formdef.name = 'foo'
formdef.workflow = workflow
formdef.workflow_options = {'foo': 'bar'}
formdef.store()
bundle = create_bundle(
[
{'type': 'forms', 'slug': 'foo', 'name': 'foo'},
{'type': 'workflows', 'slug': 'variables', 'name': 'variables'},
],
('forms/foo', formdef),
('workflows/variables', workflow),
)
FormDef.wipe()
Workflow.wipe()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
assert resp.json['data']['completion_status'] == '6/6 (100%)'
# check workflow options are set on first install
formdef = FormDef.get_by_slug('foo')
assert formdef.workflow_options == {'foo': 'bar'}
# check workflow options are not reset on further installs
formdef.workflow_options = {'foo': 'bar2'}
formdef.store()
resp = get_app(pub).put(sign_uri('/api/export-import/bundle-import/'), bundle)
afterjob_url = resp.json['url']
resp = get_app(pub).put(sign_uri(afterjob_url))
assert resp.json['data']['status'] == 'completed'
formdef = FormDef.get_by_slug('foo')
assert formdef.workflow_options == {'foo': 'bar2'}
def test_api_export_import_invalid_slug(pub):
pub.role_class.wipe()
role1 = pub.role_class(name='Test role 1')
role1.store()
role2 = pub.role_class(name='Test role 2')
role2.store()
carddef = CardDef()
carddef.name = 'Test'
carddef.workflow_roles = {'_receiver': role1.id}
carddef.backoffice_submission_roles = [role2.id]
carddef.store()
resp = get_app(pub).get(sign_uri('/api/export-import/cards/test/dependencies/'))
assert {x['text'] for x in resp.json['data']} == {'Test role 1', 'Test role 2'}
role2.slug = 'test role 2' # invalid slug
role2.store()
resp = get_app(pub).get(sign_uri('/api/export-import/cards/test/dependencies/'))
assert {x['text'] for x in resp.json['data']} == {'Test role 1'}