Changeset 8884 for main/waeup.kofa/trunk/src/waeup/kofa/students

Timestamp:

3 Jul 2012, 07:13:59 (12 years ago)

Author:

Henrik Bettermann

Message:

Import payment tickets without p_id and generate p_id in checkConversion.

Increase resolution of timestamp.

Location:

main/waeup.kofa/trunk/src/waeup/kofa/students

Files:

• batching.py (modified) (6 diffs)
• tests/sample_create_payment_data.csv (added)
• tests/test_batching.py (modified) (7 diffs)

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

@@ -26 +26 @@
 import grok
 import csv
+from time import time
 from zope.interface import Interface
 from zope.schema import getFields
@@ -291 +292 @@
     grok.baseclass()
 
-    #: required fields beside 'student_id', 'reg_number' and 'matric_number'
+    # additional available  fields
+    # beside 'student_id', 'reg_number' and 'matric_number'
     additional_fields = []
 
-    #: header fields additional required
+    #: header fields additionally required
     additional_headers = []
 
@@ -558 +560 @@
     location_fields = []
     additional_fields = ['p_id']
-    additional_headers = ['p_id']
+    additional_headers = []
+
+    def checkHeaders(self, headerfields, mode='ignore'):
+        super(StudentOnlinePaymentProcessor, self).checkHeaders(headerfields)
+        if mode == 'update' and not 'p_id' in headerfields:
+            raise FatalCSVError(
+                "Need p_id for import in update mode!")
+        return True
 
     def parentsExist(self, row, site):
@@ -573 +582 @@
         if payments is None:
             return None
+        p_id = row.get('p_id', None)
+        if p_id is None:
+            return None
         # We can use the hash symbol at the end of p_id in import files
         # to avoid annoying automatic number transformation
         # by Excel or Calc
-        p_id = row['p_id'].strip('#')
-        if p_id.startswith('p'):
-            entry = payments.get(p_id)
-        else:
-            # For data migration from old SRP
-            entry = payments.get('p' + p_id[6:])
+        p_id = p_id.strip('#')
+        if not p_id.startswith('p'):
+            # For data migration from old SRP only
+            p_id = 'p' + p_id[7:] + '0'
+        entry = payments.get(p_id)
         return entry
 
@@ -600 +611 @@
         if not p_id.startswith('p'):
             # For data migration from old SRP
-            obj.p_id = 'p' + p_id[6:]
+            obj.p_id = 'p' + p_id[7:] + '0'
             parent[obj.p_id] = obj
         else:
@@ -613 +624 @@
 
         # We have to check p_id.
-        p_id = row['p_id'].strip('#')
+        p_id = row.get('p_id', None)
+        if p_id is None:
+            timestamp = ("%d" % int(time()*10000))[1:]
+            p_id = "p%s" % timestamp
+            conv_dict['p_id'] = p_id
+            return errs, inv_errs, conv_dict
+        else:
+            p_id = p_id.strip('#')
         if p_id.startswith('p'):
             if not len(p_id) == 14:

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

@@ -124 +124 @@
 
 PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
+    '\n')[0].split(',')
+
+PAYMENT_CREATE_SAMPLE_DATA = open(
+    os.path.join(os.path.dirname(__file__), 'sample_create_payment_data.csv'),
+    'rb').read()
+
+PAYMENT_CREATE_HEADER_FIELDS = PAYMENT_CREATE_SAMPLE_DATA.split(
     '\n')[0].split(',')
 
@@ -784 +791 @@
         self.student = self.app['students'][student.student_id]
         payment = createObject(u'waeup.StudentOnlinePayment')
-        payment.p_id = 'p123'
+        payment.p_id = 'p120'
         self.student['payments'][payment.p_id] = payment
 
@@ -798 +805 @@
             self.workdir, 'sample_payment_data.csv')
         open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
+        self.csv_file2 = os.path.join(
+            self.workdir, 'sample_create_payment_data.csv')
+        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
 
     def test_interface(self):
@@ -809 +819 @@
             dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
         assert self.processor.getEntry(
-            dict(student_id=self.student.student_id, p_id='p123'),
-            self.app) is self.student['payments']['p123']
+            dict(student_id=self.student.student_id, p_id='p120'),
+            self.app) is self.student['payments']['p120']
         assert self.processor.getEntry(
-            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
-            self.app) is self.student['payments']['p123']
+            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
+            self.app) is self.student['payments']['p120']
 
     def test_addEntry(self):
@@ -831 +841 @@
             self.app)
         self.assertEqual(len(self.student['payments'].keys()),3)
-        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
+        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
 
     def test_checkConversion(self):
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(reg_number='1', p_id='3816951266236341955'))
+            dict(p_id='3816951266236341955'))
         self.assertEqual(len(errs),0)
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(reg_number='1', p_id='p1266236341955'))
+            dict(p_id='p1266236341955'))
         self.assertEqual(len(errs),0)
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(reg_number='1', p_id='nonsense'))
+            dict(p_id='nonsense'))
         self.assertEqual(len(errs),1)
-        timestamp = "%d" % int(time()*1000)
+        timestamp = ("%d" % int(time()*10000))[1:]
         p_id = "p%s" % timestamp
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(reg_number='1', p_id=p_id))
+            dict(p_id=p_id))
         self.assertEqual(len(errs),0)
 
@@ -854 +864 @@
         self.assertEqual(num_warns,0)
         payment = self.processor.getEntry(dict(reg_number='1',
-            p_id='p1290797973744'), self.app)
-        self.assertEqual(payment.p_id, 'p1290797973744')
+            p_id='p2907979737440'), self.app)
+        self.assertEqual(payment.p_id, 'p2907979737440')
         cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
         self.assertEqual(cdate, "2010-11-26 18:59:33")
@@ -881 +891 @@
         self.assertEqual(num_warns,0)
         shutil.rmtree(os.path.dirname(fin_file))
+
+    def test_import_wo_pid(self):
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
+        self.assertEqual(num_warns,0)
+        shutil.rmtree(os.path.dirname(fin_file))
+        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
+
 
 def test_suite():