iparapheur: wcs style parameters for file (#33869)

passerelle/contrib/iparapheur/models.py

@@ -14,8 +14,6 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import base64
-import json
-import magic
 import urllib
 
 from requests.exceptions import ConnectionError
@@ -32,6 +30,53 @@ from passerelle.utils.api import endpoint
 from passerelle.utils.jsonresponse import APIError
 
 
+CREATE_FILE_SCHEMA = {
+    '$schema': 'http://json-schema.org/draft-03/schema#',
+    'title': 'Iparapheur create file',
+    'definitions': {
+        'file': {
+            'type': 'object',
+            'properties': {
+                'content': {
+                    'type': 'string',
+                    'required': True
+                },
+                'content_type': {
+                    'type': 'string',
+                    'required': True
+                }
+            },
+            'required': True
+        }
+    },
+    'type': 'object',
+    'properties': {
+        'file': {
+            '$ref': '#/definitions/file'
+        },
+        'title': {
+            'type': 'string',
+            'required': True
+        },
+        'type': {
+            'type': 'string',
+            'required': True
+        },
+        'subtype': {
+            'type': 'string',
+            'required': True
+        },
+        'email': {
+            'type': 'string',
+        },
+        'visibility': {
+            'type': 'string',
+            'required': True
+        }
+    }
+}
+
+
 def get_client(model):
     try:
         soap_client = model.soap_client()
@@ -56,16 +101,6 @@ def format_file(f):
     return {'status': f.status, 'id': f.nom, 'timestamp': f.timestamp}
 
 
-def get_magic_mime(data):
-    if hasattr(magic, 'open'):
-        # original python-magic
-        mime = magic.open(magic.MAGIC_MIME_TYPE)
-        mime.load()
-        return mime.buffer(data)
-    # pypi python-magic
-    return magic.from_buffer(data, mime=True)
-
-
 class FileError(Exception):
     pass
 
@@ -131,33 +166,39 @@ class IParapheur(BaseResource, HTTPResource):
             return {'data': [format_file(f) for f in self.call('RechercherDossiers', Status=status)]}
         return {'data': [format_file(f) for f in self.call('RechercherDossiers')]}
 
-    @endpoint(perm='can_access', name='create-file', methods=['post'])
-    def create_file(self, request, email=None):
-        data = json.loads(request.body)
-        title = data['title']
-        typ = data['type']
-        subtyp = data['subtype']
-        email = data.get('email')
-        visibility = data['visibility']
-        content = base64.b64decode(data['data'])
-        content_type = data.get('content_type') if data.get('content_type') \
-            else get_magic_mime(content)
+    @endpoint(
+        perm='can_access', name='create-file',
+        post={
+            'description': _('Create file'),
+            'request_body': {
+                'schema': {
+                    'application/json': CREATE_FILE_SCHEMA
+                }
+            }
+        }
+    )
+    def create_file(self, request, post_data):
+        try:
+            content = base64.b64decode(post_data['file']['content'])
+        except TypeError:
+            raise APIError('Invalid base64 string')
+        content_type = post_data['file']['content_type']
 
         soap_client = get_client(self)
-        if visibility not in ['PUBLIC', 'SERVICE', 'CONFIDENTIEL']:
+        if post_data['visibility'] not in ['PUBLIC', 'SERVICE', 'CONFIDENTIEL']:
             raise FileError('Unknown value for "visibility". Should be "PUBLIC", "SERVICE" or "CONFIDENTIEL"')
 
         doc_type = soap_client.get_type('ns0:TypeDoc')
         doc = doc_type(content, content_type)
-        parameters = {'TypeTechnique': typ,
-                      'DossierID': slugify(title),
-                      'DossierTitre': title,
-                      'SousType': subtyp,
-                      'Visibilite': visibility,
-                      'DocumentPrincipal' : doc,
+        parameters = {'TypeTechnique': post_data['type'],
+                      'DossierID': slugify(post_data['title']),
+                      'DossierTitre': post_data['title'],
+                      'SousType': post_data['subtype'],
+                      'Visibilite': post_data['visibility'],
+                      'DocumentPrincipal': doc,
                       }
-        if email:
-            parameters['EmailEmetteur'] = email
+        if 'email' in post_data:
+            parameters['EmailEmetteur'] = post_data['email']
         resp = soap_client.overridden_service.CreerDossier(**parameters)
         if not resp or not resp.MessageRetour:
             raise FileError('unknown error, no response')

tests/test_iparapheur.py

@@ -109,8 +109,15 @@ def test_create_file(mocked_post, mocked_get, app, conn):
         mocked_post.return_value = response
         title, ext = filename.split('.')
         base64_data = 'VGVzdCBEb2N1bWVudA=='
-        data = {'type': typ, 'subtype': subtyp, 'visibility': visibility,
-                'title': title, 'data': base64_data, 'content-type':'application/pdf'}
+        data = {
+            'type': typ, 'subtype': subtyp, 'visibility': visibility,
+            'title': title,
+            'file': {
+                'content': base64_data,
+                'content_type': 'application/pdf'
+            }
+        }
+
         url = reverse('generic-endpoint', kwargs={'connector': 'iparapheur',
                                 'endpoint': 'create-file', 'slug': conn.slug})
         resp = app.post_json(url, params=data, status=403)
@@ -159,19 +166,17 @@ def test_create_file(mocked_post, mocked_get, app, conn):
     assert 'Server returned HTTP status 200 (<nada>)' in resp.json['err_desc']
 
     # Unknown value for "visibility"
-    data = {'type': typ, 'subtype': subtyp, 'visibility': 'UNKNOWN_VISIBILITY',
-            'title': title, 'data': base64_data, 'content-type':'application/pdf'}
+    err_data = data.copy()
+    err_data['visibility'] = 'UNKNOWN_VISIBILITY'
     url = reverse('generic-endpoint', kwargs={'connector': 'iparapheur',
                                     'endpoint': 'create-file', 'slug': conn.slug})
     url += '?apikey=%s' % API_KEY
-    resp = app.post_json(url, params=data, status=500)
+    resp = app.post_json(url, params=err_data, status=500)
     assert resp.json['err'] == 1
     assert 'FileError' in resp.json['err_class']
 
     # OK, providing email
-    data = {'type': typ, 'subtype': subtyp, 'visibility': visibility,
-            'title': title, 'data': base64_data, 'content-type':'application/pdf',
-            'email': email}
+    data['email'] = email
     url = reverse('generic-endpoint', kwargs={'connector': 'iparapheur',
                                     'endpoint': 'create-file', 'slug': conn.slug})
     url += '?apikey=%s' % API_KEY
@@ -436,21 +441,6 @@ def test_webfault_response(mocked_post, mocked_get, app, conn):
     assert 'ServiceError:' in resp.json['err_desc']
     assert 'Test server error' in resp.json['err_desc']
 
-def test_get_magic_mime(tmpdir):
-    from PIL import Image
-    from passerelle.contrib.iparapheur.models import get_magic_mime
-
-    image = Image.new("RGB", (10, 10), (255, 255, 255))
-    image_path = tmpdir.join('image.jpeg')
-    image.save(image_path.strpath)
-    with image_path.open() as f:
-        assert 'image/jpeg' == get_magic_mime(f.read())
-
-    text_path = tmpdir.join('text_file.txt')
-    with text_path.open('w') as f:
-        f.write('some text')
-    with text_path.open() as f:
-        assert 'text/plain' == get_magic_mime(f.read())
 
 @mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get)
 def test_call_wsdl(mocked_get, app, conn):