Changeset 13949 for main/waeup.kofa/trunk/src/waeup/kofa

Timestamp:

17 Jun 2016, 07:46:34 (8 years ago)

Author:

Henrik Bettermann

Message:

Add ApplicantPaymentExporter.

Location:

main/waeup.kofa/trunk/src/waeup/kofa/applicants

Files:

• export.py (modified) (3 diffs)
• tests/test_export.py (modified) (2 diffs)

main/waeup.kofa/trunk/src/waeup/kofa/applicants/export.py

@@ -22 +22 @@
 from zope.component import queryUtility
 from waeup.kofa.applicants.interfaces import (
-    IApplicantBaseData, IApplicantsContainer)
+    IApplicantBaseData, IApplicantsContainer, IApplicantOnlinePayment)
 from waeup.kofa.interfaces import ICSVExporter
 from waeup.kofa.interfaces import MessageFactory as _
@@ -128 +128 @@
             # We therefore search for applicant_id.
             applicant_id=(None, None))
-        used = [value for value in applicants if value.container_code.endswith('+')]
+        used = [value for value in applicants
+                if value.container_code.endswith('+')]
         return self.export(used, filepath=filepath)
 
@@ -138 +139 @@
         container = grok.getSite()['applicants'][kw['container']]
         container_values = container.values()
-        used = [value for value in container_values if value.container_code.endswith('+')]
+        used = [value for value in container_values
+                if value.container_code.endswith('+')]
         return self.export(used, filepath=filepath)
+
+
+class ApplicantPaymentExporter(grok.GlobalUtility, ExporterBase):
+    """The Applicant Payments Exporter exports all payments made by applicants.
+    In other words, it exports payment tickets in state 'paid'. The exporter
+    searches :class:`ApplicantsCatalog` and iterates over all payment tickets
+    which are stored in an applicant (container).
+
+    The exporter exports all applicant payments if started in the Data Center
+    which means in the context of the `DataCenter` object. The exporter can also
+    be started 'locally' which means in the context of an
+    `ApplicantsContainer` container, see `ApplicantExporter` above.
+    """
+    grok.implements(ICSVExporter)
+    grok.name('applicantpayments')
+
+    fields = tuple(sorted(iface_names(
+        IApplicantOnlinePayment,
+        exclude_attribs=False,
+        omit=['display_item']))) + ('applicant_id',)
+    title = _(u'Applicant Payments')
+
+    def mangle_value(self, value, name, context=None):
+        """The mangler determines the applicant's id.
+        """
+        if name == 'applicant_id' and context is not None:
+            applicant = context.__parent__
+            value = getattr(applicant, name, None)
+        return super(
+            ApplicantPaymentExporter, self).mangle_value(
+            value, name, context=context)
+
+    def export(self, payments, filepath=None):
+        """
+        """
+        writer, outfile = self.get_csv_writer(filepath)
+        for payment in payments:
+            self.write_item(payment, writer)
+        return self.close_outfile(filepath, outfile)
+
+    def export_all(self, site, filepath=None):
+        """
+        """
+        catalog = queryUtility(
+            ICatalog, context=site, name='applicants_catalog', default=None)
+        if catalog is None:
+            return self.export([], filepath)
+        applicants = catalog.searchResults(
+            # reg_num might not be set and then would not be found.
+            # We therefore search for applicant_id.
+            applicant_id=(None, None))
+        used = [value for value in applicants
+                if value.container_code.endswith('+')]
+        payments = []
+        for applicant in used:
+            for payment in applicant.values():
+                if payment.p_state == 'paid':
+                    payments.append(payment)
+        return self.export(payments, filepath=filepath)
+
+    def export_filtered(self, site, filepath=None, **kw):
+        """
+        """
+        container = grok.getSite()['applicants'][kw['container']]
+        container_values = container.values()
+        used = [value for value in container_values
+                if value.container_code.endswith('+')]
+        payments = []
+        for applicant in used:
+            for payment in applicant.values():
+                if payment.p_state == 'paid':
+                    payments.append(payment)
+        return self.export(payments, filepath=filepath)

main/waeup.kofa/trunk/src/waeup/kofa/applicants/tests/test_export.py

@@ -6 +6 @@
 import unittest
 from zope.catalog.interfaces import ICatalog
-from zope.component import queryUtility, getUtility
+from zope.component import queryUtility, getUtility, createObject
 from zope.interface.verify import verifyObject, verifyClass
 from zope.intid.interfaces import IIntIds
 from waeup.kofa.applicants import ApplicantsContainer
 from waeup.kofa.applicants.export import (
-    ApplicantsContainerExporter, ApplicantExporter)
+    ApplicantsContainerExporter, ApplicantExporter,
+    ApplicantPaymentExporter)
 from waeup.kofa.applicants.interfaces import (
     AppCatSource, ApplicationTypeSource)
@@ -233 +234 @@
             in result)
         return
+
+
+class ApplicantPaymentExporterTest(ApplicantImportExportSetup):
+
+    layer = FunctionalLayer
+
+    def setUp(self):
+        super(ApplicantPaymentExporterTest, self).setUp()
+        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
+        self.cat = getUtility(ICatalog, name='applicants_catalog')
+        self.intids = getUtility(IIntIds)
+        return
+
+    def test_ifaces(self):
+        # make sure we fullfill interface contracts
+        obj = ApplicantPaymentExporter()
+        verifyObject(ICSVExporter, obj)
+        verifyClass(ICSVExporter, ApplicantPaymentExporter)
+        return
+
+    def test_get_as_utility(self):
+        # we can get an applicant exporter as utility
+        result = queryUtility(ICSVExporter, name="applicantpayments")
+        self.assertTrue(result is not None)
+        return
+
+    def setup_applicant(self, applicant):
+        # set predictable values for `applicant`
+        applicant.reg_number = u'123456'
+        applicant.applicant_id = u'dp2011_654321'
+        applicant.firstname = u'Anna'
+        applicant.lastname = u'Tester'
+        applicant.middlename = u'M.'
+        applicant.date_of_birth = datetime.date(1981, 2, 4)
+        applicant.sex = 'f'
+        applicant.email = 'anna@sample.com'
+        applicant.phone = u'+234-123-12345'
+        # Add payment
+        payment = createObject(u'waeup.ApplicantOnlinePayment')
+        payment.p_id = 'p120'
+        payment.p_session = 2012
+        payment.p_category = 'application'
+        payment.p_state = 'paid'
+        applicant['p120'] = payment
+        return applicant
+
+    def test_export(self):
+        applicant = self.setup_applicant(self.applicant)
+        exporter = ApplicantPaymentExporter()
+        exporter.export(applicant.values(), self.outfile)
+        result = open(self.outfile, 'rb').read()
+        cdate = str('%s#' % self.applicant['p120'].creation_date)
+        self.assertEqual(
+            'ac,amount_auth,creation_date,p_category,p_id,p_item,p_session,'
+            'p_state,payment_date,r_amount_approved,r_code,r_desc,'
+            'applicant_id\r\n,'
+            '0.0,%s,application,p120,,2012,'
+            'paid,,0.0,,,dp2011_654321\r\n' % cdate, result)
+        return
+
+    def test_export_all(self):
+        self.applicant = self.setup_applicant(self.applicant)
+        exporter = ApplicantPaymentExporter()
+        exporter.export_all(self.app, self.outfile)
+        result = open(self.outfile, 'rb').read()
+        cdate = str('%s#' % self.applicant['p120'].creation_date)
+        self.assertEqual(
+            'ac,amount_auth,creation_date,p_category,p_id,p_item,p_session,'
+            'p_state,payment_date,r_amount_approved,r_code,r_desc,'
+            'applicant_id\r\n,'
+            '0.0,%s,application,p120,,2012,'
+            'paid,,0.0,,,dp2011_654321\r\n' % cdate, result)
+        return
+
+    def test_export_filtered(self):
+        self.applicant = self.setup_applicant(self.applicant)
+        exporter = ApplicantPaymentExporter()
+        exporter.export_filtered(
+            self.app, self.outfile, container=self.container.code)
+        result = open(self.outfile, 'rb').read()
+        cdate = str('%s#' % self.applicant['p120'].creation_date)
+        self.assertEqual(
+            'ac,amount_auth,creation_date,p_category,p_id,p_item,p_session,'
+            'p_state,payment_date,r_amount_approved,r_code,r_desc,'
+            'applicant_id\r\n,'
+            '0.0,%s,application,p120,,2012,'
+            'paid,,0.0,,,dp2011_654321\r\n' % cdate, result)
+        return