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

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

Current level must be in range of certificate. This must be checked in checkUpdateRequirements.
  • Property svn:keywords set to Id
File size: 49.3 KB
Line 
1## $Id: test_batching.py 9735 2012-11-28 07:52:56Z 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        # Current level must be in range of certificate.
578        # Since row has passed the converter, current_level is an integer.
579        err = self.processor.checkUpdateRequirements(
580            self.student['studycourse'],
581            dict(reg_number='1', current_level=100), self.app)
582        self.assertEqual(err, 'current_level not in range.')
583        err = self.processor.checkUpdateRequirements(
584            self.student['studycourse'],
585            dict(reg_number='1', current_level=200), self.app)
586        self.assertTrue(err is None)
587        # We can update pg students.
588        self.student['studycourse'].certificate.start_level=999
589        self.student['studycourse'].certificate.end_level=999
590        err = self.processor.checkUpdateRequirements(
591            self.student['studycourse'],
592            dict(reg_number='1', current_level=999), self.app)
593        self.assertTrue(err is None)
594        # Make sure that pg students can't be updated with wrong transition.
595        IWorkflowState(self.student).setState('returning')
596        err = self.processor.checkUpdateRequirements(
597            self.student['studycourse'],
598            dict(reg_number='1', current_level=999), self.app)
599        self.assertEqual(err, 'Not a pg student.')
600        # If certificate is not given in row (and has thus
601        # successfully passed checkConversion) the certificate
602        # attribute must be set.
603        self.student['studycourse'].certificate = None
604        err = self.processor.checkUpdateRequirements(
605            self.student['studycourse'],
606            dict(reg_number='1', current_level=100), self.app)
607        self.assertEqual(err, 'No certificate to check level.')
608
609    def test_import(self):
610        num, num_warns, fin_file, fail_file = self.processor.doImport(
611            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
612        self.assertEqual(num_warns,1)
613        content = open(fail_file).read()
614        self.assertTrue('current_level: not in range' in content)
615        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
616        self.assertEqual(studycourse.certificate.code, u'CERT1')
617        shutil.rmtree(os.path.dirname(fin_file))
618
619class StudentStudyLevelProcessorTest(StudentImportExportSetup):
620
621    def setUp(self):
622        super(StudentStudyLevelProcessorTest, self).setUp()
623
624        # Import students with subobjects
625        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
626        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
627        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
628            student_file, STUDENT_HEADER_FIELDS)
629        shutil.rmtree(os.path.dirname(fin_file))
630
631        # Update study courses
632        studycourse_file = os.path.join(
633            self.workdir, 'sample_studycourse_data.csv')
634        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
635        processor = StudentStudyCourseProcessor()
636        num, num_warns, fin_file, fail_file = processor.doImport(
637            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
638        shutil.rmtree(os.path.dirname(fin_file))
639
640        self.processor = StudentStudyLevelProcessor()
641        self.csv_file = os.path.join(
642            self.workdir, 'sample_studylevel_data.csv')
643        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
644
645    def test_interface(self):
646        # Make sure we fulfill the interface contracts.
647        assert verifyObject(IBatchProcessor, self.processor) is True
648        assert verifyClass(
649            IBatchProcessor, StudentStudyLevelProcessor) is True
650
651    def test_checkConversion(self):
652        errs, inv_errs, conv_dict = self.processor.checkConversion(
653            dict(reg_number='1', level='220'))
654        self.assertEqual(len(errs),0)
655        errs, inv_errs, conv_dict = self.processor.checkConversion(
656            dict(reg_number='1', level='999'))
657        self.assertEqual(len(errs),0)
658        errs, inv_errs, conv_dict = self.processor.checkConversion(
659            dict(reg_number='1', level='1000'))
660        self.assertEqual(len(errs),1)
661        self.assertTrue(('level','no valid integer') in errs)
662        errs, inv_errs, conv_dict = self.processor.checkConversion(
663            dict(reg_number='1', level='xyz'))
664        self.assertEqual(len(errs),1)
665        self.assertTrue(('level','no integer') in errs)
666
667    def test_import(self):
668        num, num_warns, fin_file, fail_file = self.processor.doImport(
669            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
670        self.assertEqual(num_warns,2)
671        assert self.processor.entryExists(
672            dict(reg_number='1', level='100'), self.app) is True
673        studylevel = self.processor.getEntry(
674            dict(reg_number='1', level='100'), self.app)
675        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
676        self.assertEqual(studylevel.level_session, 2008)
677        self.assertEqual(studylevel.level_verdict, None)
678        self.assertEqual(studylevel.level, 100)
679        shutil.rmtree(os.path.dirname(fin_file))
680
681        logcontent = open(self.logfile).read()
682        # Logging message from updateEntry,
683        self.assertTrue(
684            'INFO - system - StudentStudyLevel Processor - '
685            'sample_studylevel_data - K1000000 - updated: '
686            'level=100, level_verdict=C, level_session=2009'
687            in logcontent)
688
689    def test_import_update(self):
690        # We perform the same import twice,
691        # the second time in update mode. The number
692        # of warnings must be the same.
693        num, num_warns, fin_file, fail_file = self.processor.doImport(
694            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
695        shutil.rmtree(os.path.dirname(fin_file))
696        num, num_warns, fin_file, fail_file = self.processor.doImport(
697            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
698        self.assertEqual(num_warns,2)
699        shutil.rmtree(os.path.dirname(fin_file))
700
701    def test_import_remove(self):
702        # We perform the same import twice,
703        # the second time in remove mode. The number
704        # of warnings must be the same.
705        num, num_warns, fin_file, fail_file = self.processor.doImport(
706            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
707        shutil.rmtree(os.path.dirname(fin_file))
708        num, num_warns, fin_file, fail_file = self.processor.doImport(
709            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'remove')
710        assert self.processor.entryExists(
711            dict(reg_number='1', level='100'), self.app) is False
712        self.assertEqual(num_warns,2)
713
714        shutil.rmtree(os.path.dirname(fin_file))
715
716class CourseTicketProcessorTest(StudentImportExportSetup):
717
718    def setUp(self):
719        super(CourseTicketProcessorTest, self).setUp()
720
721        # Import students with subobjects
722        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
723        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
724        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
725            student_file, STUDENT_HEADER_FIELDS)
726        shutil.rmtree(os.path.dirname(fin_file))
727
728        # Add course and certificate course
729        self.course = createObject('waeup.Course')
730        self.course.code = 'COURSE1'
731        self.course.semester = 1
732        self.course.credits = 10
733        self.course.passmark = 40
734        self.app['faculties']['fac1']['dep1'].courses.addCourse(
735            self.course)
736        self.app['faculties']['fac1']['dep1'].certificates[
737            'CERT1'].addCertCourse(
738            self.course, level=100)
739
740        # Update study courses
741        studycourse_file = os.path.join(
742            self.workdir, 'sample_studycourse_data.csv')
743        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
744        processor = StudentStudyCourseProcessor()
745        num, num_warns, fin_file, fail_file = processor.doImport(
746            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
747        shutil.rmtree(os.path.dirname(fin_file))
748
749        # Import study levels
750        processor = StudentStudyLevelProcessor()
751        studylevel_file = os.path.join(
752            self.workdir, 'sample_studylevel_data.csv')
753        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
754        num, num_warns, fin_file, fail_file = processor.doImport(
755            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
756        shutil.rmtree(os.path.dirname(fin_file))
757
758        self.processor = CourseTicketProcessor()
759        self.csv_file = os.path.join(
760            self.workdir, 'sample_courseticket_data.csv')
761        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
762
763    def test_interface(self):
764        # Make sure we fulfill the interface contracts.
765        assert verifyObject(IBatchProcessor, self.processor) is True
766        assert verifyClass(
767            IBatchProcessor, CourseTicketProcessor) is True
768
769    def test_checkConversion(self):
770        errs, inv_errs, conv_dict = self.processor.checkConversion(
771            dict(reg_number='1', code='COURSE1', level='220'))
772        self.assertEqual(len(errs),0)
773        errs, inv_errs, conv_dict = self.processor.checkConversion(
774            dict(reg_number='1', code='COURSE2', level='220'))
775        self.assertEqual(len(errs),1)
776        self.assertTrue(('code','non-existent') in errs)
777
778    def test_import(self):
779        num, num_warns, fin_file, fail_file = self.processor.doImport(
780            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
781        fail_file = open(fail_file).read()
782        self.assertEqual(num_warns,5)
783        self.assertEqual(fail_file,
784            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
785            '1,COURSE1,,nonsense,,5,,Not all parents do exist yet. Skipping\r\n'
786            '1,NONSENSE,,100,,5,,code: non-existent\r\n'
787            '1,COURSE1,,200,2004,6,,level_session: does not match 2008\r\n'
788            '1,COURSE1,,300,2008,6,,level: does not exist\r\n'
789            '1,COURSE1,,300,2008X,6,,level_session: Invalid value\r\n')
790        assert self.processor.entryExists(
791            dict(reg_number='1', level='100', code='COURSE1'),
792            self.app) is True
793        courseticket = self.processor.getEntry(
794            dict(reg_number='1', level='100', code='COURSE1'), self.app)
795        self.assertEqual(courseticket.__parent__.__parent__.certificate.code,
796                         u'CERT1')
797        self.assertEqual(courseticket.score, 1)
798        self.assertEqual(courseticket.mandatory, True)
799        self.assertEqual(courseticket.fcode, 'NA')
800        self.assertEqual(courseticket.dcode, 'NA')
801        self.assertEqual(courseticket.code, 'COURSE1')
802        self.assertEqual(courseticket.title, 'Unnamed Course')
803        self.assertEqual(courseticket.credits, 10)
804        self.assertEqual(courseticket.passmark, 40)
805        self.assertEqual(courseticket.semester, 1)
806        shutil.rmtree(os.path.dirname(fin_file))
807        logcontent = open(self.logfile).read()
808        # Logging message from updateEntry,
809        self.assertTrue(
810            'INFO - system - CourseTicket Processor - '
811            'sample_courseticket_data - K1000000 - 100 - '
812            'updated: code=COURSE1, '
813            'mandatory=False, score=3'
814            in logcontent)
815
816    def test_import_update(self):
817        # We perform the same import twice,
818        # the second time in update mode. The number
819        # of warnings must be the same.
820        num, num_warns, fin_file, fail_file = self.processor.doImport(
821            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
822        shutil.rmtree(os.path.dirname(fin_file))
823        num, num_warns, fin_file, fail_file = self.processor.doImport(
824            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
825        fail_file = open(fail_file).read()
826        self.assertEqual(num_warns,5)
827        self.assertEqual(fail_file,
828            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
829            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,5,<IGNORE>,Cannot update: no such entry\r\n'
830            '1,NONSENSE,<IGNORE>,100,<IGNORE>,5,<IGNORE>,code: non-existent\r\n'
831            '1,COURSE1,<IGNORE>,200,2004,6,<IGNORE>,level_session: does not match 2008\r\n'
832            '1,COURSE1,<IGNORE>,300,2008,6,<IGNORE>,level: does not exist\r\n'
833            '1,COURSE1,<IGNORE>,300,2008X,6,<IGNORE>,level_session: Invalid value\r\n')
834        shutil.rmtree(os.path.dirname(fin_file))
835
836    def test_import_remove(self):
837        # We perform the same import twice,
838        # the second time in remove mode. The number
839        # of warnings must be the same.
840        num, num_warns, fin_file, fail_file = self.processor.doImport(
841            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
842        shutil.rmtree(os.path.dirname(fin_file))
843        assert self.processor.entryExists(
844            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
845        num, num_warns, fin_file, fail_file = self.processor.doImport(
846            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
847        self.assertEqual(num_warns,5)
848        assert self.processor.entryExists(
849            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
850        shutil.rmtree(os.path.dirname(fin_file))
851        logcontent = open(self.logfile).read()
852        self.assertTrue(
853            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
854            in logcontent)
855
856class PaymentProcessorTest(StudentImportExportSetup):
857
858    def setUp(self):
859        super(PaymentProcessorTest, self).setUp()
860
861        # Add student with payment
862        student = Student()
863        student.firstname = u'Anna'
864        student.lastname = u'Tester'
865        student.reg_number = u'123'
866        student.matric_number = u'234'
867        self.app['students'].addStudent(student)
868        self.student = self.app['students'][student.student_id]
869        payment = createObject(u'waeup.StudentOnlinePayment')
870        payment.p_id = 'p120'
871        self.student['payments'][payment.p_id] = payment
872
873        # Import students with subobjects
874        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
875        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
876        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
877            student_file, STUDENT_HEADER_FIELDS)
878        shutil.rmtree(os.path.dirname(fin_file))
879
880        self.processor = StudentOnlinePaymentProcessor()
881        self.csv_file = os.path.join(
882            self.workdir, 'sample_payment_data.csv')
883        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
884        self.csv_file2 = os.path.join(
885            self.workdir, 'sample_create_payment_data.csv')
886        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
887
888    def test_interface(self):
889        # Make sure we fulfill the interface contracts.
890        assert verifyObject(IBatchProcessor, self.processor) is True
891        assert verifyClass(
892            IBatchProcessor, StudentOnlinePaymentProcessor) is True
893
894    def test_getEntry(self):
895        assert self.processor.getEntry(
896            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
897        assert self.processor.getEntry(
898            dict(student_id=self.student.student_id, p_id='p120'),
899            self.app) is self.student['payments']['p120']
900        assert self.processor.getEntry(
901            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
902            self.app) is self.student['payments']['p120']
903
904    def test_delEntry(self):
905        assert self.processor.getEntry(
906            dict(student_id=self.student.student_id, p_id='p120'),
907            self.app) is self.student['payments']['p120']
908        self.assertEqual(len(self.student['payments'].keys()),1)
909        self.processor.delEntry(
910            dict(student_id=self.student.student_id, p_id='p120'),
911            self.app)
912        assert self.processor.getEntry(
913            dict(student_id=self.student.student_id, p_id='p120'),
914            self.app) is None
915        self.assertEqual(len(self.student['payments'].keys()),0)
916
917    def test_addEntry(self):
918        self.assertEqual(len(self.student['payments'].keys()),1)
919        payment1 = createObject(u'waeup.StudentOnlinePayment')
920        payment1.p_id = 'p234'
921        self.processor.addEntry(
922            payment1, dict(student_id=self.student.student_id, p_id='p234'),
923            self.app)
924        self.assertEqual(len(self.student['payments'].keys()),2)
925        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
926        payment2 = createObject(u'waeup.StudentOnlinePayment')
927        payment1.p_id = 'nonsense'
928        # payment1.p_id will be replaced if p_id doesn't start with 'p'
929        # and is not an old PIN
930        self.processor.addEntry(
931            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
932            self.app)
933        self.assertEqual(len(self.student['payments'].keys()),3)
934        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
935
936    def test_checkConversion(self):
937        errs, inv_errs, conv_dict = self.processor.checkConversion(
938            dict(p_id='3816951266236341955'))
939        self.assertEqual(len(errs),0)
940        errs, inv_errs, conv_dict = self.processor.checkConversion(
941            dict(p_id='p1266236341955'))
942        self.assertEqual(len(errs),0)
943        errs, inv_errs, conv_dict = self.processor.checkConversion(
944            dict(p_id='ABC-11-1234567890'))
945        self.assertEqual(len(errs),0)
946        errs, inv_errs, conv_dict = self.processor.checkConversion(
947            dict(p_id='nonsense'))
948        self.assertEqual(len(errs),1)
949        timestamp = ("%d" % int(time()*10000))[1:]
950        p_id = "p%s" % timestamp
951        errs, inv_errs, conv_dict = self.processor.checkConversion(
952            dict(p_id=p_id))
953        self.assertEqual(len(errs),0)
954
955    def test_import(self):
956        num, num_warns, fin_file, fail_file = self.processor.doImport(
957            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
958        self.assertEqual(num_warns,0)
959
960        payment = self.processor.getEntry(dict(reg_number='1',
961            p_id='p2907979737440'), self.app)
962        self.assertEqual(payment.p_id, 'p2907979737440')
963        self.assertTrue(payment.p_current)
964        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
965        self.assertEqual(cdate, "2010-11-26 18:59:33")
966        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
967
968        payment = self.processor.getEntry(dict(matric_number='100001',
969            p_id='p2907125937570'), self.app)
970        self.assertEqual(payment.p_id, 'p2907125937570')
971        self.assertEqual(payment.amount_auth, 19500.1)
972        self.assertFalse(payment.p_current)
973        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
974        # Ooooh, still the old problem, see
975        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
976        # WAT is interpreted as GMT-1 and not GMT+1
977        self.assertEqual(cdate, "2010-11-25 21:16:33")
978        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
979
980        payment = self.processor.getEntry(dict(reg_number='3',
981            p_id='ABC-11-1234567890'), self.app)
982        self.assertEqual(payment.amount_auth, 19500.6)
983
984        shutil.rmtree(os.path.dirname(fin_file))
985        logcontent = open(self.logfile).read()
986        # Logging message from updateEntry
987        self.assertTrue(
988            'INFO - system - Student Payment Processor - '
989            'sample_payment_data - K1000001 - updated: '
990            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
991            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
992            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
993            'p_state=paid'
994            in logcontent)
995        self.assertTrue(
996            'INFO - system - Student Payment Processor - '
997            'sample_payment_data - K1000001 - updated: '
998            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
999            'p_category=schoolfee, amount_auth=19500.6, p_current=True, '
1000            'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, '
1001            'p_state=paid'
1002            in logcontent)
1003
1004    def test_import_update(self):
1005        # We perform the same import twice,
1006        # the second time in update mode. The number
1007        # of warnings must be the same.
1008        num, num_warns, fin_file, fail_file = self.processor.doImport(
1009            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1010        shutil.rmtree(os.path.dirname(fin_file))
1011        num, num_warns, fin_file, fail_file = self.processor.doImport(
1012            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
1013        self.assertEqual(num_warns,0)
1014        shutil.rmtree(os.path.dirname(fin_file))
1015
1016    def test_import_remove(self):
1017        num, num_warns, fin_file, fail_file = self.processor.doImport(
1018            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1019        shutil.rmtree(os.path.dirname(fin_file))
1020        num, num_warns, fin_file, fail_file = self.processor.doImport(
1021            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
1022        self.assertEqual(num_warns,0)
1023        shutil.rmtree(os.path.dirname(fin_file))
1024        logcontent = open(self.logfile).read()
1025        self.assertTrue(
1026            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
1027            in logcontent)
1028
1029    def test_import_wo_pid(self):
1030        num, num_warns, fin_file, fail_file = self.processor.doImport(
1031            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
1032        self.assertEqual(num_warns,0)
1033        shutil.rmtree(os.path.dirname(fin_file))
1034        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
1035
1036class StudentVerdictProcessorTest(StudentImportExportSetup):
1037
1038    def setUp(self):
1039        super(StudentVerdictProcessorTest, self).setUp()
1040
1041        # Import students with subobjects
1042        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
1043        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
1044        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
1045            student_file, STUDENT_HEADER_FIELDS)
1046        shutil.rmtree(os.path.dirname(fin_file))
1047
1048        # Update study courses
1049        studycourse_file = os.path.join(
1050            self.workdir, 'sample_studycourse_data.csv')
1051        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
1052        processor = StudentStudyCourseProcessor()
1053        num, num_warns, fin_file, fail_file = processor.doImport(
1054            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
1055        shutil.rmtree(os.path.dirname(fin_file))
1056        # Import study levels
1057        self.csv_file = os.path.join(
1058            self.workdir, 'sample_studylevel_data.csv')
1059        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
1060        processor = StudentStudyLevelProcessor()
1061        num, num_warns, fin_file, fail_file = processor.doImport(
1062            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
1063        content = open(fail_file).read()
1064        shutil.rmtree(os.path.dirname(fin_file))
1065
1066        self.processor = StudentVerdictProcessor()
1067        self.csv_file = os.path.join(
1068            self.workdir, 'sample_verdict_data.csv')
1069        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
1070        return
1071
1072    def test_import(self):
1073        studycourse = self.processor.getEntry(dict(matric_number='100000'),
1074                                              self.app)
1075        self.assertEqual(studycourse['200'].level_verdict, None)
1076        student = self.processor.getParent(
1077            dict(matric_number='100000'), self.app)
1078        num, num_warns, fin_file, fail_file = self.processor.doImport(
1079            self.csv_file, VERDICT_HEADER_FIELDS,'update')
1080        #content = open(fail_file).read()
1081        #import pdb; pdb.set_trace()
1082        self.assertEqual(num_warns,5)
1083        self.assertEqual(studycourse.current_verdict, '0')
1084        self.assertEqual(student.state, 'returning')
1085        self.assertEqual(studycourse.current_level, 200)
1086        self.assertEqual(studycourse['200'].level_verdict, '0')
1087        student = self.processor.getParent(
1088            dict(matric_number='100005'), self.app)
1089        self.assertEqual(student.state, 'returning')
1090        self.assertEqual(student['studycourse'].current_verdict, 'A')
1091        self.assertEqual(studycourse.current_level, 200)
1092        self.assertEqual(student['studycourse']['200'].validated_by, 'System')
1093        self.assertTrue(isinstance(
1094            student['studycourse']['200'].validation_date, datetime.datetime))
1095        student = self.processor.getParent(
1096            dict(matric_number='100008'), self.app)
1097        self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana')
1098        content = open(fail_file).read()
1099        self.assertEqual(
1100            content,
1101            'current_session,current_level,bypass_validation,current_verdict,'
1102            'matric_number,validated_by,--ERRORS--\r\n'
1103            '2008,100,False,B,100001,<IGNORE>,Current level does not correspond.\r\n'
1104            '2007,200,<IGNORE>,C,100002,<IGNORE>,Current session does not correspond.\r\n'
1105            '2008,200,<IGNORE>,A,100003,<IGNORE>,Student in wrong state.\r\n'
1106            '2008,200,<IGNORE>,<IGNORE>,100004,<IGNORE>,No verdict in import file.\r\n'
1107            '2008,200,True,A,100007,<IGNORE>,Study level object is missing.\r\n'
1108            )
1109        logcontent = open(self.logfile).read()
1110        self.assertMatches(
1111            '... INFO - system - Verdict Processor (special processor, '
1112            'update only) - sample_verdict_data - X666666 - '
1113            'updated: current_verdict=0...',
1114            logcontent)
1115        self.assertMatches(
1116            '... INFO - system - X666666 - Returned...',
1117            logcontent)
1118
1119        shutil.rmtree(os.path.dirname(fin_file))
1120
1121def test_suite():
1122    suite = unittest.TestSuite()
1123    for testcase in [
1124        StudentProcessorTest,StudentStudyCourseProcessorTest,
1125        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
1126        PaymentProcessorTest,StudentVerdictProcessorTest]:
1127        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
1128                testcase
1129                )
1130        )
1131    return suite
1132
1133
Note: See TracBrowser for help on using the repository browser.