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

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

Add regression test. The level attribute of a StudentStudyLevel? object must be integer after update.

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