source: main/waeup.kofa/trunk/src/waeup/kofa/students/tests/test_batching.py @ 9029

Last change on this file since 9029 was 9029, checked in by Henrik Bettermann, 12 years ago

Add test and fix checkUpdateRequirements.
  • Property svn:keywords set to Id
File size: 41.5 KB
Line 
1## $Id: test_batching.py 9029 2012-07-20 09:34:55Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18"""Unit tests for students-related data processors.
19"""
20import os
21import shutil
22import tempfile
23import unittest
24import datetime
25import grok
26from time import time
27from zope.event import notify
28from zope.component import createObject
29from zope.component.hooks import setSite, clearSite
30from zope.interface.verify import verifyClass, verifyObject
31from hurry.workflow.interfaces import IWorkflowState
32
33from waeup.kofa.app import University
34from waeup.kofa.interfaces import IBatchProcessor, FatalCSVError, IUserAccount
35from waeup.kofa.students.batching import (
36    StudentProcessor, StudentStudyCourseProcessor,
37    StudentStudyLevelProcessor, CourseTicketProcessor,
38    StudentOnlinePaymentProcessor, StudentVerdictProcessor)
39from waeup.kofa.students.payments import StudentOnlinePayment
40from waeup.kofa.students.student import Student
41from waeup.kofa.students.studylevel import StudentStudyLevel, CourseTicket
42from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
43from waeup.kofa.university.faculty import Faculty
44from waeup.kofa.university.department import Department
45
46
47STUDENT_SAMPLE_DATA = open(
48    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
49    'rb').read()
50
51STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
52    '\n')[0].split(',')
53
54STUDENT_SAMPLE_DATA_UPDATE = open(
55    os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'),
56    'rb').read()
57
58STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split(
59    '\n')[0].split(',')
60
61STUDENT_SAMPLE_DATA_UPDATE2 = open(
62    os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'),
63    'rb').read()
64
65STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split(
66    '\n')[0].split(',')
67
68STUDENT_SAMPLE_DATA_UPDATE3 = open(
69    os.path.join(os.path.dirname(__file__), 'sample_student_data_update3.csv'),
70    'rb').read()
71
72STUDENT_HEADER_FIELDS_UPDATE3 = STUDENT_SAMPLE_DATA_UPDATE3.split(
73    '\n')[0].split(',')
74
75STUDENT_SAMPLE_DATA_UPDATE4 = open(
76    os.path.join(os.path.dirname(__file__), 'sample_student_data_update4.csv'),
77    'rb').read()
78
79STUDENT_HEADER_FIELDS_UPDATE4 = STUDENT_SAMPLE_DATA_UPDATE4.split(
80    '\n')[0].split(',')
81
82STUDYCOURSE_SAMPLE_DATA = open(
83    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
84    'rb').read()
85
86STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
87    '\n')[0].split(',')
88
89VERDICT_SAMPLE_DATA = open(
90    os.path.join(os.path.dirname(__file__), 'sample_verdict_data.csv'),
91    'rb').read()
92
93VERDICT_HEADER_FIELDS = VERDICT_SAMPLE_DATA.split(
94    '\n')[0].split(',')
95
96STUDENT_SAMPLE_DATA_MIGRATION = open(
97    os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'),
98    'rb').read()
99
100STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split(
101    '\n')[0].split(',')
102
103STUDENT_SAMPLE_DATA_DUPLICATES = open(
104    os.path.join(os.path.dirname(__file__), 'sample_student_data_duplicates.csv'),
105    'rb').read()
106
107STUDENT_HEADER_FIELDS_DUPLICATES = STUDENT_SAMPLE_DATA_DUPLICATES.split(
108    '\n')[0].split(',')
109
110STUDYLEVEL_SAMPLE_DATA = open(
111    os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'),
112    'rb').read()
113
114STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split(
115    '\n')[0].split(',')
116
117COURSETICKET_SAMPLE_DATA = open(
118    os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'),
119    'rb').read()
120
121COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
122    '\n')[0].split(',')
123
124PAYMENT_SAMPLE_DATA = open(
125    os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'),
126    'rb').read()
127
128PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
129    '\n')[0].split(',')
130
131PAYMENT_CREATE_SAMPLE_DATA = open(
132    os.path.join(os.path.dirname(__file__), 'sample_create_payment_data.csv'),
133    'rb').read()
134
135PAYMENT_CREATE_HEADER_FIELDS = PAYMENT_CREATE_SAMPLE_DATA.split(
136    '\n')[0].split(',')
137
138class StudentImportExportSetup(FunctionalTestCase):
139
140    layer = FunctionalLayer
141
142    def setUp(self):
143        super(StudentImportExportSetup, self).setUp()
144        self.dc_root = tempfile.mkdtemp()
145        self.workdir = tempfile.mkdtemp()
146        app = University()
147        app['datacenter'].setStoragePath(self.dc_root)
148        self.getRootFolder()['app'] = app
149        self.app = self.getRootFolder()['app']
150        setSite(app)
151
152        # Populate university
153        self.certificate = createObject('waeup.Certificate')
154        self.certificate.code = 'CERT1'
155        self.certificate.application_category = 'basic'
156        self.certificate.start_level = 200
157        self.certificate.end_level = 500
158        self.app['faculties']['fac1'] = Faculty()
159        self.app['faculties']['fac1']['dep1'] = Department()
160        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
161            self.certificate)
162
163        self.logfile = os.path.join(
164            self.app['datacenter'].storage, 'logs', 'students.log')
165        return
166
167    def tearDown(self):
168        super(StudentImportExportSetup, self).tearDown()
169        shutil.rmtree(self.workdir)
170        shutil.rmtree(self.dc_root)
171        clearSite()
172        return
173
174    def setup_for_export(self):
175        student = Student()
176        student.student_id = u'A111111'
177        self.app['students'][student.student_id] = self.student = student
178        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
179        return
180
181    def setup_student(self, student):
182        # set predictable values for `student`
183        student.matric_number = u'234'
184        student.adm_code = u'my adm code'
185        student.clearance_locked = False
186        student.clr_code = u'my clr code'
187        student.perm_address = u'Studentroad 21\nLagos 123456\n'
188        student.reg_number = u'123'
189        student.firstname = u'Anna'
190        student.lastname = u'Tester'
191        student.middlename = u'M.'
192        student.date_of_birth = datetime.date(1981, 2, 4)
193        student.sex = 'f'
194        student.email = 'anna@sample.com'
195        student.phone = u'+234-123-12345'
196        student.notice = u'Some notice\nin lines.'
197        student.nationality = u'NG'
198
199        student['studycourse'].certificate = self.certificate
200        student['studycourse'].entry_mode = 'ug_ft'
201        student['studycourse'].entry_session = 2010
202        student['studycourse'].current_session = 2012
203        student['studycourse'].current_level = int(self.certificate.start_level)
204
205        study_level = StudentStudyLevel()
206        study_level.level_session = 2012
207        study_level.level_verdict = "A"
208        study_level.level = 100
209        student['studycourse'].addStudentStudyLevel(
210            self.certificate, study_level)
211
212        ticket = CourseTicket()
213        ticket.automatic = True
214        ticket.carry_over = True
215        ticket.code = u'CRS1'
216        ticket.title = u'Course 1'
217        ticket.fcode = u'FAC1'
218        ticket.dcode = u'DEP1'
219        ticket.credits = 100
220        ticket.passmark = 100
221        ticket.semester = 2
222        study_level[ticket.code] = ticket
223        self.add_payment(student)
224        return student
225
226    def add_payment(self, student):
227        # get a payment with all fields set
228        payment = StudentOnlinePayment()
229        payment.creation_date = datetime.datetime(2012, 4, 1, 13, 12, 1)
230        payment.p_id = 'my-id'
231        payment.ac = u'666'
232        payment.p_item = u'p-item'
233        payment.p_level = 100
234        payment.p_session = 2012
235        payment.payment_date = datetime.datetime(2012, 4, 1, 14, 12, 1)
236        payment.r_amount_approved = 12.12
237        payment.r_code = u'r-code'
238        # XXX: there is no addPayment method to give predictable names
239        student['payments']['my-payment'] = payment
240        return payment
241
242
243class StudentProcessorTest(StudentImportExportSetup):
244
245    layer = FunctionalLayer
246
247    def setUp(self):
248        super(StudentProcessorTest, self).setUp()
249
250        # Add student with subobjects
251        student = Student()
252        self.app['students'].addStudent(student)
253        student = self.setup_student(student)
254        notify(grok.ObjectModifiedEvent(student))
255        self.student = self.app['students'][student.student_id]
256
257        self.processor = StudentProcessor()
258        self.workdir = tempfile.mkdtemp()
259        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
260        self.csv_file_update = os.path.join(
261            self.workdir, 'sample_student_data_update.csv')
262        self.csv_file_update2 = os.path.join(
263            self.workdir, 'sample_student_data_update2.csv')
264        self.csv_file_update3 = os.path.join(
265            self.workdir, 'sample_student_data_update3.csv')
266        self.csv_file_update4 = os.path.join(
267            self.workdir, 'sample_student_data_update4.csv')
268        self.csv_file_migration = os.path.join(
269            self.workdir, 'sample_student_data_migration.csv')
270        self.csv_file_duplicates = os.path.join(
271            self.workdir, 'sample_student_data_duplicates.csv')
272        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
273        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
274        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
275        open(self.csv_file_update3, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE3)
276        open(self.csv_file_update4, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE4)
277        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
278        open(self.csv_file_duplicates, 'wb').write(STUDENT_SAMPLE_DATA_DUPLICATES)
279
280    def test_interface(self):
281        # Make sure we fulfill the interface contracts.
282        assert verifyObject(IBatchProcessor, self.processor) is True
283        assert verifyClass(
284            IBatchProcessor, StudentProcessor) is True
285
286    def test_parentsExist(self):
287        self.assertFalse(self.processor.parentsExist(None, dict()))
288        self.assertTrue(self.processor.parentsExist(None, self.app))
289
290    def test_entryExists(self):
291        assert self.processor.entryExists(
292            dict(student_id='ID_NONE'), self.app) is False
293        assert self.processor.entryExists(
294            dict(reg_number='123'), self.app) is True
295
296    def test_getParent(self):
297        parent = self.processor.getParent(None, self.app)
298        assert parent is self.app['students']
299
300    def test_getEntry(self):
301        assert self.processor.getEntry(
302            dict(student_id='ID_NONE'), self.app) is None
303        assert self.processor.getEntry(
304            dict(student_id=self.student.student_id), self.app) is self.student
305
306    def test_addEntry(self):
307        new_student = Student()
308        self.processor.addEntry(
309            new_student, dict(), self.app)
310        assert len(self.app['students'].keys()) == 2
311
312    def test_checkConversion(self):
313        # Make sure we can check conversions and that the stud_id
314        # counter is not raised during such checks.
315        initial_stud_id = self.app['students']._curr_stud_id
316        errs, inv_errs, conv_dict = self.processor.checkConversion(
317            dict(reg_number='1', state='admitted'))
318        self.assertEqual(len(errs),0)
319        # Empty state is allowed
320        errs, inv_errs, conv_dict = self.processor.checkConversion(
321            dict(reg_number='1', state=''))
322        self.assertEqual(len(errs),0)
323        #self.assertTrue(('state', 'no value provided') in errs)
324        errs, inv_errs, conv_dict = self.processor.checkConversion(
325            dict(reg_number='1', state='nonsense'))
326        self.assertEqual(len(errs),1)
327        self.assertTrue(('state', 'not allowed') in errs)
328        new_stud_id = self.app['students']._curr_stud_id
329        self.assertEqual(initial_stud_id, new_stud_id)
330        return
331
332    def test_checkUpdateRequirements(self):
333        # Make sure that pg students can't be updated with wrong transition.
334        err = self.processor.checkUpdateRequirements(self.student,
335            dict(reg_number='1', state='returning'), self.app)
336        self.assertTrue(err is None)
337        self.certificate.study_mode = 'pg_ft'
338        err = self.processor.checkUpdateRequirements(self.student,
339            dict(reg_number='1', state='returning'), self.app)
340        self.assertEqual(err, 'State not allowed (pg student).')
341        IWorkflowState(self.student).setState('school fee paid')
342        err = self.processor.checkUpdateRequirements(self.student,
343            dict(reg_number='1', transition='reset6'), self.app)
344        self.assertEqual(err, 'Transition not allowed (pg student).')
345        err = self.processor.checkUpdateRequirements(self.student,
346            dict(reg_number='1', transition='register_courses'), self.app)
347        self.assertEqual(err, 'Transition not allowed (pg student).')
348
349
350    def test_delEntry(self):
351        assert self.student.student_id in self.app['students'].keys()
352        self.processor.delEntry(
353            dict(reg_number=self.student.reg_number), self.app)
354        assert self.student.student_id not in self.app['students'].keys()
355
356    def test_import(self):
357        self.assertEqual(self.app['students']._curr_stud_id, 1000001)
358        num, num_warns, fin_file, fail_file = self.processor.doImport(
359            self.csv_file, STUDENT_HEADER_FIELDS)
360        self.assertEqual(num_warns,0)
361        assert len(self.app['students'].keys()) == 5
362        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
363        self.assertEqual(
364            self.app['students']['X666666'].state, 'courses validated')
365        # Two new student_ids have been created.
366        self.assertEqual(self.app['students']._curr_stud_id, 1000003)
367        shutil.rmtree(os.path.dirname(fin_file))
368
369    def test_import_update(self):
370        num, num_warns, fin_file, fail_file = self.processor.doImport(
371            self.csv_file, STUDENT_HEADER_FIELDS)
372        shutil.rmtree(os.path.dirname(fin_file))
373        num, num_warns, fin_file, fail_file = self.processor.doImport(
374            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
375        self.assertEqual(num_warns,0)
376        # state has changed
377        self.assertEqual(self.app['students']['X666666'].state,'admitted')
378        # state has not changed
379        self.assertEqual(self.app['students']['Y777777'].state,'courses validated')
380        shutil.rmtree(os.path.dirname(fin_file))
381
382    def test_import_update2(self):
383        num, num_warns, fin_file, fail_file = self.processor.doImport(
384            self.csv_file, STUDENT_HEADER_FIELDS)
385        shutil.rmtree(os.path.dirname(fin_file))
386        num, num_warns, fin_file, fail_file = self.processor.doImport(
387            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
388        self.assertEqual(num_warns,0)
389        # The phone import value of Pieri was None.
390        # Confirm that phone has not been cleared.
391        container = self.app['students']
392        for key in container.keys():
393            if container[key].firstname == 'Aaren':
394                aaren = container[key]
395                break
396        self.assertEqual(aaren.phone, '--1234')
397        # The phone import value of Claus was a deletion marker.
398        # Confirm that phone has been cleared.
399        for key in container.keys():
400            if container[key].firstname == 'Claus':
401                claus = container[key]
402                break
403        assert claus.phone is None
404        shutil.rmtree(os.path.dirname(fin_file))
405
406    def test_import_update3(self):
407        num, num_warns, fin_file, fail_file = self.processor.doImport(
408            self.csv_file, STUDENT_HEADER_FIELDS)
409        shutil.rmtree(os.path.dirname(fin_file))
410        num, num_warns, fin_file, fail_file = self.processor.doImport(
411            self.csv_file_update3, STUDENT_HEADER_FIELDS_UPDATE3, 'update')
412        content = open(fail_file).read()
413        self.assertEqual(
414            content,
415            'reg_number,student_id,transition,--ERRORS--\r\n'
416            '<IGNORE>,X666666,request_clearance,Transition not allowed.\r\n'
417            )
418        self.assertEqual(num_warns,1)
419        self.assertEqual(self.app['students']['Y777777'].state,'returning')
420
421    def test_import_update4(self):
422        num, num_warns, fin_file, fail_file = self.processor.doImport(
423            self.csv_file, STUDENT_HEADER_FIELDS)
424        shutil.rmtree(os.path.dirname(fin_file))
425        self.assertRaises(
426            FatalCSVError, self.processor.doImport, self.csv_file_update4,
427            STUDENT_HEADER_FIELDS_UPDATE4, 'update')
428
429    def test_import_remove(self):
430        num, num_warns, fin_file, fail_file = self.processor.doImport(
431            self.csv_file, STUDENT_HEADER_FIELDS)
432        shutil.rmtree(os.path.dirname(fin_file))
433        num, num_warns, fin_file, fail_file = self.processor.doImport(
434            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
435        self.assertEqual(num_warns,0)
436        shutil.rmtree(os.path.dirname(fin_file))
437
438    def test_import_migration_data(self):
439        num, num_warns, fin_file, fail_file = self.processor.doImport(
440            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
441        content = open(fail_file).read()
442        self.assertEqual(num_warns,2)
443        assert len(self.app['students'].keys()) == 5
444        self.assertEqual(
445            content,
446            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
447            '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n'
448            '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n'
449            )
450        students = self.app['students']
451        self.assertTrue('A123456' in students.keys())
452        self.assertEqual(students['A123456'].state, 'clearance started')
453        self.assertEqual(students['A123456'].date_of_birth,
454                         datetime.date(1990, 1, 2))
455        self.assertFalse(students['A123456'].clearance_locked)
456        self.assertEqual(students['B123456'].state, 'cleared')
457        self.assertEqual(students['B123456'].date_of_birth,
458                         datetime.date(1990, 1, 3))
459        self.assertTrue(students['B123456'].clearance_locked)
460        history = ' '.join(students['A123456'].history.messages)
461        self.assertTrue(
462            "State 'clearance started' set by system" in history)
463        # state was empty and student is thus in state created
464        self.assertEqual(students['F123456'].state,'created')
465        # passwords were set correctly
466        self.assertEqual(
467            IUserAccount(students['A123456']).checkPassword('mypw1'), True)
468        self.assertEqual(
469            IUserAccount(students['C123456']).checkPassword('mypw1'), True)
470        shutil.rmtree(os.path.dirname(fin_file))
471
472    def test_import_duplicate_data(self):
473        num, num_warns, fin_file, fail_file = self.processor.doImport(
474            self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES)
475        content = open(fail_file).read()
476        self.assertEqual(num_warns,4)
477        self.assertEqual(
478            content,
479            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
480            '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n'
481            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n'
482            '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n'
483            '3,Uli,A123456,m,aa@aa.ng,1234,,1990-01-07,Schulz,,100002,This object already exists. Skipping.\r\n'
484            )
485        shutil.rmtree(os.path.dirname(fin_file))
486
487class StudentStudyCourseProcessorTest(StudentImportExportSetup):
488
489    def setUp(self):
490        super(StudentStudyCourseProcessorTest, self).setUp()
491
492        # Add student with subobjects
493        student = Student()
494        self.app['students'].addStudent(student)
495        student = self.setup_student(student)
496        notify(grok.ObjectModifiedEvent(student))
497        self.student = self.app['students'][student.student_id]
498
499        # Import students with subobjects
500        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
501        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
502        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
503            student_file, STUDENT_HEADER_FIELDS)
504        shutil.rmtree(os.path.dirname(fin_file))
505
506        self.processor = StudentStudyCourseProcessor()
507        self.csv_file = os.path.join(
508            self.workdir, 'sample_studycourse_data.csv')
509        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
510        return
511
512    def test_interface(self):
513        # Make sure we fulfill the interface contracts.
514        assert verifyObject(IBatchProcessor, self.processor) is True
515        assert verifyClass(
516            IBatchProcessor, StudentStudyCourseProcessor) is True
517
518    def test_entryExists(self):
519        assert self.processor.entryExists(
520            dict(reg_number='REG_NONE'), self.app) is False
521        assert self.processor.entryExists(
522            dict(reg_number='1'), self.app) is True
523
524    def test_getEntry(self):
525        student = self.processor.getEntry(
526            dict(reg_number='1'), self.app).__parent__
527        self.assertEqual(student.reg_number,'1')
528
529    def test_checkConversion(self):
530        errs, inv_errs, conv_dict = self.processor.checkConversion(
531            dict(reg_number='1', certificate='CERT1', current_level='200'))
532        self.assertEqual(len(errs),0)
533        errs, inv_errs, conv_dict = self.processor.checkConversion(
534            dict(reg_number='1', certificate='CERT999'))
535        self.assertEqual(len(errs),1)
536        self.assertTrue(('certificate', u'Invalid value') in errs)
537        errs, inv_errs, conv_dict = self.processor.checkConversion(
538            dict(reg_number='1', certificate='CERT1', current_level='100'))
539        self.assertEqual(len(errs),1)
540        self.assertTrue(('current_level','not in range') in errs)
541        # If we import only current_level, no conversion checking is done.
542        errs, inv_errs, conv_dict = self.processor.checkConversion(
543            dict(reg_number='1', current_level='100'))
544        self.assertEqual(len(errs),0)
545
546    def test_checkUpdateRequirements(self):
547        # Make sure that pg students can't be updated with wrong transition.
548        err = self.processor.checkUpdateRequirements(self.student['studycourse'],
549            dict(reg_number='1', current_level='100'), self.app)
550        self.assertTrue(err is None)
551        err = self.processor.checkUpdateRequirements(self.student['studycourse'],
552            dict(reg_number='1', current_level='999'), self.app)
553        self.assertTrue(err is None)
554        IWorkflowState(self.student).setState('returning')
555        err = self.processor.checkUpdateRequirements(self.student['studycourse'],
556            dict(reg_number='1', current_level='999'), self.app)
557        self.assertEqual(err, 'Not a pg student.')
558
559    def test_import(self):
560        num, num_warns, fin_file, fail_file = self.processor.doImport(
561            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
562        self.assertEqual(num_warns,1)
563        content = open(fail_file).read()
564        self.assertTrue('current_level: not in range' in content)
565        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
566        self.assertEqual(studycourse.certificate.code, u'CERT1')
567        shutil.rmtree(os.path.dirname(fin_file))
568
569class StudentVerdictProcessorTest(StudentImportExportSetup):
570
571
572    def setUp(self):
573        super(StudentVerdictProcessorTest, self).setUp()
574
575        # Import students with subobjects
576        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
577        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
578        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
579            student_file, STUDENT_HEADER_FIELDS)
580        shutil.rmtree(os.path.dirname(fin_file))
581
582        # Update study courses
583        studycourse_file = os.path.join(
584            self.workdir, 'sample_studycourse_data.csv')
585        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
586        processor = StudentStudyCourseProcessor()
587        num, num_warns, fin_file, fail_file = processor.doImport(
588            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
589        shutil.rmtree(os.path.dirname(fin_file))
590
591        self.processor = StudentVerdictProcessor()
592        self.csv_file = os.path.join(
593            self.workdir, 'sample_verdict_data.csv')
594        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
595        return
596
597    def test_import(self):
598        num, num_warns, fin_file, fail_file = self.processor.doImport(
599            self.csv_file, VERDICT_HEADER_FIELDS,'update')
600        self.assertEqual(num_warns,3)
601        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
602        student = self.processor.getParent(dict(matric_number='100000'), self.app)
603        self.assertEqual(studycourse.current_verdict, 'A')
604        self.assertEqual(student.state, 'returning')
605        self.assertEqual(studycourse.current_level, 200)
606        content = open(fail_file).read()
607        self.assertEqual(
608            content,
609            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
610            '2008,B,100001,100,Current level does not correspond.\r\n'
611            '2007,C,100002,200,Current session does not correspond.\r\n'
612            '2008,A,100003,200,Student in wrong state.\r\n'
613            )
614        shutil.rmtree(os.path.dirname(fin_file))
615
616
617class StudentStudyLevelProcessorTest(StudentImportExportSetup):
618
619    def setUp(self):
620        super(StudentStudyLevelProcessorTest, self).setUp()
621
622        # Import students with subobjects
623        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
624        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
625        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
626            student_file, STUDENT_HEADER_FIELDS)
627        shutil.rmtree(os.path.dirname(fin_file))
628
629        # Update study courses
630        studycourse_file = os.path.join(
631            self.workdir, 'sample_studycourse_data.csv')
632        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
633        processor = StudentStudyCourseProcessor()
634        num, num_warns, fin_file, fail_file = processor.doImport(
635            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
636        shutil.rmtree(os.path.dirname(fin_file))
637
638        self.processor = StudentStudyLevelProcessor()
639        self.csv_file = os.path.join(
640            self.workdir, 'sample_studylevel_data.csv')
641        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
642
643    def test_interface(self):
644        # Make sure we fulfill the interface contracts.
645        assert verifyObject(IBatchProcessor, self.processor) is True
646        assert verifyClass(
647            IBatchProcessor, StudentStudyLevelProcessor) is True
648
649    def test_checkConversion(self):
650        errs, inv_errs, conv_dict = self.processor.checkConversion(
651            dict(reg_number='1', level='220'))
652        self.assertEqual(len(errs),0)
653        errs, inv_errs, conv_dict = self.processor.checkConversion(
654            dict(reg_number='1', level='900'))
655        self.assertEqual(len(errs),1)
656        self.assertTrue(('level','no valid integer') in errs)
657        errs, inv_errs, conv_dict = self.processor.checkConversion(
658            dict(reg_number='1', level='xyz'))
659        self.assertEqual(len(errs),1)
660        self.assertTrue(('level','no integer') in errs)
661
662    def test_import(self):
663        num, num_warns, fin_file, fail_file = self.processor.doImport(
664            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
665        self.assertEqual(num_warns,2)
666        assert self.processor.entryExists(
667            dict(reg_number='1', level='100'), self.app) is True
668        studylevel = self.processor.getEntry(
669            dict(reg_number='1', level='100'), self.app)
670        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
671        self.assertEqual(studylevel.level_session, 2008)
672        self.assertEqual(studylevel.level_verdict, 'A')
673        self.assertEqual(studylevel.level, 100)
674        shutil.rmtree(os.path.dirname(fin_file))
675
676        logcontent = open(self.logfile).read()
677        # Logging message from updateEntry,
678        self.assertTrue(
679            'INFO - system - K1000000 - Study level '
680            'updated: level=100, level_verdict=C, level_session=2009'
681            in logcontent)
682
683    def test_import_update(self):
684        # We perform the same import twice,
685        # the second time in update mode. The number
686        # of warnings must be the same.
687        num, num_warns, fin_file, fail_file = self.processor.doImport(
688            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
689        num, num_warns, fin_file, fail_file = self.processor.doImport(
690            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
691        self.assertEqual(num_warns,2)
692        shutil.rmtree(os.path.dirname(fin_file))
693
694class CourseTicketProcessorTest(StudentImportExportSetup):
695
696    def setUp(self):
697        super(CourseTicketProcessorTest, self).setUp()
698
699        # Import students with subobjects
700        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
701        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
702        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
703            student_file, STUDENT_HEADER_FIELDS)
704        shutil.rmtree(os.path.dirname(fin_file))
705
706        # Add course and certificate course
707        self.course = createObject('waeup.Course')
708        self.course.code = 'COURSE1'
709        self.course.semester = 1
710        self.course.credits = 10
711        self.course.passmark = 40
712        self.app['faculties']['fac1']['dep1'].courses.addCourse(
713            self.course)
714        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
715            self.course, level=100)
716
717        # Update study courses
718        studycourse_file = os.path.join(
719            self.workdir, 'sample_studycourse_data.csv')
720        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
721        processor = StudentStudyCourseProcessor()
722        num, num_warns, fin_file, fail_file = processor.doImport(
723            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
724        shutil.rmtree(os.path.dirname(fin_file))
725
726        # Import study levels
727        processor = StudentStudyLevelProcessor()
728        studylevel_file = os.path.join(
729            self.workdir, 'sample_studylevel_data.csv')
730        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
731        num, num_warns, fin_file, fail_file = processor.doImport(
732            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
733        shutil.rmtree(os.path.dirname(fin_file))
734
735        self.processor = CourseTicketProcessor()
736        self.csv_file = os.path.join(
737            self.workdir, 'sample_courseticket_data.csv')
738        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
739
740    def test_interface(self):
741        # Make sure we fulfill the interface contracts.
742        assert verifyObject(IBatchProcessor, self.processor) is True
743        assert verifyClass(
744            IBatchProcessor, CourseTicketProcessor) is True
745
746    def test_checkConversion(self):
747        errs, inv_errs, conv_dict = self.processor.checkConversion(
748            dict(reg_number='1', code='COURSE1', level='220'))
749        self.assertEqual(len(errs),0)
750        errs, inv_errs, conv_dict = self.processor.checkConversion(
751            dict(reg_number='1', code='COURSE2', level='220'))
752        self.assertEqual(len(errs),1)
753        self.assertTrue(('code','non-existent') in errs)
754
755    def test_import(self):
756
757        num, num_warns, fin_file, fail_file = self.processor.doImport(
758            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
759
760        self.assertEqual(num_warns,2)
761        assert self.processor.entryExists(
762            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
763        courseticket = self.processor.getEntry(
764            dict(reg_number='1', level='100', code='COURSE1'), self.app)
765        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
766        self.assertEqual(courseticket.score, 1)
767        self.assertEqual(courseticket.mandatory, True)
768        self.assertEqual(courseticket.fcode, 'NA')
769        self.assertEqual(courseticket.dcode, 'NA')
770        self.assertEqual(courseticket.code, 'COURSE1')
771        self.assertEqual(courseticket.title, 'Unnamed Course')
772        self.assertEqual(courseticket.credits, 10)
773        self.assertEqual(courseticket.passmark, 40)
774        self.assertEqual(courseticket.semester, 1)
775        shutil.rmtree(os.path.dirname(fin_file))
776
777        logcontent = open(self.logfile).read()
778        # Logging message from updateEntry,
779        self.assertTrue(
780            'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, '
781            'mandatory=False, score=3'
782            in logcontent)
783
784    def test_import_update(self):
785        # We perform the same import twice,
786        # the second time in update mode. The number
787        # of warnings must be the same.
788        num, num_warns, fin_file, fail_file = self.processor.doImport(
789            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
790        num, num_warns, fin_file, fail_file = self.processor.doImport(
791            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
792        self.assertEqual(num_warns,2)
793        shutil.rmtree(os.path.dirname(fin_file))
794
795    def test_import_remove(self):
796        # We perform the same import twice,
797        # the second time in remove mode. The number
798        # of warnings must be the same.
799        num, num_warns, fin_file, fail_file = self.processor.doImport(
800            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
801        assert self.processor.entryExists(
802            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
803        num, num_warns, fin_file, fail_file = self.processor.doImport(
804            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
805        self.assertEqual(num_warns,2)
806        assert self.processor.entryExists(
807            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
808        shutil.rmtree(os.path.dirname(fin_file))
809        logcontent = open(self.logfile).read()
810        self.assertTrue(
811            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
812            in logcontent)
813
814class PaymentProcessorTest(StudentImportExportSetup):
815
816    def setUp(self):
817        super(PaymentProcessorTest, self).setUp()
818
819        # Add student with payment
820        student = Student()
821        student.firstname = u'Anna'
822        student.lastname = u'Tester'
823        student.reg_number = u'123'
824        student.matric_number = u'234'
825        self.app['students'].addStudent(student)
826        self.student = self.app['students'][student.student_id]
827        payment = createObject(u'waeup.StudentOnlinePayment')
828        payment.p_id = 'p120'
829        self.student['payments'][payment.p_id] = payment
830
831        # Import students with subobjects
832        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
833        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
834        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
835            student_file, STUDENT_HEADER_FIELDS)
836        shutil.rmtree(os.path.dirname(fin_file))
837
838        self.processor = StudentOnlinePaymentProcessor()
839        self.csv_file = os.path.join(
840            self.workdir, 'sample_payment_data.csv')
841        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
842        self.csv_file2 = os.path.join(
843            self.workdir, 'sample_create_payment_data.csv')
844        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
845
846    def test_interface(self):
847        # Make sure we fulfill the interface contracts.
848        assert verifyObject(IBatchProcessor, self.processor) is True
849        assert verifyClass(
850            IBatchProcessor, StudentOnlinePaymentProcessor) is True
851
852    def test_getEntry(self):
853        assert self.processor.getEntry(
854            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
855        assert self.processor.getEntry(
856            dict(student_id=self.student.student_id, p_id='p120'),
857            self.app) is self.student['payments']['p120']
858        assert self.processor.getEntry(
859            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
860            self.app) is self.student['payments']['p120']
861
862    def test_delEntry(self):
863        assert self.processor.getEntry(
864            dict(student_id=self.student.student_id, p_id='p120'),
865            self.app) is self.student['payments']['p120']
866        self.assertEqual(len(self.student['payments'].keys()),1)
867        self.processor.delEntry(
868            dict(student_id=self.student.student_id, p_id='p120'),
869            self.app)
870        assert self.processor.getEntry(
871            dict(student_id=self.student.student_id, p_id='p120'),
872            self.app) is None
873        self.assertEqual(len(self.student['payments'].keys()),0)
874
875    def test_addEntry(self):
876        self.assertEqual(len(self.student['payments'].keys()),1)
877        payment1 = createObject(u'waeup.StudentOnlinePayment')
878        payment1.p_id = 'p234'
879        self.processor.addEntry(
880            payment1, dict(student_id=self.student.student_id, p_id='p234'),
881            self.app)
882        self.assertEqual(len(self.student['payments'].keys()),2)
883        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
884        payment2 = createObject(u'waeup.StudentOnlinePayment')
885        payment1.p_id = 'nonsense'
886        # payment1.p_id will be replaced if p_id doesn't start with 'p'
887        self.processor.addEntry(
888            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
889            self.app)
890        self.assertEqual(len(self.student['payments'].keys()),3)
891        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
892
893    def test_checkConversion(self):
894        errs, inv_errs, conv_dict = self.processor.checkConversion(
895            dict(p_id='3816951266236341955'))
896        self.assertEqual(len(errs),0)
897        errs, inv_errs, conv_dict = self.processor.checkConversion(
898            dict(p_id='p1266236341955'))
899        self.assertEqual(len(errs),0)
900        errs, inv_errs, conv_dict = self.processor.checkConversion(
901            dict(p_id='nonsense'))
902        self.assertEqual(len(errs),1)
903        timestamp = ("%d" % int(time()*10000))[1:]
904        p_id = "p%s" % timestamp
905        errs, inv_errs, conv_dict = self.processor.checkConversion(
906            dict(p_id=p_id))
907        self.assertEqual(len(errs),0)
908
909    def test_import(self):
910        num, num_warns, fin_file, fail_file = self.processor.doImport(
911            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
912        self.assertEqual(num_warns,0)
913        payment = self.processor.getEntry(dict(reg_number='1',
914            p_id='p2907979737440'), self.app)
915        self.assertEqual(payment.p_id, 'p2907979737440')
916        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
917        self.assertEqual(cdate, "2010-11-26 18:59:33")
918        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
919        shutil.rmtree(os.path.dirname(fin_file))
920        logcontent = open(self.logfile).read()
921        # Logging message from updateEntry
922        self.assertTrue(
923            'INFO - system - K1000001 - Payment ticket updated: '
924            'p_id=p1266236341955, p_item=BTECHBDT, '
925            'creation_date=2010-02-15 13:19:01+00:00, r_code=00, '
926            'r_amount_approved=19500.0, p_category=schoolfee, '
927            'amount_auth=19500.0, p_state=paid'
928            in logcontent)
929
930    def test_import_update(self):
931        # We perform the same import twice,
932        # the second time in update mode. The number
933        # of warnings must be the same.
934        num, num_warns, fin_file, fail_file = self.processor.doImport(
935            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
936        num, num_warns, fin_file, fail_file = self.processor.doImport(
937            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
938        self.assertEqual(num_warns,0)
939        shutil.rmtree(os.path.dirname(fin_file))
940
941    def test_import_remove(self):
942        num, num_warns, fin_file, fail_file = self.processor.doImport(
943            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
944        num, num_warns, fin_file, fail_file = self.processor.doImport(
945            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
946        self.assertEqual(num_warns,0)
947        shutil.rmtree(os.path.dirname(fin_file))
948        logcontent = open(self.logfile).read()
949        self.assertTrue(
950            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
951            in logcontent)
952
953    def test_import_wo_pid(self):
954        num, num_warns, fin_file, fail_file = self.processor.doImport(
955            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
956        self.assertEqual(num_warns,0)
957        shutil.rmtree(os.path.dirname(fin_file))
958        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
959
960
961def test_suite():
962    suite = unittest.TestSuite()
963    for testcase in [
964        StudentProcessorTest,StudentStudyCourseProcessorTest,
965        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
966        PaymentProcessorTest,StudentVerdictProcessorTest]:
967        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
968                testcase
969                )
970        )
971    return suite
972
973
Note: See TracBrowser for help on using the repository browser.