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

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

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