## $Id: test_batching.py 8354 2012-05-04 23:26:58Z uli $ ## ## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2 of the License, or ## (at your option) any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License ## along with this program; if not, write to the Free Software ## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ## """Unit tests for students-related data processors. """ import os import shutil import tempfile import unittest import datetime from time import time from zope.component import createObject from zope.component.hooks import setSite, clearSite from zope.interface.verify import verifyClass, verifyObject from waeup.kofa.app import University from waeup.kofa.interfaces import IBatchProcessor, FatalCSVError, IUserAccount from waeup.kofa.students.batching import ( StudentProcessor, StudentStudyCourseProcessor, StudentStudyLevelProcessor, CourseTicketProcessor, StudentOnlinePaymentProcessor, StudentVerdictProcessor) from waeup.kofa.students.student import Student from waeup.kofa.students.studylevel import StudentStudyLevel, CourseTicket from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase from waeup.kofa.university.faculty import Faculty from waeup.kofa.university.department import Department STUDENT_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'), 'rb').read() STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split( '\n')[0].split(',') STUDENT_SAMPLE_DATA_UPDATE = open( os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'), 'rb').read() STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split( '\n')[0].split(',') STUDENT_SAMPLE_DATA_UPDATE2 = open( os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'), 'rb').read() STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split( '\n')[0].split(',') STUDENT_SAMPLE_DATA_UPDATE3 = open( os.path.join(os.path.dirname(__file__), 'sample_student_data_update3.csv'), 'rb').read() STUDENT_HEADER_FIELDS_UPDATE3 = STUDENT_SAMPLE_DATA_UPDATE3.split( '\n')[0].split(',') STUDENT_SAMPLE_DATA_UPDATE4 = open( os.path.join(os.path.dirname(__file__), 'sample_student_data_update4.csv'), 'rb').read() STUDENT_HEADER_FIELDS_UPDATE4 = STUDENT_SAMPLE_DATA_UPDATE4.split( '\n')[0].split(',') STUDYCOURSE_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'), 'rb').read() STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split( '\n')[0].split(',') VERDICT_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_verdict_data.csv'), 'rb').read() VERDICT_HEADER_FIELDS = VERDICT_SAMPLE_DATA.split( '\n')[0].split(',') STUDENT_SAMPLE_DATA_MIGRATION = open( os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'), 'rb').read() STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split( '\n')[0].split(',') STUDENT_SAMPLE_DATA_DUPLICATES = open( os.path.join(os.path.dirname(__file__), 'sample_student_data_duplicates.csv'), 'rb').read() STUDENT_HEADER_FIELDS_DUPLICATES = STUDENT_SAMPLE_DATA_DUPLICATES.split( '\n')[0].split(',') STUDYLEVEL_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'), 'rb').read() STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split( '\n')[0].split(',') COURSETICKET_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'), 'rb').read() 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(',') class StudentImportExportSetup(FunctionalTestCase): layer = FunctionalLayer def setUp(self): super(StudentImportExportSetup, 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) # Populate university self.certificate = createObject('waeup.Certificate') self.certificate.code = 'CERT1' self.certificate.application_category = 'basic' self.certificate.start_level = 200 self.certificate.end_level = 500 self.app['faculties']['fac1'] = Faculty() self.app['faculties']['fac1']['dep1'] = Department() self.app['faculties']['fac1']['dep1'].certificates.addCertificate( self.certificate) return def tearDown(self): super(StudentImportExportSetup, self).tearDown() shutil.rmtree(self.workdir) shutil.rmtree(self.dc_root) clearSite() return def setup_for_export(self): student = Student() student.student_id = u'A111111' self.app['students'][student.student_id] = self.student = student self.outfile = os.path.join(self.workdir, 'myoutput.csv') return def setup_student(self, student): # set predictable values for `student` 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'123456' student.student_id = u'A111111' student.firstname = u'Anna' student.lastname = u'Tester' student.middlename = u'M.' student.date_of_birth = datetime.date(1981, 2, 4) student.sex = 'f' student.email = 'anna@sample.com' student.phone = u'+234-123-12345' student.notice = u'Some notice\nin lines.' student.nationality = u'NG' student['studycourse'].certificate = self.certificate student['studycourse'].entry_mode = 'ug_ft' student['studycourse'].entry_session = 2010 student['studycourse'].current_session = 2012 student['studycourse'].current_level = int(self.certificate.start_level) study_level = StudentStudyLevel() study_level.level_session = 2012 study_level.level_verdict = "A" study_level.level = 100 student['studycourse'].addStudentStudyLevel( self.certificate, study_level) ticket = CourseTicket() ticket.automatic = True ticket.carry_over = True ticket.code = u'CRS1' ticket.title = u'Course 1' ticket.fcode = u'FAC1' ticket.dcode = u'DEP1' ticket.credits = 100 ticket.passmark = 100 ticket.semester = 2 study_level.addCourseTicket(ticket) return student class StudentProcessorTest(FunctionalTestCase): layer = FunctionalLayer 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) self.student = self.app['students'][student.student_id] self.processor = StudentProcessor() self.workdir = tempfile.mkdtemp() self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv') self.csv_file_update = os.path.join( self.workdir, 'sample_student_data_update.csv') self.csv_file_update2 = os.path.join( self.workdir, 'sample_student_data_update2.csv') self.csv_file_update3 = os.path.join( self.workdir, 'sample_student_data_update3.csv') self.csv_file_update4 = os.path.join( self.workdir, 'sample_student_data_update4.csv') self.csv_file_migration = os.path.join( self.workdir, 'sample_student_data_migration.csv') self.csv_file_duplicates = os.path.join( self.workdir, 'sample_student_data_duplicates.csv') open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA) open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE) open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2) open(self.csv_file_update3, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE3) open(self.csv_file_update4, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE4) open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION) 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. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, StudentProcessor) is True def test_parentsExist(self): self.assertFalse(self.processor.parentsExist(None, dict())) self.assertTrue(self.processor.parentsExist(None, self.app)) def test_entryExists(self): assert self.processor.entryExists( dict(student_id='ID_NONE'), self.app) is False assert self.processor.entryExists( dict(reg_number='123'), self.app) is True def test_getParent(self): parent = self.processor.getParent(None, self.app) assert parent is self.app['students'] def test_getEntry(self): assert self.processor.getEntry( dict(student_id='ID_NONE'), self.app) is None assert self.processor.getEntry( dict(student_id=self.student.student_id), self.app) is self.student def test_addEntry(self): new_student = Student() self.processor.addEntry( new_student, dict(), self.app) assert len(self.app['students'].keys()) == 2 def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', state='admitted')) self.assertEqual(len(errs),0) # Empty state is allowed errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', state='')) self.assertEqual(len(errs),0) #self.assertTrue(('state', 'no value provided') in errs) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', state='nonsense')) self.assertEqual(len(errs),1) self.assertTrue(('state', 'not allowed') in errs) def test_delEntry(self): assert self.student.student_id in self.app['students'].keys() self.processor.delEntry( dict(reg_number=self.student.reg_number), self.app) assert self.student.student_id not in self.app['students'].keys() def test_import(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDENT_HEADER_FIELDS) self.assertEqual(num_warns,0) assert len(self.app['students'].keys()) == 5 self.assertEqual(self.app['students']['X666666'].reg_number,'1') self.assertEqual( self.app['students']['X666666'].state, 'courses validated') shutil.rmtree(os.path.dirname(fin_file)) def test_import_update(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDENT_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update') self.assertEqual(num_warns,0) # state has changed self.assertEqual(self.app['students']['X666666'].state,'admitted') # state has not changed self.assertEqual(self.app['students']['Y777777'].state,'courses validated') shutil.rmtree(os.path.dirname(fin_file)) def test_import_update2(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDENT_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update') self.assertEqual(num_warns,0) # The phone import value of Pieri was None. # Confirm that phone has not been cleared. container = self.app['students'] for key in container.keys(): if container[key].firstname == 'Aaren': aaren = container[key] break self.assertEqual(aaren.phone, '--1234') # The phone import value of Claus was a deletion marker. # Confirm that phone has been cleared. for key in container.keys(): if container[key].firstname == 'Claus': claus = container[key] break assert claus.phone is None shutil.rmtree(os.path.dirname(fin_file)) def test_import_update3(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDENT_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_update3, STUDENT_HEADER_FIELDS_UPDATE3, 'update') content = open(fail_file).read() self.assertEqual( content, 'reg_number,student_id,transition,--ERRORS--\r\n' ',X666666,request_clearance,Transition not allowed.\r\n' ) self.assertEqual(num_warns,1) self.assertEqual(self.app['students']['Y777777'].state,'returning') def test_import_update4(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDENT_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) self.assertRaises( FatalCSVError, self.processor.doImport, self.csv_file_update4, STUDENT_HEADER_FIELDS_UPDATE4, 'update') def test_import_remove(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDENT_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove') self.assertEqual(num_warns,0) shutil.rmtree(os.path.dirname(fin_file)) def test_import_migration_data(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION) content = open(fail_file).read() self.assertEqual(num_warns,2) assert len(self.app['students'].keys()) == 5 self.assertEqual( content, 'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n' '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n' '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n' ) students = self.app['students'] self.assertTrue('A123456' in students.keys()) self.assertEqual(students['A123456'].state, 'clearance started') self.assertEqual(students['A123456'].date_of_birth, datetime.date(1990, 1, 2)) self.assertFalse(students['A123456'].clearance_locked) self.assertEqual(students['B123456'].state, 'cleared') self.assertEqual(students['B123456'].date_of_birth, datetime.date(1990, 1, 3)) self.assertTrue(students['B123456'].clearance_locked) history = ' '.join(students['A123456'].history.messages) self.assertTrue( "State 'clearance started' set by system" in history) # state was empty and student is thus in state created self.assertEqual(students['F123456'].state,'created') # passwords were set correctly self.assertEqual( IUserAccount(students['A123456']).checkPassword('mypw1'), True) self.assertEqual( IUserAccount(students['C123456']).checkPassword('mypw1'), True) shutil.rmtree(os.path.dirname(fin_file)) def test_import_duplicate_data(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES) content = open(fail_file).read() self.assertEqual(num_warns,3) self.assertEqual( content, 'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n' '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n' '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n' '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n' ) shutil.rmtree(os.path.dirname(fin_file)) class StudentStudyCourseProcessorTest(StudentImportExportSetup): def setUp(self): super(StudentStudyCourseProcessorTest, self).setUp() # 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.processor = StudentStudyCourseProcessor() self.csv_file = os.path.join( self.workdir, 'sample_studycourse_data.csv') open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) return def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, StudentStudyCourseProcessor) is True def test_entryExists(self): assert self.processor.entryExists( dict(reg_number='REG_NONE'), self.app) is False assert self.processor.entryExists( dict(reg_number='1'), self.app) is True def test_getEntry(self): student = self.processor.getEntry( dict(reg_number='1'), self.app).__parent__ self.assertEqual(student.reg_number,'1') def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', certificate='CERT1', current_level='200')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', certificate='CERT999')) self.assertEqual(len(errs),1) self.assertTrue(('certificate', u'Invalid value') in errs) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', certificate='CERT1', current_level='100')) self.assertEqual(len(errs),1) self.assertTrue(('current_level','not in range') in errs) # If we import only current_level, no conversion checking is done. errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', current_level='100')) self.assertEqual(len(errs),0) def test_import(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update') self.assertEqual(num_warns,1) content = open(fail_file).read() self.assertTrue('current_level: not in range' in content) studycourse = self.processor.getEntry(dict(reg_number='1'), self.app) self.assertEqual(studycourse.certificate.code, u'CERT1') shutil.rmtree(os.path.dirname(fin_file)) class StudentVerdictProcessorTest(StudentImportExportSetup): def setUp(self): super(StudentVerdictProcessorTest, self).setUp() # 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)) # Update study courses studycourse_file = os.path.join( self.workdir, 'sample_studycourse_data.csv') open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) processor = StudentStudyCourseProcessor() num, num_warns, fin_file, fail_file = processor.doImport( studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') shutil.rmtree(os.path.dirname(fin_file)) self.processor = StudentVerdictProcessor() self.csv_file = os.path.join( self.workdir, 'sample_verdict_data.csv') open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA) return def test_import(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, VERDICT_HEADER_FIELDS,'update') self.assertEqual(num_warns,3) studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app) student = self.processor.getParent(dict(matric_number='100000'), self.app) self.assertEqual(studycourse.current_verdict, 'A') self.assertEqual(student.state, 'returning') self.assertEqual(studycourse.current_level, 200) content = open(fail_file).read() self.assertEqual( content, 'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n' '2008,B,100001,100,Current level does not correspond.\r\n' '2007,C,100002,200,Current session does not correspond.\r\n' '2008,A,100003,200,Student in wrong state.\r\n' ) shutil.rmtree(os.path.dirname(fin_file)) class StudentStudyLevelProcessorTest(StudentImportExportSetup): def setUp(self): super(StudentStudyLevelProcessorTest, self).setUp() # 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)) # Update study courses studycourse_file = os.path.join( self.workdir, 'sample_studycourse_data.csv') open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) processor = StudentStudyCourseProcessor() num, num_warns, fin_file, fail_file = processor.doImport( studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') shutil.rmtree(os.path.dirname(fin_file)) self.processor = StudentStudyLevelProcessor() self.csv_file = os.path.join( self.workdir, 'sample_studylevel_data.csv') open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, StudentStudyLevelProcessor) is True def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', level='220')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', level='900')) self.assertEqual(len(errs),1) self.assertTrue(('level','no valid integer') in errs) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', level='xyz')) self.assertEqual(len(errs),1) self.assertTrue(('level','no integer') in errs) def test_import(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') self.assertEqual(num_warns,2) assert self.processor.entryExists( dict(reg_number='1', level='100'), self.app) is True studylevel = self.processor.getEntry( dict(reg_number='1', level='100'), self.app) self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1') self.assertEqual(studylevel.level_session, 2008) self.assertEqual(studylevel.level_verdict, 'A') self.assertEqual(studylevel.level, 100) shutil.rmtree(os.path.dirname(fin_file)) def test_import_update(self): # We perform the same import twice, # the second time in update mode. The number # of warnings must be the same. num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update') self.assertEqual(num_warns,2) shutil.rmtree(os.path.dirname(fin_file)) class CourseTicketProcessorTest(StudentImportExportSetup): def setUp(self): super(CourseTicketProcessorTest, self).setUp() # 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)) # Add course and course referrer self.course = createObject('waeup.Course') self.course.code = 'COURSE1' self.course.semester = 1 self.course.credits = 10 self.course.passmark = 40 self.app['faculties']['fac1']['dep1'].courses.addCourse( self.course) self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef( self.course, level=100) # Update study courses studycourse_file = os.path.join( self.workdir, 'sample_studycourse_data.csv') open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA) processor = StudentStudyCourseProcessor() num, num_warns, fin_file, fail_file = processor.doImport( studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update') shutil.rmtree(os.path.dirname(fin_file)) # Import study levels processor = StudentStudyLevelProcessor() studylevel_file = os.path.join( self.workdir, 'sample_studylevel_data.csv') open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) num, num_warns, fin_file, fail_file = processor.doImport( studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) self.processor = CourseTicketProcessor() self.csv_file = os.path.join( self.workdir, 'sample_courseticket_data.csv') open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA) def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, CourseTicketProcessor) is True def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', code='COURSE1', level='220')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', code='COURSE2', level='220')) self.assertEqual(len(errs),1) self.assertTrue(('code','non-existent') in errs) def test_import(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, COURSETICKET_HEADER_FIELDS,'create') self.assertEqual(num_warns,2) assert self.processor.entryExists( dict(reg_number='1', level='100', code='COURSE1'), self.app) is True courseticket = self.processor.getEntry( dict(reg_number='1', level='100', code='COURSE1'), self.app) self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1') self.assertEqual(courseticket.score, 1) self.assertEqual(courseticket.mandatory, True) self.assertEqual(courseticket.fcode, 'NA') self.assertEqual(courseticket.dcode, 'NA') self.assertEqual(courseticket.code, 'COURSE1') self.assertEqual(courseticket.title, 'Unnamed Course') self.assertEqual(courseticket.credits, 10) self.assertEqual(courseticket.passmark, 40) self.assertEqual(courseticket.semester, 1) shutil.rmtree(os.path.dirname(fin_file)) def test_import_update(self): # We perform the same import twice, # the second time in update 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') num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, COURSETICKET_HEADER_FIELDS,'update') self.assertEqual(num_warns,2) shutil.rmtree(os.path.dirname(fin_file)) class PaymentProcessorTest(StudentImportExportSetup): def setUp(self): super(PaymentProcessorTest, self).setUp() # 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.processor = StudentOnlinePaymentProcessor() self.csv_file = os.path.join( self.workdir, 'sample_payment_data.csv') open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA) def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.processor) is True assert verifyClass( IBatchProcessor, StudentOnlinePaymentProcessor) is True def test_getEntry(self): assert self.processor.getEntry( 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'] assert self.processor.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.processor.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.processor.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.processor.checkConversion( dict(reg_number='1', p_id='3816951266236341955')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', p_id='p1266236341955')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.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.processor.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.processor.doImport( self.csv_file, PAYMENT_HEADER_FIELDS,'create') self.assertEqual(num_warns,0) payment = self.processor.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 18:59:33") self.assertEqual(str(payment.creation_date.tzinfo),'UTC') shutil.rmtree(os.path.dirname(fin_file)) def test_import_update(self): # We perform the same import twice, # the second time in update mode. The number # of warnings must be the same. 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,'update') self.assertEqual(num_warns,0) shutil.rmtree(os.path.dirname(fin_file)) def test_suite(): suite = unittest.TestSuite() for testcase in [ StudentProcessorTest,StudentStudyCourseProcessorTest, StudentStudyLevelProcessorTest,CourseTicketProcessorTest, PaymentProcessorTest,StudentVerdictProcessorTest]: suite.addTest(unittest.TestLoader().loadTestsFromTestCase( testcase ) ) return suite