Ignore:
Timestamp:
4 Oct 2012, 15:40:40 (12 years ago)
Author:
Henrik Bettermann
Message:

Check logfile. Change order of tests.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • main/waeup.kofa/trunk/src/waeup/kofa/students/tests/test_batching.py

    r9293 r9294  
    574574        shutil.rmtree(os.path.dirname(fin_file))
    575575
     576class StudentStudyLevelProcessorTest(StudentImportExportSetup):
     577
     578    def setUp(self):
     579        super(StudentStudyLevelProcessorTest, self).setUp()
     580
     581        # Import students with subobjects
     582        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
     583        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
     584        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
     585            student_file, STUDENT_HEADER_FIELDS)
     586        shutil.rmtree(os.path.dirname(fin_file))
     587
     588        # Update study courses
     589        studycourse_file = os.path.join(
     590            self.workdir, 'sample_studycourse_data.csv')
     591        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
     592        processor = StudentStudyCourseProcessor()
     593        num, num_warns, fin_file, fail_file = processor.doImport(
     594            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
     595        shutil.rmtree(os.path.dirname(fin_file))
     596
     597        self.processor = StudentStudyLevelProcessor()
     598        self.csv_file = os.path.join(
     599            self.workdir, 'sample_studylevel_data.csv')
     600        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
     601
     602    def test_interface(self):
     603        # Make sure we fulfill the interface contracts.
     604        assert verifyObject(IBatchProcessor, self.processor) is True
     605        assert verifyClass(
     606            IBatchProcessor, StudentStudyLevelProcessor) is True
     607
     608    def test_checkConversion(self):
     609        errs, inv_errs, conv_dict = self.processor.checkConversion(
     610            dict(reg_number='1', level='220'))
     611        self.assertEqual(len(errs),0)
     612        errs, inv_errs, conv_dict = self.processor.checkConversion(
     613            dict(reg_number='1', level='900'))
     614        self.assertEqual(len(errs),1)
     615        self.assertTrue(('level','no valid integer') in errs)
     616        errs, inv_errs, conv_dict = self.processor.checkConversion(
     617            dict(reg_number='1', level='xyz'))
     618        self.assertEqual(len(errs),1)
     619        self.assertTrue(('level','no integer') in errs)
     620
     621    def test_import(self):
     622        num, num_warns, fin_file, fail_file = self.processor.doImport(
     623            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
     624        self.assertEqual(num_warns,2)
     625        assert self.processor.entryExists(
     626            dict(reg_number='1', level='100'), self.app) is True
     627        studylevel = self.processor.getEntry(
     628            dict(reg_number='1', level='100'), self.app)
     629        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
     630        self.assertEqual(studylevel.level_session, 2008)
     631        self.assertEqual(studylevel.level_verdict, None)
     632        self.assertEqual(studylevel.level, 100)
     633        shutil.rmtree(os.path.dirname(fin_file))
     634
     635        logcontent = open(self.logfile).read()
     636        # Logging message from updateEntry,
     637        self.assertTrue(
     638            'INFO - system - K1000000 - Study level '
     639            'updated: level=100, level_verdict=C, level_session=2009'
     640            in logcontent)
     641
     642    def test_import_update(self):
     643        # We perform the same import twice,
     644        # the second time in update mode. The number
     645        # of warnings must be the same.
     646        num, num_warns, fin_file, fail_file = self.processor.doImport(
     647            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
     648        shutil.rmtree(os.path.dirname(fin_file))
     649        num, num_warns, fin_file, fail_file = self.processor.doImport(
     650            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
     651        self.assertEqual(num_warns,2)
     652        shutil.rmtree(os.path.dirname(fin_file))
     653
     654class CourseTicketProcessorTest(StudentImportExportSetup):
     655
     656    def setUp(self):
     657        super(CourseTicketProcessorTest, self).setUp()
     658
     659        # Import students with subobjects
     660        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
     661        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
     662        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
     663            student_file, STUDENT_HEADER_FIELDS)
     664        shutil.rmtree(os.path.dirname(fin_file))
     665
     666        # Add course and certificate course
     667        self.course = createObject('waeup.Course')
     668        self.course.code = 'COURSE1'
     669        self.course.semester = 1
     670        self.course.credits = 10
     671        self.course.passmark = 40
     672        self.app['faculties']['fac1']['dep1'].courses.addCourse(
     673            self.course)
     674        self.app['faculties']['fac1']['dep1'].certificates[
     675            'CERT1'].addCertCourse(
     676            self.course, level=100)
     677
     678        # Update study courses
     679        studycourse_file = os.path.join(
     680            self.workdir, 'sample_studycourse_data.csv')
     681        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
     682        processor = StudentStudyCourseProcessor()
     683        num, num_warns, fin_file, fail_file = processor.doImport(
     684            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
     685        shutil.rmtree(os.path.dirname(fin_file))
     686
     687        # Import study levels
     688        processor = StudentStudyLevelProcessor()
     689        studylevel_file = os.path.join(
     690            self.workdir, 'sample_studylevel_data.csv')
     691        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
     692        num, num_warns, fin_file, fail_file = processor.doImport(
     693            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
     694        shutil.rmtree(os.path.dirname(fin_file))
     695
     696        self.processor = CourseTicketProcessor()
     697        self.csv_file = os.path.join(
     698            self.workdir, 'sample_courseticket_data.csv')
     699        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
     700
     701    def test_interface(self):
     702        # Make sure we fulfill the interface contracts.
     703        assert verifyObject(IBatchProcessor, self.processor) is True
     704        assert verifyClass(
     705            IBatchProcessor, CourseTicketProcessor) is True
     706
     707    def test_checkConversion(self):
     708        errs, inv_errs, conv_dict = self.processor.checkConversion(
     709            dict(reg_number='1', code='COURSE1', level='220'))
     710        self.assertEqual(len(errs),0)
     711        errs, inv_errs, conv_dict = self.processor.checkConversion(
     712            dict(reg_number='1', code='COURSE2', level='220'))
     713        self.assertEqual(len(errs),1)
     714        self.assertTrue(('code','non-existent') in errs)
     715
     716    def test_import(self):
     717
     718        num, num_warns, fin_file, fail_file = self.processor.doImport(
     719            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
     720
     721        self.assertEqual(num_warns,2)
     722        assert self.processor.entryExists(
     723            dict(reg_number='1', level='100', code='COURSE1'),
     724            self.app) is True
     725        courseticket = self.processor.getEntry(
     726            dict(reg_number='1', level='100', code='COURSE1'), self.app)
     727        self.assertEqual(courseticket.__parent__.__parent__.certificate.code,
     728                         u'CERT1')
     729        self.assertEqual(courseticket.score, 1)
     730        self.assertEqual(courseticket.mandatory, True)
     731        self.assertEqual(courseticket.fcode, 'NA')
     732        self.assertEqual(courseticket.dcode, 'NA')
     733        self.assertEqual(courseticket.code, 'COURSE1')
     734        self.assertEqual(courseticket.title, 'Unnamed Course')
     735        self.assertEqual(courseticket.credits, 10)
     736        self.assertEqual(courseticket.passmark, 40)
     737        self.assertEqual(courseticket.semester, 1)
     738        shutil.rmtree(os.path.dirname(fin_file))
     739
     740        logcontent = open(self.logfile).read()
     741        # Logging message from updateEntry,
     742        self.assertTrue(
     743            'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, '
     744            'mandatory=False, score=3'
     745            in logcontent)
     746
     747    def test_import_update(self):
     748        # We perform the same import twice,
     749        # the second time in update mode. The number
     750        # of warnings must be the same.
     751        num, num_warns, fin_file, fail_file = self.processor.doImport(
     752            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
     753        shutil.rmtree(os.path.dirname(fin_file))
     754        num, num_warns, fin_file, fail_file = self.processor.doImport(
     755            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
     756        self.assertEqual(num_warns,2)
     757        shutil.rmtree(os.path.dirname(fin_file))
     758
     759    def test_import_remove(self):
     760        # We perform the same import twice,
     761        # the second time in remove mode. The number
     762        # of warnings must be the same.
     763        num, num_warns, fin_file, fail_file = self.processor.doImport(
     764            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
     765        shutil.rmtree(os.path.dirname(fin_file))
     766        assert self.processor.entryExists(
     767            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
     768        num, num_warns, fin_file, fail_file = self.processor.doImport(
     769            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
     770        self.assertEqual(num_warns,2)
     771        assert self.processor.entryExists(
     772            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
     773        shutil.rmtree(os.path.dirname(fin_file))
     774        logcontent = open(self.logfile).read()
     775        self.assertTrue(
     776            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
     777            in logcontent)
     778
     779class PaymentProcessorTest(StudentImportExportSetup):
     780
     781    def setUp(self):
     782        super(PaymentProcessorTest, self).setUp()
     783
     784        # Add student with payment
     785        student = Student()
     786        student.firstname = u'Anna'
     787        student.lastname = u'Tester'
     788        student.reg_number = u'123'
     789        student.matric_number = u'234'
     790        self.app['students'].addStudent(student)
     791        self.student = self.app['students'][student.student_id]
     792        payment = createObject(u'waeup.StudentOnlinePayment')
     793        payment.p_id = 'p120'
     794        self.student['payments'][payment.p_id] = payment
     795
     796        # Import students with subobjects
     797        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
     798        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
     799        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
     800            student_file, STUDENT_HEADER_FIELDS)
     801        shutil.rmtree(os.path.dirname(fin_file))
     802
     803        self.processor = StudentOnlinePaymentProcessor()
     804        self.csv_file = os.path.join(
     805            self.workdir, 'sample_payment_data.csv')
     806        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
     807        self.csv_file2 = os.path.join(
     808            self.workdir, 'sample_create_payment_data.csv')
     809        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
     810
     811    def test_interface(self):
     812        # Make sure we fulfill the interface contracts.
     813        assert verifyObject(IBatchProcessor, self.processor) is True
     814        assert verifyClass(
     815            IBatchProcessor, StudentOnlinePaymentProcessor) is True
     816
     817    def test_getEntry(self):
     818        assert self.processor.getEntry(
     819            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
     820        assert self.processor.getEntry(
     821            dict(student_id=self.student.student_id, p_id='p120'),
     822            self.app) is self.student['payments']['p120']
     823        assert self.processor.getEntry(
     824            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
     825            self.app) is self.student['payments']['p120']
     826
     827    def test_delEntry(self):
     828        assert self.processor.getEntry(
     829            dict(student_id=self.student.student_id, p_id='p120'),
     830            self.app) is self.student['payments']['p120']
     831        self.assertEqual(len(self.student['payments'].keys()),1)
     832        self.processor.delEntry(
     833            dict(student_id=self.student.student_id, p_id='p120'),
     834            self.app)
     835        assert self.processor.getEntry(
     836            dict(student_id=self.student.student_id, p_id='p120'),
     837            self.app) is None
     838        self.assertEqual(len(self.student['payments'].keys()),0)
     839
     840    def test_addEntry(self):
     841        self.assertEqual(len(self.student['payments'].keys()),1)
     842        payment1 = createObject(u'waeup.StudentOnlinePayment')
     843        payment1.p_id = 'p234'
     844        self.processor.addEntry(
     845            payment1, dict(student_id=self.student.student_id, p_id='p234'),
     846            self.app)
     847        self.assertEqual(len(self.student['payments'].keys()),2)
     848        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
     849        payment2 = createObject(u'waeup.StudentOnlinePayment')
     850        payment1.p_id = 'nonsense'
     851        # payment1.p_id will be replaced if p_id doesn't start with 'p'
     852        self.processor.addEntry(
     853            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
     854            self.app)
     855        self.assertEqual(len(self.student['payments'].keys()),3)
     856        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
     857
     858    def test_checkConversion(self):
     859        errs, inv_errs, conv_dict = self.processor.checkConversion(
     860            dict(p_id='3816951266236341955'))
     861        self.assertEqual(len(errs),0)
     862        errs, inv_errs, conv_dict = self.processor.checkConversion(
     863            dict(p_id='p1266236341955'))
     864        self.assertEqual(len(errs),0)
     865        errs, inv_errs, conv_dict = self.processor.checkConversion(
     866            dict(p_id='nonsense'))
     867        self.assertEqual(len(errs),1)
     868        timestamp = ("%d" % int(time()*10000))[1:]
     869        p_id = "p%s" % timestamp
     870        errs, inv_errs, conv_dict = self.processor.checkConversion(
     871            dict(p_id=p_id))
     872        self.assertEqual(len(errs),0)
     873
     874    def test_import(self):
     875        num, num_warns, fin_file, fail_file = self.processor.doImport(
     876            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
     877        self.assertEqual(num_warns,0)
     878        payment = self.processor.getEntry(dict(reg_number='1',
     879            p_id='p2907979737440'), self.app)
     880        self.assertEqual(payment.p_id, 'p2907979737440')
     881        self.assertTrue(payment.p_current)
     882        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
     883        self.assertEqual(cdate, "2010-11-26 18:59:33")
     884        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
     885        payment = self.processor.getEntry(dict(matric_number='100001',
     886            p_id='p2907125937570'), self.app)
     887        self.assertEqual(payment.p_id, 'p2907125937570')
     888        self.assertFalse(payment.p_current)
     889        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
     890        # Ooooh, still the old problem, see
     891        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
     892        # WAT is interpreted as GMT-1 and not GMT+1
     893        self.assertEqual(cdate, "2010-11-25 21:16:33")
     894        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
     895        shutil.rmtree(os.path.dirname(fin_file))
     896        logcontent = open(self.logfile).read()
     897        # Logging message from updateEntry
     898        self.assertTrue(
     899            'INFO - system - K1000001 - Payment ticket updated: '
     900            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
     901            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
     902            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
     903            'p_state=paid'
     904            in logcontent)
     905
     906    def test_import_update(self):
     907        # We perform the same import twice,
     908        # the second time in update mode. The number
     909        # of warnings must be the same.
     910        num, num_warns, fin_file, fail_file = self.processor.doImport(
     911            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
     912        shutil.rmtree(os.path.dirname(fin_file))
     913        num, num_warns, fin_file, fail_file = self.processor.doImport(
     914            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
     915        self.assertEqual(num_warns,0)
     916        shutil.rmtree(os.path.dirname(fin_file))
     917
     918    def test_import_remove(self):
     919        num, num_warns, fin_file, fail_file = self.processor.doImport(
     920            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
     921        shutil.rmtree(os.path.dirname(fin_file))
     922        num, num_warns, fin_file, fail_file = self.processor.doImport(
     923            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
     924        self.assertEqual(num_warns,0)
     925        shutil.rmtree(os.path.dirname(fin_file))
     926        logcontent = open(self.logfile).read()
     927        self.assertTrue(
     928            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
     929            in logcontent)
     930
     931    def test_import_wo_pid(self):
     932        num, num_warns, fin_file, fail_file = self.processor.doImport(
     933            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
     934        self.assertEqual(num_warns,0)
     935        shutil.rmtree(os.path.dirname(fin_file))
     936        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
     937
    576938class StudentVerdictProcessorTest(StudentImportExportSetup):
    577939
     
    6401002            '2008,A,True,100007,200,Study level object is missing.\r\n'
    6411003            )
    642         shutil.rmtree(os.path.dirname(fin_file))
    643 
    644 
    645 class StudentStudyLevelProcessorTest(StudentImportExportSetup):
    646 
    647     def setUp(self):
    648         super(StudentStudyLevelProcessorTest, self).setUp()
    649 
    650         # Import students with subobjects
    651         student_file = os.path.join(self.workdir, 'sample_student_data.csv')
    652         open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
    653         num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
    654             student_file, STUDENT_HEADER_FIELDS)
    655         shutil.rmtree(os.path.dirname(fin_file))
    656 
    657         # Update study courses
    658         studycourse_file = os.path.join(
    659             self.workdir, 'sample_studycourse_data.csv')
    660         open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
    661         processor = StudentStudyCourseProcessor()
    662         num, num_warns, fin_file, fail_file = processor.doImport(
    663             studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
    664         shutil.rmtree(os.path.dirname(fin_file))
    665 
    666         self.processor = StudentStudyLevelProcessor()
    667         self.csv_file = os.path.join(
    668             self.workdir, 'sample_studylevel_data.csv')
    669         open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
    670 
    671     def test_interface(self):
    672         # Make sure we fulfill the interface contracts.
    673         assert verifyObject(IBatchProcessor, self.processor) is True
    674         assert verifyClass(
    675             IBatchProcessor, StudentStudyLevelProcessor) is True
    676 
    677     def test_checkConversion(self):
    678         errs, inv_errs, conv_dict = self.processor.checkConversion(
    679             dict(reg_number='1', level='220'))
    680         self.assertEqual(len(errs),0)
    681         errs, inv_errs, conv_dict = self.processor.checkConversion(
    682             dict(reg_number='1', level='900'))
    683         self.assertEqual(len(errs),1)
    684         self.assertTrue(('level','no valid integer') in errs)
    685         errs, inv_errs, conv_dict = self.processor.checkConversion(
    686             dict(reg_number='1', level='xyz'))
    687         self.assertEqual(len(errs),1)
    688         self.assertTrue(('level','no integer') in errs)
    689 
    690     def test_import(self):
    691         num, num_warns, fin_file, fail_file = self.processor.doImport(
    692             self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
    693         self.assertEqual(num_warns,2)
    694         assert self.processor.entryExists(
    695             dict(reg_number='1', level='100'), self.app) is True
    696         studylevel = self.processor.getEntry(
    697             dict(reg_number='1', level='100'), self.app)
    698         self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
    699         self.assertEqual(studylevel.level_session, 2008)
    700         self.assertEqual(studylevel.level_verdict, None)
    701         self.assertEqual(studylevel.level, 100)
    702         shutil.rmtree(os.path.dirname(fin_file))
    703 
    7041004        logcontent = open(self.logfile).read()
    705         # Logging message from updateEntry,
    706         self.assertTrue(
    707             'INFO - system - K1000000 - Study level '
    708             'updated: level=100, level_verdict=C, level_session=2009'
    709             in logcontent)
    710 
    711     def test_import_update(self):
    712         # We perform the same import twice,
    713         # the second time in update mode. The number
    714         # of warnings must be the same.
    715         num, num_warns, fin_file, fail_file = self.processor.doImport(
    716             self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
    717         shutil.rmtree(os.path.dirname(fin_file))
    718         num, num_warns, fin_file, fail_file = self.processor.doImport(
    719             self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
    720         self.assertEqual(num_warns,2)
    721         shutil.rmtree(os.path.dirname(fin_file))
    722 
    723 class CourseTicketProcessorTest(StudentImportExportSetup):
    724 
    725     def setUp(self):
    726         super(CourseTicketProcessorTest, self).setUp()
    727 
    728         # Import students with subobjects
    729         student_file = os.path.join(self.workdir, 'sample_student_data.csv')
    730         open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
    731         num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
    732             student_file, STUDENT_HEADER_FIELDS)
    733         shutil.rmtree(os.path.dirname(fin_file))
    734 
    735         # Add course and certificate course
    736         self.course = createObject('waeup.Course')
    737         self.course.code = 'COURSE1'
    738         self.course.semester = 1
    739         self.course.credits = 10
    740         self.course.passmark = 40
    741         self.app['faculties']['fac1']['dep1'].courses.addCourse(
    742             self.course)
    743         self.app['faculties']['fac1']['dep1'].certificates[
    744             'CERT1'].addCertCourse(
    745             self.course, level=100)
    746 
    747         # Update study courses
    748         studycourse_file = os.path.join(
    749             self.workdir, 'sample_studycourse_data.csv')
    750         open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
    751         processor = StudentStudyCourseProcessor()
    752         num, num_warns, fin_file, fail_file = processor.doImport(
    753             studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
    754         shutil.rmtree(os.path.dirname(fin_file))
    755 
    756         # Import study levels
    757         processor = StudentStudyLevelProcessor()
    758         studylevel_file = os.path.join(
    759             self.workdir, 'sample_studylevel_data.csv')
    760         open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
    761         num, num_warns, fin_file, fail_file = processor.doImport(
    762             studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
    763         shutil.rmtree(os.path.dirname(fin_file))
    764 
    765         self.processor = CourseTicketProcessor()
    766         self.csv_file = os.path.join(
    767             self.workdir, 'sample_courseticket_data.csv')
    768         open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
    769 
    770     def test_interface(self):
    771         # Make sure we fulfill the interface contracts.
    772         assert verifyObject(IBatchProcessor, self.processor) is True
    773         assert verifyClass(
    774             IBatchProcessor, CourseTicketProcessor) is True
    775 
    776     def test_checkConversion(self):
    777         errs, inv_errs, conv_dict = self.processor.checkConversion(
    778             dict(reg_number='1', code='COURSE1', level='220'))
    779         self.assertEqual(len(errs),0)
    780         errs, inv_errs, conv_dict = self.processor.checkConversion(
    781             dict(reg_number='1', code='COURSE2', level='220'))
    782         self.assertEqual(len(errs),1)
    783         self.assertTrue(('code','non-existent') in errs)
    784 
    785     def test_import(self):
    786 
    787         num, num_warns, fin_file, fail_file = self.processor.doImport(
    788             self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
    789 
    790         self.assertEqual(num_warns,2)
    791         assert self.processor.entryExists(
    792             dict(reg_number='1', level='100', code='COURSE1'),
    793             self.app) is True
    794         courseticket = self.processor.getEntry(
    795             dict(reg_number='1', level='100', code='COURSE1'), self.app)
    796         self.assertEqual(courseticket.__parent__.__parent__.certificate.code,
    797                          u'CERT1')
    798         self.assertEqual(courseticket.score, 1)
    799         self.assertEqual(courseticket.mandatory, True)
    800         self.assertEqual(courseticket.fcode, 'NA')
    801         self.assertEqual(courseticket.dcode, 'NA')
    802         self.assertEqual(courseticket.code, 'COURSE1')
    803         self.assertEqual(courseticket.title, 'Unnamed Course')
    804         self.assertEqual(courseticket.credits, 10)
    805         self.assertEqual(courseticket.passmark, 40)
    806         self.assertEqual(courseticket.semester, 1)
    807         shutil.rmtree(os.path.dirname(fin_file))
    808 
    809         logcontent = open(self.logfile).read()
    810         # Logging message from updateEntry,
    811         self.assertTrue(
    812             'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, '
    813             'mandatory=False, score=3'
    814             in logcontent)
    815 
    816     def test_import_update(self):
    817         # We perform the same import twice,
    818         # the second time in update mode. The number
    819         # of warnings must be the same.
    820         num, num_warns, fin_file, fail_file = self.processor.doImport(
    821             self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
    822         shutil.rmtree(os.path.dirname(fin_file))
    823         num, num_warns, fin_file, fail_file = self.processor.doImport(
    824             self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
    825         self.assertEqual(num_warns,2)
    826         shutil.rmtree(os.path.dirname(fin_file))
    827 
    828     def test_import_remove(self):
    829         # We perform the same import twice,
    830         # the second time in remove mode. The number
    831         # of warnings must be the same.
    832         num, num_warns, fin_file, fail_file = self.processor.doImport(
    833             self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
    834         shutil.rmtree(os.path.dirname(fin_file))
    835         assert self.processor.entryExists(
    836             dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
    837         num, num_warns, fin_file, fail_file = self.processor.doImport(
    838             self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
    839         self.assertEqual(num_warns,2)
    840         assert self.processor.entryExists(
    841             dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
    842         shutil.rmtree(os.path.dirname(fin_file))
    843         logcontent = open(self.logfile).read()
    844         self.assertTrue(
    845             'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
    846             in logcontent)
    847 
    848 class PaymentProcessorTest(StudentImportExportSetup):
    849 
    850     def setUp(self):
    851         super(PaymentProcessorTest, self).setUp()
    852 
    853         # Add student with payment
    854         student = Student()
    855         student.firstname = u'Anna'
    856         student.lastname = u'Tester'
    857         student.reg_number = u'123'
    858         student.matric_number = u'234'
    859         self.app['students'].addStudent(student)
    860         self.student = self.app['students'][student.student_id]
    861         payment = createObject(u'waeup.StudentOnlinePayment')
    862         payment.p_id = 'p120'
    863         self.student['payments'][payment.p_id] = payment
    864 
    865         # Import students with subobjects
    866         student_file = os.path.join(self.workdir, 'sample_student_data.csv')
    867         open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
    868         num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
    869             student_file, STUDENT_HEADER_FIELDS)
    870         shutil.rmtree(os.path.dirname(fin_file))
    871 
    872         self.processor = StudentOnlinePaymentProcessor()
    873         self.csv_file = os.path.join(
    874             self.workdir, 'sample_payment_data.csv')
    875         open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
    876         self.csv_file2 = os.path.join(
    877             self.workdir, 'sample_create_payment_data.csv')
    878         open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
    879 
    880     def test_interface(self):
    881         # Make sure we fulfill the interface contracts.
    882         assert verifyObject(IBatchProcessor, self.processor) is True
    883         assert verifyClass(
    884             IBatchProcessor, StudentOnlinePaymentProcessor) is True
    885 
    886     def test_getEntry(self):
    887         assert self.processor.getEntry(
    888             dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
    889         assert self.processor.getEntry(
    890             dict(student_id=self.student.student_id, p_id='p120'),
    891             self.app) is self.student['payments']['p120']
    892         assert self.processor.getEntry(
    893             dict(student_id=self.student.student_id, p_id='XXXXXX112'),
    894             self.app) is self.student['payments']['p120']
    895 
    896     def test_delEntry(self):
    897         assert self.processor.getEntry(
    898             dict(student_id=self.student.student_id, p_id='p120'),
    899             self.app) is self.student['payments']['p120']
    900         self.assertEqual(len(self.student['payments'].keys()),1)
    901         self.processor.delEntry(
    902             dict(student_id=self.student.student_id, p_id='p120'),
    903             self.app)
    904         assert self.processor.getEntry(
    905             dict(student_id=self.student.student_id, p_id='p120'),
    906             self.app) is None
    907         self.assertEqual(len(self.student['payments'].keys()),0)
    908 
    909     def test_addEntry(self):
    910         self.assertEqual(len(self.student['payments'].keys()),1)
    911         payment1 = createObject(u'waeup.StudentOnlinePayment')
    912         payment1.p_id = 'p234'
    913         self.processor.addEntry(
    914             payment1, dict(student_id=self.student.student_id, p_id='p234'),
    915             self.app)
    916         self.assertEqual(len(self.student['payments'].keys()),2)
    917         self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
    918         payment2 = createObject(u'waeup.StudentOnlinePayment')
    919         payment1.p_id = 'nonsense'
    920         # payment1.p_id will be replaced if p_id doesn't start with 'p'
    921         self.processor.addEntry(
    922             payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
    923             self.app)
    924         self.assertEqual(len(self.student['payments'].keys()),3)
    925         self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
    926 
    927     def test_checkConversion(self):
    928         errs, inv_errs, conv_dict = self.processor.checkConversion(
    929             dict(p_id='3816951266236341955'))
    930         self.assertEqual(len(errs),0)
    931         errs, inv_errs, conv_dict = self.processor.checkConversion(
    932             dict(p_id='p1266236341955'))
    933         self.assertEqual(len(errs),0)
    934         errs, inv_errs, conv_dict = self.processor.checkConversion(
    935             dict(p_id='nonsense'))
    936         self.assertEqual(len(errs),1)
    937         timestamp = ("%d" % int(time()*10000))[1:]
    938         p_id = "p%s" % timestamp
    939         errs, inv_errs, conv_dict = self.processor.checkConversion(
    940             dict(p_id=p_id))
    941         self.assertEqual(len(errs),0)
    942 
    943     def test_import(self):
    944         num, num_warns, fin_file, fail_file = self.processor.doImport(
    945             self.csv_file, PAYMENT_HEADER_FIELDS,'create')
    946         self.assertEqual(num_warns,0)
    947         payment = self.processor.getEntry(dict(reg_number='1',
    948             p_id='p2907979737440'), self.app)
    949         self.assertEqual(payment.p_id, 'p2907979737440')
    950         self.assertTrue(payment.p_current)
    951         cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
    952         self.assertEqual(cdate, "2010-11-26 18:59:33")
    953         self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
    954         payment = self.processor.getEntry(dict(matric_number='100001',
    955             p_id='p2907125937570'), self.app)
    956         self.assertEqual(payment.p_id, 'p2907125937570')
    957         self.assertFalse(payment.p_current)
    958         cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
    959         # Ooooh, still the old problem, see
    960         # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
    961         # WAT is interpreted as GMT-1 and not GMT+1
    962         self.assertEqual(cdate, "2010-11-25 21:16:33")
    963         self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
    964         shutil.rmtree(os.path.dirname(fin_file))
    965         logcontent = open(self.logfile).read()
    966         # Logging message from updateEntry
    967         self.assertTrue(
    968             'INFO - system - K1000001 - Payment ticket updated: '
    969             'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
    970             'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
    971             'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
    972             'p_state=paid'
    973             in logcontent)
    974 
    975     def test_import_update(self):
    976         # We perform the same import twice,
    977         # the second time in update mode. The number
    978         # of warnings must be the same.
    979         num, num_warns, fin_file, fail_file = self.processor.doImport(
    980             self.csv_file, PAYMENT_HEADER_FIELDS,'create')
    981         shutil.rmtree(os.path.dirname(fin_file))
    982         num, num_warns, fin_file, fail_file = self.processor.doImport(
    983             self.csv_file, PAYMENT_HEADER_FIELDS,'update')
    984         self.assertEqual(num_warns,0)
    985         shutil.rmtree(os.path.dirname(fin_file))
    986 
    987     def test_import_remove(self):
    988         num, num_warns, fin_file, fail_file = self.processor.doImport(
    989             self.csv_file, PAYMENT_HEADER_FIELDS,'create')
    990         shutil.rmtree(os.path.dirname(fin_file))
    991         num, num_warns, fin_file, fail_file = self.processor.doImport(
    992             self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
    993         self.assertEqual(num_warns,0)
    994         shutil.rmtree(os.path.dirname(fin_file))
    995         logcontent = open(self.logfile).read()
    996         self.assertTrue(
    997             'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
    998             in logcontent)
    999 
    1000     def test_import_wo_pid(self):
    1001         num, num_warns, fin_file, fail_file = self.processor.doImport(
    1002             self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
    1003         self.assertEqual(num_warns,0)
    1004         shutil.rmtree(os.path.dirname(fin_file))
    1005         self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
    1006 
     1005        self.assertMatches(
     1006            '... INFO - system - X666666 - Study course updated: current_verdict=0...',
     1007            logcontent)
     1008        self.assertMatches(
     1009            '... INFO - system - X666666 - Returned...',
     1010            logcontent)
     1011
     1012        shutil.rmtree(os.path.dirname(fin_file))
    10071013
    10081014def test_suite():
Note: See TracChangeset for help on using the changeset viewer.