0001-iparapheur-wcs-style-parameters-for-file-33869.patch
passerelle/contrib/iparapheur/models.py | ||
---|---|---|
14 | 14 |
# You should have received a copy of the GNU Affero General Public License |
15 | 15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
16 | 16 |
import base64 |
17 |
import json |
|
18 |
import magic |
|
19 | 17 |
import urllib |
20 | 18 | |
21 | 19 |
from requests.exceptions import ConnectionError |
... | ... | |
32 | 30 |
from passerelle.utils.jsonresponse import APIError |
33 | 31 | |
34 | 32 | |
33 |
CREATE_FILE_SCHEMA = { |
|
34 |
'$schema': 'http://json-schema.org/draft-03/schema#', |
|
35 |
'title': 'Iparapheur create file', |
|
36 |
'definitions': { |
|
37 |
'file': { |
|
38 |
'type': 'object', |
|
39 |
'properties': { |
|
40 |
'content': { |
|
41 |
'type': 'string', |
|
42 |
'required': True |
|
43 |
}, |
|
44 |
'content_type': { |
|
45 |
'type': 'string', |
|
46 |
'required': True |
|
47 |
} |
|
48 |
}, |
|
49 |
'required': True |
|
50 |
} |
|
51 |
}, |
|
52 |
'type': 'object', |
|
53 |
'properties': { |
|
54 |
'file': { |
|
55 |
'$ref': '#/definitions/file' |
|
56 |
}, |
|
57 |
'title': { |
|
58 |
'type': 'string', |
|
59 |
'required': True |
|
60 |
}, |
|
61 |
'type': { |
|
62 |
'type': 'string', |
|
63 |
'required': True |
|
64 |
}, |
|
65 |
'subtype': { |
|
66 |
'type': 'string', |
|
67 |
'required': True |
|
68 |
}, |
|
69 |
'email': { |
|
70 |
'type': 'string', |
|
71 |
}, |
|
72 |
'visibility': { |
|
73 |
'type': 'string', |
|
74 |
'required': True |
|
75 |
} |
|
76 |
} |
|
77 |
} |
|
78 | ||
79 | ||
35 | 80 |
def get_client(model): |
36 | 81 |
try: |
37 | 82 |
soap_client = model.soap_client() |
... | ... | |
56 | 101 |
return {'status': f.status, 'id': f.nom, 'timestamp': f.timestamp} |
57 | 102 | |
58 | 103 | |
59 |
def get_magic_mime(data): |
|
60 |
if hasattr(magic, 'open'): |
|
61 |
# original python-magic |
|
62 |
mime = magic.open(magic.MAGIC_MIME_TYPE) |
|
63 |
mime.load() |
|
64 |
return mime.buffer(data) |
|
65 |
# pypi python-magic |
|
66 |
return magic.from_buffer(data, mime=True) |
|
67 | ||
68 | ||
69 | 104 |
class FileError(Exception): |
70 | 105 |
pass |
71 | 106 | |
... | ... | |
131 | 166 |
return {'data': [format_file(f) for f in self.call('RechercherDossiers', Status=status)]} |
132 | 167 |
return {'data': [format_file(f) for f in self.call('RechercherDossiers')]} |
133 | 168 | |
134 |
@endpoint(perm='can_access', name='create-file', methods=['post']) |
|
135 |
def create_file(self, request, email=None): |
|
136 |
data = json.loads(request.body) |
|
137 |
title = data['title'] |
|
138 |
typ = data['type'] |
|
139 |
subtyp = data['subtype'] |
|
140 |
email = data.get('email') |
|
141 |
visibility = data['visibility'] |
|
142 |
content = base64.b64decode(data['data']) |
|
143 |
content_type = data.get('content_type') if data.get('content_type') \ |
|
144 |
else get_magic_mime(content) |
|
169 |
@endpoint( |
|
170 |
perm='can_access', name='create-file', |
|
171 |
post={ |
|
172 |
'description': _('Create file'), |
|
173 |
'request_body': { |
|
174 |
'schema': { |
|
175 |
'application/json': CREATE_FILE_SCHEMA |
|
176 |
} |
|
177 |
} |
|
178 |
} |
|
179 |
) |
|
180 |
def create_file(self, request, post_data): |
|
181 |
try: |
|
182 |
content = base64.b64decode(post_data['file']['content']) |
|
183 |
except TypeError: |
|
184 |
raise APIError('Invalid base64 string') |
|
185 |
content_type = post_data['file']['content_type'] |
|
145 | 186 | |
146 | 187 |
soap_client = get_client(self) |
147 |
if visibility not in ['PUBLIC', 'SERVICE', 'CONFIDENTIEL']:
|
|
188 |
if post_data['visibility'] not in ['PUBLIC', 'SERVICE', 'CONFIDENTIEL']:
|
|
148 | 189 |
raise FileError('Unknown value for "visibility". Should be "PUBLIC", "SERVICE" or "CONFIDENTIEL"') |
149 | 190 | |
150 | 191 |
doc_type = soap_client.get_type('ns0:TypeDoc') |
151 | 192 |
doc = doc_type(content, content_type) |
152 |
parameters = {'TypeTechnique': typ,
|
|
153 |
'DossierID': slugify(title),
|
|
154 |
'DossierTitre': title,
|
|
155 |
'SousType': subtyp,
|
|
156 |
'Visibilite': visibility,
|
|
157 |
'DocumentPrincipal' : doc,
|
|
193 |
parameters = {'TypeTechnique': post_data['type'],
|
|
194 |
'DossierID': slugify(post_data['title']),
|
|
195 |
'DossierTitre': post_data['title'],
|
|
196 |
'SousType': post_data['subtype'],
|
|
197 |
'Visibilite': post_data['visibility'],
|
|
198 |
'DocumentPrincipal': doc, |
|
158 | 199 |
} |
159 |
if email:
|
|
160 |
parameters['EmailEmetteur'] = email
|
|
200 |
if 'email' in post_data:
|
|
201 |
parameters['EmailEmetteur'] = post_data['email']
|
|
161 | 202 |
resp = soap_client.overridden_service.CreerDossier(**parameters) |
162 | 203 |
if not resp or not resp.MessageRetour: |
163 | 204 |
raise FileError('unknown error, no response') |
tests/test_iparapheur.py | ||
---|---|---|
109 | 109 |
mocked_post.return_value = response |
110 | 110 |
title, ext = filename.split('.') |
111 | 111 |
base64_data = 'VGVzdCBEb2N1bWVudA==' |
112 |
data = {'type': typ, 'subtype': subtyp, 'visibility': visibility, |
|
113 |
'title': title, 'data': base64_data, 'content-type':'application/pdf'} |
|
112 |
data = { |
|
113 |
'type': typ, 'subtype': subtyp, 'visibility': visibility, |
|
114 |
'title': title, |
|
115 |
'file': { |
|
116 |
'content': base64_data, |
|
117 |
'content_type': 'application/pdf' |
|
118 |
} |
|
119 |
} |
|
120 | ||
114 | 121 |
url = reverse('generic-endpoint', kwargs={'connector': 'iparapheur', |
115 | 122 |
'endpoint': 'create-file', 'slug': conn.slug}) |
116 | 123 |
resp = app.post_json(url, params=data, status=403) |
... | ... | |
128 | 135 |
assert req.find('ns1:DocumentPrincipal', SOAP_NAMESPACES).text == base64_data |
129 | 136 |
assert resp.json['data']['RecordId'] == file_id |
130 | 137 | |
138 |
# data = { |
|
139 |
# 'type': typ, 'subtype': subtyp, 'visibility': visibility, |
|
140 |
# 'title': title, 'data': base64_data, 'content-type':'application/pdf' |
|
141 |
# } |
|
131 | 142 |
# KO |
132 | 143 |
soap_response = """<?xml version='1.0' encoding='UTF-8'?><S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body><CreerDossierResponse xmlns="http://www.adullact.org/spring-ws/iparapheur/1.0" xmlns:xmime="http://www.w3.org/2005/05/xmlmime"><MessageRetour><codeRetour>KO</codeRetour><message>KOmessage</message><severite>INFO</severite></MessageRetour></CreerDossierResponse></S:Body></S:Envelope>""" |
133 | 144 |
response._content = soap_response |
... | ... | |
159 | 170 |
assert 'Server returned HTTP status 200 (<nada>)' in resp.json['err_desc'] |
160 | 171 | |
161 | 172 |
# Unknown value for "visibility" |
162 |
data = {'type': typ, 'subtype': subtyp, 'visibility': 'UNKNOWN_VISIBILITY',
|
|
163 |
'title': title, 'data': base64_data, 'content-type':'application/pdf'}
|
|
173 |
err_data = data.copy()
|
|
174 |
err_data['visibility'] = 'UNKNOWN_VISIBILITY'
|
|
164 | 175 |
url = reverse('generic-endpoint', kwargs={'connector': 'iparapheur', |
165 | 176 |
'endpoint': 'create-file', 'slug': conn.slug}) |
166 | 177 |
url += '?apikey=%s' % API_KEY |
167 |
resp = app.post_json(url, params=data, status=500) |
|
178 |
resp = app.post_json(url, params=err_data, status=500)
|
|
168 | 179 |
assert resp.json['err'] == 1 |
169 | 180 |
assert 'FileError' in resp.json['err_class'] |
170 | 181 | |
171 | 182 |
# OK, providing email |
172 |
data = {'type': typ, 'subtype': subtyp, 'visibility': visibility, |
|
173 |
'title': title, 'data': base64_data, 'content-type':'application/pdf', |
|
174 |
'email': email} |
|
183 |
data['email'] = email |
|
175 | 184 |
url = reverse('generic-endpoint', kwargs={'connector': 'iparapheur', |
176 | 185 |
'endpoint': 'create-file', 'slug': conn.slug}) |
177 | 186 |
url += '?apikey=%s' % API_KEY |
... | ... | |
436 | 445 |
assert 'ServiceError:' in resp.json['err_desc'] |
437 | 446 |
assert 'Test server error' in resp.json['err_desc'] |
438 | 447 | |
439 |
def test_get_magic_mime(tmpdir): |
|
440 |
from PIL import Image |
|
441 |
from passerelle.contrib.iparapheur.models import get_magic_mime |
|
442 | ||
443 |
image = Image.new("RGB", (10, 10), (255, 255, 255)) |
|
444 |
image_path = tmpdir.join('image.jpeg') |
|
445 |
image.save(image_path.strpath) |
|
446 |
with image_path.open() as f: |
|
447 |
assert 'image/jpeg' == get_magic_mime(f.read()) |
|
448 | ||
449 |
text_path = tmpdir.join('text_file.txt') |
|
450 |
with text_path.open('w') as f: |
|
451 |
f.write('some text') |
|
452 |
with text_path.open() as f: |
|
453 |
assert 'text/plain' == get_magic_mime(f.read()) |
|
454 | 448 | |
455 | 449 |
@mock.patch('passerelle.utils.Request.get', side_effect=iph_mocked_get) |
456 | 450 |
def test_call_wsdl(mocked_get, app, conn): |
457 |
- |