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

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

Disable states and transitions which are not allowed for pg students. Not yet fully tested.
  • Property svn:keywords set to Id
File size: 40.5 KB
Line 
1## $Id: test_batching.py 9028 2012-07-20 08:56:34Z 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
243
244class StudentProcessorTest(StudentImportExportSetup):
245
246    layer = FunctionalLayer
247
248    def setUp(self):
249        super(StudentProcessorTest, self).setUp()
250
251        # Add student with subobjects
252        student = Student()
253        self.app['students'].addStudent(student)
254        student = self.setup_student(student)
255        notify(grok.ObjectModifiedEvent(student))
256        self.student = self.app['students'][student.student_id]
257
258        self.processor = StudentProcessor()
259        self.workdir = tempfile.mkdtemp()
260        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
261        self.csv_file_update = os.path.join(
262            self.workdir, 'sample_student_data_update.csv')
263        self.csv_file_update2 = os.path.join(
264            self.workdir, 'sample_student_data_update2.csv')
265        self.csv_file_update3 = os.path.join(
266            self.workdir, 'sample_student_data_update3.csv')
267        self.csv_file_update4 = os.path.join(
268            self.workdir, 'sample_student_data_update4.csv')
269        self.csv_file_migration = os.path.join(
270            self.workdir, 'sample_student_data_migration.csv')
271        self.csv_file_duplicates = os.path.join(
272            self.workdir, 'sample_student_data_duplicates.csv')
273        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
274        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
275        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
276        open(self.csv_file_update3, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE3)
277        open(self.csv_file_update4, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE4)
278        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
279        open(self.csv_file_duplicates, 'wb').write(STUDENT_SAMPLE_DATA_DUPLICATES)
280
281    def test_interface(self):
282        # Make sure we fulfill the interface contracts.
283        assert verifyObject(IBatchProcessor, self.processor) is True
284        assert verifyClass(
285            IBatchProcessor, StudentProcessor) is True
286
287    def test_parentsExist(self):
288        self.assertFalse(self.processor.parentsExist(None, dict()))
289        self.assertTrue(self.processor.parentsExist(None, self.app))
290
291    def test_entryExists(self):
292        assert self.processor.entryExists(
293            dict(student_id='ID_NONE'), self.app) is False
294        assert self.processor.entryExists(
295            dict(reg_number='123'), self.app) is True
296
297    def test_getParent(self):
298        parent = self.processor.getParent(None, self.app)
299        assert parent is self.app['students']
300
301    def test_getEntry(self):
302        assert self.processor.getEntry(
303            dict(student_id='ID_NONE'), self.app) is None
304        assert self.processor.getEntry(
305            dict(student_id=self.student.student_id), self.app) is self.student
306
307    def test_addEntry(self):
308        new_student = Student()
309        self.processor.addEntry(
310            new_student, dict(), self.app)
311        assert len(self.app['students'].keys()) == 2
312
313    def test_checkConversion(self):
314        # Make sure we can check conversions and that the stud_id
315        # counter is not raised during such checks.
316        initial_stud_id = self.app['students']._curr_stud_id
317        errs, inv_errs, conv_dict = self.processor.checkConversion(
318            dict(reg_number='1', state='admitted'))
319        self.assertEqual(len(errs),0)
320        # Empty state is allowed
321        errs, inv_errs, conv_dict = self.processor.checkConversion(
322            dict(reg_number='1', state=''))
323        self.assertEqual(len(errs),0)
324        #self.assertTrue(('state', 'no value provided') in errs)
325        errs, inv_errs, conv_dict = self.processor.checkConversion(
326            dict(reg_number='1', state='nonsense'))
327        self.assertEqual(len(errs),1)
328        self.assertTrue(('state', 'not allowed') in errs)
329        new_stud_id = self.app['students']._curr_stud_id
330        self.assertEqual(initial_stud_id, new_stud_id)
331        return
332
333    def test_checkUpdateRequirements(self):
334        # Make sure that pg students can't be updated with wrong transition.
335        err = self.processor.checkUpdateRequirements(self.student,
336            dict(reg_number='1', state='returning'), self.app)
337        self.assertTrue(err is None)
338        self.certificate.study_mode = 'pg_ft'
339        err = self.processor.checkUpdateRequirements(self.student,
340            dict(reg_number='1', state='returning'), self.app)
341        self.assertEqual(err, 'State not allowed (pg student).')
342        IWorkflowState(self.student).setState('school fee paid')
343        err = self.processor.checkUpdateRequirements(self.student,
344            dict(reg_number='1', transition='reset6'), self.app)
345        self.assertEqual(err, 'Transition not allowed (pg student).')
346        err = self.processor.checkUpdateRequirements(self.student,
347            dict(reg_number='1', transition='register_courses'), self.app)
348        self.assertEqual(err, 'Transition not allowed (pg student).')
349
350
351    def test_delEntry(self):
352        assert self.student.student_id in self.app['students'].keys()
353        self.processor.delEntry(
354            dict(reg_number=self.student.reg_number), self.app)
355        assert self.student.student_id not in self.app['students'].keys()
356
357    def test_import(self):
358        self.assertEqual(self.app['students']._curr_stud_id, 1000001)
359        num, num_warns, fin_file, fail_file = self.processor.doImport(
360            self.csv_file, STUDENT_HEADER_FIELDS)
361        self.assertEqual(num_warns,0)
362        assert len(self.app['students'].keys()) == 5
363        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
364        self.assertEqual(
365            self.app['students']['X666666'].state, 'courses validated')
366        # Two new student_ids have been created.
367        self.assertEqual(self.app['students']._curr_stud_id, 1000003)
368        shutil.rmtree(os.path.dirname(fin_file))
369
370    def test_import_update(self):
371        num, num_warns, fin_file, fail_file = self.processor.doImport(
372            self.csv_file, STUDENT_HEADER_FIELDS)
373        shutil.rmtree(os.path.dirname(fin_file))
374        num, num_warns, fin_file, fail_file = self.processor.doImport(
375            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
376        self.assertEqual(num_warns,0)
377        # state has changed
378        self.assertEqual(self.app['students']['X666666'].state,'admitted')
379        # state has not changed
380        self.assertEqual(self.app['students']['Y777777'].state,'courses validated')
381        shutil.rmtree(os.path.dirname(fin_file))
382
383    def test_import_update2(self):
384        num, num_warns, fin_file, fail_file = self.processor.doImport(
385            self.csv_file, STUDENT_HEADER_FIELDS)
386        shutil.rmtree(os.path.dirname(fin_file))
387        num, num_warns, fin_file, fail_file = self.processor.doImport(
388            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
389        self.assertEqual(num_warns,0)
390        # The phone import value of Pieri was None.
391        # Confirm that phone has not been cleared.
392        container = self.app['students']
393        for key in container.keys():
394            if container[key].firstname == 'Aaren':
395                aaren = container[key]
396                break
397        self.assertEqual(aaren.phone, '--1234')
398        # The phone import value of Claus was a deletion marker.
399        # Confirm that phone has been cleared.
400        for key in container.keys():
401            if container[key].firstname == 'Claus':
402                claus = container[key]
403                break
404        assert claus.phone is None
405        shutil.rmtree(os.path.dirname(fin_file))
406
407    def test_import_update3(self):
408        num, num_warns, fin_file, fail_file = self.processor.doImport(
409            self.csv_file, STUDENT_HEADER_FIELDS)
410        shutil.rmtree(os.path.dirname(fin_file))
411        num, num_warns, fin_file, fail_file = self.processor.doImport(
412            self.csv_file_update3, STUDENT_HEADER_FIELDS_UPDATE3, 'update')
413        content = open(fail_file).read()
414        self.assertEqual(
415            content,
416            'reg_number,student_id,transition,--ERRORS--\r\n'
417            '<IGNORE>,X666666,request_clearance,Transition not allowed.\r\n'
418            )
419        self.assertEqual(num_warns,1)
420        self.assertEqual(self.app['students']['Y777777'].state,'returning')
421
422    def test_import_update4(self):
423        num, num_warns, fin_file, fail_file = self.processor.doImport(
424            self.csv_file, STUDENT_HEADER_FIELDS)
425        shutil.rmtree(os.path.dirname(fin_file))
426        self.assertRaises(
427            FatalCSVError, self.processor.doImport, self.csv_file_update4,
428            STUDENT_HEADER_FIELDS_UPDATE4, 'update')
429
430    def test_import_remove(self):
431        num, num_warns, fin_file, fail_file = self.processor.doImport(
432            self.csv_file, STUDENT_HEADER_FIELDS)
433        shutil.rmtree(os.path.dirname(fin_file))
434        num, num_warns, fin_file, fail_file = self.processor.doImport(
435            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
436        self.assertEqual(num_warns,0)
437        shutil.rmtree(os.path.dirname(fin_file))
438
439    def test_import_migration_data(self):
440        num, num_warns, fin_file, fail_file = self.processor.doImport(
441            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
442        content = open(fail_file).read()
443        self.assertEqual(num_warns,2)
444        assert len(self.app['students'].keys()) == 5
445        self.assertEqual(
446            content,
447            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
448            '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n'
449            '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n'
450            )
451        students = self.app['students']
452        self.assertTrue('A123456' in students.keys())
453        self.assertEqual(students['A123456'].state, 'clearance started')
454        self.assertEqual(students['A123456'].date_of_birth,
455                         datetime.date(1990, 1, 2))
456        self.assertFalse(students['A123456'].clearance_locked)
457        self.assertEqual(students['B123456'].state, 'cleared')
458        self.assertEqual(students['B123456'].date_of_birth,
459                         datetime.date(1990, 1, 3))
460        self.assertTrue(students['B123456'].clearance_locked)
461        history = ' '.join(students['A123456'].history.messages)
462        self.assertTrue(
463            "State 'clearance started' set by system" in history)
464        # state was empty and student is thus in state created
465        self.assertEqual(students['F123456'].state,'created')
466        # passwords were set correctly
467        self.assertEqual(
468            IUserAccount(students['A123456']).checkPassword('mypw1'), True)
469        self.assertEqual(
470            IUserAccount(students['C123456']).checkPassword('mypw1'), True)
471        shutil.rmtree(os.path.dirname(fin_file))
472
473    def test_import_duplicate_data(self):
474        num, num_warns, fin_file, fail_file = self.processor.doImport(
475            self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES)
476        content = open(fail_file).read()
477        self.assertEqual(num_warns,4)
478        self.assertEqual(
479            content,
480            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
481            '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n'
482            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n'
483            '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n'
484            '3,Uli,A123456,m,aa@aa.ng,1234,,1990-01-07,Schulz,,100002,This object already exists. Skipping.\r\n'
485            )
486        shutil.rmtree(os.path.dirname(fin_file))
487
488class StudentStudyCourseProcessorTest(StudentImportExportSetup):
489
490    def setUp(self):
491        super(StudentStudyCourseProcessorTest, self).setUp()
492
493        # Import students with subobjects
494        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
495        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
496        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
497            student_file, STUDENT_HEADER_FIELDS)
498        shutil.rmtree(os.path.dirname(fin_file))
499
500        self.processor = StudentStudyCourseProcessor()
501        self.csv_file = os.path.join(
502            self.workdir, 'sample_studycourse_data.csv')
503        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
504        return
505
506    def test_interface(self):
507        # Make sure we fulfill the interface contracts.
508        assert verifyObject(IBatchProcessor, self.processor) is True
509        assert verifyClass(
510            IBatchProcessor, StudentStudyCourseProcessor) is True
511
512    def test_entryExists(self):
513        assert self.processor.entryExists(
514            dict(reg_number='REG_NONE'), self.app) is False
515        assert self.processor.entryExists(
516            dict(reg_number='1'), self.app) is True
517
518    def test_getEntry(self):
519        student = self.processor.getEntry(
520            dict(reg_number='1'), self.app).__parent__
521        self.assertEqual(student.reg_number,'1')
522
523    def test_checkConversion(self):
524        errs, inv_errs, conv_dict = self.processor.checkConversion(
525            dict(reg_number='1', certificate='CERT1', current_level='200'))
526        self.assertEqual(len(errs),0)
527        errs, inv_errs, conv_dict = self.processor.checkConversion(
528            dict(reg_number='1', certificate='CERT999'))
529        self.assertEqual(len(errs),1)
530        self.assertTrue(('certificate', u'Invalid value') in errs)
531        errs, inv_errs, conv_dict = self.processor.checkConversion(
532            dict(reg_number='1', certificate='CERT1', current_level='100'))
533        self.assertEqual(len(errs),1)
534        self.assertTrue(('current_level','not in range') in errs)
535        # If we import only current_level, no conversion checking is done.
536        errs, inv_errs, conv_dict = self.processor.checkConversion(
537            dict(reg_number='1', current_level='100'))
538        self.assertEqual(len(errs),0)
539
540    def test_import(self):
541        num, num_warns, fin_file, fail_file = self.processor.doImport(
542            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
543        self.assertEqual(num_warns,1)
544        content = open(fail_file).read()
545        self.assertTrue('current_level: not in range' in content)
546        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
547        self.assertEqual(studycourse.certificate.code, u'CERT1')
548        shutil.rmtree(os.path.dirname(fin_file))
549
550class StudentVerdictProcessorTest(StudentImportExportSetup):
551
552
553    def setUp(self):
554        super(StudentVerdictProcessorTest, self).setUp()
555
556        # Import students with subobjects
557        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
558        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
559        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
560            student_file, STUDENT_HEADER_FIELDS)
561        shutil.rmtree(os.path.dirname(fin_file))
562
563        # Update study courses
564        studycourse_file = os.path.join(
565            self.workdir, 'sample_studycourse_data.csv')
566        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
567        processor = StudentStudyCourseProcessor()
568        num, num_warns, fin_file, fail_file = processor.doImport(
569            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
570        shutil.rmtree(os.path.dirname(fin_file))
571
572        self.processor = StudentVerdictProcessor()
573        self.csv_file = os.path.join(
574            self.workdir, 'sample_verdict_data.csv')
575        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
576        return
577
578    def test_import(self):
579        num, num_warns, fin_file, fail_file = self.processor.doImport(
580            self.csv_file, VERDICT_HEADER_FIELDS,'update')
581        self.assertEqual(num_warns,3)
582        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
583        student = self.processor.getParent(dict(matric_number='100000'), self.app)
584        self.assertEqual(studycourse.current_verdict, 'A')
585        self.assertEqual(student.state, 'returning')
586        self.assertEqual(studycourse.current_level, 200)
587        content = open(fail_file).read()
588        self.assertEqual(
589            content,
590            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
591            '2008,B,100001,100,Current level does not correspond.\r\n'
592            '2007,C,100002,200,Current session does not correspond.\r\n'
593            '2008,A,100003,200,Student in wrong state.\r\n'
594            )
595        shutil.rmtree(os.path.dirname(fin_file))
596
597
598class StudentStudyLevelProcessorTest(StudentImportExportSetup):
599
600    def setUp(self):
601        super(StudentStudyLevelProcessorTest, self).setUp()
602
603        # Import students with subobjects
604        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
605        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
606        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
607            student_file, STUDENT_HEADER_FIELDS)
608        shutil.rmtree(os.path.dirname(fin_file))
609
610        # Update study courses
611        studycourse_file = os.path.join(
612            self.workdir, 'sample_studycourse_data.csv')
613        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
614        processor = StudentStudyCourseProcessor()
615        num, num_warns, fin_file, fail_file = processor.doImport(
616            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
617        shutil.rmtree(os.path.dirname(fin_file))
618
619        self.processor = StudentStudyLevelProcessor()
620        self.csv_file = os.path.join(
621            self.workdir, 'sample_studylevel_data.csv')
622        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
623
624    def test_interface(self):
625        # Make sure we fulfill the interface contracts.
626        assert verifyObject(IBatchProcessor, self.processor) is True
627        assert verifyClass(
628            IBatchProcessor, StudentStudyLevelProcessor) is True
629
630    def test_checkConversion(self):
631        errs, inv_errs, conv_dict = self.processor.checkConversion(
632            dict(reg_number='1', level='220'))
633        self.assertEqual(len(errs),0)
634        errs, inv_errs, conv_dict = self.processor.checkConversion(
635            dict(reg_number='1', level='900'))
636        self.assertEqual(len(errs),1)
637        self.assertTrue(('level','no valid integer') in errs)
638        errs, inv_errs, conv_dict = self.processor.checkConversion(
639            dict(reg_number='1', level='xyz'))
640        self.assertEqual(len(errs),1)
641        self.assertTrue(('level','no integer') in errs)
642
643    def test_import(self):
644        num, num_warns, fin_file, fail_file = self.processor.doImport(
645            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
646        self.assertEqual(num_warns,2)
647        assert self.processor.entryExists(
648            dict(reg_number='1', level='100'), self.app) is True
649        studylevel = self.processor.getEntry(
650            dict(reg_number='1', level='100'), self.app)
651        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
652        self.assertEqual(studylevel.level_session, 2008)
653        self.assertEqual(studylevel.level_verdict, 'A')
654        self.assertEqual(studylevel.level, 100)
655        shutil.rmtree(os.path.dirname(fin_file))
656
657        logcontent = open(self.logfile).read()
658        # Logging message from updateEntry,
659        self.assertTrue(
660            'INFO - system - K1000000 - Study level '
661            'updated: level=100, level_verdict=C, level_session=2009'
662            in logcontent)
663
664    def test_import_update(self):
665        # We perform the same import twice,
666        # the second time in update mode. The number
667        # of warnings must be the same.
668        num, num_warns, fin_file, fail_file = self.processor.doImport(
669            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
670        num, num_warns, fin_file, fail_file = self.processor.doImport(
671            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
672        self.assertEqual(num_warns,2)
673        shutil.rmtree(os.path.dirname(fin_file))
674       
675
676class CourseTicketProcessorTest(StudentImportExportSetup):
677
678    def setUp(self):
679        super(CourseTicketProcessorTest, self).setUp()
680
681        # Import students with subobjects
682        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
683        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
684        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
685            student_file, STUDENT_HEADER_FIELDS)
686        shutil.rmtree(os.path.dirname(fin_file))
687
688        # Add course and certificate course
689        self.course = createObject('waeup.Course')
690        self.course.code = 'COURSE1'
691        self.course.semester = 1
692        self.course.credits = 10
693        self.course.passmark = 40
694        self.app['faculties']['fac1']['dep1'].courses.addCourse(
695            self.course)
696        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
697            self.course, level=100)
698
699        # Update study courses
700        studycourse_file = os.path.join(
701            self.workdir, 'sample_studycourse_data.csv')
702        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
703        processor = StudentStudyCourseProcessor()
704        num, num_warns, fin_file, fail_file = processor.doImport(
705            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
706        shutil.rmtree(os.path.dirname(fin_file))
707
708        # Import study levels
709        processor = StudentStudyLevelProcessor()
710        studylevel_file = os.path.join(
711            self.workdir, 'sample_studylevel_data.csv')
712        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
713        num, num_warns, fin_file, fail_file = processor.doImport(
714            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
715        shutil.rmtree(os.path.dirname(fin_file))
716
717        self.processor = CourseTicketProcessor()
718        self.csv_file = os.path.join(
719            self.workdir, 'sample_courseticket_data.csv')
720        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
721
722    def test_interface(self):
723        # Make sure we fulfill the interface contracts.
724        assert verifyObject(IBatchProcessor, self.processor) is True
725        assert verifyClass(
726            IBatchProcessor, CourseTicketProcessor) is True
727
728    def test_checkConversion(self):
729        errs, inv_errs, conv_dict = self.processor.checkConversion(
730            dict(reg_number='1', code='COURSE1', level='220'))
731        self.assertEqual(len(errs),0)
732        errs, inv_errs, conv_dict = self.processor.checkConversion(
733            dict(reg_number='1', code='COURSE2', level='220'))
734        self.assertEqual(len(errs),1)
735        self.assertTrue(('code','non-existent') in errs)
736
737    def test_import(self):
738
739        num, num_warns, fin_file, fail_file = self.processor.doImport(
740            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
741
742        self.assertEqual(num_warns,2)
743        assert self.processor.entryExists(
744            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
745        courseticket = self.processor.getEntry(
746            dict(reg_number='1', level='100', code='COURSE1'), self.app)
747        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
748        self.assertEqual(courseticket.score, 1)
749        self.assertEqual(courseticket.mandatory, True)
750        self.assertEqual(courseticket.fcode, 'NA')
751        self.assertEqual(courseticket.dcode, 'NA')
752        self.assertEqual(courseticket.code, 'COURSE1')
753        self.assertEqual(courseticket.title, 'Unnamed Course')
754        self.assertEqual(courseticket.credits, 10)
755        self.assertEqual(courseticket.passmark, 40)
756        self.assertEqual(courseticket.semester, 1)
757        shutil.rmtree(os.path.dirname(fin_file))
758
759        logcontent = open(self.logfile).read()
760        # Logging message from updateEntry,
761        self.assertTrue(
762            'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, '
763            'mandatory=False, score=3'
764            in logcontent)
765
766    def test_import_update(self):
767        # We perform the same import twice,
768        # the second time in update mode. The number
769        # of warnings must be the same.
770        num, num_warns, fin_file, fail_file = self.processor.doImport(
771            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
772        num, num_warns, fin_file, fail_file = self.processor.doImport(
773            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
774        self.assertEqual(num_warns,2)
775        shutil.rmtree(os.path.dirname(fin_file))
776
777    def test_import_remove(self):
778        # We perform the same import twice,
779        # the second time in remove mode. The number
780        # of warnings must be the same.
781        num, num_warns, fin_file, fail_file = self.processor.doImport(
782            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
783        assert self.processor.entryExists(
784            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
785        num, num_warns, fin_file, fail_file = self.processor.doImport(
786            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
787        self.assertEqual(num_warns,2)
788        assert self.processor.entryExists(
789            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
790        shutil.rmtree(os.path.dirname(fin_file))
791        logcontent = open(self.logfile).read()
792        self.assertTrue(
793            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
794            in logcontent)
795
796class PaymentProcessorTest(StudentImportExportSetup):
797
798    def setUp(self):
799        super(PaymentProcessorTest, self).setUp()
800
801        # Add student with payment
802        student = Student()
803        student.firstname = u'Anna'
804        student.lastname = u'Tester'
805        student.reg_number = u'123'
806        student.matric_number = u'234'
807        self.app['students'].addStudent(student)
808        self.student = self.app['students'][student.student_id]
809        payment = createObject(u'waeup.StudentOnlinePayment')
810        payment.p_id = 'p120'
811        self.student['payments'][payment.p_id] = payment
812
813        # Import students with subobjects
814        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
815        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
816        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
817            student_file, STUDENT_HEADER_FIELDS)
818        shutil.rmtree(os.path.dirname(fin_file))
819
820        self.processor = StudentOnlinePaymentProcessor()
821        self.csv_file = os.path.join(
822            self.workdir, 'sample_payment_data.csv')
823        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
824        self.csv_file2 = os.path.join(
825            self.workdir, 'sample_create_payment_data.csv')
826        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
827
828    def test_interface(self):
829        # Make sure we fulfill the interface contracts.
830        assert verifyObject(IBatchProcessor, self.processor) is True
831        assert verifyClass(
832            IBatchProcessor, StudentOnlinePaymentProcessor) is True
833
834    def test_getEntry(self):
835        assert self.processor.getEntry(
836            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
837        assert self.processor.getEntry(
838            dict(student_id=self.student.student_id, p_id='p120'),
839            self.app) is self.student['payments']['p120']
840        assert self.processor.getEntry(
841            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
842            self.app) is self.student['payments']['p120']
843
844    def test_delEntry(self):
845        assert self.processor.getEntry(
846            dict(student_id=self.student.student_id, p_id='p120'),
847            self.app) is self.student['payments']['p120']
848        self.assertEqual(len(self.student['payments'].keys()),1)
849        self.processor.delEntry(
850            dict(student_id=self.student.student_id, p_id='p120'),
851            self.app)
852        assert self.processor.getEntry(
853            dict(student_id=self.student.student_id, p_id='p120'),
854            self.app) is None
855        self.assertEqual(len(self.student['payments'].keys()),0)
856
857    def test_addEntry(self):
858        self.assertEqual(len(self.student['payments'].keys()),1)
859        payment1 = createObject(u'waeup.StudentOnlinePayment')
860        payment1.p_id = 'p234'
861        self.processor.addEntry(
862            payment1, dict(student_id=self.student.student_id, p_id='p234'),
863            self.app)
864        self.assertEqual(len(self.student['payments'].keys()),2)
865        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
866        payment2 = createObject(u'waeup.StudentOnlinePayment')
867        payment1.p_id = 'nonsense'
868        # payment1.p_id will be replaced if p_id doesn't start with 'p'
869        self.processor.addEntry(
870            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
871            self.app)
872        self.assertEqual(len(self.student['payments'].keys()),3)
873        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
874
875    def test_checkConversion(self):
876        errs, inv_errs, conv_dict = self.processor.checkConversion(
877            dict(p_id='3816951266236341955'))
878        self.assertEqual(len(errs),0)
879        errs, inv_errs, conv_dict = self.processor.checkConversion(
880            dict(p_id='p1266236341955'))
881        self.assertEqual(len(errs),0)
882        errs, inv_errs, conv_dict = self.processor.checkConversion(
883            dict(p_id='nonsense'))
884        self.assertEqual(len(errs),1)
885        timestamp = ("%d" % int(time()*10000))[1:]
886        p_id = "p%s" % timestamp
887        errs, inv_errs, conv_dict = self.processor.checkConversion(
888            dict(p_id=p_id))
889        self.assertEqual(len(errs),0)
890
891    def test_import(self):
892        num, num_warns, fin_file, fail_file = self.processor.doImport(
893            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
894        self.assertEqual(num_warns,0)
895        payment = self.processor.getEntry(dict(reg_number='1',
896            p_id='p2907979737440'), self.app)
897        self.assertEqual(payment.p_id, 'p2907979737440')
898        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
899        self.assertEqual(cdate, "2010-11-26 18:59:33")
900        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
901        shutil.rmtree(os.path.dirname(fin_file))
902        logcontent = open(self.logfile).read()
903        # Logging message from updateEntry
904        self.assertTrue(
905            'INFO - system - K1000001 - Payment ticket updated: '
906            'p_id=p1266236341955, p_item=BTECHBDT, '
907            'creation_date=2010-02-15 13:19:01+00:00, r_code=00, '
908            'r_amount_approved=19500.0, p_category=schoolfee, '
909            'amount_auth=19500.0, p_state=paid'
910            in logcontent)
911
912    def test_import_update(self):
913        # We perform the same import twice,
914        # the second time in update mode. The number
915        # of warnings must be the same.
916        num, num_warns, fin_file, fail_file = self.processor.doImport(
917            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
918        num, num_warns, fin_file, fail_file = self.processor.doImport(
919            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
920        self.assertEqual(num_warns,0)
921        shutil.rmtree(os.path.dirname(fin_file))
922
923    def test_import_remove(self):
924        num, num_warns, fin_file, fail_file = self.processor.doImport(
925            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
926        num, num_warns, fin_file, fail_file = self.processor.doImport(
927            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
928        self.assertEqual(num_warns,0)
929        shutil.rmtree(os.path.dirname(fin_file))
930        logcontent = open(self.logfile).read()
931        self.assertTrue(
932            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
933            in logcontent)
934
935    def test_import_wo_pid(self):
936        num, num_warns, fin_file, fail_file = self.processor.doImport(
937            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
938        self.assertEqual(num_warns,0)
939        shutil.rmtree(os.path.dirname(fin_file))
940        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
941
942
943def test_suite():
944    suite = unittest.TestSuite()
945    for testcase in [
946        StudentProcessorTest,StudentStudyCourseProcessorTest,
947        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
948        PaymentProcessorTest,StudentVerdictProcessorTest]:
949        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
950                testcase
951                )
952        )
953    return suite
954
955
Note: See TracBrowser for help on using the repository browser.