Produits Entr'ouvert » w.c.s.

import os

import pytest

from wcs import fields

from wcs.blocks import BlockDef

from wcs.carddef import CardDef

from wcs.data_sources import NamedDataSource

from wcs.formdef import FormDef

from wcs.qommon.http_request import HTTPRequest

from wcs.wf.backoffice_fields import SetBackofficeFieldsWorkflowStatusItem

from wcs.wf.create_formdata import CreateFormdataWorkflowStatusItem, Mapping

from wcs.wf.export_to_model import ExportToModel

from wcs.wf.external_workflow import ExternalWorkflowGlobalAction

from wcs.wf.geolocate import GeolocateWorkflowStatusItem

from wcs.wf.jump import JumpWorkflowStatusItem

from wcs.wf.notification import SendNotificationWorkflowStatusItem

from wcs.wf.profile import UpdateUserProfileStatusItem

from wcs.wf.redirect_to_url import RedirectToUrlWorkflowStatusItem

from wcs.wf.sendmail import SendmailWorkflowStatusItem

from wcs.wf.wscall import WebserviceCallStatusItem

from wcs.workflows import (

    ChoiceWorkflowStatusItem,

    DisplayMessageWorkflowStatusItem,

    SendSMSWorkflowStatusItem,

    Workflow,

)

from wcs.wscalls import NamedWsCall

from ..utilities import clean_temporary_pub, create_temporary_pub, get_app, login

from .test_all import create_superuser

@pytest.fixture

def pub(request):

    pub = create_temporary_pub(sql_mode=True)

    req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})

    pub.set_app_dir(req)

    pub.cfg['identification'] = {'methods': ['password']}

    pub.cfg['language'] = {'language': 'en'}

    pub.write_cfg()

    if os.path.exists(os.path.join(pub.app_dir, 'deprecations.json')):

        os.remove(os.path.join(pub.app_dir, 'deprecations.json'))

    BlockDef.wipe()

    CardDef.wipe()

    FormDef.wipe()

    NamedDataSource.wipe()

    NamedWsCall.wipe()

    Workflow.wipe()

    return pub

def teardown_module(module):

    clean_temporary_pub()

def test_no_deprecations(pub):

    create_superuser(pub)

    app = login(get_app(pub))

    # first time, it's a redirect to the scanning job

    resp = app.get('/backoffice/studio/deprecations/', status=302)

    resp = resp.follow()

    resp = resp.click('Go to deprecation report')

    # second time, the page stays on

    resp = app.get('/backoffice/studio/deprecations/', status=200)

    assert 'No deprecated items were found on this site.' in resp.text

def test_deprecations(pub):

    create_superuser(pub)

    formdef = FormDef()

    formdef.name = 'foobar'

    formdef.fields = [

        fields.PageField(id='1', label='page1', type='page', condition={'type': 'python', 'value': 'True'}),

        fields.StringField(id='2', label='python_prefill', prefill={'type': 'formula', 'value': '1 + 2'}),

        fields.StringField(

            id='3', label='ezt_prefill', prefill={'type': 'string', 'value': '[form_var_test]'}

        ),

        fields.StringField(id='4', label='jsonp_data', data_source={'type': 'jsonp', 'value': 'xxx'}),

        fields.StringField(id='5', label='ezt_in_datasource', data_source={'type': 'json', 'value': '[xxx]'}),

        fields.CommentField(id='6', label='[ezt] in label', type='comment'),

        fields.CommentField(id='7', label='{{script.usage}} in template', type='comment'),

        fields.PageField(

            id='10',

            label='page2',

            type='page',

            post_conditions=[

                {'condition': {'type': 'python', 'value': 'False'}, 'error_message': 'You shall not pass.'}

            ],

        ),

    ]

    formdef.store()

    workflow = Workflow(name='test')

    st0 = workflow.add_status('Status0', 'st0')

    display = DisplayMessageWorkflowStatusItem()

    display.message = 'message with [ezt] info'

    display.parent = st0

    st0.items.append(display)

    wscall = WebserviceCallStatusItem()

    wscall.id = '_wscall'

    wscall.varname = 'xxx'

    wscall.url = 'http://remote.example.net/xml'

    wscall.post_data = {'str': 'abcd', 'evalme': '=form_number'}

    wscall.parent = st0

    st0.items.append(wscall)

    sendsms = SendSMSWorkflowStatusItem()

    sendsms.id = '_sendsms'

    sendsms.to = 'xxx'

    sendsms.parent = st0

    st0.items.append(sendsms)

    accept = ChoiceWorkflowStatusItem()

    accept.id = '_choice'

    accept.label = '[plop]'

    accept.parent = st0

    st0.items.append(accept)

    item = SetBackofficeFieldsWorkflowStatusItem()

    item.id = '_item'

    item.fields = [{'field_id': 'bo1', 'value': '=form_var_foo'}]

    item.parent = st0

    st0.items.append(item)

    create_formdata = CreateFormdataWorkflowStatusItem()

    create_formdata.id = '_create_formdata'

    create_formdata.varname = 'resubmitted'

    create_formdata.mappings = [

        Mapping(field_id='0', expression='=form_var_toto_string'),

    ]

    create_formdata.parent = st0

    st0.items.append(create_formdata)

    item = UpdateUserProfileStatusItem()

    item.id = '_item'

    item.fields = [{'field_id': '__email', 'value': '=form_var_foo'}]

    item.parent = st0

    st0.items.append(item)

    sendmail = SendmailWorkflowStatusItem()

    sendmail.id = '_sendmail'

    sendmail.to = ['=plop']

    sendmail.parent = st0

    st0.items.append(sendmail)

    for klass in (

        ExportToModel,

        ExternalWorkflowGlobalAction,

        GeolocateWorkflowStatusItem,

        JumpWorkflowStatusItem,

        SendNotificationWorkflowStatusItem,

        RedirectToUrlWorkflowStatusItem,

    ):

        action = klass()

        action.parent = st0

        st0.items.append(action)

    workflow.store()

    data_source = NamedDataSource(name='ds_python')

    data_source.data_source = {'type': 'formula', 'value': repr([('1', 'un'), ('2', 'deux')])}

    data_source.store()

    data_source = NamedDataSource(name='ds_jsonp')

    data_source.data_source = {'type': 'jsonp', 'value': 'xxx'}

    data_source.store()

    NamedWsCall.wipe()

    wscall = NamedWsCall()

    wscall.name = 'Hello'

    wscall.request = {'url': 'http://example.net', 'qs_data': {'a': '=1+2'}}

    wscall.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/studio/deprecations/', status=302)

    resp = resp.follow()

    resp = resp.click('Go to deprecation report')

    assert [x.text for x in resp.pyquery('.section--ezt li a')] == [

        'foobar / Field "ezt_prefill"',

        'foobar / Field "ezt_in_datasource"',

        'foobar / Field "[ezt] in label"',

        'test / Alert',

        'test / Manual Jump',

    ]

    assert [x.text for x in resp.pyquery('.section--jsonp li a')] == [

        'foobar / Field "jsonp_data"',

        'Data source "ds_jsonp"',

    ]

    assert [x.text for x in resp.pyquery('.section--python-data-source li a')] == ['Data source "ds_python"']

    assert [x.text for x in resp.pyquery('.section--python-condition li a')] == [

        'foobar / Field "page1"',

        'foobar / Field "page2"',

    ]

    assert [x.text for x in resp.pyquery('.section--python-prefill li a')] == [

        'foobar / Field "python_prefill"'

    ]

    assert [x.text for x in resp.pyquery('.section--python-expression li a')] == [

        'test / Webservice',

        'test / Backoffice Data',

        'test / New Form Creation',

        'test / User Profile Update',

        'test / Email',

        'Webservice "Hello"',

    ]

    assert [x.text for x in resp.pyquery('.section--script li a')] == [

        'foobar / Field "{{script.usage}} in template"'

    ]

def test_deprecations_cronjob(pub):

    assert not os.path.exists(os.path.join(pub.app_dir, 'deprecations.json'))

    pub.update_deprecations_report()

    assert os.path.exists(os.path.join(pub.app_dir, 'deprecations.json'))

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

# Copyright (C) 2005-2022  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 datetime

import json

import os

import re

from quixote import get_publisher, get_request, get_response, redirect

from quixote.directory import Directory

from wcs.data_sources import NamedDataSource

from wcs.formdef import get_formdefs_of_all_kinds

from wcs.qommon import _, template

from wcs.qommon.afterjobs import AfterJob

from wcs.qommon.backoffice.menu import html_top

from wcs.workflows import Workflow

from wcs.wscalls import NamedWsCall

class DeprecationsDirectory(Directory):

    do_not_call_in_templates = True

    _q_exports = ['', 'scan']

    def _q_index(self):

        report_path = os.path.join(get_publisher().app_dir, 'deprecations.json')

        if not os.path.exists(report_path):

            # create report if necessary

            return self.scan()

        html_top('studio', _('Deprecations Report'))

        get_response().breadcrumb.append(('deprecations/', _('Deprecations Report')))

        context = {'has_sidebar': False, 'view': self}

        with open(report_path) as fd:

            context['report'] = json.load(fd)

        context['report']['report_lines'].sort(key=lambda x: x['category'])

        return template.QommonTemplateResponse(

            templates=['wcs/backoffice/deprecations.html'], context=context, is_django_native=True

        )

    def scan(self):

        job = get_response().add_after_job(

            DeprecationsScanAfterJob(

                label=_('Scanning for deprecations'),

                user_id=get_request().user.id,

                return_url='/backoffice/studio/deprecations/',

            )

        )

        job.store()

        return redirect(job.get_processing_url())

    @property

    def titles(self):

        return {

            'ezt': _('EZT text'),

            'jsonp': _('JSONP data source'),

            'python-condition': _('Python condition'),

            'python-expression': _('Python expression'),

            'python-prefill': _('Python prefill'),

            'python-data-source': _('Python data source'),

            'script': _('Filesystem Script'),

        }

    @property

    def short_docs(self):

        return {

            'ezt': _('Use Django templates.'),

            'jsonp': _('Use JSON sources with id and query parameters.'),

            'python-condition': _('Use Django condition.'),

            'python-expression': _('Use Django templates.'),

            'python-prefill': _('Use Django templates.'),

            'python-data-source': _('Use cards.'),

            'script': _('Use a dedicated template tags application.'),

        }

    @property

    def help_urls(self):

        return {

            'ezt': 'https://doc-publik.entrouvert.com/admin-fonctionnel/elements-deprecies/',

            'jsonp': 'https://doc-publik.entrouvert.com/admin-fonctionnel/elements-deprecies/',

            'python-condition': 'https://doc-publik.entrouvert.com/admin-fonctionnel/elements-deprecies/',

            'python-expression': 'https://doc-publik.entrouvert.com/admin-fonctionnel/elements-deprecies/',

            'python-prefill': 'https://doc-publik.entrouvert.com/admin-fonctionnel/elements-deprecies/',

            'python-data-source': 'https://doc-publik.entrouvert.com/admin-fonctionnel/elements-deprecies/',

            'script': 'https://doc-publik.entrouvert.com/admin-fonctionnel/elements-deprecies/',

        }

class DeprecationsScanAfterJob(AfterJob):

    def done_action_url(self):

        return self.kwargs['return_url']

    def done_action_label(self):

        return _('Go to deprecation report')

    def done_button_attributes(self):

        return {'data-redirect-auto': 'true'}

    def execute(self):

        self.report_lines = []

        formdefs = get_formdefs_of_all_kinds()

        workflows = Workflow.select(ignore_errors=True, ignore_migration=True)

        named_data_sources = NamedDataSource.select(ignore_errors=True, ignore_migration=True)

        named_ws_calls = NamedWsCall.select(ignore_errors=True, ignore_migration=True)

        # extra step to build report file

        self.total_count = len(formdefs) + len(workflows) + len(named_data_sources) + len(named_ws_calls) + 1

        self.store()

        for formdef in formdefs:

            for field in formdef.fields or []:

                location_label = _('%s / Field "%s"') % (formdef.name, field.ellipsized_label)

                url = formdef.get_field_admin_url(field)

                self.check_data_source(

                    getattr(field, 'data_source', None), location_label=location_label, url=url

                )

                prefill = getattr(field, 'prefill', None)

                if prefill:

                    if prefill.get('type') == 'formula':

                        self.add_report_line(

                            location_label=location_label,

                            url=url,

                            category='python-prefill',

                        )

                    else:

                        self.check_string(

                            prefill.get('value'), location_label=location_label, url=url, python_check=False

                        )

                if field.type == 'page':

                    for condition in field.get_conditions():

                        if condition and condition.get('type') == 'python':

                            self.add_report_line(

                                location_label=location_label,

                                url=url,

                                category='python-condition',

                            )

                            break

                if field.type in ('title', 'subtitle', 'comment'):

                    self.check_string(field.label, location_label=location_label, url=url, python_check=False)

            self.increment_count()

        for workflow in workflows:

            for action in workflow.get_all_items():

                location_label = '%s / %s' % (workflow.name, action.description)

                url = action.get_admin_url()

                for string in action.get_computed_strings():

                    self.check_string(string, location_label=location_label, url=url)

            self.increment_count()

        for named_data_source in named_data_sources:

            location_label = _('%s "%s"') % (_('Data source'), named_data_source.name)

            url = named_data_source.get_admin_url()

            self.check_data_source(

                getattr(named_data_source, 'data_source', None), location_label=location_label, url=url

            )

            self.increment_count()

        for named_ws_call in named_ws_calls:

            location_label = _('%s "%s"') % (_('Webservice'), named_ws_call.name)

            url = named_ws_call.get_admin_url()

            for string in named_ws_call.get_computed_strings():

                self.check_string(string, location_label=location_label, url=url)

            self.increment_count()

        self.build_report_file()

        self.increment_count()

    def check_data_source(self, data_source, location_label, url):

        if not data_source:

            return

        if data_source.get('type') == 'jsonp':

            self.add_report_line(

                location_label=location_label,

                url=url,

                category='jsonp',

            )

        if data_source.get('type') == 'formula':

            self.add_report_line(

                location_label=location_label,

                url=url,

                category='python-data-source',

            )

        if data_source.get('type') == 'json':

            self.check_string(data_source.get('value'), location_label, url, python_check=False)

    def check_string(self, string, location_label, url, python_check=True):

        if not isinstance(string, str):

            return

        if python_check and string.startswith('='):  # python expression

            self.add_report_line(location_label=location_label, url=url, category='python-expression')

        else:

            if template.Template(string).format == 'ezt':

                self.add_report_line(location_label=location_label, url=url, category='ezt')

            if re.findall(r'\Wscript\.\w', string):

                self.add_report_line(location_label=location_label, url=url, category='script')

    def add_report_line(self, **kwargs):

        if kwargs not in self.report_lines:

            self.report_lines.append(kwargs)

    def build_report_file(self):

        with open(os.path.join(get_publisher().app_dir, 'deprecations.json'), 'w') as fd:

            json.dump(

                {

                    'now': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),

                    'report_lines': self.report_lines,

                },

                fd,

                indent=2,

            )

from wcs.backoffice.deprecations import DeprecationsDirectory

    _q_exports = ['', 'deprecations', ('logged-errors', 'logged_errors_dir')]

    deprecations = DeprecationsDirectory()

    def get_field_admin_url(self, field):

        return self.get_admin_url() + '%s/' % field.id

        if get_request():

            for section in ('settings', 'forms', 'workflows'):

                if get_publisher().get_backoffice_root().is_accessible(section):

                    return '%s/%s/data-sources/%s/' % (base_url, section, self.id)

    def get_conditions(self):

        if self.condition:

            yield self.condition

        for post_condition in self.post_conditions or []:

            yield post_condition.get('condition')

    def get_field_admin_url(self, field):

        return self.get_admin_url() + 'fields/%s/' % field.id

        # once a day : update deprecations report

        cls.register_cronjob(

            CronJob(cls.update_deprecations_report, name='update_deprecations_report', hours=[2], minutes=[0])

        )

        # other jobs

    def update_deprecations_report(self):

        from .backoffice.deprecations import DeprecationsScanAfterJob

        DeprecationsScanAfterJob().execute()

.errors-and-deprecations {

	display: flex;

	flex-direction: column;

	.recent-errors {

		flex: 1;

		margin-bottom: 0;

	}

}

.no-deprecations {

	min-height: 80vh;

	background: transparent url(/static/css/dc2/fireworks.svg) bottom 20% left 50% no-repeat;

	background-size: contain;

}

.deprecations {

	.pk-information {

		margin-top: 0;

	}

	ul {

		column-width: 40em;

	}

}

            self.format = 'django'

            self.format = 'ezt'

            self.format = 'plain'

{% extends "wcs/backoffice.html" %}

{% load i18n %}

{% block content %}

<div id="appbar">

<h2>{% trans "Deprecations Report" %}</h2>

<span class="actions">

<a href="scan">{% trans "Rescan for deprecations" %}</a>

</span>

</div>

{% if not report.report_lines %}

<div class="no-deprecations">

        <p>{% trans "No deprecated items were found on this site." %}</p>

</div>

{% else %}

<div class="section deprecations">

{% regroup report.report_lines by category as category_report_lines %}

{% for category in category_report_lines %}

<h3>{{ view.titles|get:category.grouper }}</h3>

<div class="section--{{ category.grouper }}">

<div class="pk-information">

  <p>{{ view.short_docs|get:category.grouper }}

  <a href="{{ view.help_urls|get:category.grouper }}">{% trans "Read more" %}</a></p>

</div>

<ul>

 {% for report_line in category.list %}

 <li><a href="{{ report_line.url }}">{{ report_line.location_label }}</a></li>

 {% endfor %}

</ul>

</div>

{% endfor %}

</div>

<p>{% trans "Generated on:" %} {{ report.now|datetime }}</p>

{% endif %}

{% endblock %}

    <div class="errors-and-deprecations">

    <div class="paragraph recent-errors {% if not recent_errors %}no-recent-errors{% endif %}">

    <div>

      <p><a class="button button-paragraph" href="deprecations/">{% trans "Deprecations Report" %}</a></p>

    </div>

    </div>

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        for field in self.fields or []:

            yield field.get('value')

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.user_association_template

        for mapping in self.mappings or []:

            yield mapping.expression

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield from [self.role_id, self.variable]

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.filename

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.target_id

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield from [self.address_string, self.map_variable, self.photo_variable]

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.timeout

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield from [self.title, self.body, self.users_template]

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        for field in self.fields or []:

            yield field.get('value')

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.url

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        if not self.mail_template:

            yield self.subject

            yield self.body

        if self.to:

            yield from self.to

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.url

        if self.qs_data:

            yield from self.qs_data.values()

        if self.post_data:

            yield from self.post_data.values()

    def get_admin_url(self):

        return '%s/global-actions/%s/' % (self.parent.get_admin_url(), self.id)

    def get_admin_url(self):

        return self.parent.get_admin_url() + 'items/%s/' % self.id

    def get_computed_strings(self):

        # get list of computed strings, to check for deprecations

        return []

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.label

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.body

        if self.to:

            yield from self.to

    def get_computed_strings(self):

        yield from super().get_computed_strings()

        yield self.message

    def get_computed_strings(self):

        if self.request:

            yield self.request.get('url')

            yield self.request.get('request_signature_key')

            if self.request.get('qs_data'):

                yield from self.request.get('qs_data').values()

            if self.request.get('post_data'):

                yield from self.request.get('post_data').values()