0002-mdel-control-zipfile-structure-36870.patch
| passerelle/apps/mdel/models.py | ||
|---|---|---|
| 198 | 198 | |
| 199 | 199 |
inputs_dir = os.path.join(resource_base_dir, self.resource.slug, |
| 200 | 200 |
'inputs', self.name) |
| 201 | ||
| 201 |
input_files = {}
|
|
| 202 | 202 |
attached_files = [] |
| 203 | 203 | |
| 204 | 204 |
if flow_type == 'ILE-LA': |
| ... | ... | |
| 245 | 245 | |
| 246 | 246 |
doc = mdel.ILEData(self.demand_id, formdata) |
| 247 | 247 |
doc.save(inputs_dir) |
| 248 |
input_files['demande'] = doc.filename |
|
| 248 | 249 | |
| 249 | 250 |
for attached_file in attached_files: |
| 250 | 251 |
attached_file.save(inputs_dir) |
| ... | ... | |
| 319 | 320 | |
| 320 | 321 |
doc = mdel.AECData(self.demand_id, formdata) |
| 321 | 322 |
doc.save(inputs_dir) |
| 323 |
input_files['demande'] = doc.filename |
|
| 322 | 324 | |
| 323 | 325 |
submission_date = formdata.get('receipt_time', None)
|
| 324 | 326 | |
| 325 | 327 |
message = mdel.Message(flow_type, demand_num, code_insee, date=submission_date) |
| 326 | 328 |
message.save(inputs_dir) |
| 329 |
input_files['message'] = message.filename |
|
| 327 | 330 | |
| 328 | 331 |
description = mdel.Description(flow_type, demand_num, code_insee, date=submission_date, |
| 329 | 332 |
attached_files=attached_files) |
| 330 | 333 |
description.save(inputs_dir) |
| 334 |
input_files['enveloppe'] = description.filename |
|
| 331 | 335 | |
| 332 | 336 |
# create zip file |
| 333 |
return zipdir(inputs_dir) |
|
| 337 |
return zipdir(inputs_dir, input_files)
|
|
| 334 | 338 | |
| 335 | 339 |
def get_status(self): |
| 336 | 340 |
"""Read demand' statuses from file |
| passerelle/apps/mdel/utils.py | ||
|---|---|---|
| 69 | 69 |
return self |
| 70 | 70 | |
| 71 | 71 | |
| 72 |
def zipdir(path): |
|
| 72 |
def zipdir(path, input_files):
|
|
| 73 | 73 |
"""Zip directory |
| 74 | 74 |
""" |
| 75 |
basefiles = [] |
|
| 76 |
for key in ('message', 'enveloppe', 'demande'):
|
|
| 77 |
if key in input_files: |
|
| 78 |
basefiles.append(input_files[key]) |
|
| 79 | ||
| 75 | 80 |
archname = path + '.zip' |
| 76 | 81 |
with zipfile.ZipFile(archname, 'w', zipfile.ZIP_DEFLATED) as zipf: |
| 77 |
for root, dirs, files in os.walk(path): |
|
| 78 |
for f in files: |
|
| 79 |
fpath = os.path.join(root, f) |
|
| 82 |
for file_ in basefiles: |
|
| 83 |
fpath = os.path.join(path, file_) |
|
| 84 |
if os.path.exists(fpath): |
|
| 85 |
zipf.write(fpath, file_) |
|
| 86 | ||
| 87 |
for file_ in os.listdir(path): |
|
| 88 |
if file_ not in basefiles: |
|
| 89 |
fpath = os.path.join(path, file_) |
|
| 80 | 90 |
zipf.write(fpath, os.path.basename(fpath)) |
| 91 | ||
| 81 | 92 |
return archname |
| 82 | 93 | |
| 83 | 94 | |
| tests/test_mdel.py | ||
|---|---|---|
| 21 | 21 |
import json |
| 22 | 22 |
import base64 |
| 23 | 23 |
import copy |
| 24 |
import zipfile |
|
| 24 | 25 | |
| 25 | 26 |
from xml.etree import ElementTree as etree |
| 26 | 27 |
from lxml import etree as letree |
| ... | ... | |
| 55 | 56 |
letree.parse(doc, parser=parser) # Will raise an exception if schema not respected |
| 56 | 57 | |
| 57 | 58 | |
| 59 |
def check_zip_file(zipdir, expected_files): |
|
| 60 |
# check files order in zip |
|
| 61 |
zipres = zipfile.ZipFile(zipdir, 'r') |
|
| 62 |
files = [f.filename for f in zipres.infolist()] |
|
| 63 |
assert files[:len(expected_files)] == expected_files |
|
| 64 | ||
| 58 | 65 | |
| 59 | 66 |
@pytest.fixture |
| 60 | 67 |
def setup(db): |
| ... | ... | |
| 136 | 143 |
if aec_payload['display_id'] == '15-4': |
| 137 | 144 |
assert resp.json['data']['demand_id'] == '15-4-AEC-LA' |
| 138 | 145 | |
| 139 |
doc = os.path.join(get_resource_base_dir(), 'test', 'inputs', '15-4-AEC-LA--0', '15-4-AEC-LA-doc-.xml') |
|
| 146 |
basedir = os.path.join(get_resource_base_dir(), 'test', 'inputs', '15-4-AEC-LA--0') |
|
| 147 |
doc = os.path.join(basedir, '15-4-AEC-LA-doc-.xml') |
|
| 140 | 148 |
validate_schema(doc, AEC_XSD) |
| 149 |
check_zip_file( |
|
| 150 |
basedir + '.zip', ['message.xml', '15-4-AEC-LA-ent-.xml', '15-4-AEC-LA-doc-.xml'] |
|
| 151 |
) |
|
| 141 | 152 | |
| 142 | 153 |
root = etree.parse(doc).getroot() |
| 143 | 154 | |
| ... | ... | |
| 184 | 195 | |
| 185 | 196 |
assert resp.json['data']['demand_id'] == '16-1-AEC-LA' |
| 186 | 197 | |
| 187 |
doc = os.path.join(get_resource_base_dir(), 'test', 'inputs', '16-1-AEC-LA--0', '16-1-AEC-LA-doc-.xml') |
|
| 198 |
basedir = os.path.join(get_resource_base_dir(), 'test', 'inputs', '16-1-AEC-LA--0') |
|
| 199 |
doc = os.path.join(basedir, '16-1-AEC-LA-doc-.xml') |
|
| 200 |
validate_schema(doc, AEC_XSD) |
|
| 201 |
check_zip_file( |
|
| 202 |
basedir + '.zip', ['message.xml', '16-1-AEC-LA-ent-.xml', '16-1-AEC-LA-doc-.xml'] |
|
| 203 |
) |
|
| 188 | 204 |
validate_schema(doc, AEC_XSD) |
| 189 | 205 | |
| 190 | 206 |
root = etree.parse(doc).getroot() |
| ... | ... | |
| 238 | 254 |
else: |
| 239 | 255 |
assert resp.json['data']['demand_id'] == '17-1-AEC-LA' |
| 240 | 256 | |
| 241 |
doc = os.path.join(get_resource_base_dir(), 'test', 'inputs', '17-1-AEC-LA--0', '17-1-AEC-LA-doc-.xml') |
|
| 257 |
basedir = os.path.join(get_resource_base_dir(), 'test', 'inputs', '17-1-AEC-LA--0') |
|
| 258 |
doc = os.path.join(basedir, '17-1-AEC-LA-doc-.xml') |
|
| 259 |
validate_schema(doc, AEC_XSD) |
|
| 260 |
check_zip_file( |
|
| 261 |
basedir + '.zip', ['message.xml', '17-1-AEC-LA-ent-.xml', '17-1-AEC-LA-doc-.xml'] |
|
| 262 |
) |
|
| 242 | 263 |
validate_schema(doc, AEC_XSD) |
| 243 | 264 | |
| 244 | 265 |
root = etree.parse(doc).getroot() |
| 245 |
- |
|