family: add Egee Thonon invoices loader (#48070)

passerelle/apps/family/loaders/egee_thonon.py

@@ -0,0 +1,54 @@
+# Passerelle - uniform access to data and services
+# Copyright (C) 2020  Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# 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 datetime
+from xml.etree import ElementTree as ET
+
+from django.core.exceptions import ValidationError
+from django.utils.translation import ugettext_lazy as _
+
+from ..models import Invoice
+
+
+class Loader:
+    def __init__(self, connector):
+        self.connector = connector
+
+    def clean(self, archive):
+        if not 'factures.xml' in archive.namelist():
+            raise ValidationError(_('Missing factures.xml file in zip.'))
+
+    def load(self, archive):
+        with archive.open('factures.xml') as fdxml:
+            factures = ET.fromstring(fdxml.read())
+        external_ids = []
+        for facture in factures:
+            external_id = facture.attrib['Reference']
+            amount = facture.attrib['Montant'].replace(',', '.')
+            date = datetime.datetime.strptime(facture.attrib['Date'], '%d/%m/%Y')
+            limit_date = date + datetime.timedelta(days=62)
+            invoice = {
+                'total_amount': amount,
+                'amount': amount,
+                'issue_date': date,
+                'pay_limit_date': limit_date,
+                'online_payment': True,
+                'no_online_payment_reason': None,
+            }
+            obj, created = Invoice.objects.update_or_create(resource=self.connector,
+                    external_id=external_id, defaults=invoice)
+            external_ids.append(external_id)
+        Invoice.objects.filter(resource=self.connector).exclude(external_id__in=external_ids).delete()

passerelle/apps/family/models.py

@@ -146,6 +146,7 @@ class GenericFamily(BaseResource):
             ('native', _('Native')),
             ('concerto_fondettes', _('Concerto extract from Fondettes')),
             ('concerto_orleans', _(u'Concerto extract from Orléans')),
+            ('egee_thonon', _(u'Egee Invoices from Thonon Agglomération')),
         ),
         default='native')
 

tests/test_family.py

@@ -358,3 +358,41 @@ def test_incorrect_orleans_data(caplog):
         assert 'Error occured while importing data:' in record.message
         assert record.name == 'passerelle.resource.family.test-orleans'
         assert record.levelno == logging.ERROR
+
+def test_egee_thonon_loader():
+    Invoice.objects.all().delete()
+    filepath = os.path.join(os.path.dirname(__file__), 'data', 'family_data_egee_thonon.zip')
+    with open(filepath, 'rb') as fd:
+        resource = GenericFamily.objects.create(title='test-egee-thonon',
+                slug='test-egee-thonon', archive=File(fd, 'family_data_egee_thonon.zip'),
+                file_format='egee_thonon')
+    assert Invoice.objects.filter(resource=resource).count() == 4
+    for invoice in Invoice.objects.all():
+        assert len(invoice.external_id) == 13
+        assert invoice.online_payment is True
+        assert invoice.no_online_payment_reason is None
+        assert invoice.paid is False
+        assert (invoice.pay_limit_date - invoice.issue_date).days == 62
+
+    # modify invoice list, and reimport
+    Invoice.objects.first().delete()
+    Invoice.objects.first().delete()
+    paid_invoice = Invoice.objects.first()
+    paid_invoice.paid = True
+    paid_invoice.save()
+    Invoice.objects.create(resource=resource, external_id='1234', amount='1234')
+    call_command('update_families_from_zip')
+    assert Invoice.objects.filter(resource=resource).count() == 4
+    for invoice in Invoice.objects.all():
+        assert len(invoice.external_id) == 13
+        if invoice.id == paid_invoice.id:
+            assert invoice.paid is True  # already paid, not updated
+        else:
+            assert invoice.paid is False
+
+    # invalid zip file
+    filepath = os.path.join(os.path.dirname(__file__), 'data', 'family_data.zip')
+    with open(filepath, 'rb') as fd:
+        resource.archive = File(fd, 'family_data.zip')
+        with pytest.raises(ValidationError, match=r'Missing factures.xml file in zip'):
+            resource.clean()