Produits Entr'ouvert » w.c.s.

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

# Copyright (C) 2005-2020  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 xml.etree.ElementTree as ET

from django.utils.six import StringIO

import pytest

from webtest import Upload

from wcs.qommon.http_request import HTTPRequest

from wcs.api_access import ApiAccess

from utilities import get_app, login, create_temporary_pub, clean_temporary_pub

from .test_all import create_superuser

def pytest_generate_tests(metafunc):

    if 'pub' in metafunc.fixturenames:

        metafunc.parametrize('pub', ['pickle', 'sql', 'pickle-templates'], indirect=True)

@pytest.fixture

def pub(request):

    pub = create_temporary_pub(

        sql_mode=bool('sql' in request.param),

        templates_mode=bool('templates' in request.param)

    )

    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()

    return pub

def teardown_module(module):

    clean_temporary_pub()

@pytest.fixture

def api_access():

    ApiAccess.wipe()

    api_access = ApiAccess(name='Jhon')

    api_access.description = 'API key for Jhon'

    api_access.access_key = '12345'

    api_access.store()

    return api_access

def test_api_access_new(pub):

    create_superuser(pub)

    ApiAccess.wipe()

    app = login(get_app(pub))

    # go to the page and cancel

    resp = app.get('/backoffice/settings/api-access/')

    resp = resp.click('New API access')

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

    assert resp.location == 'http://example.net/backoffice/settings/api-access/'

    # go to the page and add an API access

    resp = app.get('/backoffice/settings/api-access/')

    resp = resp.click('New API access')

    resp.form['name'] = 'a new API access'

    resp.form['description'] = 'description'

    resp.form['access_id'] = 'jhon'

    resp.form['access_key'] = '1234'

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

    assert resp.location == 'http://example.net/backoffice/settings/api-access/'

    resp = resp.follow()

    assert 'a new API access' in resp.text

    resp = resp.click('a new API access')

    assert 'API access - a new API access' in resp.text

    resp = resp.click('Edit')

    assert 'Edit API access' in resp.text

    assert ApiAccess.get('1').name == 'a new API access'

    # a key must be defined

    resp = app.get('/backoffice/settings/api-access/new')

    resp.form['name'] = 'a new API access'

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

    assert resp.html.find('div', {'class': 'error'}).text == 'required field'

    # can't have 2 api access with same name

    resp.form['access_key'] = '1234'

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

    assert resp.html.find('div', {'class': 'error'}).text == 'This name is already used.'

    # can't have 2 api access with same access_id

    resp.form['name'] = 'new name'

    resp.form['access_id'] = 'jhon'

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

    assert resp.html.find('div', {'class': 'error'}).text == 'This value is already used.'

def test_api_access_view(pub, api_access):

    create_superuser(pub)

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/api-access/%s/' % api_access.id)

    assert '12345' in resp.text

    resp = app.get('/backoffice/settings/api-access/wrong-id/', status=404)

def test_api_access_edit(pub, api_access):

    create_superuser(pub)

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/api-access/1/')

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

    assert resp.form['name'].value == 'Jhon'

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

    assert resp.location == 'http://example.net/backoffice/settings/api-access/1/'

    resp = resp.follow()

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

    assert 'access_id' in resp.form.fields

    resp.form['name'] = 'Smith Robert'

    resp.form['description'] = 'bla bla bla'

    resp.form['access_id'] = 'smith2'

    resp.form['access_key'] = '5678'

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

    assert resp.location == 'http://example.net/backoffice/settings/api-access/1/'

    resp = resp.follow()

    assert ApiAccess.get('1').name == 'Smith Robert'

    assert ApiAccess.get('1').description == 'bla bla bla'

    assert ApiAccess.get('1').access_id == 'smith2'

    assert ApiAccess.get('1').access_key == '5678'

    # can't have 2 api access with same name

    resp = app.get('/backoffice/settings/api-access/new')

    resp.form['name'] = 'Jhon'

    resp.form['access_key'] = '1234'

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

    resp = app.get('/backoffice/settings/api-access/1/edit')

    resp.form['name'] = 'Jhon'

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

    assert resp.html.find('div', {'class': 'error'}).text == 'This name is already used.'

def test_api_access_delete(pub, api_access):

    create_superuser(pub)

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/api-access/1/')

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

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

    assert resp.location == 'http://example.net/backoffice/settings/api-access/'

    resp = app.get('/backoffice/settings/api-access/1/')

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

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

    assert resp.location == 'http://example.net/backoffice/settings/api-access/'

    assert ApiAccess.count() == 0

def test_api_access_export(pub, api_access):

    create_superuser(pub)

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/api-access/1/')

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

    xml_export = resp.text

    ds = StringIO(xml_export)

    api_access2 = ApiAccess.import_from_xml(ds)

    assert api_access2.name == 'Jhon'

def test_api_access_import(pub, api_access):

    create_superuser(pub)

    api_access.access_id = 'foobar'

    api_access.store()

    api_access_xml = ET.tostring(api_access.export_to_xml(include_id=True))

    ApiAccess.wipe()

    assert ApiAccess.count() == 0

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/api-access/')

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

    resp.forms[0]['file'] = Upload('api_access.wcs', api_access_xml)

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

    assert ApiAccess.count() == 1

    assert set([aa.access_id for aa in ApiAccess.select()]) == set(['foobar'])

    # check access_id

    resp = app.get('/backoffice/settings/api-access/')

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

    resp.forms[0]['file'] = Upload('api_access.wcs', api_access_xml)

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

    assert ApiAccess.count() == 2

    assert set([aa.access_id for aa in ApiAccess.select()]) == set(['foobar', 'jhon'])

    resp = app.get('/backoffice/settings/api-access/')

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

    resp.forms[0]['file'] = Upload('api_access.wcs', api_access_xml)

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

    assert ApiAccess.count() == 3

    assert set([aa.access_id for aa in ApiAccess.select()]) == set(

        ['foobar', 'jhon', 'jhon_1'])

    # import an invalid file

    resp = app.get('/backoffice/settings/api-access/')

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

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

    assert resp.location == 'http://example.net/backoffice/settings/api-access/'

    resp = resp.follow()

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

    resp.form['file'] = Upload('api_access.wcs', b'garbage')

    resp = resp.form.submit()

    assert 'Invalid File' in resp.text

def test_api_access_new_access_id(pub, api_access):

    api_access.access_id = None

    api_access.store()

    assert api_access.access_id

from wcs.api_access import ApiAccess

def test_api_access_snapshot_restore(pub):

    create_superuser(pub)

    create_role()

    app = login(get_app(pub))

    api_access = ApiAccess(name='test')

    api_access.store()

    for i in range(2):

        api_access.name = 'test %s' % i

        api_access.store()

    assert pub.snapshot_class.count() == 3

    # restore as new

    resp = app.get('/backoffice/settings/api-access/%s/history/' % api_access.id)

    snapshot = pub.snapshot_class.select_object_history(api_access)[2]

    resp = resp.click(href='%s/restore' % snapshot.id)

    assert resp.form['action'].value == 'as-new'

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

    assert ApiAccess.count() == 2

    api_access2 = ApiAccess.get(resp.location.split('/')[-2])

    assert api_access2.name == 'test'

    assert api_access2.id != api_access.id

    # restore over

    resp = app.get('/backoffice/settings/api-access/%s/history/' % api_access.id)

    snapshot = pub.snapshot_class.select_object_history(api_access)[2]

    resp = resp.click(href='%s/restore' % snapshot.id)

    resp.form['action'].value = 'overwrite'

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

    assert ApiAccess.count() == 2

    api_access2 = ApiAccess.get(resp.location.split('/')[-2])

    assert api_access2.id == api_access.id

def test_api_access_snapshot_browse(pub):

    create_superuser(pub)

    create_role()

    ApiAccess.wipe()

    api_access = ApiAccess(name='test')

    api_access.store()

    assert pub.snapshot_class.count() == 1

    # check calling .store() without changes doesn't create snapshots

    api_access.store()

    assert pub.snapshot_class.count() == 1

    app = login(get_app(pub))

    resp = app.get('/backoffice/settings/api-access/%s/history/' % api_access.id)

    snapshot = pub.snapshot_class.select_object_history(api_access)[0]

    resp = resp.click(href='%s/view/' % snapshot.id)

    assert 'This API access is readonly' in resp

    with pytest.raises(IndexError):

        resp = resp.click('Edit')

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

# Copyright (C) 2005-2020  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 xml.etree.ElementTree as ET

from quixote import get_publisher, get_response, redirect

from quixote.directory import Directory

from quixote.html import TemplateIO, htmltext

from wcs.qommon import _, errors, template

from wcs.qommon import misc

from wcs.qommon.form import (

    Form, StringWidget, TextWidget, HtmlWidget, FileWidget, get_session, force_str)

from wcs.qommon.backoffice.menu import html_top

from wcs.api_access import ApiAccess

from wcs.backoffice.snapshots import SnapshotsDirectory

class ApiAccessUI(object):

    def __init__(self, api_access):

        self.api_access = api_access

        if self.api_access is None:

            self.api_access = ApiAccess()

    def get_form(self):

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

                    advanced_label=_('Additional options'))

        form.add(StringWidget, 'name', title=_('Name'), required=True, size=30,

                 value=self.api_access.name)

        form.add(TextWidget, 'description', title=_('Description'),

                 cols=40, rows=5,

                 value=self.api_access.description)

        form.add(StringWidget, 'access_id', title=_('Access identifier'), size=30,

                 value=self.api_access.access_id)

        form.add(StringWidget, 'access_key', title=_('Access key'), required=True, size=30,

                 value=self.api_access.access_key)

        if not self.api_access.is_readonly():

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

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

        return form

    def submit_form(self, form):

        name = form.get_widget('name').parse()

        access_id = form.get_widget('access_id').parse()

        for api_access in ApiAccess.select():

            if api_access.id == self.api_access.id:

                continue

            if name == api_access.name:

                form.get_widget('name').set_error(_('This name is already used.'))

            if access_id == api_access.access_id:

                form.get_widget('access_id').set_error(_('This value is already used.'))

        if form.has_errors():

            raise ValueError()

        self.api_access.name = name

        self.api_access.description = form.get_widget('description').parse()

        self.api_access.access_id = access_id

        self.api_access.access_key = form.get_widget('access_key').parse()

        self.api_access.store()

class ApiAccessPage(Directory):

    _q_exports = ['', 'edit', 'delete', 'export',

                  ('history', 'snapshots_dir'),]

    def __init__(self, component, instance=None):

        try:

            self.api_access = instance or ApiAccess.get(component)

        except KeyError:

            raise errors.TraversalError()

        self.api_access_ui = ApiAccessUI(self.api_access)

        get_response().breadcrumb.append((component + '/', self.api_access.name))

        self.snapshots_dir = SnapshotsDirectory(self.api_access)

    def get_sidebar(self):

        r = TemplateIO(html=True)

        if self.api_access.is_readonly():

            r += htmltext('<div class="infonotice"><p>%s</p></div>') % _(

                'This API access is readonly.')

        r += htmltext('<ul id="sidebar-actions">')

        if not self.api_access.is_readonly():

            r += htmltext('<li><a href="export">%s</a></li>') % _('Export')

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

            if get_publisher().snapshot_class:

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

                    'Save snapshot')

                r += htmltext('<li><a href="history/">%s</a></li>') % _('History')

        r += htmltext('</ul>')

        return r.getvalue()

    def _q_index(self):

        html_top('api-accesss', title=self.api_access.name)

        get_response().filter['sidebar'] = self.get_sidebar()

        return template.QommonTemplateResponse(

            templates=['wcs/backoffice/api_access.html'],

            context={'view': self, 'api_access': self.api_access})

    def edit(self):

        form = self.api_access_ui.get_form()

        if form.get_submit() == 'cancel':

            return redirect('.')

        if form.get_submit() == 'submit' and not form.has_errors():

            try:

                self.api_access_ui.submit_form(form)

            except ValueError:

                pass

            else:

                return redirect('../%s/' % self.api_access.id)

        get_response().breadcrumb.append(('edit', _('Edit')))

        html_top('api-access', title=_('Edit API access'))

        r = TemplateIO(html=True)

        r += htmltext('<h2>%s</h2>') % _('Edit API access')

        r += form.render()

        return r.getvalue()

    def delete(self):

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

        form.widgets.append(HtmlWidget('<p>%s</p>' % _(

            'You are about to irrevocably delete this API access.')))

        form.add_submit('delete', _('Delete'))

        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(('delete', _('Delete')))

            html_top('api_accesss', title=_('Delete API access'))

            r = TemplateIO(html=True)

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

                _('Deleting API access:'), self.api_access.name)

            r += form.render()

            return r.getvalue()

        else:

            self.api_access.remove_self()

            return redirect('..')

    def export(self):

        x = self.api_access.export_to_xml(include_id=True)

        misc.indent_xml(x)

        response = get_response()

        response.set_content_type('application/x-wcs-api-access')

        response.set_header(

            'content-disposition',

            'attachment; filename=api-access-%s.wcs' % self.api_access.access_id)

        return '<?xml version="1.0"?>\n' + force_str(ET.tostring(x))

class ApiAccessDirectory(Directory):

    _q_exports = ['', 'new', ('import', 'p_import')]

    def _q_traverse(self, path):

        get_response().breadcrumb.append(('api-access/', _('API access')))

        return super(ApiAccessDirectory, self)._q_traverse(path)

    def _q_index(self):

        html_top('api-access', title=_('API access'))

        return template.QommonTemplateResponse(

            templates=['wcs/backoffice/api_accesses.html'],

            context={'view': self, 'api_accesses': ApiAccess.select(order_by='name')})

    def new(self):

        get_response().breadcrumb.append(('new', _('New')))

        api_access_ui = ApiAccessUI(None)

        form = api_access_ui.get_form()

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

            return redirect('.')

        if form.get_submit() == 'submit' and not form.has_errors():

            try:

                api_access_ui.submit_form(form)

            except ValueError:

                pass

            else:

                return redirect('.')

        html_top('api-access', title=_('New API access'))

        r = TemplateIO(html=True)

        r += htmltext('<h2>%s</h2>') % _('New API access')

        r += form.render()

        return r.getvalue()

    def _q_lookup(self, component):

        return ApiAccessPage(component)

    def p_import(self):

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

        import_title = _('Import API access')

        form.add(FileWidget, 'file', title=_('File'), required=True)

        form.add_submit('submit', import_title)

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

        if form.get_submit() == 'cancel':

            return redirect('.')

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

            try:

                return self.import_submit(form)

            except ValueError:

                pass

        get_response().breadcrumb.append(('import', _('Import')))

        html_top('api-access', title=import_title)

        r = TemplateIO(html=True)

        r += htmltext('<h2>%s</h2>') % import_title

        r += htmltext('<p>%s</p>') % _(

            'You can install a new API access by uploading a file.')

        r += form.render()

        return r.getvalue()

    def import_submit(self, form):

        fp = form.get_widget('file').parse().fp

        error = False

        try:

            api_access = ApiAccess.import_from_xml(fp)

            get_session().message = (

                'info', _('This API access has been successfully imported.'))

        except ValueError:

            error = True

        if error:

            form.set_error('file', _('Invalid File'))

            raise ValueError()

        # check access_id unicity

        for obj in ApiAccess.select():

            if obj.access_id == api_access.access_id:

                api_access.access_id = None  # a new one will be set in .store()

                break

        api_access.store()

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

from .api_access import ApiAccessDirectory

            ('data-sources', 'data_sources'), 'wscalls', 'logs',

            ('api-access', 'api_access')]

    api_access = ApiAccessDirectory()

        if enabled('api-access'):

            r += htmltext('<dt><a href="api-access">%s</a></dt> <dd>%s</dd>') % (

                _('API access'), _('Configure access to the API endpoints'))

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

# Copyright (C) 2005-2020  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/>.

from quixote import get_publisher

from .qommon.misc import simplify

from .qommon.xml_storage import XmlStorableObject

class ApiAccess(XmlStorableObject):

    _names = 'apiaccess'

    xml_root_node = 'apiaccess'

    name = None

    access_id = None

    access_key = None

    description = None

    # declarations for serialization

    XML_NODES = [

        ('name', 'str'),

        ('description', 'str'),

        ('access_id', 'str'),

        ('access_key', 'str'),

    ]

    def __init__(self, name=None):

        XmlStorableObject.__init__(self)

        self.name = name

    def get_admin_url(self):

        base_url = get_publisher().get_backoffice_url()

        return '%s/settings/api-access/%s/' % (base_url, self.id)

    def store(self, comment=None):

        assert not self.is_readonly()

        if self.access_id is None:

            # set access_id if it's not yet there

            self.access_id = self.get_new_access_id()

        super(ApiAccess, self).store()

        if get_publisher().snapshot_class:

            get_publisher().snapshot_class.snap(instance=self, comment=comment)

    def get_new_access_id(self):

        new_access_id = simplify(self.name, space='_')

        base_new_access_id = new_access_id

        suffix_no = 0

        while True:

            for obj in self.select():

                if obj.access_id == new_access_id:

                    break

            else:

                break

            if obj.id == self.id:

                break

            suffix_no += 1

            new_access_id = '%s_%s' % (base_new_access_id, suffix_no)

        return new_access_id

        from wcs.api_access import ApiAccess

        if klass is ApiAccess:

            from wcs.admin.api_access import ApiAccessPage

            return ApiAccessPage(component='view',

                    instance=self.snapshot.instance)

        from wcs.api_access import ApiAccess

        for klass in (BlockDef, CardDef, NamedDataSource, FormDef, Workflow, NamedWsCall, ApiAccess):

{% load i18n %}

{% block body %}

<div id="appbar">

  <h2>{% trans "API access" %} - {{ api_access.name }}</h2>

{% if not api_access.is_readonly %}

<span class="actions">

  <a href="edit">{% trans "Edit" %}</a>

</span>

{% endif %}

</div>

{% if api_access.description %}

<div class="bo-block">{{ api_access.description }}</div>

{% endif %}

<div class="bo-block">

  <h3>{% trans "Parameters" %}</h3>

  <ul>

    <li>{% trans "Access identifier:" %} {{ api_access.access_id }}</li>

    <li>{% trans "Access key:" %} {{ api_access.access_key }}</li>

  </ul>

</div>

{% endblock %}

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

{% load i18n %}

{% block appbar-title %}{% trans "API access" %}{% endblock %}

{% block appbar-actions %}

<a rel="popup" href="import">{% trans "Import" %}</a>

<a rel="popup" href="new">{% trans "New API access" %}</a>

{% endblock %}

{% block content %}

{% if api_accesses %}

<ul class="objects-list single-links">

  {% for api_access in api_accesses %}

  <li><a href="{{ api_access.id }}/">{{ api_access.name }} ({{ api_access.access_id }})</a></li>

  {% endfor %}

</ul>

{% else %}

<div class="infonotice">

{% trans "There are no API access defined." %}

</div>

{% endif %}

{% endblock %}