api: expose /ods as endpoint (#40995)

tests/test_api.py

@@ -13,15 +13,18 @@ import datetime
 import time
 import json
 import sys
+import xml.etree.ElementTree as ET
+import zipfile
 
 from django.utils.encoding import force_bytes, force_text
-from django.utils.six import StringIO
+from django.utils.six import StringIO, BytesIO
 from django.utils.six.moves.urllib import parse as urllib
 from django.utils.six.moves.urllib import parse as urlparse
 
 from quixote import cleanup, get_publisher
 from wcs.qommon.http_request import HTTPRequest
 from wcs.qommon.form import PicklableUpload
+from wcs.qommon import ods
 from wcs.users import User
 from wcs.roles import Role
 from wcs.carddef import CardDef
@@ -1970,6 +1973,64 @@ def test_api_geojson_formdata(pub, local_user):
     resp = get_app(pub).get(sign_uri('/api/forms/test/geojson', user=local_user), status=404)
 
 
+def test_api_ods_formdata(pub, local_user):
+    Role.wipe()
+    role = Role(name='test')
+    role.store()
+
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.name = 'test'
+    formdef.workflow_roles = {'_receiver': role.id}
+    formdef.fields = [
+        fields.StringField(id='0', label='foobar', varname='foobar', type='string'),
+    ]
+    formdef.store()
+
+    data_class = formdef.data_class()
+    data_class.wipe()
+
+    # check access is denied if the user has not the appropriate role
+    resp = get_app(pub).get(sign_uri('/api/forms/test/ods', user=local_user), status=403)
+    # even if there's an anonymise parameter
+    resp = get_app(pub).get(sign_uri('/api/forms/test/ods?anonymise', user=local_user), status=403)
+
+    data = {'0': 'foobar'}
+    for i in range(30):
+        formdata = data_class()
+        formdata.data = data
+        formdata.user_id = local_user.id
+        formdata.just_created()
+        if i % 3 == 0:
+            formdata.jump_status('new')
+        else:
+            formdata.jump_status('finished')
+        formdata.store()
+
+    # add proper role to user
+    local_user.roles = [role.id]
+    local_user.store()
+
+    # check it gets the data
+    resp = get_app(pub).get(sign_uri('/api/forms/test/ods', user=local_user))
+    assert resp.content_type == 'application/vnd.oasis.opendocument.spreadsheet'
+
+    # check it still gives a ods file when there is more date
+    for i in range(300):
+        formdata = data_class()
+        formdata.data = data
+        formdata.user_id = local_user.id
+        formdata.just_created()
+        formdata.jump_status('new')
+        formdata.store()
+
+    resp = get_app(pub).get(sign_uri('/api/forms/test/ods', user=local_user))
+    assert resp.content_type == 'application/vnd.oasis.opendocument.spreadsheet'
+    zipf = zipfile.ZipFile(BytesIO(resp.body))
+    ods_sheet = ET.parse(zipf.open('content.xml'))
+    assert len(ods_sheet.findall('.//{%s}table-row' % ods.NS['table'])) == 311
+
+
 def test_api_global_geojson(pub, local_user):
     Role.wipe()
     role = Role(name='test')

wcs/api.py

@@ -164,7 +164,7 @@ class ApiFormdataPage(FormStatusPage):
 
 
 class ApiFormPage(BackofficeFormPage):
-    _q_exports = [('list', 'json'), 'geojson'] # restrict to API endpoints
+    _q_exports = [('list', 'json'), 'geojson', 'ods']  # restrict to API endpoints
     formdef_class = FormDef
 
     def __init__(self, component):

wcs/backoffice/management.py

@@ -1771,9 +1771,12 @@ class FormPage(Directory):
 
     def ods(self):
         self.check_access()
+        if 'anonymise' in get_request().form:
+            # api/ will let this pass but we don't want that.
+            raise errors.AccessForbiddenError()
         fields = self.get_fields_from_query()
         selected_filter = self.get_filter_from_query()
-        user = get_request().user
+        user = get_user_from_api_query_string() or get_request().user
         query = get_request().form.get('q')
         criterias = self.get_criterias_from_query()
 
@@ -1814,7 +1817,8 @@ class FormPage(Directory):
 
         count = self.formdef.data_class().count()
         exporter = Exporter(self, self.formdef, fields, selected_filter)
-        if count > 100: # Arbitrary threshold
+        if count > 100 and not get_request().is_api_url():
+            # The "100" threshold is arbitrary
             job = get_response().add_after_job(
                 str(N_('Exporting forms in Open Document format')),
                 exporter.export)