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

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

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