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

Last change on this file since 9843 was 9799, checked in by uli, 12 years ago

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