csvdatasource: add upload-csv-file endpoint (#29713)

passerelle/apps/csvdatasource/models.py

@@ -16,6 +16,7 @@
 
 import csv
 import datetime
+import mimetypes
 import os
 import re
 import tempfile
@@ -26,6 +27,7 @@ import six
 from django.conf import settings
 from django.contrib.postgres.fields import JSONField
 from django.core.exceptions import ValidationError
+from django.core.files.base import ContentFile
 from django.db import models, transaction
 from django.urls import reverse
 from django.utils.encoding import force_str, force_text, smart_text
@@ -144,6 +146,8 @@ class CsvDataSource(BaseResource):
         'https://doc-publik.entrouvert.com/admin-fonctionnel/parametrage-avance/source-de-donnees-csv/'
     )
 
+    _can_update_file_description = _('Uploading spreadsheet file is limited to the following API users:')
+
     class Meta:
         verbose_name = _('Spreadsheet File')
 
@@ -518,6 +522,22 @@ class CsvDataSource(BaseResource):
                 os.makedirs(unused_dir, exist_ok=True)
                 os.rename(filepath, os.path.join(unused_dir, filename))
 
+    @endpoint(perm='can_update_file', methods=['put'])
+    def update(self, request):
+        ext = mimetypes.guess_extension(request.content_type)
+        if not ext:
+            raise APIError(
+                "can't guess filename extension for '%s' content type" % request.content_type, http_status=400
+            )
+        name = self.csv_file.storage.get_available_name('api-uploaded-file' + ext)
+        self.csv_file = ContentFile(content=request.body, name=name)
+        try:
+            self.clean()
+        except ValidationError as e:
+            raise APIError(e, http_status=400)
+        self.save()
+        return {'created': self.csv_file.name}
+
 
 class TableRow(models.Model):
     resource = models.ForeignKey('CsvDataSource', on_delete=models.CASCADE)

passerelle/apps/csvdatasource/templates/csvdatasource/csvdatasource_detail.html

@@ -13,6 +13,7 @@
 {% endblock %}
 
 {% block endpoints %}
+<h4>{% trans "Access" %}</h4>
 <ul class="endpoints">
 {% url 'generic-endpoint' connector='csvdatasource' slug=object.slug endpoint='data' as csvdatasource_data_url %}
 <li class="get-method">
@@ -30,4 +31,12 @@
 </li>
 {% endfor %}
 </ul>
+<h4>{% trans "Management" %}</h4>
+<ul class="endpoints">
+  <li class="put-method">
+    <div class="description">
+      <span class="description--label">{% trans "Modify spreadsheet file:" %}</span>
+      <a href="/csvdatasource/test/update" class="example-url">/csvdatasource/test/update</a></div>
+  </li>
+</ul>
 {% endblock %}

tests/test_csv_datasource.py

@@ -29,6 +29,7 @@ import six
 import webtest
 from django.contrib.contenttypes.models import ContentType
 from django.core.files import File
+from django.core.files.storage import default_storage
 from django.core.management import call_command
 from django.test import Client, override_settings
 from django.urls import reverse
@@ -38,7 +39,7 @@ from django.utils.six.moves.urllib.parse import urlencode
 from django.utils.timezone import now
 from test_manager import login
 
-from passerelle.apps.csvdatasource.models import CsvDataSource, Query, TableRow
+from passerelle.apps.csvdatasource.models import CsvDataSource, Query, TableRow, upload_to
 from passerelle.base.models import AccessRight, ApiUser
 from passerelle.compat import json_loads
 
@@ -867,6 +868,63 @@ def test_change_csv_command(setup):
     assert list(csv.get_rows()) != []
 
 
+def test_update(admin_user, app, setup):
+    csv, url = setup(data=StringIO(data), filename='api-uploaded-file.csv')
+    upload_dir = default_storage.path(upload_to(csv, ''))
+    url = reverse(
+        'generic-endpoint',
+        kwargs={
+            'connector': 'csvdatasource',
+            'slug': csv.slug,
+            'endpoint': 'update',
+        },
+    )
+
+    assert CsvDataSource.objects.get().get_rows()[0]['fam'] == '121'
+    assert [files for root, dirs, files in os.walk(upload_dir)][0] == ['api-uploaded-file.csv']
+
+    # curl -H "Content-Type: text/csv" -T data.csv
+    headers = {'CONTENT-TYPE': 'text/csv'}
+    body = '212' + data[3:]
+
+    # restricted access
+    resp = app.put(url, params=body, headers=headers, status=403)
+    assert resp.json['err']
+    assert 'PermissionDenied' in resp.json['err_class']
+    assert resp.json['err_class'] == "django.core.exceptions.PermissionDenied"
+
+    # add can_update_file access
+    api = ApiUser.objects.get()
+    obj_type = ContentType.objects.get_for_model(csv)
+    AccessRight.objects.create(
+        codename='can_update_file', apiuser=api, resource_type=obj_type, resource_pk=csv.pk
+    )
+
+    resp = app.put(url, params=body, headers=headers)
+    assert not resp.json['err']
+    assert CsvDataSource.objects.get().get_rows()[0]['fam'] == '212'
+    assert len([files for root, dirs, files in os.walk(upload_dir)][0]) == 2
+
+    resp = app.put(url, params=body, headers={}, status=400)
+    assert resp.json['err']
+    assert "can't guess filename extension" in resp.json['err_desc']
+
+    resp = app.put(url, params='\n', headers=headers, status=400)
+    assert resp.json['err']
+    assert 'Could not detect CSV dialect' in resp.json['err_desc']
+
+    headers = {'CONTENT-TYPE': 'application/vnd.oasis.opendocument.spreadsheet'}
+    resp = app.put(url, params=body, headers=headers, status=400)
+    assert resp.json['err']
+    assert 'Invalid CSV file: File is not a zip file' in resp.json['err_desc']
+
+    csv.sheet_name = ''
+    csv.save()
+    resp = app.put(url, params=body, headers=headers, status=400)
+    assert resp.json['err']
+    assert 'You must specify a sheet name' in resp.json['err_desc']
+
+
 @pytest.mark.parametrize(
     'payload,expected',
     [

tests/test_manager.py

@@ -632,13 +632,13 @@ def test_manager_add_open_access_warning(app, admin_user):
     app = login(app)
     resp = app.get(csv.get_absolute_url())
     assert 'Access is limited' in resp.html.find('div', {'id': 'security'}).div.text.strip()
-    resp = resp.click('Add')
+    resp = resp.click('Add', href='/can_access/')
     resp.form['apiuser'] = private.pk
     resp = resp.form.submit().follow()
     assert AccessRight.objects.count() == 1
 
     # adding public user displays a warning
-    resp = resp.click('Add')
+    resp = resp.click('Add', href='/can_access/')
     resp.form['apiuser'] = public.pk
     resp = resp.form.submit()
     assert AccessRight.objects.count() == 1