Projet

Général

Profil

0001-api-add-data-and-geojson-views-covering-all-formdata.patch

Frédéric Péters, 28 mai 2017 00:48

Télécharger (11 ko)

Voir les différences:

Subject: [PATCH] api: add data and geojson views covering all formdatas
 (#14260)

 tests/test_api.py            | 106 +++++++++++++++++++++++++++++++++++++++++++
 wcs/api.py                   | 100 +++++++++++++++++++++++++++++-----------
 wcs/backoffice/management.py |   3 +-
 wcs/formdata.py              |   3 +-
 4 files changed, 183 insertions(+), 29 deletions(-)
tests/test_api.py
1446 1446
    formdef.store()
1447 1447
    resp = get_app(pub).get(sign_uri('/api/forms/test/geojson', user=local_user), status=404)
1448 1448

  
1449
def test_api_global_geojson(pub, local_user):
1450
    Role.wipe()
1451
    role = Role(name='test')
1452
    role.store()
1453

  
1454
    FormDef.wipe()
1455
    formdef = FormDef()
1456
    formdef.name = 'test'
1457
    formdef.workflow_roles = {'_receiver': role.id}
1458
    formdef.fields = []
1459
    formdef.store()
1460

  
1461
    data_class = formdef.data_class()
1462
    data_class.wipe()
1463

  
1464
    formdef.geolocations = {'base': 'Location'}
1465
    formdef.store()
1466

  
1467
    for i in range(30):
1468
        formdata = data_class()
1469
        date = time.strptime('2014-01-20', '%Y-%m-%d')
1470
        formdata.geolocations = {'base': {'lat': 48, 'lon': 2}}
1471
        formdata.user_id = local_user.id
1472
        formdata.just_created()
1473
        if i%3 == 0:
1474
            formdata.jump_status('new')
1475
        else:
1476
            formdata.jump_status('finished')
1477
        formdata.store()
1478

  
1479
    if not pub.is_using_postgresql():
1480
        resp = get_app(pub).get(sign_uri('/api/forms/geojson', user=local_user), status=404)
1481
        pytest.skip('this requires SQL')
1482
        return
1483

  
1484
    # check empty content if user doesn't have the appropriate role
1485
    resp = get_app(pub).get(sign_uri('/api/forms/geojson', user=local_user))
1486
    assert 'features' in resp.json
1487
    assert len(resp.json['features']) == 0
1488

  
1489
    # add proper role to user
1490
    local_user.roles = [role.id]
1491
    local_user.store()
1492

  
1493
    # check it gets the data
1494
    resp = get_app(pub).get(sign_uri('/api/forms/geojson', user=local_user))
1495
    assert 'features' in resp.json
1496
    assert len(resp.json['features']) == 10
1497

  
1498
    # check with a filter
1499
    resp = get_app(pub).get(sign_uri('/api/forms/geojson?status=done', user=local_user))
1500
    assert 'features' in resp.json
1501
    assert len(resp.json['features']) == 20
1502

  
1503
def test_api_global_listing(pub, local_user):
1504
    Role.wipe()
1505
    role = Role(name='test')
1506
    role.store()
1507

  
1508
    FormDef.wipe()
1509
    formdef = FormDef()
1510
    formdef.name = 'test'
1511
    formdef.workflow_roles = {'_receiver': role.id}
1512
    formdef.fields = [
1513
        fields.StringField(id='0', label='foobar', varname='foobar'),
1514
        ]
1515
    formdef.store()
1516

  
1517
    data_class = formdef.data_class()
1518
    data_class.wipe()
1519

  
1520
    formdef.store()
1521

  
1522
    for i in range(30):
1523
        formdata = data_class()
1524
        date = time.strptime('2014-01-20', '%Y-%m-%d')
1525
        formdata.data = {'0': 'FOO BAR'}
1526
        formdata.user_id = local_user.id
1527
        formdata.just_created()
1528
        if i%3 == 0:
1529
            formdata.jump_status('new')
1530
        else:
1531
            formdata.jump_status('finished')
1532
        formdata.store()
1533

  
1534
    if not pub.is_using_postgresql():
1535
        resp = get_app(pub).get(sign_uri('/api/forms/geojson', user=local_user), status=404)
1536
        pytest.skip('this requires SQL')
1537
        return
1538

  
1539
    # check empty content if user doesn't have the appropriate role
1540
    resp = get_app(pub).get(sign_uri('/api/forms/', user=local_user))
1541
    assert len(resp.json['data']) == 0
1542

  
1543
    # add proper role to user
1544
    local_user.roles = [role.id]
1545
    local_user.store()
1546

  
1547
    # check it gets the data
1548
    resp = get_app(pub).get(sign_uri('/api/forms/', user=local_user))
1549
    assert len(resp.json['data']) == 10
1550

  
1551
    # check with a filter
1552
    resp = get_app(pub).get(sign_uri('/api/forms/?status=done', user=local_user))
1553
    assert len(resp.json['data']) == 20
1554

  
1449 1555
def test_roles(pub, local_user):
1450 1556
    Role.wipe()
1451 1557
    role = Role(name='Hello World')
wcs/api.py
22 22
import sys
23 23

  
24 24
from quixote import get_request, get_publisher, get_response, get_session, redirect
25
from quixote.directory import Directory
25
from quixote.directory import Directory, AccessControlled
26 26

  
27 27
from qommon import _
28 28
from qommon import misc
......
39 39
from wcs.api_utils import is_url_signed, get_user_from_api_query_string
40 40

  
41 41
from backoffice.management import FormPage as BackofficeFormPage
42
from backoffice.management import ManagementDirectory
42 43

  
43 44
def posted_json_data_to_formdata_data(formdef, data):
44 45
    # remap fields from varname to field id
......
68 69
    return data
69 70

  
70 71

  
72
def get_formdata_dict(formdata, user, consider_status_visibility=True):
73
    if consider_status_visibility:
74
        status = formdata.get_visible_status(user=user)
75
        if not status:
76
            # skip hidden forms
77
            return None
78
    else:
79
        status = formdata.get_status()
80

  
81
    title = _('%(name)s #%(id)s (%(status)s)') % {
82
            'name': formdata.formdef.name,
83
            'id': formdata.get_display_id(),
84
            'status': status.name,
85
    }
86
    d = {'title': title,
87
        'name': formdata.formdef.name,
88
        'url': formdata.get_url(),
89
        'datetime': misc.strftime.strftime('%Y-%m-%d %H:%M:%S', formdata.receipt_time),
90
        'status': status.name,
91
        'status_css_class': status.extra_css_class,
92
        'keywords': formdata.formdef.keywords_list,
93
    }
94
    d.update(formdata.get_substitution_variables(minimal=True))
95
    if get_request().form.get('full') == 'on':
96
        d.update(formdata.get_json_export_dict(include_files=False))
97
    return d
98

  
99

  
71 100
class ApiFormdataPage(FormStatusPage):
72 101
    _q_exports_orig = ['', 'download']
73 102

  
......
145 174
        return ApiFormdataPage(self.formdef, formdata)
146 175

  
147 176

  
148
class ApiFormsDirectory(Directory):
149
    def _q_lookup(self, component):
177
class ApiFormsDirectory(AccessControlled, Directory):
178
    _q_exports = ['', 'geojson']
179

  
180
    def _q_access(self):
150 181
        if not is_url_signed():
151 182
            # grant access to admins, to ease debug
152 183
            if not (get_request().user and get_request().user.is_admin):
153 184
                raise AccessForbiddenError('user not authenticated')
185

  
186
    def _q_index(self):
187
        if not get_publisher().is_using_postgresql():
188
            raise TraversalError()
189

  
190
        get_request().user = get_user_from_api_query_string() or get_request().user
191

  
192
        from wcs import sql
193

  
194
        management_directory = ManagementDirectory()
195
        criterias = management_directory.get_global_listing_criterias()
196

  
197
        limit = int(get_request().form.get('limit',
198
            get_publisher().get_site_option('default-page-size') or 20))
199
        offset = int(get_request().form.get('offset', 0))
200
        order_by = get_request().form.get('order_by',
201
            get_publisher().get_site_option('default-sort-order') or '-receipt_time')
202

  
203
        output = [get_formdata_dict(x, user=get_request().user, consider_status_visibility=False)
204
                  for x in sql.AnyFormData.select(
205
                      criterias, order_by=order_by, limit=limit, offset=offset)]
206

  
207
        get_response().set_content_type('application/json')
208
        return json.dumps({'data': output},
209
                cls=misc.JSONEncoder,
210
                encoding=get_publisher().site_charset)
211

  
212

  
213
    def geojson(self):
214
        if not get_publisher().is_using_postgresql():
215
            raise TraversalError()
216
        get_request().user = get_user_from_api_query_string() or get_request().user
217
        return ManagementDirectory().geojson()
218

  
219
    def _q_lookup(self, component):
154 220
        return ApiFormPage(component)
155 221

  
156 222

  
......
485 551
        for form in self.get_user_forms(user):
486 552
            if form.is_draft():
487 553
                continue
488
            visible_status = form.get_visible_status(user=user)
489
            # skip hidden forms
490
            if not visible_status:
554
            formdata_dict = get_formdata_dict(form, user)
555
            if not formdata_dict:
556
                # skip hidden forms
491 557
                continue
492
            name = form.formdef.name
493
            id = form.get_display_id()
494
            status = visible_status.name
495
            title = _('%(name)s #%(id)s (%(status)s)') % {
496
                    'name': name,
497
                    'id': id,
498
                    'status': status
499
            }
500
            url = form.get_url()
501
            d = {'title': title,
502
                'name': form.formdef.name,
503
                'url': url,
504
                'datetime': misc.strftime.strftime('%Y-%m-%d %H:%M:%S', form.receipt_time),
505
                'status': status,
506
                'status_css_class': visible_status.extra_css_class,
507
                'keywords': form.formdef.keywords_list,
508
            }
509
            d.update(form.get_substitution_variables(minimal=True))
510
            if get_request().form.get('full') == 'on':
511
                d.update(form.get_json_export_dict(include_files=False))
512
            forms.append(d)
558
            forms.append(formdata_dict)
513 559

  
514 560
        return json.dumps(forms,
515 561
                cls=misc.JSONEncoder,
wcs/backoffice/management.py
730 730

  
731 731
    def get_global_listing_criterias(self):
732 732
        parsed_values = {}
733
        user_roles = [logged_users_role().id] + (get_request().user.roles or [])
733
        user_roles = [logged_users_role().id] + (
734
                get_request().user.roles if get_request().user else [])
734 735
        criterias = get_global_criteria(get_request(), parsed_values)
735 736
        query_parameters = (get_request().form or {}).copy()
736 737
        query_parameters.pop('callback', None) # when using jsonp
wcs/formdata.py
30 30
from qommon.storage import StorableObject, Intersects, Contains
31 31
import qommon.misc
32 32
from qommon import ezt
33
from qommon.evalutils import make_datetime
33 34
from qommon.substitution import Substitutions
34 35

  
35 36
from roles import Role
......
555 556
                'form_criticality_level': self.criticality_level,
556 557
            })
557 558
            if self.receipt_time:
558
                d['form_receipt_datetime'] = datetime.datetime(*self.receipt_time[:6])
559
                d['form_receipt_datetime'] = make_datetime(self.receipt_time)
559 560

  
560 561
        d['form_status'] = self.get_status_label()
561 562

  
562
-