Produits Entr'ouvert » w.c.s.

import io

import json

import os

import tarfile

import xml.etree.ElementTree as ET

import pytest

from wcs.blocks import BlockDef

from wcs.categories import Category

from wcs.data_sources import NamedDataSource

from wcs.fields import BlockField, StringField

from wcs.formdef import FormDef

from wcs.wf.form import FormWorkflowStatusItem, WorkflowFormFieldsFormDef

from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef, WorkflowVariablesFieldsFormDef

from ..utilities import clean_temporary_pub, create_temporary_pub, get_app

from .utils import sign_uri

@pytest.fixture

def pub():

    pub = create_temporary_pub(sql_mode=True)

    with open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w') as fd:

        fd.write(

            '''\

[api-secrets]

coucou = 1234

'''

        )

    Category.wipe()

    FormDef.wipe()

    BlockDef.wipe()

    Workflow.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'

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)

def test_export_import_dependencies(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']['dependencies']))

    assert not resp.json['data']

    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', type='block:test'),

    ]

    workflow.variables_formdef = WorkflowVariablesFieldsFormDef(workflow=workflow)

    workflow.variables_formdef.fields = [StringField(label='Test', id='1')]

    status = workflow.add_status('New')

    form = FormWorkflowStatusItem()

    form.parent = status

    status.items.append(form)

    data_source = NamedDataSource(name='foobar')

    data_source.store()

    display_form = FormWorkflowStatusItem()

    display_form.id = '_x'

    display_form.formdef = WorkflowFormFieldsFormDef(item=display_form)

    display_form.formdef.fields.append(StringField(label='Test', data_source={'type': 'foobar'}))

    status.items.append(display_form)

    display_form.parent = status

    workflow.store()

    formdef.fields = [

        BlockField(id='1', label='test', type='block:test'),

    ]

    formdef.workflow = workflow

    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 resp.json['data']

    assert {(x['id'], x['type']) for x in resp.json['data']} == {('test', 'workflows'), ('test', 'blocks')}

    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 resp.json['data']

    assert {(x['id'], x['type']) for x in resp.json['data']} == {

        ('foobar', 'data-sources'),

        ('test', 'blocks'),

    }

def create_bundle(*args):

    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'},

                {'type': 'blocks', 'slug': 'test', 'name': 'test'},

                {'type': 'workflows', 'slug': 'test', 'name': 'test'},

                {'type': 'forms-categories', 'slug': 'test', 'name': 'test'},

            ],

        }

        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)

        for (path, obj) in args:

            tarinfo = tarfile.TarInfo(path)

            xml_export = ET.tostring(obj.export_to_xml(include_id=True))

            tarinfo.size = len(xml_export)

            tar.addfile(tarinfo, fileobj=io.BytesIO(xml_export))

    return tar_io.getvalue()

def test_export_import_bundle_import(pub):

    workflow = Workflow(name='test')

    workflow.store()

    block = BlockDef(name='test')

    block.store()

    formdef = FormDef()

    formdef.name = 'Test'

    formdef.fields = [

        BlockField(id='1', label='test', type='block:test'),

    ]

    formdef.workflow = workflow

    formdef.disabled = False

    formdef.store()

    category = Category(name='Test')

    category.store()

    bundle = create_bundle(

        ('forms/test', formdef),

        ('blocks/test', block),

        ('workflows/test', workflow),

        ('forms-categories/test', category),

    )

    Category.wipe()

    FormDef.wipe()

    BlockDef.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 Category.count() == 1

    assert FormDef.count() == 1

    assert BlockDef.count() == 1

    assert Workflow.count() == 1

    assert FormDef.select()[0].fields[0].type == 'block:test'

    # run new import to check it doesn't duplicate objects

    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 Category.count() == 1

    assert FormDef.count() == 1

    assert BlockDef.count() == 1

    assert Workflow.count() == 1

    # change immutable attribute and check it's not reset

    formdef = FormDef.select()[0]

    formdef.disabled = True

    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.select()[0]

    assert formdef.disabled is True

# w.c.s. - web application for online forms

# Copyright (C) 2005-2021  Entr'ouvert

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, see <http://www.gnu.org/licenses/>.

import io

import json

import tarfile

import xml.etree.ElementTree as ET

from django.http import Http404, HttpResponse, HttpResponseForbidden, JsonResponse

from django.urls import reverse

from wcs.api_utils import is_url_signed

from wcs.blocks import BlockDef

from wcs.carddef import CardDef

from wcs.categories import BlockCategory, CardDefCategory, Category, WorkflowCategory

from wcs.data_sources import NamedDataSource, StubNamedDataSource

from wcs.formdef import FormDef

from wcs.mail_templates import MailTemplate

from wcs.workflows import Workflow

from wcs.wscalls import NamedWsCall

from .qommon import _

from .qommon.afterjobs import AfterJob

from .qommon.misc import indent_xml, xml_node_text

klasses = {

    'blocks': BlockDef,

    'blocks-categories': BlockCategory,

    'cards': CardDef,

    'cards-categories': CardDefCategory,

    'data-sources': NamedDataSource,

    'forms-categories': Category,

    'forms': FormDef,

    'mail-templates': MailTemplate,

    'workflows-categories': WorkflowCategory,

    'workflows': Workflow,

    'wscalls': NamedWsCall,

}

klass_to_slug = {y: x for x, y in klasses.items()}

def signature_required(func):

    def f(*args, **kwargs):

        if not is_url_signed():

            return HttpResponseForbidden()

        return func(*args, **kwargs)

    return f

@signature_required

def index(request):

    response = [

        {'id': 'forms', 'text': _('Forms'), 'singular': _('Form')},

        {'id': 'cards', 'text': _('Cards'), 'singular': _('Card')},

        {'id': 'workflows', 'text': _('Workflows'), 'singular': _('Workflow')},

        {'id': 'blocks', 'text': _('Blocks'), 'singular': _('Block of fields'), 'minor': True},

        {'id': 'data-sources', 'text': _('Data Sources'), 'singular': _('Data Source'), 'minor': True},

        {'id': 'mail-templates', 'text': _('Mail Templates'), 'singular': _('Mail Template'), 'minor': True},

        {'id': 'wscalls', 'text': _('Webservice Calls'), 'singular': _('Webservice Call'), 'minor': True},

        {

            'id': 'blocks-categories',

            'text': _('Categories (blocks)'),

            'singular': _('Category (block)'),

            'minor': True,

        },

        {

            'id': 'cards-categories',

            'text': _('Categories (cards)'),

            'singular': _('Category (cards)'),

            'minor': True,

        },

        {

            'id': 'forms-categories',

            'text': _('Categories (forms)'),

            'singular': _('Category (forms)'),

            'minor': True,

        },

        {

            'id': 'workflow-categories',

            'text': _('Categories (workflows)'),

            'singular': _('Category (workflows)'),

            'minor': True,

        },

    ]

    for obj in response:

        obj['urls'] = {

            'list': request.build_absolute_uri(

                reverse('api-export-import-objects-list', kwargs={'objects': obj['id']})

            ),

        }

    return JsonResponse({'data': response})

@signature_required

def export_object_ref(request, obj):

    slug = obj.slug

    objects = klass_to_slug[obj.__class__]

    return {

        'id': slug,

        'text': obj.name,

        'type': objects,

        'urls': {

            'export': request.build_absolute_uri(

                reverse('api-export-import-object-export', kwargs={'objects': objects, 'slug': slug})

            ),

            'dependencies': request.build_absolute_uri(

                reverse('api-export-import-object-dependencies', kwargs={'objects': objects, 'slug': slug})

            ),

        },

    }

@signature_required

def objects_list(request, objects):

    klass = klasses.get(objects)

    if not klass:

        raise Http404()

    return JsonResponse({'data': [export_object_ref(request, x) for x in klass.select()]})

def get_object(objects, slug):

    klass = klasses.get(objects)

    if not klass:

        raise Http404()

    return klass.get_by_slug(slug)

@signature_required

def object_export(request, objects, slug):

    obj = get_object(objects, slug)

    etree = obj.export_to_xml(include_id=True)

    indent_xml(etree)

    return HttpResponse(ET.tostring(etree), content_type='text/xml')

@signature_required

def object_dependencies(request, objects, slug):

    obj = get_object(objects, slug)

    dependencies = []

    if hasattr(obj, 'get_dependencies'):

        for dependency in obj.get_dependencies():

            if dependency is None or isinstance(dependency, StubNamedDataSource):

                continue

            dependencies.append(export_object_ref(request, dependency))

    return JsonResponse({'data': dependencies})

class BundleImportJob(AfterJob):

    def __init__(self, tar_content, **kwargs):

        super().__init__(**kwargs)

        self.tar_content = tar_content

    def execute(self):

        tar_io = io.BytesIO(self.tar_content)

        with tarfile.open(fileobj=tar_io) as self.tar:

            manifest = json.loads(self.tar.extractfile('manifest.json').read().decode())

            self.app_name = manifest.get('application')

            # first pass on formdef/carddef/blockdef/workflows to create them empty

            # (name and slug); so they can be found for sure in import pass

            for type in ('forms', 'cards', 'blocks', 'workflows'):

                self.pre_install([x for x in manifest.get('elements') if x.get('type') == type])

            # real installation pass

            for type in (

                'blocks-categories',

                'cards-categories',

                'forms-categories',

                'workflows-categories',

                'data-sources',

                'wscalls',

                'mail-templates',

                'forms',

                'cards',

                'blocks',

                'workflows',

            ):

                self.install([x for x in manifest.get('elements') if x.get('type') == type])

    def pre_install(self, elements):

        for element in elements:

            element_klass = klasses[element['type']]

            element_content = self.tar.extractfile('%s/%s' % (element['type'], element['slug'])).read()

            tree = ET.fromstring(element_content)

            if hasattr(element_klass, 'url_name'):

                slug = xml_node_text(tree.find('url_name'))

            else:

                slug = xml_node_text(tree.find('slug'))

            try:

                existing_object = element_klass.get_by_slug(slug)

            except KeyError:

                pass

            else:

                if existing_object:

                    continue

            new_object = element_klass()

            new_object.slug = slug

            new_object.name = '[pre-import] %s' % xml_node_text(tree.find('name'))

            new_object.store(comment=_('Application (%s)') % self.app_name)

    def install(self, elements):

        for element in elements:

            element_klass = klasses[element['type']]

            element_content = self.tar.extractfile('%s/%s' % (element['type'], element['slug'])).read()

            new_object = element_klass.import_from_xml_tree(

                ET.fromstring(element_content), include_id=False, check_datasources=False

            )

            try:

                existing_object = element_klass.get_by_slug(new_object.slug)

                if existing_object is None:

                    raise KeyError()

            except KeyError:

                new_object.store(comment=_('Application (%s)') % self.app_name)

                continue

            # replace

            new_object.id = existing_object.id

            if element['type'] in ('forms', 'cards'):

                # keep some settings

                for attr in (

                    'workflow_options',

                    'workflow_roles',

                    'roles',

                    'required_authentication_contexts',

                    'backoffice_submission_roles',

                    'publication_date',

                    'expiration_date',

                    'disabled',

                ):

                    setattr(new_object, attr, getattr(existing_object, attr))

            new_object.store(comment=_('Application (%s) update') % self.app_name)

@signature_required

def bundle_import(request):

    job = BundleImportJob(tar_content=request.body)

    job.store()

    job.run(spool=True)

    return JsonResponse({'err': 0, 'url': job.get_api_status_url()})

    def get_dependencies(self):

        yield self.category

        for field in self.fields or []:

            yield from field.get_dependencies()

    def build_absolute_uri(self, *args):

        return self.django_request.build_absolute_uri(*args)

    def get_dependencies(self):

        if getattr(self, 'data_source', None):

            data_source_type = self.data_source.get('type')

            if data_source_type and data_source_type.startswith('carddef:'):

                from .carddef import CardDef

                carddef_slug = data_source_type.split(':')[1]

                try:

                    yield CardDef.get_by_urlname(carddef_slug)

                except KeyError:

                    pass

            else:

                from .data_sources import NamedDataSource

                yield NamedDataSource.get_by_slug(data_source_type, ignore_errors=True)

    def get_dependencies(self):

        yield self.block

    def get_dependencies(self):

        yield self.category

        if self.workflow_id:

            yield self.workflow

        for field in self.fields or []:

            yield from field.get_dependencies()

from quixote import get_publisher, get_request, get_response

    def get_api_status_url(self):

        return get_request().build_absolute_uri('/api/jobs/%s/' % self.id)

from . import api, api_export_import, compat, views

    url(r'^api/export-import/$', api_export_import.index, name='api-export-import'),

    url(r'^api/export-import/bundle-import/$', api_export_import.bundle_import),

    url(

        r'^api/export-import/(?P<objects>[\w-]+)/$',

        api_export_import.objects_list,

        name='api-export-import-objects-list',

    ),

    url(

        r'^api/export-import/(?P<objects>[\w-]+)/(?P<slug>[\w_-]+)/$',

        api_export_import.object_export,

        name='api-export-import-object-export',

    ),

    url(

        r'^api/export-import/(?P<objects>[\w-]+)/(?P<slug>[\w_-]+)/dependencies/$',

        api_export_import.object_dependencies,

        name='api-export-import-object-dependencies',

    ),

    def get_dependencies(self):

        if self.formdef and self.formdef.fields:

            for field in self.formdef.fields:

                yield from field.get_dependencies()

    def get_dependencies(self):

        yield self.category

        if self.variables_formdef and self.variables_formdef.fields:

            for field in self.variables_formdef.fields:

                yield from field.get_dependencies()

        if self.backoffice_fields_formdef and self.backoffice_fields_formdef.fields:

            for field in self.backoffice_fields_formdef.fields:

                yield from field.get_dependencies()

        if self.possible_status:

            for status in self.possible_status:

                yield from status.get_dependencies()

    def get_dependencies(self):

        for action in self.items or []:

            yield from action.get_dependencies()

    def get_dependencies(self):

        return []