Changeset 7623 for main/waeup.sirp/trunk/src/waeup/sirp/students

Timestamp:

10 Feb 2012, 12:08:47 (13 years ago)

Author:

Henrik Bettermann

Message:

Implement first version of payment importer.

Location:

main/waeup.sirp/trunk/src/waeup/sirp/students

Files:

• batching.py (modified) (2 diffs)
• tests/sample_payment_data.csv (added)
• tests/test_batching.py (modified) (5 diffs)

main/waeup.sirp/trunk/src/waeup/sirp/students/batching.py

@@ -38 +38 @@
     IStudent, IStudentStudyCourse,
     IStudentUpdateByRegNo, IStudentUpdateByMatricNo,
-    IStudentStudyLevel, ICourseTicket)
+    IStudentStudyLevel, ICourseTicket,
+    IStudentOnlinePayment)
 from waeup.sirp.students.workflow import  IMPORTABLE_STATES
 from waeup.sirp.utils.batching import BatchProcessor
@@ -510 +511 @@
             return errs, inv_errs, conv_dict
         return errs, inv_errs, conv_dict
+
+class StudentOnlinePaymentProcessor(BatchProcessor):
+    """A batch processor for IStudentOnlinePayment objects.
+    """
+    grok.implements(IBatchProcessor)
+    grok.provides(IBatchProcessor)
+    grok.context(Interface)
+    util_name = 'paymentimporter'
+    grok.name(util_name)
+
+    name = u'Payment Importer'
+    iface = IStudentOnlinePayment
+    factory_name = 'waeup.StudentOnlinePayment'
+
+    location_fields = []
+
+    mode = None
+
+    @property
+    def available_fields(self):
+        return sorted(list(set(
+            ['student_id','reg_number','matric_number','p_id'] + getFields(
+                self.iface).keys())))
+
+    def checkHeaders(self, headerfields, mode='ignore'):
+        if not 'reg_number' in headerfields and not 'student_id' \
+            in headerfields and not 'matric_number' in headerfields:
+            raise FatalCSVError(
+                "Need at least columns student_id " +
+                "or reg_number or matric_number for import!")
+        if not 'p_id' in headerfields:
+            raise FatalCSVError(
+                "Need p_id for import!")
+        # Check for fields to be ignored...
+        not_ignored_fields = [x for x in headerfields
+                              if not x.startswith('--')]
+        if len(set(not_ignored_fields)) < len(not_ignored_fields):
+            raise FatalCSVError(
+                "Double headers: each column name may only appear once.")
+        return True
+
+    def getParent(self, row, site):
+        if not 'students' in site.keys():
+            return None
+        if 'student_id' in row.keys() and row['student_id']:
+            if row['student_id'] in site['students']:
+                student = site['students'][row['student_id']]
+                return student['payments']
+        elif 'reg_number' in row.keys() and row['reg_number']:
+            reg_number = row['reg_number']
+            #import pdb; pdb.set_trace()
+            cat = queryUtility(ICatalog, name='students_catalog')
+            results = list(
+                cat.searchResults(reg_number=(reg_number, reg_number)))
+            if results:
+                return results[0]['payments']
+        elif 'matric_number' in row.keys() and row['matric_number']:
+            matric_number = row['matric_number']
+            cat = queryUtility(ICatalog, name='students_catalog')
+            results = list(
+                cat.searchResults(matric_number=(matric_number, matric_number)))
+            if results:
+                return results[0]['payments']
+        return None
+
+    def parentsExist(self, row, site):
+        return self.getParent(row, site) is not None
+
+    def entryExists(self, row, site):
+        return self.getEntry(row, site) is not None
+
+    def getEntry(self, row, site):
+        payments = self.getParent(row, site)
+        if payments is None:
+            return None
+        if row['p_id'].startswith('p'):
+            entry = payments.get(row['p_id'])
+        else:
+            # For data migration from old SRP
+            entry = payments.get('p' + row['p_id'][6:])
+        return entry
+
+    def addEntry(self, obj, row, site):
+        parent = self.getParent(row, site)
+        if not row['p_id'].startswith('p'):
+            # For data migration from old SRP
+            obj.p_id = 'p' + row['p_id'][6:]
+            parent[obj.p_id] = obj
+        else:
+            parent[row['p_id']] = obj
+        return
+
+    def checkConversion(self, row, mode='ignore'):
+        """Validates all values in row.
+        """
+        converter = IObjectConverter(self.iface)
+        errs, inv_errs, conv_dict =  converter.fromStringDict(
+            row, self.factory_name)
+        # We have to check p_id.
+        if row['p_id'].startswith('p'):
+            if not len(row['p_id']) == 14:
+                errs.append(('p_id','invalid length'))
+                return errs, inv_errs, conv_dict
+        else:
+            if not len(row['p_id']) == 19:
+                errs.append(('p_id','invalid length'))
+                return errs, inv_errs, conv_dict
+        return errs, inv_errs, conv_dict

main/waeup.sirp/trunk/src/waeup/sirp/students/tests/test_batching.py

@@ -23 +23 @@
 import unittest
 import datetime
+from time import time
 from zope.component import createObject
 from zope.component.hooks import setSite, clearSite
@@ -32 +33 @@
 from waeup.sirp.students.batching import (
     StudentProcessor, StudentStudyCourseProcessor,
-    StudentStudyLevelProcessor, CourseTicketProcessor)
+    StudentStudyLevelProcessor, CourseTicketProcessor,
+    StudentOnlinePaymentProcessor)
 from waeup.sirp.students.student import Student
 from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
@@ -84 +86 @@
 
 COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+PAYMENT_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'),
+    'rb').read()
+
+PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
     '\n')[0].split(',')
 
@@ -524 +533 @@
         shutil.rmtree(os.path.dirname(fin_file))
 
+class PaymentImporterTest(FunctionalTestCase):
+
+    layer = FunctionalLayer
+
+    def setUp(self):
+        super(PaymentImporterTest, self).setUp()
+        self.dc_root = tempfile.mkdtemp()
+        self.workdir = tempfile.mkdtemp()
+        app = University()
+        app['datacenter'].setStoragePath(self.dc_root)
+        self.getRootFolder()['app'] = app
+        self.app = self.getRootFolder()['app']
+        setSite(app)
+
+        # Add student with payment
+        student = Student()
+        student.firstname = u'Anna'
+        student.lastname = u'Tester'
+        student.reg_number = u'123'
+        student.matric_number = u'234'
+        self.app['students'].addStudent(student)
+        self.student = self.app['students'][student.student_id]
+        payment = createObject(u'waeup.StudentOnlinePayment')
+        payment.p_id = 'p123'
+        self.student['payments'][payment.p_id] = payment
+
+        # Import students with subobjects
+        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
+        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
+        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
+            student_file, STUDENT_HEADER_FIELDS)
+        shutil.rmtree(os.path.dirname(fin_file))
+
+        self.importer = StudentOnlinePaymentProcessor()
+        self.csv_file = os.path.join(
+            self.workdir, 'sample_payment_data.csv')
+        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
+
+    def tearDown(self):
+        super(PaymentImporterTest, self).tearDown()
+        shutil.rmtree(self.workdir)
+        shutil.rmtree(self.dc_root)
+        clearSite()
+        return
+
+    def test_interface(self):
+        # Make sure we fulfill the interface contracts.
+        assert verifyObject(IBatchProcessor, self.importer) is True
+        assert verifyClass(
+            IBatchProcessor, StudentOnlinePaymentProcessor) is True
+
+    def test_getEntry(self):
+        assert self.importer.getEntry(
+            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
+        assert self.importer.getEntry(
+            dict(student_id=self.student.student_id, p_id='p123'),
+            self.app) is self.student['payments']['p123']
+        assert self.importer.getEntry(
+            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
+            self.app) is self.student['payments']['p123']
+
+    def test_addEntry(self):
+        self.assertEqual(len(self.student['payments'].keys()),1)
+        payment1 = createObject(u'waeup.StudentOnlinePayment')
+        payment1.p_id = 'p234'
+        self.importer.addEntry(
+            payment1, dict(student_id=self.student.student_id, p_id='p234'),
+            self.app)
+        self.assertEqual(len(self.student['payments'].keys()),2)
+        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
+        payment2 = createObject(u'waeup.StudentOnlinePayment')
+        payment1.p_id = 'nonsense'
+        # payment1.p_id will be replaced if p_id doesn't start with 'p'
+        self.importer.addEntry(
+            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
+            self.app)
+        self.assertEqual(len(self.student['payments'].keys()),3)
+        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
+
+    def test_checkConversion(self):
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', p_id='3816951266236341955'))
+        self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', p_id='p1266236341955'))
+        self.assertEqual(len(errs),0)
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', p_id='nonsense'))
+        self.assertEqual(len(errs),1)
+        timestamp = "%d" % int(time()*1000)
+        p_id = "p%s" % timestamp
+        errs, inv_errs, conv_dict = self.importer.checkConversion(
+            dict(reg_number='1', p_id=p_id))
+        self.assertEqual(len(errs),0)
+
+    def test_import(self):
+        num, num_warns, fin_file, fail_file = self.importer.doImport(
+            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
+        self.assertEqual(num_warns,0)
+        payment = self.importer.getEntry(dict(reg_number='1',
+            p_id='p1290797973744'), self.app)
+        self.assertEqual(payment.p_id, 'p1290797973744')
+        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
+        self.assertEqual(cdate, "2010-11-26 19:59:33")
+        shutil.rmtree(os.path.dirname(fin_file))
 
 def test_suite():
@@ -529 +643 @@
     for testcase in [
         StudentImporterTest,StudentStudyCourseImporterTest,
-        StudentStudyLevelImporterTest,CourseTicketImporterTest,]:
+        StudentStudyLevelImporterTest,CourseTicketImporterTest,
+        PaymentImporterTest,]:
         suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
                 testcase