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

Last change on this file since 15066 was 15066, checked in by Henrik Bettermann, 6 years ago

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