# -*- coding: utf-8 -*- ## $Id: test_batching.py 14901 2017-11-16 07:02:47Z henrik $ ## ## 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 import grok from time import time from zope.event import notify from zope.component import createObject, queryUtility from zope.component.hooks import setSite, clearSite from zope.catalog.interfaces import ICatalog from zope.interface.verify import verifyClass, verifyObject from hurry.workflow.interfaces import IWorkflowState from waeup.kofa.app import University from waeup.kofa.interfaces import ( IBatchProcessor, FatalCSVError, IUserAccount, DuplicationError) from waeup.kofa.students.batching import ( StudentProcessor, StudentStudyCourseProcessor, StudentStudyLevelProcessor, CourseTicketProcessor, StudentOnlinePaymentProcessor, StudentVerdictProcessor) from waeup.kofa.students.payments import StudentOnlinePayment from waeup.kofa.students.student import Student from waeup.kofa.students.studylevel import StudentStudyLevel, CourseTicket from waeup.kofa.students.accommodation import BedTicket from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase from waeup.kofa.university.faculty import Faculty from waeup.kofa.university.department import Department from waeup.kofa.hostels.hostel import Hostel, Bed, NOT_OCCUPIED 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(',') TRANSFER_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_transfer_data.csv'), 'rb').read() TRANSFER_HEADER_FIELDS = TRANSFER_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(',') STUDENT_SAMPLE_DATA_EXTASCII = open( os.path.join(os.path.dirname(__file__), 'sample_student_data_extascii.csv'), 'rb').read() STUDENT_HEADER_FIELDS_EXTASCII = STUDENT_SAMPLE_DATA_EXTASCII.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(',') 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(',') curr_year = datetime.datetime.now().year 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.certificate.study_mode = u'ug_ft' self.app['faculties']['fac1'] = Faculty() self.app['faculties']['fac1']['dep1'] = Department() self.app['faculties']['fac1']['dep1'].certificates.addCertificate( self.certificate) # Create a hostel with two beds hostel = Hostel() hostel.hostel_id = u'hall-1' hostel.hostel_name = u'Hall 1' self.app['hostels'].addHostel(hostel) bed = Bed() bed.bed_id = u'hall-1_A_101_A' bed.bed_number = 1 bed.owner = NOT_OCCUPIED bed.bed_type = u'regular_male_fr' self.app['hostels'][hostel.hostel_id].addBed(bed) bed = Bed() bed.bed_id = u'hall-1_A_101_B' bed.bed_number = 2 bed.owner = NOT_OCCUPIED bed.bed_type = u'regular_female_fr' self.app['hostels'][hostel.hostel_id].addBed(bed) self.logfile = os.path.join( self.app['datacenter'].storage, 'logs', 'students.log') 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'234' student.adm_code = u'my adm code' student.clr_code = u'my clr code' student.perm_address = u'Studentroad 21\nLagos 123456\n' student.reg_number = u'123' 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[ticket.code] = ticket bedticket = BedTicket() bedticket.booking_session = 2004 bedticket.bed_type = u'any bed type' bedticket.bed = self.app['hostels']['hall-1']['hall-1_A_101_A'] student['accommodation'].addBedTicket(bedticket) self.add_payment(student) return student def add_payment(self, student): # get a payment with all fields set payment = StudentOnlinePayment() payment.creation_date = datetime.datetime(curr_year-6, 4, 1, 13, 12, 1) payment.p_id = 'my-id' payment.p_category = u'schoolfee' payment.p_state = 'paid' payment.ac = u'666' payment.p_item = u'p-item' payment.p_level = 100 payment.p_session = curr_year - 6 payment.payment_date = datetime.datetime(curr_year-6, 4, 1, 14, 12, 1) payment.amount_auth = 12.12 payment.r_amount_approved = 12.12 payment.r_code = u'r-code' # XXX: there is no addPayment method to give predictable names self.payment = student['payments']['my-payment'] = payment return payment class StudentProcessorTest(StudentImportExportSetup): layer = FunctionalLayer def setUp(self): super(StudentProcessorTest, 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] self.processor = StudentProcessor() 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') self.csv_file_extascii = os.path.join( self.workdir, 'sample_student_data_extascii.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) open(self.csv_file_extascii, 'wb').write(STUDENT_SAMPLE_DATA_EXTASCII) 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): # Make sure we can check conversions and that the stud_id # counter is not raised during such checks. initial_stud_id = self.app['students']._curr_stud_id 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) new_stud_id = self.app['students']._curr_stud_id 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): 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): self.assertEqual(self.app['students']._curr_stud_id, 1000001) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDENT_HEADER_FIELDS) self.assertEqual(num_warns, 0) # Nine students have been added. self.assertEqual(len(self.app['students']), 10) # Three empty rows have been skipped. self.assertEqual(num, 12) self.assertEqual(self.app['students']['X666666'].reg_number,'1') self.assertEqual( self.app['students']['X666666'].state, 'courses validated') # Two new student_ids have been created. self.assertEqual(self.app['students']._curr_stud_id, 1000003) shutil.rmtree(os.path.dirname(fin_file)) def test_import_extascii(self): self.assertEqual(self.app['students']._curr_stud_id, 1000001) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_extascii, STUDENT_HEADER_FIELDS_EXTASCII) self.assertEqual(num_warns,0) self.assertEqual(len(self.app['students']), 3) self.assertEqual(self.app['students']['X111111'].reg_number,'1') 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() shutil.rmtree(os.path.dirname(fin_file)) self.assertEqual( content, 'reg_number,student_id,transition,firstname,--ERRORS--\r\n' ',X666666,request_clearance,,Transition not allowed.\r\n' ',X666666,,XXX,RequiredMissing: firstname\r\n' ) self.assertEqual(num_warns,2) 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,4) 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: Invalid input\r\n' '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: Invalid input\r\n' '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: Invalid input; matric_number: Invalid input\r\n' '3,Uli,A123456,m,aa@aa.ng,1234,,1990-01-07,Schulz,,100002,This object already exists.\r\n' ) shutil.rmtree(os.path.dirname(fin_file)) class StudentStudyCourseProcessorTest(StudentImportExportSetup): def setUp(self): 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') 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) self.csv_file_transfer = os.path.join( self.workdir, 'sample_transfer_data.csv') open(self.csv_file_transfer, 'wb').write(TRANSFER_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_checkUpdateRequirements(self): # Current level must be in range of certificate. # Since row has passed the converter, current_level is an integer. err = self.processor.checkUpdateRequirements( self.student['studycourse'], dict(reg_number='1', current_level=100), self.app) self.assertEqual(err, 'current_level not in range.') err = self.processor.checkUpdateRequirements( self.student['studycourse'], dict(reg_number='1', current_level=200), self.app) self.assertTrue(err is None) # We can update pg students. self.student['studycourse'].certificate.start_level=999 self.student['studycourse'].certificate.end_level=999 err = self.processor.checkUpdateRequirements( self.student['studycourse'], dict(reg_number='1', current_level=999), self.app) self.assertTrue(err is None) # Make sure that pg students can't be updated with wrong transition. 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.') # If certificate is not given in row (and has thus # successfully passed checkConversion) the certificate # attribute must be set. self.student['studycourse'].certificate = None err = self.processor.checkUpdateRequirements( self.student['studycourse'], dict(reg_number='1', current_level=100), self.app) self.assertEqual(err, 'No certificate to check level.') # When transferring students the method also checks # if the former studycourse is complete. err = self.processor.checkUpdateRequirements( self.student['studycourse'], dict(reg_number='1', certificate='CERT1', current_level=200, entry_mode='transfer'), self.app) self.assertEqual(err, 'Former study course record incomplete.') self.student['studycourse'].certificate = self.certificate self.student['studycourse'].entry_session = 2005 # The method doesn't care if current_level # is not in range of CERT1. This is done by checkConversion # if certificate is in row. err = self.processor.checkUpdateRequirements( self.student['studycourse'], dict(reg_number='1', certificate='CERT1', current_level=200, entry_mode='transfer'), self.app) self.assertTrue(err is None) 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)) def test_import_transfer(self): self.certificate2 = createObject('waeup.Certificate') self.certificate2.code = 'CERT2' self.certificate2.application_category = 'basic' self.certificate2.start_level = 200 self.certificate2.end_level = 500 self.certificate2.study_mode = u'ug_pt' self.app['faculties']['fac1']['dep1'].certificates.addCertificate( self.certificate2) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file_transfer, TRANSFER_HEADER_FIELDS,'update') self.assertEqual(num_warns,0) self.assertEqual(self.student['studycourse'].certificate.code, 'CERT2') self.assertEqual(self.student['studycourse_1'].certificate.code, 'CERT1') self.assertEqual(self.student['studycourse'].entry_mode, 'transfer') self.assertEqual(self.student['studycourse_1'].entry_mode, 'ug_ft') self.assertEqual(self.student.current_mode, 'ug_pt') shutil.rmtree(os.path.dirname(fin_file)) # Transer has bee logged. logcontent = open(self.logfile).read() self.assertTrue( 'INFO - system - K1000000 - transferred from CERT1 to CERT2\n' in logcontent) self.assertTrue( 'INFO - system - ' 'StudentStudyCourse Processor (update only) - ' 'sample_transfer_data - K1000000 - updated: entry_mode=transfer, ' 'certificate=CERT2, current_session=2009, current_level=300' in logcontent) # A history message has been added. history = ' '.join(self.student.history.messages) self.assertTrue( "Transferred from CERT1 to CERT2 by system" in history) # The catalog has been updated cat = queryUtility(ICatalog, name='students_catalog') results = list( cat.searchResults( certcode=('CERT2', 'CERT2'))) self.assertTrue(results[0] is self.student) results = list( cat.searchResults( current_session=(2009, 2009))) self.assertTrue(results[0] is self.student) results = list( cat.searchResults( certcode=('CERT1', 'CERT1'))) self.assertEqual(len(results), 0) 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='999')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', level='1000')) self.assertEqual(len(errs),1) self.assertTrue(('level', u'Invalid value') in errs) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(reg_number='1', level='xyz')) self.assertEqual(len(errs),1) self.assertTrue(('level', u'Invalid value') 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, '0') self.assertEqual(studylevel.level, 100) shutil.rmtree(os.path.dirname(fin_file)) logcontent = open(self.logfile).read() # Logging message from updateEntry, self.assertTrue( 'INFO - system - StudentStudyLevel Processor - ' 'sample_studylevel_data - K1000000 - updated: ' 'level=100, level_verdict=C, level_session=2009' in logcontent) 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') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update') self.assertEqual(num_warns,2) studylevel = self.processor.getEntry( dict(reg_number='1', level='100'), self.app) self.assertEqual(studylevel.level, 100) 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, STUDYLEVEL_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, STUDYLEVEL_HEADER_FIELDS,'remove') assert self.processor.entryExists( dict(reg_number='1', level='100'), self.app) is False 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 certificate course 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'].addCertCourse( 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') fail_file = open(fail_file).read() self.assertEqual(num_warns,5) self.assertEqual(fail_file, 'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n' '1,COURSE1,,nonsense,,5,,Not all parents do exist yet.\r\n' '1,NONSENSE,,100,,5,,code: non-existent\r\n' '1,COURSE1,,200,2004,6,,level_session: does not match 2008\r\n' '1,COURSE1,,300,2008,6,,level object: does not exist\r\n' '1,COURSE1,,300,2008X,6,,level_session: Invalid value\r\n') 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) self.assertEqual(courseticket.level, 100) self.assertEqual(courseticket.level_session, 2008) shutil.rmtree(os.path.dirname(fin_file)) logcontent = open(self.logfile).read() # Logging message from updateEntry, self.assertTrue( 'INFO - system - CourseTicket Processor - ' 'sample_courseticket_data - K1000000 - 100 - ' 'updated: code=COURSE1, ' 'mandatory=False, score=3' in logcontent) # The catalog has been updated cat = queryUtility(ICatalog, name='coursetickets_catalog') results = list( cat.searchResults( level=(100, 100))) self.assertEqual(len(results),3) 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') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, COURSETICKET_HEADER_FIELDS,'update') fail_file = open(fail_file).read() self.assertEqual(num_warns,5) self.assertEqual(fail_file, 'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n' '1,COURSE1,,nonsense,,5,,Cannot update: no such entry\r\n' '1,NONSENSE,,100,,5,,code: non-existent\r\n' '1,COURSE1,,200,2004,6,,level_session: does not match 2008\r\n' '1,COURSE1,,300,2008,6,,level object: does not exist\r\n' '1,COURSE1,,300,2008X,6,,level_session: Invalid value\r\n') 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') shutil.rmtree(os.path.dirname(fin_file)) 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,5) 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): 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 = 'p120' payment.p_session = 2012 payment.p_category = 'schoolfee' payment.p_state = 'paid' 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) 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): # 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='p120'), self.app) is self.student['payments']['p120'] 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) 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' # and is not an old PIN 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']['p560'].p_id, 'p560') # Requirement added on 19/02/2015: same payment must not exist. payment3 = createObject(u'waeup.StudentOnlinePayment') payment3.p_id = 'p456' payment3.p_session = 2012 payment3.p_category = 'schoolfee' self.assertRaises( DuplicationError, self.processor.addEntry, payment3, dict(student_id=self.student.student_id, p_id='p456'), self.app) logcontent = open(self.logfile).read() self.assertTrue( 'INFO - system - StudentOnlinePayment Processor - K1000000 - ' 'previous update cancelled' in logcontent) def test_checkConversion(self): errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id=''), mode='create') self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id=''), mode='update') self.assertEqual(len(errs),1) self.assertEqual(errs[0], ('p_id', u'missing')) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id='3816951266236341955')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id='p1266236341955')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id='ABC-11-1234567890')) self.assertEqual(len(errs),0) errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id='nonsense')) self.assertEqual(len(errs),1) self.assertEqual(errs[0], ('p_id', u'invalid format')) timestamp = ("%d" % int(time()*10000))[1:] p_id = "p%s" % timestamp errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id=p_id)) self.assertEqual(len(errs),0) dup_payment = createObject(u'waeup.StudentOnlinePayment') dup_payment.p_id = 'XYZ-99-1234567890' self.student['payments'][dup_payment.p_id] = dup_payment errs, inv_errs, conv_dict = self.processor.checkConversion( dict(p_id='XYZ-99-1234567890'), mode='create') self.assertEqual(len(errs),1) self.assertEqual(errs[0], ('p_id', u'p_id exists in K1000000 ')) 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='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') payment = self.processor.getEntry(dict(matric_number='100001', p_id='p2907125937570'), self.app) self.assertEqual(payment.p_id, 'p2907125937570') self.assertEqual(payment.amount_auth, 19500.1) 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') payment = self.processor.getEntry(dict(reg_number='3', p_id='ABC-11-1234567890'), self.app) self.assertEqual(payment.amount_auth, 19500.6) shutil.rmtree(os.path.dirname(fin_file)) logcontent = open(self.logfile).read() # Logging message from updateEntry self.assertTrue( 'INFO - system - StudentOnlinePayment Processor - ' 'sample_payment_data - K1000001 - 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_session=2009, ' 'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, ' 'p_state=paid' in logcontent) self.assertTrue( 'INFO - system - StudentOnlinePayment Processor - ' 'sample_payment_data - K1000001 - updated: ' 'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, ' 'p_category=schoolfee, amount_auth=19500.6, p_current=True, ' 'p_session=2011, ' 'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, ' 'p_state=paid' in logcontent) def test_import_update(self): # We perform the same import twice, # the second time in update mode. The number # of warnings increases becaus one p_id is missing. num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, PAYMENT_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, PAYMENT_HEADER_FIELDS,'update') self.assertEqual(num_warns,1) content = open(fail_file).read() shutil.rmtree(os.path.dirname(fin_file)) self.assertTrue('p_id: missing' in content) def test_import_remove(self): # We perform the same import twice, # the second time in remove mode. The number # of warnings increases becaus one p_id is missing. num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, PAYMENT_HEADER_FIELDS,'create') shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, PAYMENT_HEADER_FIELDS,'remove') self.assertEqual(num_warns,1) content = open(fail_file).read() self.assertTrue('p_id: missing' in content) 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_same_payment_exists(self): num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create') # One payment with same session and category exists self.assertEqual(num_warns,1) content = open(fail_file).read() self.assertTrue( '1,942,online,BTECHBDT,2010/11/26 19:59:33.744 GMT+1,0,' '19500,schoolfee,19500,2015,paid,' 'Same payment has already been made.' in content) shutil.rmtree(os.path.dirname(fin_file)) self.assertEqual(len(self.app['students']['X666666']['payments']), 13) 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)) # Import study levels self.csv_file = os.path.join( self.workdir, 'sample_studylevel_data.csv') open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA) processor = StudentStudyLevelProcessor() num, num_warns, fin_file, fail_file = processor.doImport( self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create') content = open(fail_file).read() 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): studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app) self.assertEqual(studycourse['200'].level_verdict, '0') student = self.processor.getParent( dict(matric_number='100000'), self.app) num, num_warns, fin_file, fail_file = self.processor.doImport( self.csv_file, VERDICT_HEADER_FIELDS,'update') self.assertEqual(num_warns,5) self.assertEqual(studycourse.current_verdict, '0') self.assertEqual(student.state, 'returning') self.assertEqual(studycourse.current_level, 200) self.assertEqual(studycourse['200'].level_verdict, '0') student = self.processor.getParent( dict(matric_number='100005'), self.app) self.assertEqual(student.state, 'returning') self.assertEqual(student['studycourse'].current_verdict, 'A') self.assertEqual(studycourse.current_level, 200) self.assertEqual(student['studycourse']['200'].validated_by, 'System') self.assertTrue(isinstance( student['studycourse']['200'].validation_date, datetime.datetime)) student = self.processor.getParent( dict(matric_number='100008'), self.app) self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana') content = open(fail_file).read() self.assertEqual( content, 'current_session,current_level,bypass_validation,current_verdict,' 'matric_number,validated_by,--ERRORS--\r\n' '2008,100,False,B,100001,,Current level does not correspond.\r\n' '2007,200,,C,100002,,Current session does not correspond.\r\n' '2008,200,,A,100003,,Student in wrong state.\r\n' '2008,200,,,100004,,No verdict in import file.\r\n' '2008,200,True,A,100007,,Study level object is missing.\r\n' ) logcontent = open(self.logfile).read() self.assertMatches( '... INFO - system - Verdict Processor (special processor, ' 'update only) - sample_verdict_data - X666666 - ' 'updated: current_verdict=0...', logcontent) self.assertMatches( '... INFO - system - X666666 - Returned...', logcontent) 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