0001-family-add-Egee-Thonon-invoices-loader-48070.patch
passerelle/apps/family/loaders/egee_thonon.py | ||
---|---|---|
1 |
# Passerelle - uniform access to data and services |
|
2 |
# Copyright (C) 2020 Entr'ouvert |
|
3 |
# |
|
4 |
# This program is free software: you can redistribute it and/or modify it |
|
5 |
# under the terms of the GNU Affero General Public License as published |
|
6 |
# by the Free Software Foundation, either version 3 of the License, or |
|
7 |
# (at your option) any later version. |
|
8 |
# |
|
9 |
# This program is distributed in the hope that it will be useful, |
|
10 |
# but WITHOUT ANY WARRANTY; without even the implied warranty of |
|
11 |
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
|
12 |
# GNU Affero General Public License for more details. |
|
13 |
# |
|
14 |
# You should have received a copy of the GNU Affero General Public License |
|
15 |
# along with this program. If not, see <http://www.gnu.org/licenses/>. |
|
16 | ||
17 |
import datetime |
|
18 |
from xml.etree import ElementTree as ET |
|
19 | ||
20 |
from django.core.exceptions import ValidationError |
|
21 |
from django.utils.translation import ugettext_lazy as _ |
|
22 | ||
23 |
from ..models import Invoice |
|
24 | ||
25 | ||
26 |
class Loader: |
|
27 |
def __init__(self, connector): |
|
28 |
self.connector = connector |
|
29 | ||
30 |
def clean(self, archive): |
|
31 |
if not 'factures.xml' in archive.namelist(): |
|
32 |
raise ValidationError(_('Missing factures.xml file in zip.')) |
|
33 | ||
34 |
def load(self, archive): |
|
35 |
with archive.open('factures.xml') as fdxml: |
|
36 |
factures = ET.fromstring(fdxml.read()) |
|
37 |
external_ids = [] |
|
38 |
for facture in factures: |
|
39 |
external_id = facture.attrib['Reference'] |
|
40 |
amount = facture.attrib['Montant'].replace(',', '.') |
|
41 |
date = datetime.datetime.strptime(facture.attrib['Date'], '%d/%m/%Y') |
|
42 |
limit_date = date + datetime.timedelta(days=62) |
|
43 |
invoice = {} |
|
44 |
invoice['total_amount'] = amount |
|
45 |
invoice['amount'] = amount |
|
46 |
invoice['issue_date'] = date.strftime('%Y-%m-%d') |
|
47 |
invoice['pay_limit_date'] = limit_date.strftime('%Y-%m-%d') |
|
48 |
invoice['online_payment'] = True |
|
49 |
invoice['no_online_payment_reason'] = None |
|
50 |
obj, created = Invoice.objects.update_or_create(resource=self.connector, |
|
51 |
external_id=external_id, defaults=invoice) |
|
52 |
external_ids.append(external_id) |
|
53 |
Invoice.objects.filter(resource=self.connector).exclude(external_id__in=external_ids).delete() |
passerelle/apps/family/models.py | ||
---|---|---|
146 | 146 |
('native', _('Native')), |
147 | 147 |
('concerto_fondettes', _('Concerto extract from Fondettes')), |
148 | 148 |
('concerto_orleans', _(u'Concerto extract from Orléans')), |
149 |
('egee_thonon', _(u'Egee Invoices from Thonon Agglomération')), |
|
149 | 150 |
), |
150 | 151 |
default='native') |
151 | 152 |
tests/test_family.py | ||
---|---|---|
358 | 358 |
assert 'Error occured while importing data:' in record.message |
359 | 359 |
assert record.name == 'passerelle.resource.family.test-orleans' |
360 | 360 |
assert record.levelno == logging.ERROR |
361 | ||
362 |
def test_egee_thonon_loader(): |
|
363 |
Invoice.objects.all().delete() |
|
364 |
filepath = os.path.join(os.path.dirname(__file__), 'data', 'family_data_egee_thonon.zip') |
|
365 |
with open(filepath, 'rb') as fd: |
|
366 |
resource = GenericFamily.objects.create(title='test-egee-thonon', |
|
367 |
slug='test-egee-thonon', archive=File(fd, 'family_data_egee_thonon.zip'), |
|
368 |
file_format='egee_thonon') |
|
369 |
assert Invoice.objects.filter(resource=resource).count() == 4 |
|
370 |
for invoice in Invoice.objects.all(): |
|
371 |
assert len(invoice.external_id) == 13 |
|
372 |
assert invoice.online_payment is True |
|
373 |
assert invoice.no_online_payment_reason is None |
|
374 |
assert invoice.paid is False |
|
375 |
assert (invoice.pay_limit_date - invoice.issue_date).days == 62 |
|
376 | ||
377 |
# modify invoice list, and reimport |
|
378 |
Invoice.objects.first().delete() |
|
379 |
Invoice.objects.first().delete() |
|
380 |
paid_invoice = Invoice.objects.first() |
|
381 |
paid_invoice.paid = True |
|
382 |
paid_invoice.save() |
|
383 |
Invoice.objects.create(resource=resource, external_id='1234', amount='1234') |
|
384 |
call_command('update_families_from_zip') |
|
385 |
assert Invoice.objects.filter(resource=resource).count() == 4 |
|
386 |
for invoice in Invoice.objects.all(): |
|
387 |
assert len(invoice.external_id) == 13 |
|
388 |
if invoice.id == paid_invoice.id: |
|
389 |
assert invoice.paid is True # already paid, not updated |
|
390 |
else: |
|
391 |
assert invoice.paid is False |
|
392 | ||
393 |
# invalid zip file |
|
394 |
filepath = os.path.join(os.path.dirname(__file__), 'data', 'family_data.zip') |
|
395 |
with open(filepath, 'rb') as fd: |
|
396 |
resource.archive = File(fd, 'family_data.zip') |
|
397 |
with pytest.raises(ValidationError, match=r'Missing factures.xml file in zip'): |
|
398 |
resource.clean() |
|
361 |
- |