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

Timestamp:

21 Sep 2012, 08:19:35 (12 years ago)

Author:

uli

Message:

Rollback r9209. Looks like multiple merges from trunk confuse svn when merging back into trunk.

Location:

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

Files:

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

main/waeup.kofa/branches/uli-zc-async/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
@@ -129 +126 @@
     '\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):
 
@@ -181 +171 @@
     def setup_student(self, student):
         # set predictable values for `student`
-        student.matric_number = u'234'
+        student.matric_number = u'M123456'
         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'123'
+        student.reg_number = u'123456'
+        student.student_id = u'A111111'
         student.firstname = u'Anna'
         student.lastname = u'Tester'
@@ -220 +211 @@
         ticket.passmark = 100
         ticket.semester = 2
-        study_level[ticket.code] = ticket
+        study_level.addCourseTicket(ticket)
         self.add_payment(student)
         return student
@@ -241 +232 @@
 
 
-class StudentProcessorTest(StudentImportExportSetup):
+
+class StudentProcessorTest(FunctionalTestCase):
 
     layer = FunctionalLayer
@@ -247 +239 @@
     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()
@@ -278 +285 @@
         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.
@@ -329 +343 @@
         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):
@@ -490 +486 @@
         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')
@@ -543 +532 @@
             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):
@@ -692 +667 @@
         self.assertEqual(num_warns,2)
         shutil.rmtree(os.path.dirname(fin_file))
+        
 
 class CourseTicketProcessorTest(StudentImportExportSetup):
@@ -705 +681 @@
         shutil.rmtree(os.path.dirname(fin_file))
 
-        # Add course and certificate course
+        # Add course and course referrer
         self.course = createObject('waeup.Course')
         self.course.code = 'COURSE1'
@@ -713 +689 @@
         self.app['faculties']['fac1']['dep1'].courses.addCourse(
             self.course)
-        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
+        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
             self.course, level=100)
 
@@ -779 +755 @@
         # Logging message from updateEntry,
         self.assertTrue(
-            'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, '
+            'INFO - system - K1000000 - Course ticket updated: code=COURSE1, '
             'mandatory=False, score=3'
             in logcontent)
@@ -793 +769 @@
         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):
@@ -827 +784 @@
         self.student = self.app['students'][student.student_id]
         payment = createObject(u'waeup.StudentOnlinePayment')
-        payment.p_id = 'p120'
+        payment.p_id = 'p123'
         self.student['payments'][payment.p_id] = payment
 
@@ -841 +798 @@
             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):
@@ -855 +809 @@
             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='p120'),
-            self.app) is self.student['payments']['p120']
+            dict(student_id=self.student.student_id, p_id='p123'),
+            self.app) is self.student['payments']['p123']
         assert self.processor.getEntry(
-            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)
+            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
+            self.app) is self.student['payments']['p123']
 
     def test_addEntry(self):
@@ -890 +831 @@
             self.app)
         self.assertEqual(len(self.student['payments'].keys()),3)
-        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
+        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
 
     def test_checkConversion(self):
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(p_id='3816951266236341955'))
+            dict(reg_number='1', p_id='3816951266236341955'))
         self.assertEqual(len(errs),0)
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(p_id='p1266236341955'))
+            dict(reg_number='1', p_id='p1266236341955'))
         self.assertEqual(len(errs),0)
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(p_id='nonsense'))
+            dict(reg_number='1', p_id='nonsense'))
         self.assertEqual(len(errs),1)
-        timestamp = ("%d" % int(time()*10000))[1:]
+        timestamp = "%d" % int(time()*1000)
         p_id = "p%s" % timestamp
         errs, inv_errs, conv_dict = self.processor.checkConversion(
-            dict(p_id=p_id))
+            dict(reg_number='1', p_id=p_id))
         self.assertEqual(len(errs),0)
 
@@ -913 +854 @@
         self.assertEqual(num_warns,0)
         payment = self.processor.getEntry(dict(reg_number='1',
-            p_id='p2907979737440'), self.app)
-        self.assertEqual(payment.p_id, 'p2907979737440')
-        self.assertTrue(payment.p_current)
+            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 18:59:33")
         self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
-        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))
+        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_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'
+            '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'
             in logcontent)
 
@@ -950 +881 @@
         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():