Produits Entr'ouvert » w.c.s.

import json

from wcs.carddef import CardDef

from wcs.qommon.afterjobs import AfterJob

def test_backoffice_no_json(pub):

    create_superuser(pub)

    FormDef.wipe()

    formdef = FormDef()

    formdef.name = 'form title'

    formdef.fields = []

    formdef.workflow_roles = {'_receiver': 1}

    formdef.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/management/form-title/')

    resp = resp.click('Export a Spreadsheet')

    assert [x[0] for x in resp.form['format'].options] == ['ods', 'csv']

def test_backoffice_cards_json(pub):

    create_superuser(pub)

    CardDef.wipe()

    carddef = CardDef()

    carddef.name = 'card title'

    carddef.fields = [

        fields.StringField(id='1', label='1st field', type='string'),

    ]

    carddef.workflow_roles = {'_receiver': 1}

    carddef.store()

    carddef.data_class().wipe()

    for i in range(10):

        carddata = carddef.data_class()()

        carddata.data = {'1': 'foo %s' % i}

        carddata.just_created()

        carddata.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/data/card-title/')

    resp = resp.click('Export Data')

    assert [x[0] for x in resp.form['format'].options] == ['ods', 'csv', 'json']

    resp.form['format'] = 'json'

    resp = resp.form.submit('submit')

    parsed_url = urllib.parse.urlparse(resp.location)

    assert parsed_url.path == '/backoffice/processing'

    job_id = urllib.parse.parse_qs(urllib.parse.urlparse(resp.location).query)['job'][0]

    job = AfterJob.get(job_id)

    assert job.completion_time

    json_export = json.loads(job.file_content)

    assert len(json_export['data']) == 10

    export_data_label = _('Export Data')

    has_json_export_support = True

    has_json_export_support = False

    export_data_label = _('Export a Spreadsheet')

            ) % (qs, self.export_data_label)

        formats = [

            ('ods', _('OpenDocument (.ods)'), 'ods'),

            ('csv', _('Text (.csv)'), 'csv'),

        ]

        if self.has_json_export_support:

            formats.append(('json', _('JSON'), 'json'))

            options=formats,

            attrs={'data-dynamic-display-parent': 'true'},

            attrs={

                'data-dynamic-display-child-of': 'format',

                'data-dynamic-display-value-in': 'csv|ods',

            },

            r += htmltext('<h2>%s</h2>') % _('Export Options')

        elif file_format == 'json':

            return self.export_json_file()

    def export_json_file(self):

        fields = self.get_fields_from_query()

        selected_filter = self.get_filter_from_query()

        selected_filter_operator = self.get_filter_operator_from_query()

        user = get_request().user

        query = get_request().form.get('q')

        criterias = self.get_criterias_from_query()

        order_by = misc.get_order_by_or_400(get_request().form.get('order_by', None))

        job = JsonFileExportAfterJob(

            self.formdef,

            fields=fields,

            selected_filter=selected_filter,

            selected_filter_operator=selected_filter_operator,

            user_id=user.id,

            query=query,

            criterias=criterias,

            order_by=order_by,

        )

        job = get_response().add_after_job(job)

        job.store()

        return redirect(job.get_processing_url())

            output = JsonFileExportAfterJob(self.formdef).create_json_export(

                items,

                user=user,

                anonymise=anonymise,

                digest_key=digest_key,

                include_evolution=True,

                include_files=False,

                include_roles=True,

            )

class JsonFileExportAfterJob(CsvExportAfterJob):

    label = _('Exporting to JSON file')

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

        super().__init__(formdef=formdef, **kwargs)

        self.file_name = '%s.json' % formdef.url_name

    def create_json_export(

        self, items, user, anonymise, digest_key, include_evolution, include_files, include_roles

    ):

        formdef = self.kwargs['formdef_class'].get(self.kwargs['formdef_id'])

        prefetched_users = None

        prefetched_roles = None

        prefetched_users = {

            str(x.id): x

            for x in get_publisher().user_class.get_ids(

                [x.user_id for x in items if x.user_id], ignore_errors=True

            )

            if x is not None

        }

        if include_roles:

            role_ids = set((formdef.workflow_roles or {}).values())

            for formdata in items:

                if formdata.workflow_roles:

                    for value in formdata.workflow_roles.values():

                        if not isinstance(value, list):

                            value = [value]

                        role_ids |= set(value)

            prefetched_roles = {

                str(x.id): x

                for x in get_publisher().role_class.get_ids(role_ids, ignore_errors=True)

                if x is not None

            }

        output = []

        for formdata in items:

            data = formdata.get_json_export_dict(

                anonymise=anonymise,

                user=user,

                digest_key=digest_key,

                prefetched_users=prefetched_users,

                prefetched_roles=prefetched_roles,

                include_evolution=include_evolution,

                include_files=include_files,

                include_roles=include_roles,

            )

            data.pop('digests')

            if digest_key:

                data['digest'] = (formdata.digests or {}).get(digest_key)

            output.append(data)

            self.increment_count()

        return output

    def create_export(self, formdef, fields, items, total_count):

        self.file_content = json.dumps(

            {

                'data': self.create_json_export(

                    items,

                    user=None,

                    anonymise=False,

                    digest_key=None,

                    include_evolution=False,

                    include_files=True,

                    include_roles=False,

                )

            },

            indent=2,

            cls=misc.JSONEncoder,

        )

        self.content_type = 'application/json'

        self.store()

        include_evolution=True,

        include_roles=True,

        if include_roles:

            # add a roles dictionary, with workflow functions and two special

            # entries for concerned/actions roles.

            data['roles'] = {}

            workflow_roles = {}

            if self.formdef.workflow_roles:

                workflow_roles.update(self.formdef.workflow_roles)

            if self.workflow_roles:

                workflow_roles.update(self.workflow_roles)

            for workflow_role in workflow_roles:

                value = workflow_roles.get(workflow_role)

                if not isinstance(value, list):

                    value = [value]

                data['roles'][workflow_role] = value

            data['roles']['concerned'] = self.get_concerned_roles()

            data['roles']['actions'] = self.get_actions_roles()

            for role_key in data['roles']:

                # exclude special _submitter value

                role_list = [x for x in data['roles'][role_key] if x != '_submitter']

                # get role objects

                if prefetched_roles is not None:

                    role_list = [prefetched_roles.get(str(x)) for x in role_list]

                else:

                    role_list = [get_publisher().role_class.get(x, ignore_errors=True) for x in role_list]

                # export as json dicts

                role_list = [x.get_json_export_dict() for x in role_list if x is not None]

                data['roles'][role_key] = role_list

        if self.evolution and include_evolution: