Changeset 9169 for main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/students/tests/test_batching.py

Timestamp:

10 Sep 2012, 11:05:07 (12 years ago)

Author:

uli

Message:

Merge changes from trunk, r8786-HEAD

Location:

main/waeup.kofa/branches/uli-async-update

Files:

• . (modified) (1 prop)
• src/waeup/kofa/students/tests/test_batching.py (modified) (21 diffs)

main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/students/tests/test_batching.py

@@ -23 +23 @@
 import unittest
 import datetime
+import grok
 from time import time
+from zope.event import notify
 from zope.component import createObject
 from zope.component.hooks import setSite, clearSite
 from zope.interface.verify import verifyClass, verifyObject
+from hurry.workflow.interfaces import IWorkflowState
 
 from waeup.kofa.app import University
@@ -126 +129 @@
     '\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(',')
+
 class StudentImportExportSetup(FunctionalTestCase):
 
@@ -171 +181 @@
     def setup_student(self, student):
         # set predictable values for `student`
-        student.matric_number = u'M123456'
+        student.matric_number = u'234'
         student.adm_code = u'my adm code'
         student.clearance_locked = False
         student.clr_code = u'my clr code'
         student.perm_address = u'Studentroad 21\nLagos 123456\n'
-        student.reg_number = u'123456'
-        student.student_id = u'A111111'
+        student.reg_number = u'123'
         student.firstname = u'Anna'
         student.lastname = u'Tester'
@@ -211 +220 @@
         ticket.passmark = 100
         ticket.semester = 2
-        study_level.addCourseTicket(ticket)
+        study_level[ticket.code] = ticket
         self.add_payment(student)
         return student
@@ -232 +241 @@
 
 
-
-class StudentProcessorTest(FunctionalTestCase):
+class StudentProcessorTest(StudentImportExportSetup):
 
     layer = FunctionalLayer
@@ -239 +247 @@
     def setUp(self):
         super(StudentProcessorTest, self).setUp()
-        # Setup a sample site for each test
-        app = University()
-        self.dc_root = tempfile.mkdtemp()
-        app['datacenter'].setStoragePath(self.dc_root)
-
-        # Prepopulate the ZODB...
-        self.getRootFolder()['app'] = app
-        # we add the site immediately after creation to the
-        # ZODB. Catalogs and other local utilities are not setup
-        # before that step.
-        self.app = self.getRootFolder()['app']
-        # Set site here. Some of the following setup code might need
-        # to access grok.getSite() and should get our new app then
-        setSite(app)
 
         # Add student with subobjects
         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)
+        student = self.setup_student(student)
+        notify(grok.ObjectModifiedEvent(student))
         self.student = self.app['students'][student.student_id]
+
         self.processor = StudentProcessor()
         self.workdir = tempfile.mkdtemp()
@@ -285 +278 @@
         open(self.csv_file_duplicates, 'wb').write(STUDENT_SAMPLE_DATA_DUPLICATES)
 
-    def tearDown(self):
-        super(StudentProcessorTest, self).tearDown()
-        shutil.rmtree(self.workdir)
-        shutil.rmtree(self.dc_root)
-        clearSite()
-        return
-
     def test_interface(self):
         # Make sure we fulfill the interface contracts.
@@ -343 +329 @@
         self.assertEqual(initial_stud_id, new_stud_id)
         return
+
+    def test_checkUpdateRequirements(self):
+        # Make sure that pg students can't be updated with wrong transition.
+        err = self.processor.checkUpdateRequirements(self.student,
+            dict(reg_number='1', state='returning'), self.app)
+        self.assertTrue(err is None)
+        self.certificate.study_mode = 'pg_ft'
+        err = self.processor.checkUpdateRequirements(self.student,
+            dict(reg_number='1', state='returning'), self.app)
+        self.assertEqual(err, 'State not allowed (pg student).')
+        IWorkflowState(self.student).setState('school fee paid')
+        err = self.processor.checkUpdateRequirements(self.student,
+            dict(reg_number='1', transition='reset6'), self.app)
+        self.assertEqual(err, 'Transition not allowed (pg student).')
+        err = self.processor.checkUpdateRequirements(self.student,
+            dict(reg_number='1', transition='register_courses'), self.app)
+        self.assertEqual(err, 'Transition not allowed (pg student).')
+
 
     def test_delEntry(self):
@@ -486 +490 @@
         super(StudentStudyCourseProcessorTest, self).setUp()
 
+        # Add student with subobjects
+        student = Student()
+        self.app['students'].addStudent(student)
+        student = self.setup_student(student)
+        notify(grok.ObjectModifiedEvent(student))
+        self.student = self.app['students'][student.student_id]
+
         # Import students with subobjects
         student_file = os.path.join(self.workdir, 'sample_student_data.csv')
@@ -532 +543 @@
             dict(reg_number='1', current_level='100'))
         self.assertEqual(len(errs),0)
+
+    def test_checkUpdateRequirements(self):
+        # Make sure that pg students can't be updated with wrong transition.
+        err = self.processor.checkUpdateRequirements(self.student['studycourse'],
+            dict(reg_number='1', current_level='100'), self.app)
+        self.assertTrue(err is None)
+        # Since row has passed the converter, current_level is an integer.
+        err = self.processor.checkUpdateRequirements(self.student['studycourse'],
+            dict(reg_number='1', current_level=999), self.app)
+        self.assertTrue(err is None)
+        IWorkflowState(self.student).setState('returning')
+        err = self.processor.checkUpdateRequirements(self.student['studycourse'],
+            dict(reg_number='1', current_level=999), self.app)
+        self.assertEqual(err, 'Not a pg student.')
 
     def test_import(self):
@@ -667 +692 @@
         self.assertEqual(num_warns,2)
         shutil.rmtree(os.path.dirname(fin_file))
-        
 
 class CourseTicketProcessorTest(StudentImportExportSetup):
@@ -681 +705 @@
         shutil.rmtree(os.path.dirname(fin_file))
 
-        # Add course and course referrer
+        # Add course and certificate course
         self.course = createObject('waeup.Course')
         self.course.code = 'COURSE1'
@@ -689 +713 @@
         self.app['faculties']['fac1']['dep1'].courses.addCourse(
             self.course)
-        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
+        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
             self.course, level=100)
 
@@ -755 +779 @@
         # Logging message from updateEntry,
         self.assertTrue(
-            'INFO - system - K1000000 - Course ticket updated: code=COURSE1, '
+            'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, '
             'mandatory=False, score=3'
             in logcontent)
@@ -769 +793 @@
         self.assertEqual(num_warns,2)
         shutil.rmtree(os.path.dirname(fin_file))
+
+    def test_import_remove(self):
+        # We perform the same import twice,
+        # the second time in remove mode. The number
+        # of warnings must be the same.
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
+        assert self.processor.entryExists(
+            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
+        self.assertEqual(num_warns,2)
+        assert self.processor.entryExists(
+            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
+        shutil.rmtree(os.path.dirname(fin_file))
+        logcontent = open(self.logfile).read()
+        self.assertTrue(
+            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
+            in logcontent)
 
 class PaymentProcessorTest(StudentImportExportSetup):
@@ -784 +827 @@
         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 +841 @@
             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 +855 @@
             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_delEntry(self):
+        assert self.processor.getEntry(
+            dict(student_id=self.student.student_id, p_id='p120'),
+            self.app) is self.student['payments']['p120']
+        self.assertEqual(len(self.student['payments'].keys()),1)
+        self.processor.delEntry(
+            dict(student_id=self.student.student_id, p_id='p120'),
+            self.app)
+        assert self.processor.getEntry(
+            dict(student_id=self.student.student_id, p_id='p120'),
+            self.app) is None
+        self.assertEqual(len(self.student['payments'].keys()),0)
 
     def test_addEntry(self):
@@ -831 +890 @@
             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 +913 @@
         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')
+        self.assertTrue(payment.p_current)
         cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
         self.assertEqual(cdate, "2010-11-26 18:59:33")
         self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
-        shutil.rmtree(os.path.dirname(fin_file))
-
+        payment = self.processor.getEntry(dict(matric_number='100001',
+            p_id='p2907125937570'), self.app)
+        self.assertEqual(payment.p_id, 'p2907125937570')
+        self.assertFalse(payment.p_current)
+        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
+        # Ooooh, still the old problem, see
+        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
+        # WAT is interpreted as GMT-1 and not GMT+1
+        self.assertEqual(cdate, "2010-11-25 21:16:33")
+        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
+        shutil.rmtree(os.path.dirname(fin_file))
         logcontent = open(self.logfile).read()
-        # Logging message from updateEntry,
+        # Logging message from updateEntry
         self.assertTrue(
             'INFO - system - K1000001 - Payment ticket updated: '
-            'p_id=p1266236341955, p_item=BTECHBDT, '
-            'creation_date=2010-02-15 13:19:01+00:00, r_code=00, '
-            'r_amount_approved=19500.0, p_category=schoolfee, '
-            'amount_auth=19500.0, p_state=paid'
+            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
+            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
+            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
+            'p_state=paid'
             in logcontent)
 
@@ -881 +950 @@
         self.assertEqual(num_warns,0)
         shutil.rmtree(os.path.dirname(fin_file))
+
+    def test_import_remove(self):
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
+        num, num_warns, fin_file, fail_file = self.processor.doImport(
+            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
+        self.assertEqual(num_warns,0)
+        shutil.rmtree(os.path.dirname(fin_file))
+        logcontent = open(self.logfile).read()
+        self.assertTrue(
+            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
+            in logcontent)
+
+    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():