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

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

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