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

Last change on this file since 14443 was 14366, checked in by Henrik Bettermann, 8 years ago

Fix tests after turn of the year.
  • Property svn:keywords set to Id
File size: 57.0 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 14366 2017-01-05 08:31:51Z 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,--ERRORS--\r\n'
484            '<IGNORE>,X666666,request_clearance,Transition not allowed.\r\n'
485            )
486        self.assertEqual(num_warns,1)
487        self.assertEqual(self.app['students']['Y777777'].state,'returning')
488
489    def test_import_update4(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        self.assertRaises(
494            FatalCSVError, self.processor.doImport, self.csv_file_update4,
495            STUDENT_HEADER_FIELDS_UPDATE4, 'update')
496
497    def test_import_remove(self):
498        num, num_warns, fin_file, fail_file = self.processor.doImport(
499            self.csv_file, STUDENT_HEADER_FIELDS)
500        shutil.rmtree(os.path.dirname(fin_file))
501        num, num_warns, fin_file, fail_file = self.processor.doImport(
502            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
503        self.assertEqual(num_warns,0)
504        shutil.rmtree(os.path.dirname(fin_file))
505
506    def test_import_migration_data(self):
507        num, num_warns, fin_file, fail_file = self.processor.doImport(
508            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
509        content = open(fail_file).read()
510        self.assertEqual(num_warns,2)
511        assert len(self.app['students'].keys()) == 5
512        self.assertEqual(
513            content,
514            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
515            '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n'
516            '5,John,E123456,x,aa@aa.ng,1234,<IGNORE>,1990-01-06,Kennedy,<IGNORE>,100004,sex: Invalid value\r\n'
517            )
518        students = self.app['students']
519        self.assertTrue('A123456' in students.keys())
520        self.assertEqual(students['A123456'].state, 'clearance started')
521        self.assertEqual(students['A123456'].date_of_birth,
522                         datetime.date(1990, 1, 2))
523        self.assertFalse(students['A123456'].clearance_locked)
524        self.assertEqual(students['B123456'].state, 'cleared')
525        self.assertEqual(students['B123456'].date_of_birth,
526                         datetime.date(1990, 1, 3))
527        self.assertTrue(students['B123456'].clearance_locked)
528        history = ' '.join(students['A123456'].history.messages)
529        self.assertTrue(
530            "State 'clearance started' set by system" in history)
531        # state was empty and student is thus in state created
532        self.assertEqual(students['F123456'].state,'created')
533        # passwords were set correctly
534        self.assertEqual(
535            IUserAccount(students['A123456']).checkPassword('mypw1'), True)
536        self.assertEqual(
537            IUserAccount(students['C123456']).checkPassword('mypw1'), True)
538        shutil.rmtree(os.path.dirname(fin_file))
539
540    def test_import_duplicate_data(self):
541        num, num_warns, fin_file, fail_file = self.processor.doImport(
542            self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES)
543        content = open(fail_file).read()
544        self.assertEqual(num_warns,4)
545        self.assertEqual(
546            content,
547            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
548            '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: Invalid input\r\n'
549            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: Invalid input\r\n'
550            '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'
551            '3,Uli,A123456,m,aa@aa.ng,1234,<IGNORE>,1990-01-07,Schulz,<IGNORE>,100002,This object already exists.\r\n'
552            )
553        shutil.rmtree(os.path.dirname(fin_file))
554
555class StudentStudyCourseProcessorTest(StudentImportExportSetup):
556
557    def setUp(self):
558        super(StudentStudyCourseProcessorTest, self).setUp()
559
560        # Add student with subobjects
561        student = Student()
562        self.app['students'].addStudent(student)
563        student = self.setup_student(student)
564        notify(grok.ObjectModifiedEvent(student))
565        self.student = self.app['students'][student.student_id]
566
567        # Import students with subobjects
568        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
569        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
570        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
571            student_file, STUDENT_HEADER_FIELDS)
572        shutil.rmtree(os.path.dirname(fin_file))
573
574        self.processor = StudentStudyCourseProcessor()
575        self.csv_file = os.path.join(
576            self.workdir, 'sample_studycourse_data.csv')
577        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
578        self.csv_file_transfer = os.path.join(
579            self.workdir, 'sample_transfer_data.csv')
580        open(self.csv_file_transfer, 'wb').write(TRANSFER_SAMPLE_DATA)
581        return
582
583    def test_interface(self):
584        # Make sure we fulfill the interface contracts.
585        assert verifyObject(IBatchProcessor, self.processor) is True
586        assert verifyClass(
587            IBatchProcessor, StudentStudyCourseProcessor) is True
588
589    def test_entryExists(self):
590        assert self.processor.entryExists(
591            dict(reg_number='REG_NONE'), self.app) is False
592        assert self.processor.entryExists(
593            dict(reg_number='1'), self.app) is True
594
595    def test_getEntry(self):
596        student = self.processor.getEntry(
597            dict(reg_number='1'), self.app).__parent__
598        self.assertEqual(student.reg_number,'1')
599
600    def test_checkConversion(self):
601        errs, inv_errs, conv_dict = self.processor.checkConversion(
602            dict(reg_number='1', certificate='CERT1', current_level='200'))
603        self.assertEqual(len(errs),0)
604        errs, inv_errs, conv_dict = self.processor.checkConversion(
605            dict(reg_number='1', certificate='CERT999'))
606        self.assertEqual(len(errs),1)
607        self.assertTrue(('certificate', u'Invalid value') in errs)
608        errs, inv_errs, conv_dict = self.processor.checkConversion(
609            dict(reg_number='1', certificate='CERT1', current_level='100'))
610        self.assertEqual(len(errs),1)
611        self.assertTrue(('current_level','not in range') in errs)
612        # If we import only current_level, no conversion checking is done.
613        errs, inv_errs, conv_dict = self.processor.checkConversion(
614            dict(reg_number='1', current_level='100'))
615        self.assertEqual(len(errs),0)
616
617    def test_checkUpdateRequirements(self):
618        # Current level must be in range of certificate.
619        # Since row has passed the converter, current_level is an integer.
620        err = self.processor.checkUpdateRequirements(
621            self.student['studycourse'],
622            dict(reg_number='1', current_level=100), self.app)
623        self.assertEqual(err, 'current_level not in range.')
624        err = self.processor.checkUpdateRequirements(
625            self.student['studycourse'],
626            dict(reg_number='1', current_level=200), self.app)
627        self.assertTrue(err is None)
628        # We can update pg students.
629        self.student['studycourse'].certificate.start_level=999
630        self.student['studycourse'].certificate.end_level=999
631        err = self.processor.checkUpdateRequirements(
632            self.student['studycourse'],
633            dict(reg_number='1', current_level=999), self.app)
634        self.assertTrue(err is None)
635        # Make sure that pg students can't be updated with wrong transition.
636        IWorkflowState(self.student).setState('returning')
637        err = self.processor.checkUpdateRequirements(
638            self.student['studycourse'],
639            dict(reg_number='1', current_level=999), self.app)
640        self.assertEqual(err, 'Not a pg student.')
641        # If certificate is not given in row (and has thus
642        # successfully passed checkConversion) the certificate
643        # attribute must be set.
644        self.student['studycourse'].certificate = None
645        err = self.processor.checkUpdateRequirements(
646            self.student['studycourse'],
647            dict(reg_number='1', current_level=100), self.app)
648        self.assertEqual(err, 'No certificate to check level.')
649        # When transferring students the method also checks
650        # if the former studycourse is complete.
651        err = self.processor.checkUpdateRequirements(
652            self.student['studycourse'],
653            dict(reg_number='1', certificate='CERT1', current_level=200,
654            entry_mode='transfer'), self.app)
655        self.assertEqual(err, 'Former study course record incomplete.')
656        self.student['studycourse'].certificate = self.certificate
657        self.student['studycourse'].entry_session = 2005
658        # The method doesn't care if current_level
659        # is not in range of CERT1. This is done by checkConversion
660        # if certificate is in row.
661        err = self.processor.checkUpdateRequirements(
662            self.student['studycourse'],
663            dict(reg_number='1', certificate='CERT1', current_level=200,
664            entry_mode='transfer'), self.app)
665        self.assertTrue(err is None)
666
667    def test_import(self):
668        num, num_warns, fin_file, fail_file = self.processor.doImport(
669            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
670        self.assertEqual(num_warns,1)
671        content = open(fail_file).read()
672        self.assertTrue('current_level: not in range' in content)
673        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
674        self.assertEqual(studycourse.certificate.code, u'CERT1')
675        shutil.rmtree(os.path.dirname(fin_file))
676
677    def test_import_transfer(self):
678        self.certificate2 = createObject('waeup.Certificate')
679        self.certificate2.code = 'CERT2'
680        self.certificate2.application_category = 'basic'
681        self.certificate2.start_level = 200
682        self.certificate2.end_level = 500
683        self.certificate2.study_mode = u'ug_pt'
684        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
685            self.certificate2)
686        num, num_warns, fin_file, fail_file = self.processor.doImport(
687            self.csv_file_transfer, TRANSFER_HEADER_FIELDS,'update')
688        self.assertEqual(num_warns,0)
689        self.assertEqual(self.student['studycourse'].certificate.code, 'CERT2')
690        self.assertEqual(self.student['studycourse_1'].certificate.code, 'CERT1')
691        self.assertEqual(self.student['studycourse'].entry_mode, 'transfer')
692        self.assertEqual(self.student['studycourse_1'].entry_mode, 'ug_ft')
693        self.assertEqual(self.student.current_mode, 'ug_pt')
694        shutil.rmtree(os.path.dirname(fin_file))
695        # Transer has bee logged.
696        logcontent = open(self.logfile).read()
697        self.assertTrue(
698            'INFO - system - K1000000 - transferred from CERT1 to CERT2\n'
699            in logcontent)
700        self.assertTrue(
701            'INFO - system - '
702            'StudentStudyCourse Processor (update only) - '
703            'sample_transfer_data - K1000000 - updated: entry_mode=transfer, '
704            'certificate=CERT2, current_session=2009, current_level=300'
705            in logcontent)
706        # A history message has been added.
707        history = ' '.join(self.student.history.messages)
708        self.assertTrue(
709            "Transferred from CERT1 to CERT2 by system" in history)
710        # The catalog has been updated
711        cat = queryUtility(ICatalog, name='students_catalog')
712        results = list(
713            cat.searchResults(
714            certcode=('CERT2', 'CERT2')))
715        self.assertTrue(results[0] is self.student)
716        results = list(
717            cat.searchResults(
718            current_session=(2009, 2009)))
719        self.assertTrue(results[0] is self.student)
720        results = list(
721            cat.searchResults(
722            certcode=('CERT1', 'CERT1')))
723        self.assertEqual(len(results), 0)
724
725class StudentStudyLevelProcessorTest(StudentImportExportSetup):
726
727    def setUp(self):
728        super(StudentStudyLevelProcessorTest, self).setUp()
729
730        # Import students with subobjects
731        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
732        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
733        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
734            student_file, STUDENT_HEADER_FIELDS)
735        shutil.rmtree(os.path.dirname(fin_file))
736
737        # Update study courses
738        studycourse_file = os.path.join(
739            self.workdir, 'sample_studycourse_data.csv')
740        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
741        processor = StudentStudyCourseProcessor()
742        num, num_warns, fin_file, fail_file = processor.doImport(
743            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
744        shutil.rmtree(os.path.dirname(fin_file))
745
746        self.processor = StudentStudyLevelProcessor()
747        self.csv_file = os.path.join(
748            self.workdir, 'sample_studylevel_data.csv')
749        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
750
751    def test_interface(self):
752        # Make sure we fulfill the interface contracts.
753        assert verifyObject(IBatchProcessor, self.processor) is True
754        assert verifyClass(
755            IBatchProcessor, StudentStudyLevelProcessor) is True
756
757    def test_checkConversion(self):
758        errs, inv_errs, conv_dict = self.processor.checkConversion(
759            dict(reg_number='1', level='220'))
760        self.assertEqual(len(errs),0)
761        errs, inv_errs, conv_dict = self.processor.checkConversion(
762            dict(reg_number='1', level='999'))
763        self.assertEqual(len(errs),0)
764        errs, inv_errs, conv_dict = self.processor.checkConversion(
765            dict(reg_number='1', level='1000'))
766        self.assertEqual(len(errs),1)
767        self.assertTrue(('level', u'Invalid value') in errs)
768        errs, inv_errs, conv_dict = self.processor.checkConversion(
769            dict(reg_number='1', level='xyz'))
770        self.assertEqual(len(errs),1)
771        self.assertTrue(('level', u'Invalid value') in errs)
772
773    def test_import(self):
774        num, num_warns, fin_file, fail_file = self.processor.doImport(
775            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
776        self.assertEqual(num_warns,2)
777        assert self.processor.entryExists(
778            dict(reg_number='1', level='100'), self.app) is True
779        studylevel = self.processor.getEntry(
780            dict(reg_number='1', level='100'), self.app)
781        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
782        self.assertEqual(studylevel.level_session, 2008)
783        self.assertEqual(studylevel.level_verdict, '0')
784        self.assertEqual(studylevel.level, 100)
785        shutil.rmtree(os.path.dirname(fin_file))
786
787        logcontent = open(self.logfile).read()
788        # Logging message from updateEntry,
789        self.assertTrue(
790            'INFO - system - StudentStudyLevel Processor - '
791            'sample_studylevel_data - K1000000 - updated: '
792            'level=100, level_verdict=C, level_session=2009'
793            in logcontent)
794
795    def test_import_update(self):
796        # We perform the same import twice,
797        # the second time in update mode. The number
798        # of warnings must be the same.
799        num, num_warns, fin_file, fail_file = self.processor.doImport(
800            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
801        shutil.rmtree(os.path.dirname(fin_file))
802        num, num_warns, fin_file, fail_file = self.processor.doImport(
803            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
804        self.assertEqual(num_warns,2)
805        studylevel = self.processor.getEntry(
806            dict(reg_number='1', level='100'), self.app)
807        self.assertEqual(studylevel.level, 100)
808        shutil.rmtree(os.path.dirname(fin_file))
809
810    def test_import_remove(self):
811        # We perform the same import twice,
812        # the second time in remove mode. The number
813        # of warnings must be the same.
814        num, num_warns, fin_file, fail_file = self.processor.doImport(
815            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
816        shutil.rmtree(os.path.dirname(fin_file))
817        num, num_warns, fin_file, fail_file = self.processor.doImport(
818            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'remove')
819        assert self.processor.entryExists(
820            dict(reg_number='1', level='100'), self.app) is False
821        self.assertEqual(num_warns,2)
822
823        shutil.rmtree(os.path.dirname(fin_file))
824
825class CourseTicketProcessorTest(StudentImportExportSetup):
826
827    def setUp(self):
828        super(CourseTicketProcessorTest, self).setUp()
829
830        # Import students with subobjects
831        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
832        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
833        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
834            student_file, STUDENT_HEADER_FIELDS)
835        shutil.rmtree(os.path.dirname(fin_file))
836
837        # Add course and certificate course
838        self.course = createObject('waeup.Course')
839        self.course.code = 'COURSE1'
840        self.course.semester = 1
841        self.course.credits = 10
842        self.course.passmark = 40
843        self.app['faculties']['fac1']['dep1'].courses.addCourse(
844            self.course)
845        self.app['faculties']['fac1']['dep1'].certificates[
846            'CERT1'].addCertCourse(
847            self.course, level=100)
848
849        # Update study courses
850        studycourse_file = os.path.join(
851            self.workdir, 'sample_studycourse_data.csv')
852        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
853        processor = StudentStudyCourseProcessor()
854        num, num_warns, fin_file, fail_file = processor.doImport(
855            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
856        shutil.rmtree(os.path.dirname(fin_file))
857
858        # Import study levels
859        processor = StudentStudyLevelProcessor()
860        studylevel_file = os.path.join(
861            self.workdir, 'sample_studylevel_data.csv')
862        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
863        num, num_warns, fin_file, fail_file = processor.doImport(
864            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
865        shutil.rmtree(os.path.dirname(fin_file))
866
867        self.processor = CourseTicketProcessor()
868        self.csv_file = os.path.join(
869            self.workdir, 'sample_courseticket_data.csv')
870        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
871
872    def test_interface(self):
873        # Make sure we fulfill the interface contracts.
874        assert verifyObject(IBatchProcessor, self.processor) is True
875        assert verifyClass(
876            IBatchProcessor, CourseTicketProcessor) is True
877
878    def test_checkConversion(self):
879        errs, inv_errs, conv_dict = self.processor.checkConversion(
880            dict(reg_number='1', code='COURSE1', level='220'))
881        self.assertEqual(len(errs),0)
882        errs, inv_errs, conv_dict = self.processor.checkConversion(
883            dict(reg_number='1', code='COURSE2', level='220'))
884        self.assertEqual(len(errs),1)
885        self.assertTrue(('code','non-existent') in errs)
886
887    def test_import(self):
888        num, num_warns, fin_file, fail_file = self.processor.doImport(
889            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
890        fail_file = open(fail_file).read()
891        self.assertEqual(num_warns,5)
892        self.assertEqual(fail_file,
893            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
894            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,5,<IGNORE>,Not all parents do exist yet.\r\n'
895            '1,NONSENSE,<IGNORE>,100,<IGNORE>,5,<IGNORE>,code: non-existent\r\n'
896            '1,COURSE1,<IGNORE>,200,2004,6,<IGNORE>,level_session: does not match 2008\r\n'
897            '1,COURSE1,<IGNORE>,300,2008,6,<IGNORE>,level object: does not exist\r\n'
898            '1,COURSE1,<IGNORE>,300,2008X,6,<IGNORE>,level_session: Invalid value\r\n')
899        assert self.processor.entryExists(
900            dict(reg_number='1', level='100', code='COURSE1'),
901            self.app) is True
902        courseticket = self.processor.getEntry(
903            dict(reg_number='1', level='100', code='COURSE1'), self.app)
904        self.assertEqual(courseticket.__parent__.__parent__.certificate.code,
905                         u'CERT1')
906        self.assertEqual(courseticket.score, 1)
907        self.assertEqual(courseticket.mandatory, True)
908        self.assertEqual(courseticket.fcode, 'NA')
909        self.assertEqual(courseticket.dcode, 'NA')
910        self.assertEqual(courseticket.code, 'COURSE1')
911        self.assertEqual(courseticket.title, 'Unnamed Course')
912        self.assertEqual(courseticket.credits, 10)
913        self.assertEqual(courseticket.passmark, 40)
914        self.assertEqual(courseticket.semester, 1)
915        self.assertEqual(courseticket.level, 100)
916        self.assertEqual(courseticket.level_session, 2008)
917        shutil.rmtree(os.path.dirname(fin_file))
918        logcontent = open(self.logfile).read()
919        # Logging message from updateEntry,
920        self.assertTrue(
921            'INFO - system - CourseTicket Processor - '
922            'sample_courseticket_data - K1000000 - 100 - '
923            'updated: code=COURSE1, '
924            'mandatory=False, score=3'
925            in logcontent)
926
927        # The catalog has been updated
928        cat = queryUtility(ICatalog, name='coursetickets_catalog')
929        results = list(
930            cat.searchResults(
931            level=(100, 100)))
932        self.assertEqual(len(results),3)
933
934    def test_import_update(self):
935        # We perform the same import twice,
936        # the second time in update mode. The number
937        # of warnings must be the same.
938        num, num_warns, fin_file, fail_file = self.processor.doImport(
939            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
940        shutil.rmtree(os.path.dirname(fin_file))
941        num, num_warns, fin_file, fail_file = self.processor.doImport(
942            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
943        fail_file = open(fail_file).read()
944        self.assertEqual(num_warns,5)
945        self.assertEqual(fail_file,
946            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
947            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,5,<IGNORE>,Cannot update: no such entry\r\n'
948            '1,NONSENSE,<IGNORE>,100,<IGNORE>,5,<IGNORE>,code: non-existent\r\n'
949            '1,COURSE1,<IGNORE>,200,2004,6,<IGNORE>,level_session: does not match 2008\r\n'
950            '1,COURSE1,<IGNORE>,300,2008,6,<IGNORE>,level object: does not exist\r\n'
951            '1,COURSE1,<IGNORE>,300,2008X,6,<IGNORE>,level_session: Invalid value\r\n')
952        shutil.rmtree(os.path.dirname(fin_file))
953
954    def test_import_remove(self):
955        # We perform the same import twice,
956        # the second time in remove mode. The number
957        # of warnings must be the same.
958        num, num_warns, fin_file, fail_file = self.processor.doImport(
959            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
960        shutil.rmtree(os.path.dirname(fin_file))
961        assert self.processor.entryExists(
962            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
963        num, num_warns, fin_file, fail_file = self.processor.doImport(
964            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
965        self.assertEqual(num_warns,5)
966        assert self.processor.entryExists(
967            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
968        shutil.rmtree(os.path.dirname(fin_file))
969        logcontent = open(self.logfile).read()
970        self.assertTrue(
971            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
972            in logcontent)
973
974class PaymentProcessorTest(StudentImportExportSetup):
975
976    def setUp(self):
977        super(PaymentProcessorTest, self).setUp()
978
979        # Add student with payment
980        student = Student()
981        student.firstname = u'Anna'
982        student.lastname = u'Tester'
983        student.reg_number = u'123'
984        student.matric_number = u'234'
985        self.app['students'].addStudent(student)
986        self.student = self.app['students'][student.student_id]
987        payment = createObject(u'waeup.StudentOnlinePayment')
988        payment.p_id = 'p120'
989        payment.p_session = 2012
990        payment.p_category = 'schoolfee'
991        payment.p_state = 'paid'
992        self.student['payments'][payment.p_id] = payment
993
994        # Import students with subobjects
995        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
996        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
997        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
998            student_file, STUDENT_HEADER_FIELDS)
999        shutil.rmtree(os.path.dirname(fin_file))
1000
1001        self.processor = StudentOnlinePaymentProcessor()
1002        self.csv_file = os.path.join(
1003            self.workdir, 'sample_payment_data.csv')
1004        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
1005        self.csv_file2 = os.path.join(
1006            self.workdir, 'sample_create_payment_data.csv')
1007        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
1008
1009    def test_interface(self):
1010        # Make sure we fulfill the interface contracts.
1011        assert verifyObject(IBatchProcessor, self.processor) is True
1012        assert verifyClass(
1013            IBatchProcessor, StudentOnlinePaymentProcessor) is True
1014
1015    def test_getEntry(self):
1016        assert self.processor.getEntry(
1017            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
1018        assert self.processor.getEntry(
1019            dict(student_id=self.student.student_id, p_id='p120'),
1020            self.app) is self.student['payments']['p120']
1021        assert self.processor.getEntry(
1022            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
1023            self.app) is self.student['payments']['p120']
1024
1025    def test_delEntry(self):
1026        assert self.processor.getEntry(
1027            dict(student_id=self.student.student_id, p_id='p120'),
1028            self.app) is self.student['payments']['p120']
1029        self.assertEqual(len(self.student['payments'].keys()),1)
1030        self.processor.delEntry(
1031            dict(student_id=self.student.student_id, p_id='p120'),
1032            self.app)
1033        assert self.processor.getEntry(
1034            dict(student_id=self.student.student_id, p_id='p120'),
1035            self.app) is None
1036        self.assertEqual(len(self.student['payments'].keys()),0)
1037
1038    def test_addEntry(self):
1039        self.assertEqual(len(self.student['payments'].keys()),1)
1040        payment1 = createObject(u'waeup.StudentOnlinePayment')
1041        payment1.p_id = 'p234'
1042        self.processor.addEntry(
1043            payment1, dict(student_id=self.student.student_id, p_id='p234'),
1044            self.app)
1045        self.assertEqual(len(self.student['payments'].keys()),2)
1046        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
1047        payment2 = createObject(u'waeup.StudentOnlinePayment')
1048        payment1.p_id = 'nonsense'
1049        # payment1.p_id will be replaced if p_id doesn't start with 'p'
1050        # and is not an old PIN
1051        self.processor.addEntry(
1052            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
1053            self.app)
1054        self.assertEqual(len(self.student['payments'].keys()),3)
1055        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
1056        # Requirement added on 19/02/2015: same payment must not exist.
1057        payment3 = createObject(u'waeup.StudentOnlinePayment')
1058        payment3.p_id = 'p456'
1059        payment3.p_session = 2012
1060        payment3.p_category = 'schoolfee'
1061        self.assertRaises(
1062            DuplicationError, self.processor.addEntry, payment3,
1063            dict(student_id=self.student.student_id, p_id='p456'), self.app)
1064        logcontent = open(self.logfile).read()
1065        self.assertTrue(
1066            'INFO - system - StudentOnlinePayment Processor - K1000000 - '
1067            'previous update cancelled'
1068            in logcontent)
1069
1070    def test_checkConversion(self):
1071        errs, inv_errs, conv_dict = self.processor.checkConversion(
1072            dict(p_id='<IGNORE>'), mode='create')
1073        self.assertEqual(len(errs),0)
1074        errs, inv_errs, conv_dict = self.processor.checkConversion(
1075            dict(p_id='<IGNORE>'), mode='update')
1076        self.assertEqual(len(errs),1)
1077        self.assertEqual(errs[0], ('p_id', u'missing'))
1078        errs, inv_errs, conv_dict = self.processor.checkConversion(
1079            dict(p_id='3816951266236341955'))
1080        self.assertEqual(len(errs),0)
1081        errs, inv_errs, conv_dict = self.processor.checkConversion(
1082            dict(p_id='p1266236341955'))
1083        self.assertEqual(len(errs),0)
1084        errs, inv_errs, conv_dict = self.processor.checkConversion(
1085            dict(p_id='ABC-11-1234567890'))
1086        self.assertEqual(len(errs),0)
1087        errs, inv_errs, conv_dict = self.processor.checkConversion(
1088            dict(p_id='nonsense'))
1089        self.assertEqual(len(errs),1)
1090        self.assertEqual(errs[0], ('p_id', u'invalid format'))
1091        timestamp = ("%d" % int(time()*10000))[1:]
1092        p_id = "p%s" % timestamp
1093        errs, inv_errs, conv_dict = self.processor.checkConversion(
1094            dict(p_id=p_id))
1095        self.assertEqual(len(errs),0)
1096        dup_payment = createObject(u'waeup.StudentOnlinePayment')
1097        dup_payment.p_id = 'XYZ-99-1234567890'
1098        self.student['payments'][dup_payment.p_id] = dup_payment
1099        errs, inv_errs, conv_dict = self.processor.checkConversion(
1100            dict(p_id='XYZ-99-1234567890'), mode='create')
1101        self.assertEqual(len(errs),1)
1102        self.assertEqual(errs[0], ('p_id', u'p_id exists in K1000000 '))
1103
1104    def test_import(self):
1105        num, num_warns, fin_file, fail_file = self.processor.doImport(
1106            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1107        self.assertEqual(num_warns,0)
1108
1109        payment = self.processor.getEntry(dict(reg_number='1',
1110            p_id='p2907979737440'), self.app)
1111        self.assertEqual(payment.p_id, 'p2907979737440')
1112        self.assertTrue(payment.p_current)
1113        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1114        self.assertEqual(cdate, "2010-11-26 18:59:33")
1115        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1116
1117        payment = self.processor.getEntry(dict(matric_number='100001',
1118            p_id='p2907125937570'), self.app)
1119        self.assertEqual(payment.p_id, 'p2907125937570')
1120        self.assertEqual(payment.amount_auth, 19500.1)
1121        self.assertFalse(payment.p_current)
1122        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1123        # Ooooh, still the old problem, see
1124        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
1125        # WAT is interpreted as GMT-1 and not GMT+1
1126        self.assertEqual(cdate, "2010-11-25 21:16:33")
1127        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1128
1129        payment = self.processor.getEntry(dict(reg_number='3',
1130            p_id='ABC-11-1234567890'), self.app)
1131        self.assertEqual(payment.amount_auth, 19500.6)
1132
1133        shutil.rmtree(os.path.dirname(fin_file))
1134        logcontent = open(self.logfile).read()
1135        # Logging message from updateEntry
1136        self.assertTrue(
1137            'INFO - system - StudentOnlinePayment Processor - '
1138            'sample_payment_data - K1000001 - updated: '
1139            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1140            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
1141            'p_session=2009, '
1142            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
1143            'p_state=paid'
1144            in logcontent)
1145        self.assertTrue(
1146            'INFO - system - StudentOnlinePayment Processor - '
1147            'sample_payment_data - K1000001 - updated: '
1148            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1149            'p_category=schoolfee, amount_auth=19500.6, p_current=True, '
1150            'p_session=2011, '
1151            'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, '
1152            'p_state=paid'
1153            in logcontent)
1154
1155    def test_import_update(self):
1156        # We perform the same import twice,
1157        # the second time in update mode. The number
1158        # of warnings increases becaus one p_id is missing.
1159        num, num_warns, fin_file, fail_file = self.processor.doImport(
1160            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1161        shutil.rmtree(os.path.dirname(fin_file))
1162        num, num_warns, fin_file, fail_file = self.processor.doImport(
1163            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
1164        self.assertEqual(num_warns,1)
1165        content = open(fail_file).read()
1166        shutil.rmtree(os.path.dirname(fin_file))
1167        self.assertTrue('p_id: missing' in content)
1168
1169    def test_import_remove(self):
1170        # We perform the same import twice,
1171        # the second time in remove mode. The number
1172        # of warnings increases becaus one p_id is missing.
1173        num, num_warns, fin_file, fail_file = self.processor.doImport(
1174            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1175        shutil.rmtree(os.path.dirname(fin_file))
1176        num, num_warns, fin_file, fail_file = self.processor.doImport(
1177            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
1178        self.assertEqual(num_warns,1)
1179        content = open(fail_file).read()
1180        self.assertTrue('p_id: missing' in content)
1181        shutil.rmtree(os.path.dirname(fin_file))
1182        logcontent = open(self.logfile).read()
1183        self.assertTrue(
1184            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
1185            in logcontent)
1186
1187    def test_import_same_payment_exists(self):
1188        num, num_warns, fin_file, fail_file = self.processor.doImport(
1189            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
1190        # One payment with same session and category exists
1191        self.assertEqual(num_warns,1)
1192        content = open(fail_file).read()
1193        self.assertTrue(
1194            '1,942,online,BTECHBDT,2010/11/26 19:59:33.744 GMT+1,0,'
1195            '19500,schoolfee,19500,2015,paid,'
1196            'Same payment has already been made.'
1197            in content)
1198        shutil.rmtree(os.path.dirname(fin_file))
1199        self.assertEqual(len(self.app['students']['X666666']['payments']), 13)
1200
1201class StudentVerdictProcessorTest(StudentImportExportSetup):
1202
1203    def setUp(self):
1204        super(StudentVerdictProcessorTest, self).setUp()
1205
1206        # Import students with subobjects
1207        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
1208        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
1209        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
1210            student_file, STUDENT_HEADER_FIELDS)
1211        shutil.rmtree(os.path.dirname(fin_file))
1212
1213        # Update study courses
1214        studycourse_file = os.path.join(
1215            self.workdir, 'sample_studycourse_data.csv')
1216        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
1217        processor = StudentStudyCourseProcessor()
1218        num, num_warns, fin_file, fail_file = processor.doImport(
1219            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
1220        shutil.rmtree(os.path.dirname(fin_file))
1221        # Import study levels
1222        self.csv_file = os.path.join(
1223            self.workdir, 'sample_studylevel_data.csv')
1224        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
1225        processor = StudentStudyLevelProcessor()
1226        num, num_warns, fin_file, fail_file = processor.doImport(
1227            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
1228        content = open(fail_file).read()
1229        shutil.rmtree(os.path.dirname(fin_file))
1230
1231        self.processor = StudentVerdictProcessor()
1232        self.csv_file = os.path.join(
1233            self.workdir, 'sample_verdict_data.csv')
1234        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
1235        return
1236
1237    def test_import(self):
1238        studycourse = self.processor.getEntry(dict(matric_number='100000'),
1239                                              self.app)
1240        self.assertEqual(studycourse['200'].level_verdict, '0')
1241        student = self.processor.getParent(
1242            dict(matric_number='100000'), self.app)
1243        num, num_warns, fin_file, fail_file = self.processor.doImport(
1244            self.csv_file, VERDICT_HEADER_FIELDS,'update')
1245        #content = open(fail_file).read()
1246        #import pdb; pdb.set_trace()
1247        self.assertEqual(num_warns,5)
1248        self.assertEqual(studycourse.current_verdict, '0')
1249        self.assertEqual(student.state, 'returning')
1250        self.assertEqual(studycourse.current_level, 200)
1251        self.assertEqual(studycourse['200'].level_verdict, '0')
1252        student = self.processor.getParent(
1253            dict(matric_number='100005'), self.app)
1254        self.assertEqual(student.state, 'returning')
1255        self.assertEqual(student['studycourse'].current_verdict, 'A')
1256        self.assertEqual(studycourse.current_level, 200)
1257        self.assertEqual(student['studycourse']['200'].validated_by, 'System')
1258        self.assertTrue(isinstance(
1259            student['studycourse']['200'].validation_date, datetime.datetime))
1260        student = self.processor.getParent(
1261            dict(matric_number='100008'), self.app)
1262        self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana')
1263        content = open(fail_file).read()
1264        self.assertEqual(
1265            content,
1266            'current_session,current_level,bypass_validation,current_verdict,'
1267            'matric_number,validated_by,--ERRORS--\r\n'
1268            '2008,100,False,B,100001,<IGNORE>,Current level does not correspond.\r\n'
1269            '2007,200,<IGNORE>,C,100002,<IGNORE>,Current session does not correspond.\r\n'
1270            '2008,200,<IGNORE>,A,100003,<IGNORE>,Student in wrong state.\r\n'
1271            '2008,200,<IGNORE>,<IGNORE>,100004,<IGNORE>,No verdict in import file.\r\n'
1272            '2008,200,True,A,100007,<IGNORE>,Study level object is missing.\r\n'
1273            )
1274        logcontent = open(self.logfile).read()
1275        self.assertMatches(
1276            '... INFO - system - Verdict Processor (special processor, '
1277            'update only) - sample_verdict_data - X666666 - '
1278            'updated: current_verdict=0...',
1279            logcontent)
1280        self.assertMatches(
1281            '... INFO - system - X666666 - Returned...',
1282            logcontent)
1283
1284        shutil.rmtree(os.path.dirname(fin_file))
1285
1286def test_suite():
1287    suite = unittest.TestSuite()
1288    for testcase in [
1289        StudentProcessorTest,StudentStudyCourseProcessorTest,
1290        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
1291        PaymentProcessorTest,StudentVerdictProcessorTest]:
1292        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
1293                testcase
1294                )
1295        )
1296    return suite
1297
1298
Note: See TracBrowser for help on using the repository browser.