Produits Entr'ouvert » w.c.s.

# -*- coding: utf-8 -*-

import json

import mock

import os

except ImportError:

    lasso = None

import pytest

from webtest import Upload

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

    xml_export = resp.text

    data_source2 = NamedDataSource.import_from_xml(ds)

    assert data_source2.name == 'foobar'

# -*- coding: utf-8 -*-

import datetime

import mock

import os

import re

import pytest

from webtest import Upload

from wcs.qommon.http_request import HTTPRequest

from wcs.qommon.errors import ConnectionError

from wcs.categories import Category

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

    xml_export = resp.text

    formdef2 = FormDef.import_from_xml(fd)

    assert formdef2.name == 'form title'

        resp.form['url'] = 'http://remote.invalid/test.wcs'

        resp = resp.form.submit()

        assert 'Error loading form' in resp

        resp.form['url'] = 'http://remote.invalid/test.wcs'

        resp = resp.form.submit()

    assert FormDef.count() == 1

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

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

    assert resp.content_type == 'application/x-wcs-archive'

    assert 'formdef' in [x.name for x in tf.getmembers()]

    assert len(tf.getmembers()) == 8  # 7 formdata + 1 formdef

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

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

    assert resp.content_type == 'application/x-wcs-archive'

    assert 'formdef' in [x.name for x in tf.getmembers()]

    assert len(tf.getmembers()) == 1  # 0 formdata + 1 formdef

# -*- coding: utf-8 -*-

import logging

import os

import zipfile

from webtest import Upload

import mock

from quixote.http_request import Upload as QuixoteUpload

    resp = app.get('/backoffice/settings/export')

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

    assert resp.location.startswith('http://example.net/backoffice/processing?job=')

    resp = resp.follow()

    assert 'completed' in resp.text

    resp = resp.click('Download Export')

    zipf = zipfile.ZipFile(zip_content, 'a')

    filelist = zipf.namelist()

    assert len(filelist) == 0

    export_to.label = 'test'

    upload = QuixoteUpload('/foo/bar', content_type='application/vnd.oasis.opendocument.text')

    file_content = b'''PK\x03\x04\x14\x00\x00\x08\x00\x00\'l\x8eG^\xc62\x0c\'\x00'''

    upload.fp.write(file_content)

    upload.fp.seek(0)

    export_to.model_file = UploadedFile('models', 'export_to_model-1.upload', upload)

    resp = app.get('/backoffice/settings/export')

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

    assert resp.location.startswith('http://example.net/backoffice/processing?job=')

    resp = resp.follow()

    resp = resp.click('Download Export')

    zipf = zipfile.ZipFile(zip_content, 'a')

    filelist = zipf.namelist()

    assert 'formdefs/1' not in filelist

    resp.form['wscalls'] = False

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

    assert resp.location.startswith('http://example.net/backoffice/processing?job=')

    resp = resp.follow()

    resp = resp.click('Download Export')

    zipf = zipfile.ZipFile(zip_content, 'a')

    filelist = zipf.namelist()

    assert 'formdefs_xml/%s' % formdef.id in filelist

    resp = app.get('/backoffice/settings/export')

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

    assert resp.location.startswith('http://example.net/backoffice/processing?job=')

    resp = resp.follow()

    resp = resp.click('Download Export')

    zipf = zipfile.ZipFile(zip_content, 'a')

    filelist = zipf.namelist()

    assert len([x for x in filelist if 'roles/' in x]) == 0

    resp = app.get('/backoffice/settings/themes')

    resp = resp.click('download', index=0)

    assert resp.headers['content-type'] == 'application/zip'

    zipf = zipfile.ZipFile(zip_content, 'a')

    filelist = zipf.namelist()

    assert 'alto/icon.png' in filelist

# -*- coding: utf-8 -*-

import mock

import os

import re

        resp.form['url'] = 'http://remote.invalid/test.wcs'

        resp = resp.form.submit()

        assert 'Error loading form' in resp

        resp.form['url'] = 'http://remote.invalid/test.wcs'

        resp = resp.form.submit()

    assert Workflow.count() == 2

# -*- coding: utf-8 -*-

import xml.etree.ElementTree as ET

import pytest

from webtest import Upload

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

    xml_export = resp.text

    wscall2 = NamedWsCall.import_from_xml(ds)

    assert wscall2.name == 'xxx'

# -*- coding: utf-8 -*-

import datetime

import json

import os

import re

import mock

import pytest

from quixote import get_publisher

from quixote.http_request import Upload as QuixoteUpload

from wcs.blocks import BlockDef

    formdef.store()

    resp = app.get('/backoffice/management/form-title/%s/' % number31.id)

    resp = resp.click('Download all files as .zip')

    zipf = zipfile.ZipFile(zip_content, 'a')

    filelist = zipf.namelist()

    assert set(filelist) == {'1_bar', '2_bar'}

        def side_effect(url, *args):

            assert '?name_id=admin' in url

        urlopen.side_effect = side_effect

        def side_effect(url, *args):

            assert '?test=foobar' in url

        urlopen.side_effect = side_effect

        def side_effect(url, *args):

            assert '?xxx=FOO BAR 30' in url

        urlopen.side_effect = side_effect

        def side_effect(url, *args):

            if 'toto' not in url:

            else:

        urlopen.side_effect = side_effect

    export_to.convert_to_pdf = False

    export_to.label = 'create doc'

    upload = QuixoteUpload('/foo/test.rtf', content_type='application/rtf')

    upload.fp.write(b'HELLO WORLD')

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

# -*- coding: utf-8 -*-

import datetime

import os

import time

import xml.etree.ElementTree as ET

import pytest

from wcs.qommon import ods

from wcs.blocks import BlockDef

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

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

    assert resp.location.startswith('http://example.net/backoffice/processing?job=')

    resp = resp.follow()

    assert 'completed' in resp.text

    resp = resp.click('Download Export')

    assert 'filename=form-title.ods' in resp.headers['content-disposition']

    assert resp.body[:2] == b'PK'  # ods has a zip container

    ods_sheet = ET.parse(zipf.open('content.xml'))

    # check the ods contains a link to the document

    elem = ods_sheet.findall('.//{%s}a' % ods.NS['text'])[0]

import time

import pytest

from utilities import clean_temporary_pub, create_temporary_pub, get_app, login

from wcs import fields

    def post_formdata():

        signed_url = sign_url(

            'http://example.net/api/formdefs/form-title/submit'

            '1234',

        )

        url = signed_url[len('http://example.net') :]

import os

import pytest

def site_options(request, pub, section, variable, value):

    path = os.path.join(pub.app_dir, 'site-options.cfg')

    if os.path.exists(path):

        config.read([path])

        config.write(site_option)

    def fin():

        if os.path.exists(path):

            config.read([path])

            config.remove_option(section, variable)

# -*- coding: utf-8 -*-

import json

import pytest

import hashlib

import os

import re

import time

import zipfile

from webtest import Upload, Hidden, Radio

import mock

import xml.etree.ElementTree as ET

except ImportError:

    Image = None

from django.utils.encoding import force_bytes, force_text

from wcs.qommon import force_str

                {'id': '2', 'text': 'world', 'extra': 'bar'},

            ]

        }

        resp = app.get('/test/')

        assert 'data-select2-url=' in resp.text

        # simulate select2 mode, with qommon.forms.js adding an extra hidden widget

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

        data = {'data': [{'id': '1', 'text': 'un'}, {'id': '2', 'text': 'deux'}]}

        assert submit_item_data_source_field(ds) == {'0': '1', '0_display': 'un'}

    # numeric identifiers

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

        data = {'data': [{'id': 1, 'text': 'un'}, {'id': 2, 'text': 'deux'}]}

        assert submit_item_data_source_field(ds) == {'0': '1', '0_display': 'un'}

    # check it's not set if it's not whitelisted

    resp = get_app(pub).get('/?session_var_foo=hello')

    resp = resp.follow()

    resp = resp.click('test')

    assert resp.forms[0]['f0'].value == ''

    )

    resp = get_app(pub).get('/?session_var_foo=hello')

    resp = resp.follow()

    resp = resp.click('test')

    assert resp.forms[0]['f0'].value == 'hello'

    # check it survives a login

    resp = get_app(pub).get('/?session_var_foo=hello2')

    resp = resp.follow()

    resp = resp.click('Login')

    resp = resp.follow()

    # check repeated options are ignored

    resp = get_app(pub).get('/?session_var_foo=hello&session_var_foo=hello2')

    resp = resp.follow()

    resp = resp.click('test')

    assert resp.forms[0]['f0'].value == ''

    # check extra query string parameters are not lost

    resp = get_app(pub).get('/?session_var_foo=hello&foo=bar')

    os.unlink(os.path.join(pub.app_dir, 'site-options.cfg'))

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

        data = {'data': [{'id': '1', 'text': 'hello'}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        assert not resp.form['f0$element1'].checked

        assert resp.form['f0$element2'].checked

    fargo_resp = app.get('/fargo/pick')  # display file picker

    assert fargo_resp.location == 'http://fargo.example.net/pick/?pick=http%3A//example.net/fargo/pick'

    with mock.patch('wcs.portfolio.urlopen') as urlopen:

        fargo_resp = app.get('/fargo/pick?url=http://www.example.org/...')

        assert 'window.top.document.fargo_set_token' in fargo_resp.text

    resp.form['f0$file'] = None

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

        data = {'data': [{'id': '1', 'text': 'hello'}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        resp.form['f0'] = '1'

        resp.form['f0'] = '2'

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

        data = {'data': [{'id': '1', 'text': 'hello', 'disabled': True}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        pq = resp.pyquery.remove_namespaces()

        assert pq('option[disabled=disabled][value="1"]').text() == 'hello'

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

        data = {'data': [{'id': '1', 'text': 'hello', 'disabled': True}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        pq = resp.pyquery.remove_namespaces()

        assert len(pq('option[disabled=disabled][value="1"]')) == 0

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

        data = {'data': [{'id': '1', 'text': 'hello'}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        resp.form['f0'] = '1'

        resp.form['f0'] = '2'

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

        data = {'data': [{'id': '1', 'text': 'hello', 'disabled': True}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        pq = resp.pyquery.remove_namespaces()

        assert len(pq('input[name="f0"][disabled=disabled][value="1"]')) == 1

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

        data = {'data': [{'id': '1', 'text': 'hello'}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        resp.form['f0$element1'].checked = True

        resp.form['f0$element2'].checked = True

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

        data = {'data': [{'id': '1', 'text': 'hello', 'disabled': True}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        assert 'disabled' in resp.form['f0$element1'].attrs

        resp.form['f0$element1'].checked = True

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

        data = {'data': [{'id': '1', 'text': 'hello', 'disabled': True}, {'id': '2', 'text': 'world'}]}

        resp = get_app(pub).get('/test/')

        assert not 'f0$element1' in resp.form.fields

        resp.form['f0$element2'].checked = True

                {'id': '2', 'text': 'world', 'extra': 'bar'},

            ]

        }

        resp = get_app(pub).get('/test/')

        assert 'data-autocomplete="true"' in resp.text

        assert resp.form['f0'].value == '1'

                {'id': '2', 'text': 'world', 'extra': 'bar'},

            ]

        }

        resp = get_app(pub).get('/test/')

        assert 'data-autocomplete="true"' in resp.text

        assert 'data-hint="help text"' in resp.text

                {'id': '2', 'text': 'world', 'extra': 'bar'},

            ]

        }

        resp = app.get('/test/')

        assert urlopen.call_count == 0

        pq = resp.pyquery.remove_namespaces()

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

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        resp2 = app.get(select2_url + '?q=hell')

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0] == 'http://remote.example.net/json?q=hell'

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

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        resp = resp.form.submit('submit')  # -> validation page

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0] == 'http://remote.example.net/json?id=1'

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

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        resp = resp.form.submit('submit')  # -> submit

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0] == 'http://remote.example.net/json?id=1'

        data = {

            'data': [{'id': 1, 'text': 'hello', 'extra': 'foo'}, {'id': 2, 'text': 'world', 'extra': 'bar'}]

        }

        resp = app.get('/test/')

        assert urlopen.call_count == 0

        pq = resp.pyquery.remove_namespaces()

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

        data = {'data': [{'id': 1, 'text': 'hello', 'extra': 'foo'}]}

        resp2 = app.get(select2_url + '?q=hell')

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0] == 'http://remote.example.net/json-numeric-id?q=hell'

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

        data = {'data': [{'id': 1, 'text': 'hello', 'extra': 'foo'}]}

        resp = resp.form.submit('submit')  # -> validation page

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0] == 'http://remote.example.net/json-numeric-id?id=1'

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

        data = {'data': [{'id': 1, 'text': 'hello', 'extra': 'foo'}]}

        resp = resp.form.submit('submit')  # -> submit

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0] == 'http://remote.example.net/json-numeric-id?id=1'

                {'id': '2', 'text': 'world', 'extra': 'bar'},

            ]

        }

        resp = app.get('/test/')

        assert urlopen.call_count == 0

        pq = resp.pyquery.remove_namespaces()

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

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        resp2 = app.get(select2_url + '?q=hell')

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0].startswith('http://remote.example.net/json?q=hell&orig=example.net&')

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

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        resp = resp.form.submit('submit')  # -> validation page

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0].startswith('http://remote.example.net/json?id=1&orig=example.net&')

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

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        resp = resp.form.submit('submit')  # -> submit

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0].startswith('http://remote.example.net/json?id=1&orig=example.net&')

                {'id': '2', 'text': 'world', 'extra': 'bar'},

            ]

        }

        resp = app.get('/test/')

        pq = resp.pyquery.remove_namespaces()

        select2_url = pq('select').attr['data-select2-url']

                {'id': '2', 'text': 'world', 'extra': 'bar'},

            ]

        }

        resp = app.get('/test/')

        assert urlopen.call_count == 0

        pq = resp.pyquery.remove_namespaces()

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

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        resp2 = app.get(select2_url + '?q=hell', status=403)

        assert urlopen.call_count == 0

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

        data = {'data': create_formdata['data']}

        app = get_app(create_formdata['pub'])

        resp = app.get('/source-form/')

                {"id": "3", "datetime": "2021-01-14 10:40:00", "text": "event 3"},

            ]

        }

        resp = app.get('/test/')

        assert 'data-date="2021-01-12"' in resp and 'data-time="10:00"' in resp

                {"id": "3", "datetime": "2021-01-14 10:40:00", "text": "event 3"},

            ]

        }

        resp = app.get('/test/')

        resp.form['f2'] = '2021-01-14'

# -*- coding: utf-8 -*-

import json

import mock

import pytest

from webtest import Upload

            payload = [{'id': '1', 'text': 'foo'}]

        elif query == 'bar':

            payload = [{'id': '2', 'text': 'bar'}]

    mock_urlopen.side_effect = lambda url: data_source(url)

# -*- coding: utf-8 -*-

import base64

import json

import locale

import os

import mock

import pytest

from django.utils.encoding import force_bytes

from quixote.http_request import Upload as QuixoteUpload

from webtest import Upload, Hidden

    export_to.convert_to_pdf = False

    export_to.label = 'create doc'

    upload = QuixoteUpload('/foo/test.rtf', content_type='application/rtf')

    upload.fp.write(b'HELLO WORLD')

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    # change export model to now be a RTF file, do the action again on the same form and

    # check that both the old .odt file and the new .rtf file are there and valid.

    upload = QuixoteUpload('/foo/test.rtf', content_type='application/rtf')

    upload.fp.write(b'HELLO NEW WORLD')

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    # use substitution variables on rtf: only ezt format is accepted

    upload = QuixoteUpload('/foo/test.rtf', content_type='application/rtf')

    upload.fp.write(b'HELLO {{DJANGO}} WORLD [form_name]')

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    template_filename = os.path.join(os.path.dirname(__file__), '..', odt_template)

    template = open(template_filename, 'rb').read()

    upload = QuixoteUpload('/foo/' + odt_template, content_type='application/octet-stream')

    upload.fp.write(template)

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    resp = resp.follow()  # $form/$id/create_doc

    resp = resp.follow()  # $form/$id/create_doc/

    with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

    resp = login(get_app(pub), username='foo', password='foo').get(form_location)

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

    resp = resp.follow()

    assert resp.content_type == 'application/octet-stream'

    with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

    # change file content, same name

    upload = QuixoteUpload('/foo/test.rtf', content_type='application/rtf')

    upload.fp.write(b'HELLO NEW WORLD')

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

        body = resp.click(odt_template, index=0).follow().body

    assert resp.click('test.rtf', index=0).follow().body == b'HELLO NEW WORLD'

    template_filename = os.path.join(os.path.dirname(__file__), '..', 'template.odt')

    template = open(template_filename, 'rb').read()

    upload = QuixoteUpload('/foo/template.odt', content_type='application/octet-stream')

    upload.fp.write(template)

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    resp = resp.follow()  # $form/$id/create_doc

    resp = resp.follow()  # $form/$id/create_doc/

    with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

    export_to.attach_to_history = True

    wf.store()

    response1 = resp = resp.follow()

    assert resp.content_type == 'application/octet-stream'

    with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

    # change file content, same name

    upload = QuixoteUpload('/foo/test.rtf', content_type='application/rtf')

    upload.fp.write(b'HELLO NEW WORLD')

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

        body = resp.click('template.odt', index=0).follow().body

    response2 = resp.click('test.rtf', index=0).follow()

    assert response2.body == b'HELLO NEW WORLD'

    # Test attachment substitution variables

    template_filename = os.path.join(os.path.dirname(__file__), '..', 'template.odt')

    template = open(template_filename, 'rb').read()

    upload = QuixoteUpload('/foo/template.odt', content_type='application/octet-stream')

    upload.fp.write(template)

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    template_filename = os.path.join(os.path.dirname(__file__), '..', 'template.odt')

    template = open(template_filename, 'rb').read()

    upload = QuixoteUpload('/foo/template.odt', content_type='application/octet-stream')

    upload.fp.write(template)

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    template_filename = os.path.join(os.path.dirname(__file__), '..', 'template.odt')

    template = open(template_filename, 'rb').read()

    upload = QuixoteUpload('/foo/template.odt', content_type='application/octet-stream')

    upload.fp.write(template)

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    resp = resp.follow()

    assert resp.content_type == 'application/octet-stream'

    with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

    assert formdef.data_class().count() == 1

    assert formdef.data_class().select()[0].status == 'wf-st2'

    template_filename = os.path.join(os.path.dirname(__file__), '..', 'template.odt')

    template = open(template_filename, 'rb').read()

    upload = QuixoteUpload('/foo/template.odt', content_type='application/octet-stream')

    upload.fp.write(template)

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

        resp = resp.follow()

        assert resp.content_type == 'application/octet-stream'

        with open(os.path.join(os.path.dirname(__file__), '..', 'template-out.odt'), 'rb') as f:

    assert formdef.data_class().count() == 1

    assert formdef.data_class().select()[0].status == 'wf-st2'

    export_to = ExportToModel()

    export_to.label = 'create doc'

    upload = QuixoteUpload('/foo/test.rtf', content_type='application/rtf')

    upload.fp.write(b'HELLO WORLD')

    upload.fp.seek(0)

    export_to.model_file = UploadedFile(pub.app_dir, None, upload)

    formdata = formdef.data_class().select()[0]

    assert 'xxx_var_yyy_raw' in formdata.workflow_data

    assert download.content_type == 'text/plain'

    assert download.body == b'foobar'

        data = {'data': [{'id': '1', 'text': 'hello', 'extra': 'foo'}]}

        def side_effect(url, *args):

        urlopen.side_effect = side_effect

# -*- coding: utf-8 -*-

import pytest

import json

import shutil

import os

import time

import json

import sys

import xml.etree.ElementTree as ET

import zipfile

from django.utils.encoding import force_bytes, force_text

from quixote import cleanup, get_publisher

from wcs.qommon.http_request import HTTPRequest

def sign_uri(uri, user=None, format='json'):

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    if query:

        query += '&'

    if format:

        query += 'format=%s&' % format

    query += 'orig=coucou&algo=sha256&timestamp=' + timestamp

    if user:

        base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha256).digest())

    )

def test_user_page_redirect(pub):

def test_get_user_from_api_query_string_error_missing_timestamp(pub):

        base64.b64encode(hmac.new(b'1234', b'format=json&orig=coucou&algo=sha1', hashlib.sha1).digest())

    )

    output = get_app(pub).get(

def test_get_user_from_api_query_string_error_missing_email(pub):

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    query = 'format=json&orig=coucou&algo=sha1&timestamp=' + timestamp

    output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature), status=403)

    assert output.json['err_desc'] == 'no user specified'

def test_get_user_from_api_query_string_error_unknown_nameid(pub):

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    query = 'format=json&orig=coucou&algo=sha1&NameID=xxx&timestamp=' + timestamp

    output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature), status=403)

    assert output.json['err_desc'] == 'unknown NameID'

    # works fine without user.

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    query = 'format=json&orig=coucou&algo=sha1&timestamp=' + timestamp

    output = get_app(pub).get('/categories?%s&signature=%s' % (query, signature))

    assert output.json == {'data': []}

    output = get_app(pub).get('/json?%s&signature=%s' % (query, signature))

    # check the categories and forms endpoints accept an unknown NameID

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    query = 'format=json&NameID=xxx&orig=coucou&algo=sha1&timestamp=' + timestamp

    output = get_app(pub).get('/categories?%s&signature=%s' % (query, signature))

    assert output.json == {'data': []}

    output = get_app(pub).get('/json?%s&signature=%s' % (query, signature))

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    query = (

        'format=json&orig=coucou&algo=sha1&email='

        + '&timestamp='

        + timestamp

    )

    output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature))

    assert output.json['user_display_name'] == u'Jean Darmette'

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    query = (

        'format=json&orig=coucou&algo=sha256&email='

        + '&timestamp='

        + timestamp

    )

    output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature), status=403)

    assert output.json['err_desc'] == 'invalid signature'

    timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'

    query = (

        'format=json&orig=coucou&algo=sha256&email='

        + '&timestamp='

        + timestamp

    )

    output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature))

    assert output.json['user_display_name'] == u'Jean Darmette'

def test_sign_url(pub, local_user):

    signed_url = sign_url(

        '1234',

    )

    url = signed_url[len('http://example.net') :]

    get_app(pub).get('%s&foo=bar' % url, status=403)

    signed_url = sign_url(

        '12345',

    )

    url = signed_url[len('http://example.net') :]

    local_user.roles = [role.id]

    local_user.store()

    signed_url = sign_url(

        '1234',

    )

    url = signed_url[len('http://example.net') :]

    local_user.store()

    signed_url = sign_url(

        'http://example.net/api/user/?format=json&orig=UNKNOWN_ACCESS&email=%s'

        '5678',

    )

    url = signed_url[len('http://example.net') :]

    assert output.json['err_desc'] == 'invalid orig'

    signed_url = sign_url(

        '5678',

    )

    url = signed_url[len('http://example.net') :]

    pub.clean_nonces(now=0)

    nonce_dir = os.path.join(pub.app_dir, 'nonces')

    assert not os.path.exists(nonce_dir) or not os.listdir(nonce_dir)

    req = HTTPRequest(

        None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net', 'QUERY_STRING': signed_url[1:]}

    )

def test_get_user_compat_endpoint(pub, local_user):

    signed_url = sign_url(

    )

    url = signed_url[len('http://example.net') :]

    output = get_app(pub).get(url)

    formdef.store()

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

            '''\

{"data": [{"id": 0, "text": "zéro", "foo": "bar"}, \

{"id": 1, "text": "uné", "foo": "bar1"}, \

    def url():

        signed_url = sign_url(

            'http://example.net/api/formdefs/test/submit'

            '1234',

        )

        return signed_url[len('http://example.net') :]

    def url():

        signed_url = sign_url(

            'http://example.net/api/formdefs/test/submit'

            '1234',

        )

        return signed_url[len('http://example.net') :]

    signed_url = sign_url(

        'http://example.net/api/formdefs/test/submit'

        '1234',

    )

    url = signed_url[len('http://example.net') :]

    resp = get_app(pub).get(sign_uri('/api/forms/test/ods', user=local_user))

    assert resp.content_type == 'application/vnd.oasis.opendocument.spreadsheet'

    ods_sheet = ET.parse(zipf.open('content.xml'))

    assert len(ods_sheet.findall('.//{%s}table-row' % ods.NS['table'])) == 311

    # check it now gets the data

    resp = get_app(pub).get(sign_uri('/api/forms/test/ods', user=local_user))

    ods_sheet = ET.parse(zipf.open('content.xml'))

    assert len(ods_sheet.findall('.//{%s}table-row' % ods.NS['table'])) == 11

    custom_view.store()

    resp = get_app(pub).get(sign_uri('/api/forms/test/ods/custom-view', user=local_user))

    ods_sheet = ET.parse(zipf.open('content.xml'))

    assert len(ods_sheet.findall('.//{%s}table-row' % ods.NS['table'])) == 21

def test_reverse_geocoding(pub):

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

        get_app(pub).get('/api/reverse-geocoding', status=400)

        resp = get_app(pub).get('/api/reverse-geocoding?lat=0&lon=0')

        assert resp.content_type == 'application/json'

    def url():

        signed_url = sign_url(

            'http://example.net/api/formdefs/test/submit'

            '1234',

        )

        return signed_url[len('http://example.net') :]

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

            '''\

{"data": [{"id": 0, "text": "zéro", "foo": "bar"}, \

{"id": 1, "text": "uné", "foo": "bar1"}, \

def test_geocoding(pub):

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

        get_app(pub).get('/api/geocoding', status=400)

        resp = get_app(pub).get('/api/geocoding?q=test')

        assert resp.content_type == 'application/json'

import pytest

import xml.etree.ElementTree as ET

from wcs.qommon.http_request import HTTPRequest

from wcs.qommon.misc import indent_xml as indent

from wcs.qommon.template import Template

    carddef.data_class().wipe()

    pub.custom_view_class.wipe()

    assert carddef2.name == 'foo'

    assert carddef2.fields[1].data_source == {'type': 'carddef:foo'}

    assert carddef2._custom_views

import os

import pickle

import shutil

import pytest

from quixote import cleanup

from wcs.categories import Category, CardDefCategory

    test.store()

    test = category_class.get(1)

    test2 = category_class.import_from_xml(fd, include_id=True)

    assert test.id == test2.id

    assert test.name == test2.name

# -*- coding: utf-8 -*-

import codecs

import pytest

import os

import json

import shutil

from quixote import cleanup

from wcs.qommon.http_request import HTTPRequest

    data_source2 = NamedDataSource.select()[0]

    assert data_source2.data_source == data_source.data_source

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

            '{"data": [{"id": 0, "text": "zéro"}, {"id": 1, "text": "uné"}, {"id": 2, "text": "deux"}]}'

        )

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

    data_source.store()

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

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

        signed_url = urlopen.call_args[0][0]

    assert signed_url.startswith('https://api.example.com/json?')

    # stupid simple (but sufficient) signature test:

    assert querystring['algo'] == ['sha256']

    assert querystring['orig'] == ['example.net']

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

    data_source.store()

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

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

        signed_url = urlopen.call_args[0][0]

    assert signed_url.startswith('https://api.example.com/json?')

    assert querystring['algo'] == ['sha256']

    assert querystring['orig'] == ['example.net']

    assert querystring['nonce'][0]

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

    data_source.store()

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

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

        unsigned_url = urlopen.call_args[0][0]

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

    datasource.store()

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

            json.dumps({'data': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]})

        )

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

        value = [{'id': '1', 'text': 'foo'}]

        assert datasource.get_structured_value('1') == value[0]

        assert urlopen.call_count == 1

        assert urlopen.call_args[0][0] == 'http://whatever/?id=1'

    get_request().datasources_cache = {}

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

        value = [{'id': '1', 'text': 'bar'}, {'id': '2', 'text': 'foo'}]

        assert datasource.get_structured_value('1') == value[0]

        assert urlopen.call_count == 1

    get_request().datasources_cache = {}

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

        assert datasource.get_structured_value('1') is None

        assert urlopen.call_count == 1

    get_request().datasources_cache = {}

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

        value = [{'id': '1', 'text': 'foo'}]

        assert datasource.get_structured_value('1') == value[0]

        assert urlopen.call_count == 1

    get_request().datasources_cache = {}

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

        value = [{'id': '1', 'text': 'foo'}]

        assert datasource.get_structured_value('1') is None

        assert urlopen.call_count == 1

        # no cache for errors

    get_request().datasources_cache = {}

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

        value = {'id': '1', 'text': 'foo'}  # not a list

        assert datasource.get_structured_value('1') is None

        assert urlopen.call_count == 1

    get_request().datasources_cache = {}

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

        assert datasource.get_structured_value('1') is None

        assert urlopen.call_count == 1

    get_request().datasources_cache = {}

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

        value = [{'id': '1', 'text': 'bar'}, {'id': '2', 'text': 'foo'}]

        assert datasource.get_structured_value('2') == value[1]

        assert urlopen.call_count == 1

        # try again, get from request.datasources_cache

    get_request().datasources_cache = {}

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

        value = [{'id': '1', 'text': 'bar'}, {'id': '2', 'text': 'foo'}]

        assert datasource.get_structured_value('3') is None

        assert urlopen.call_count == 1

        # try again, get from request.datasources_cache

# -*- coding: utf-8 -*-

import pytest

import json

import shutil

from quixote import cleanup

from wcs import fields

from wcs.data_sources import NamedDataSource, build_agenda_datasources, collect_agenda_data

    pub.load_site_options()

    NamedDataSource.wipe()

    assert collect_agenda_data(pub) == []

    assert urlopen.call_args_list == [mock.call('http://chrono.example.net/api/agenda/')]

    assert urlopen.call_args_list == [mock.call('http://chrono.example.net/api/agenda/')]

    # events agenda

    urlopen.reset_mock()

    assert collect_agenda_data(pub) == [

        {'text': 'Events A', 'url': 'http://chrono.example.net/api/agenda/events-A/datetimes/'},

    # meetings agenda

    urlopen.side_effect = [

    ]

    urlopen.reset_mock()

    assert collect_agenda_data(pub) == [

    # if meeting types could not be collected

    urlopen.side_effect = [

        ConnectionError,

    ]

    urlopen.reset_mock()

    ]

    urlopen.side_effect = [

        ConnectionError,

    ]

    urlopen.reset_mock()

import datetime

import pytest

import os

from quixote import cleanup

from wcs.qommon.ezt import (

    Template,

def test_simple_qualifier():

    template = Template()

    template.parse('<p>[foo]</p>')

    template.generate(output, {'foo': 'bar'})

    assert output.getvalue() == '<p>bar</p>'

def test_simple_qualifier_missing_variable():

    template = Template()

    template.parse('<p>[foo]</p>')

    template.generate(output, {})

    assert output.getvalue() == '<p>[foo]</p>'

    template.parse('<p>[if-any foo]bar[end]</p>')