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

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

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