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

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

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