Produits Entr'ouvert » w.c.s.

def test_data_sources_agenda_manual_qs_data_type_options(pub):

    create_superuser(pub)

    data_source = NamedDataSource(name='foobar')

    data_source.external = 'agenda_manual'

    data_source.store()

    if not pub.site_options.has_section('options'):

        pub.site_options.add_section('options')

    pub.site_options.set('options', 'disable-python-expressions', 'false')

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

        pub.site_options.write(fd)

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/data-sources/%s/edit' % data_source.id)

    assert resp.form['qs_data$element0value$type'].options == [

        ('text', False, 'Text'),

        ('template', False, 'Template'),

    ]

    pub.site_options.set('options', 'disable-python-expressions', 'true')

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

        pub.site_options.write(fd)

    resp = app.get('/backoffice/settings/data-sources/%s/edit' % data_source.id)

    assert resp.form['qs_data$element0value$type'].options == [

        ('text', False, 'Text'),

        ('template', False, 'Template'),

    ]

    assert 'This data source is readonly.' not in resp

    assert 'href="edit"' in resp

    assert 'href="delete"' in resp

    assert 'href="duplicate"' not in resp

    assert 'href="edit"' not in resp

    assert 'href="delete"' not in resp

    assert 'href="duplicate"' in resp

def test_data_sources_agenda_duplicate(pub):

    create_superuser(pub)

    NamedDataSource.wipe()

    data_source = NamedDataSource(name='foobar')

    data_source.data_source = {'type': 'json', 'value': 'http://some.url'}

    data_source.external = 'agenda'

    data_source.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/data-sources/%s/duplicate' % data_source.id)

    resp = resp.forms[0].submit('cancel')

    assert resp.location == 'http://example.net/backoffice/settings/data-sources/%s/' % data_source.id

    assert NamedDataSource.count() == 1

    resp = app.get('/backoffice/settings/data-sources/%s/duplicate' % data_source.id)

    resp = resp.click(href='duplicate')

    resp = resp.forms[0].submit()

    assert NamedDataSource.count() == 2

    new_data_source = NamedDataSource.select(order_by='id')[1]

    assert resp.location == 'http://example.net/backoffice/settings/data-sources/%s' % new_data_source.id

    assert new_data_source.data_source == {'type': 'json', 'value': 'http://some.url'}

    assert new_data_source.external == 'agenda_manual'

    assert new_data_source.qs_data is None

def test_data_sources_agenda_manual_view(pub):

    create_superuser(pub)

    NamedDataSource.wipe()

    data_source = NamedDataSource(name='foobar')

    data_source.data_source = {'type': 'json', 'value': 'http://some.url'}

    data_source.external = 'agenda_manual'

    data_source.qs_data = {'var1': 'value1', 'var2': 'value2'}

    data_source.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)

    assert 'This data source is readonly.' not in resp

    assert 'href="edit"' in resp

    assert 'href="delete"' in resp

    assert 'href="duplicate"' not in resp

    assert 'Query string data' in resp

    assert '<li>var1: value1</li>' in resp

    assert '<li>var2: value2</li>' in resp

    data_source.qs_data = None

    data_source.store()

    resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)

    assert 'Query string data' not in resp

def test_data_sources_user_view(pub):

    create_superuser(pub)

    NamedDataSource.wipe()

    data_source = NamedDataSource(name='foobar')

    data_source.data_source = {'type': 'wcs:users'}

    data_source.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/data-sources/%s/' % data_source.id)

    assert 'This data source is readonly.' not in resp

    assert 'href="edit"' in resp

    assert 'href="delete"' in resp

    assert 'href="duplicate"' not in resp

def test_data_sources_agenda_manual_edit(pub):

    create_superuser(pub)

    NamedDataSource.wipe()

    data_source = NamedDataSource(name='foobar')

    data_source.external = 'agenda_manual'

    data_source.store()

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/data-sources/%s/edit' % data_source.id)

    resp.forms[0]['qs_data$element0key'] = 'arg1'

    resp.forms[0]['qs_data$element0value$value_template'] = '{{ foobar }}'

    resp.forms[0]['qs_data$element0value$type'] = 'template'

    resp = resp.forms[0].submit('submit')

    data_source = NamedDataSource.get(data_source.id)

    assert data_source.qs_data == {'arg1': '{{ foobar }}'}

def test_json_datasource_bad_url(pub, error_email, http_requests, emails):

@pytest.mark.parametrize('notify', [True, False])

@pytest.mark.parametrize('record', [True, False])

def test_json_datasource_bad_qs_data(pub, error_email, emails, notify, record):

    datasource = {

        'type': 'json',

        'value': "https://whatever.com/json",

        'qs_data': {'foo': '{% for invalid %}', 'bar': '{{ valid }}'},

        'notify_on_errors': notify,

        'record_on_errors': record,

    }

    with mock.patch('wcs.qommon.misc.urlopen') as urlopen:

        urlopen.side_effect = lambda *args: io.StringIO(json.dumps({'data': [{'id': '1', 'text': 'foo'}]}))

        assert data_sources.get_items(datasource) == [('1', 'foo', '1', {'id': '1', 'text': 'foo'})]

    url = urlopen.call_args[0][0]

    assert url == 'https://whatever.com/json?bar='

    message = '[DATASOURCE] Failed to compute value "{% for invalid %}" for "foo" query parameter'

    if notify:

        assert emails.count() == 1

        assert message in emails.get_latest('subject')

    else:

        assert emails.count() == 0

    if pub.is_using_postgresql():

        if record:

            assert pub.loggederror_class.count() == 1

            logged_error = pub.loggederror_class.select(order_by='id')[0]

            assert logged_error.summary == message

        else:

            assert pub.loggederror_class.count() == 0

@pytest.mark.parametrize('qs_data', [{}, {'arg1': 'val1', 'arg2': 'val2'}])

def test_data_source_signed(no_request_pub, qs_data):

    data_source.qs_data = qs_data

    if qs_data:

        assert querystring['arg1'][0] == 'val1'

        assert querystring['arg2'][0] == 'val2'

    if qs_data:

        assert querystring['arg1'][0] == 'val1'

        assert querystring['arg2'][0] == 'val2'

    if qs_data:

        assert unsigned_url == 'https://no-secret.example.com/json?arg1=val1&arg2=val2'

    else:

        assert unsigned_url == 'https://no-secret.example.com/json'

    data_source.data_source = {'type': 'json', 'value': "https://no-secret.example.com/json?foo=bar"}

    data_source.store()

    with mock.patch('wcs.qommon.misc.urlopen') as urlopen:

        urlopen.side_effect = lambda *args: io.StringIO('{"data": [{"id": 0, "text": "zero"}]}')

        assert len(data_sources.get_items({'type': 'foobar'})) == 1

        unsigned_url = urlopen.call_args[0][0]

    if qs_data:

        assert unsigned_url == 'https://no-secret.example.com/json?foo=bar&arg1=val1&arg2=val2'

    else:

        assert unsigned_url == 'https://no-secret.example.com/json?foo=bar'

def test_data_source_with_qs_data(pub):

    data_source = NamedDataSource(name='test')

    data_source.qs_data = {'arg1': 'val1', 'arg2': 'val2'}

    data_source.store()

    data_source2 = assert_import_export_works(data_source, include_id=True)

    assert data_source2.qs_data == {'arg1': 'val1', 'arg2': 'val2'}

    # create some datasource, with same urls, but external != 'agenda'

    ds = NamedDataSource(name='Foo A')

    ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-A/datetimes/'}

    ds.external = 'agenda_manual'

    ds.store()

    ds = NamedDataSource(name='Foo B')

    ds.external = 'agenda_manual'

    ds.data_source = {'type': 'json', 'value': 'http://chrono.example.net/api/agenda/events-B/datetimes/'}

    ds.store()

    assert NamedDataSource.count() == 4  # no changes

    assert NamedDataSource.count() == 4  # no changes

    assert NamedDataSource.count() == 4 + 2

    datasource1 = NamedDataSource.get(4 + 1)

    datasource2 = NamedDataSource.get(4 + 2)

    assert NamedDataSource.count() == 4 + 2

    datasource1 = NamedDataSource.get(4 + 1)

    datasource2 = NamedDataSource.get(4 + 2)

    assert NamedDataSource.count() == 4 + 2

    datasource2 = NamedDataSource.get(4 + 2)

    datasource3 = NamedDataSource.get(4 + 3)

    assert NamedDataSource.count() == 4 + 3

    datasource2 = NamedDataSource.get(4 + 2)

    datasource3 = NamedDataSource.get(4 + 3)

    datasource4 = NamedDataSource.get(4 + 4)

    assert NamedDataSource.count() == 4 + 3

    datasource4 = NamedDataSource.get(4 + 4)

    with pytest.raises(IndexError):

        resp = resp.click('Duplicate')

    ComputedExpressionWidget,

    WidgetDict,

        if not self.datasource or (

            self.datasource.type != 'wcs:users' and self.datasource.external != 'agenda_manual'

        ):

        if not self.datasource or (

            self.datasource.type != 'wcs:users' and self.datasource.external != 'agenda_manual'

        ):

        if self.datasource.external == 'agenda_manual':

            form.add(

                WidgetDict,

                'qs_data',

                title=_('Query string data'),

                value=self.datasource.qs_data or {},

                element_value_type=ComputedExpressionWidget,

                element_value_kwargs={'allow_python': False},

                allow_empty_values=True,

                value_for_empty_value='',

            )

        'duplicate',

        if self.datasource.external == 'agenda':

            r += htmltext('<li><a href="duplicate" rel="popup">%s</a></li>') % _('Duplicate')

        elif not self.datasource.is_readonly():

    def duplicate(self):

        if not self.datasource.external == 'agenda' or hasattr(self.datasource, 'snapshot_object'):

            return redirect('.')

        form = Form(enctype='multipart/form-data')

        form.add_submit('duplicate', _('Submit'))

        form.add_submit('cancel', _('Cancel'))

        if form.get_widget('cancel').parse():

            return redirect('.')

        if not form.is_submitted() or form.has_errors():

            get_response().breadcrumb.append(('duplicate', _('Duplicate')))

            html_top('datasources', title=_('Duplicate Data Source'))

            r = TemplateIO(html=True)

            r += htmltext('<h2>%s %s</h2>') % (_('Duplicating Data Source:'), self.datasource.name)

            r += form.render()

            return r.getvalue()

        tree = self.datasource.export_to_xml(include_id=True)

        new_datasource = NamedDataSource.import_from_xml_tree(tree)

        new_datasource.name = _('Copy of %s' % new_datasource.name)

        new_datasource.slug = new_datasource.get_new_slug(new_datasource.slug)

        new_datasource.external = 'agenda_manual'

        new_datasource.store()

        return redirect('../%s' % new_datasource.id)

        if data_source.get('qs_data'):  # merge qs_data into url

            from wcs.workflows import WorkflowStatusItem

            parsed = urllib.parse.urlparse(url)

            qs = list(urllib.parse.parse_qsl(parsed.query))

            for key, value in data_source['qs_data'].items():

                try:

                    value = WorkflowStatusItem.compute(value, raises=True, record_errors=False)

                    value = str(value) if value is not None else ''

                except Exception as e:

                    get_publisher().record_error(

                        _(

                            'Failed to compute value "%(value)s" for "%(query)s" query parameter'

                            % {'value': value, 'query': key}

                        ),

                        context='[DATASOURCE]',

                        exception=e,

                        notify=data_source.get('notify_on_errors'),

                        record=data_source.get('record_on_errors'),

                    )

                else:

                    key = force_str(key)

                    value = force_str(value)

                    qs.append((key, value))

            qs = urllib.parse.urlencode(qs)

            url = urllib.parse.urlunparse(parsed[:4] + (qs,) + parsed[5:6])

    qs_data = None

        ('qs_data', 'qs_data'),

                    'qs_data': self.qs_data,

    def export_qs_data_to_xml(self, element, attribute_name, *args, **kwargs):

        if not self.qs_data:

            return

        for (key, value) in self.qs_data.items():

            item = ET.SubElement(element, 'item')

            if isinstance(key, str):

                ET.SubElement(item, 'name').text = force_text(key)

            else:

                raise AssertionError('unknown type for key (%r)' % key)

            if isinstance(value, str):

                ET.SubElement(item, 'value').text = force_text(value)

            else:

                raise AssertionError('unknown type for value (%r)' % key)

    def import_qs_data_from_xml(self, element, **kwargs):

        if element is None:

            return

        qs_data = {}

        for item in element.findall('item'):

            key = force_str(item.find('name').text)

            value = force_str(item.find('value').text or '')

            qs_data[key] = value

        return qs_data

        url = self.get_variadic_url()

        new_url = self.get_variadic_url()

        if new_url != url:

            token_context = {'session_id': get_session().id, 'url': new_url, 'slug': self.slug}

            token, created = get_publisher().token_class.get_or_create(

                type='autocomplete', context=token_context

            )

            if created:

                token.store()

            return '/api/geojson/%s' % token.id

            url = self.get_variadic_url()

        url = self.get_variadic_url()

  {% if datasource.qs_data %}

  <li>{% trans "Query string data:" %}

      <ul>

        {% for k, v in datasource.qs_data.items %}

          <li>{% blocktrans %}{{ k }}:{% endblocktrans %} {{ v }}</li>

        {% endfor %}

      </ul>

    </li>

  {% endif %}