source: main/waeup.kofa/branches/uli-rm-bootstrap/src/waeup/kofa/students/tests/test_batching.py @ 17930

Last change on this file since 17930 was 14901, checked in by Henrik Bettermann, 7 years ago

Set additional_fields_required.
  • Property svn:keywords set to Id
File size: 57.0 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 14901 2017-11-16 07:02:47Z henrik $
3##
4## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
5## This program is free software; you can redistribute it and/or modify
6## it under the terms of the GNU General Public License as published by
7## the Free Software Foundation; either version 2 of the License, or
8## (at your option) any later version.
9##
10## This program is distributed in the hope that it will be useful,
11## but WITHOUT ANY WARRANTY; without even the implied warranty of
12## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13## GNU General Public License for more details.
14##
15## You should have received a copy of the GNU General Public License
16## along with this program; if not, write to the Free Software
17## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18##
19"""Unit tests for students-related data processors.
20"""
21import os
22import shutil
23import tempfile
24import unittest
25import datetime
26import grok
27from time import time
28from zope.event import notify
29from zope.component import createObject, queryUtility
30from zope.component.hooks import setSite, clearSite
31from zope.catalog.interfaces import ICatalog
32from zope.interface.verify import verifyClass, verifyObject
33from hurry.workflow.interfaces import IWorkflowState
34
35from waeup.kofa.app import University
36from waeup.kofa.interfaces import (
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,2)
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        shutil.rmtree(os.path.dirname(fin_file))
787
788        logcontent = open(self.logfile).read()
789        # Logging message from updateEntry,
790        self.assertTrue(
791            'INFO - system - StudentStudyLevel Processor - '
792            'sample_studylevel_data - K1000000 - updated: '
793            'level=100, level_verdict=C, level_session=2009'
794            in logcontent)
795
796    def test_import_update(self):
797        # We perform the same import twice,
798        # the second time in update mode. The number
799        # of warnings must be the same.
800        num, num_warns, fin_file, fail_file = self.processor.doImport(
801            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
802        shutil.rmtree(os.path.dirname(fin_file))
803        num, num_warns, fin_file, fail_file = self.processor.doImport(
804            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
805        self.assertEqual(num_warns,2)
806        studylevel = self.processor.getEntry(
807            dict(reg_number='1', level='100'), self.app)
808        self.assertEqual(studylevel.level, 100)
809        shutil.rmtree(os.path.dirname(fin_file))
810
811    def test_import_remove(self):
812        # We perform the same import twice,
813        # the second time in remove mode. The number
814        # of warnings must be the same.
815        num, num_warns, fin_file, fail_file = self.processor.doImport(
816            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
817        shutil.rmtree(os.path.dirname(fin_file))
818        num, num_warns, fin_file, fail_file = self.processor.doImport(
819            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'remove')
820        assert self.processor.entryExists(
821            dict(reg_number='1', level='100'), self.app) is False
822        self.assertEqual(num_warns,2)
823
824        shutil.rmtree(os.path.dirname(fin_file))
825
826class CourseTicketProcessorTest(StudentImportExportSetup):
827
828    def setUp(self):
829        super(CourseTicketProcessorTest, self).setUp()
830
831        # Import students with subobjects
832        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
833        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
834        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
835            student_file, STUDENT_HEADER_FIELDS)
836        shutil.rmtree(os.path.dirname(fin_file))
837
838        # Add course and certificate course
839        self.course = createObject('waeup.Course')
840        self.course.code = 'COURSE1'
841        self.course.semester = 1
842        self.course.credits = 10
843        self.course.passmark = 40
844        self.app['faculties']['fac1']['dep1'].courses.addCourse(
845            self.course)
846        self.app['faculties']['fac1']['dep1'].certificates[
847            'CERT1'].addCertCourse(
848            self.course, level=100)
849
850        # Update study courses
851        studycourse_file = os.path.join(
852            self.workdir, 'sample_studycourse_data.csv')
853        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
854        processor = StudentStudyCourseProcessor()
855        num, num_warns, fin_file, fail_file = processor.doImport(
856            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
857        shutil.rmtree(os.path.dirname(fin_file))
858
859        # Import study levels
860        processor = StudentStudyLevelProcessor()
861        studylevel_file = os.path.join(
862            self.workdir, 'sample_studylevel_data.csv')
863        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
864        num, num_warns, fin_file, fail_file = processor.doImport(
865            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
866        shutil.rmtree(os.path.dirname(fin_file))
867
868        self.processor = CourseTicketProcessor()
869        self.csv_file = os.path.join(
870            self.workdir, 'sample_courseticket_data.csv')
871        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
872
873    def test_interface(self):
874        # Make sure we fulfill the interface contracts.
875        assert verifyObject(IBatchProcessor, self.processor) is True
876        assert verifyClass(
877            IBatchProcessor, CourseTicketProcessor) is True
878
879    def test_checkConversion(self):
880        errs, inv_errs, conv_dict = self.processor.checkConversion(
881            dict(reg_number='1', code='COURSE1', level='220'))
882        self.assertEqual(len(errs),0)
883        errs, inv_errs, conv_dict = self.processor.checkConversion(
884            dict(reg_number='1', code='COURSE2', level='220'))
885        self.assertEqual(len(errs),1)
886        self.assertTrue(('code','non-existent') in errs)
887
888    def test_import(self):
889        num, num_warns, fin_file, fail_file = self.processor.doImport(
890            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
891        fail_file = open(fail_file).read()
892        self.assertEqual(num_warns,5)
893        self.assertEqual(fail_file,
894            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
895            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,5,<IGNORE>,Not all parents do exist yet.\r\n'
896            '1,NONSENSE,<IGNORE>,100,<IGNORE>,5,<IGNORE>,code: non-existent\r\n'
897            '1,COURSE1,<IGNORE>,200,2004,6,<IGNORE>,level_session: does not match 2008\r\n'
898            '1,COURSE1,<IGNORE>,300,2008,6,<IGNORE>,level object: does not exist\r\n'
899            '1,COURSE1,<IGNORE>,300,2008X,6,<IGNORE>,level_session: Invalid value\r\n')
900        assert self.processor.entryExists(
901            dict(reg_number='1', level='100', code='COURSE1'),
902            self.app) is True
903        courseticket = self.processor.getEntry(
904            dict(reg_number='1', level='100', code='COURSE1'), self.app)
905        self.assertEqual(courseticket.__parent__.__parent__.certificate.code,
906                         u'CERT1')
907        self.assertEqual(courseticket.score, 1)
908        self.assertEqual(courseticket.mandatory, True)
909        self.assertEqual(courseticket.fcode, 'NA')
910        self.assertEqual(courseticket.dcode, 'NA')
911        self.assertEqual(courseticket.code, 'COURSE1')
912        self.assertEqual(courseticket.title, 'Unnamed Course')
913        self.assertEqual(courseticket.credits, 10)
914        self.assertEqual(courseticket.passmark, 40)
915        self.assertEqual(courseticket.semester, 1)
916        self.assertEqual(courseticket.level, 100)
917        self.assertEqual(courseticket.level_session, 2008)
918        shutil.rmtree(os.path.dirname(fin_file))
919        logcontent = open(self.logfile).read()
920        # Logging message from updateEntry,
921        self.assertTrue(
922            'INFO - system - CourseTicket Processor - '
923            'sample_courseticket_data - K1000000 - 100 - '
924            'updated: code=COURSE1, '
925            'mandatory=False, score=3'
926            in logcontent)
927
928        # The catalog has been updated
929        cat = queryUtility(ICatalog, name='coursetickets_catalog')
930        results = list(
931            cat.searchResults(
932            level=(100, 100)))
933        self.assertEqual(len(results),3)
934
935    def test_import_update(self):
936        # We perform the same import twice,
937        # the second time in update mode. The number
938        # of warnings must be the same.
939        num, num_warns, fin_file, fail_file = self.processor.doImport(
940            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
941        shutil.rmtree(os.path.dirname(fin_file))
942        num, num_warns, fin_file, fail_file = self.processor.doImport(
943            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
944        fail_file = open(fail_file).read()
945        self.assertEqual(num_warns,5)
946        self.assertEqual(fail_file,
947            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
948            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,5,<IGNORE>,Cannot update: no such entry\r\n'
949            '1,NONSENSE,<IGNORE>,100,<IGNORE>,5,<IGNORE>,code: non-existent\r\n'
950            '1,COURSE1,<IGNORE>,200,2004,6,<IGNORE>,level_session: does not match 2008\r\n'
951            '1,COURSE1,<IGNORE>,300,2008,6,<IGNORE>,level object: does not exist\r\n'
952            '1,COURSE1,<IGNORE>,300,2008X,6,<IGNORE>,level_session: Invalid value\r\n')
953        shutil.rmtree(os.path.dirname(fin_file))
954
955    def test_import_remove(self):
956        # We perform the same import twice,
957        # the second time in remove mode. The number
958        # of warnings must be the same.
959        num, num_warns, fin_file, fail_file = self.processor.doImport(
960            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
961        shutil.rmtree(os.path.dirname(fin_file))
962        assert self.processor.entryExists(
963            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
964        num, num_warns, fin_file, fail_file = self.processor.doImport(
965            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
966        self.assertEqual(num_warns,5)
967        assert self.processor.entryExists(
968            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
969        shutil.rmtree(os.path.dirname(fin_file))
970        logcontent = open(self.logfile).read()
971        self.assertTrue(
972            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
973            in logcontent)
974
975class PaymentProcessorTest(StudentImportExportSetup):
976
977    def setUp(self):
978        super(PaymentProcessorTest, self).setUp()
979
980        # Add student with payment
981        student = Student()
982        student.firstname = u'Anna'
983        student.lastname = u'Tester'
984        student.reg_number = u'123'
985        student.matric_number = u'234'
986        self.app['students'].addStudent(student)
987        self.student = self.app['students'][student.student_id]
988        payment = createObject(u'waeup.StudentOnlinePayment')
989        payment.p_id = 'p120'
990        payment.p_session = 2012
991        payment.p_category = 'schoolfee'
992        payment.p_state = 'paid'
993        self.student['payments'][payment.p_id] = payment
994
995        # Import students with subobjects
996        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
997        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
998        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
999            student_file, STUDENT_HEADER_FIELDS)
1000        shutil.rmtree(os.path.dirname(fin_file))
1001
1002        self.processor = StudentOnlinePaymentProcessor()
1003        self.csv_file = os.path.join(
1004            self.workdir, 'sample_payment_data.csv')
1005        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
1006        self.csv_file2 = os.path.join(
1007            self.workdir, 'sample_create_payment_data.csv')
1008        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
1009
1010    def test_interface(self):
1011        # Make sure we fulfill the interface contracts.
1012        assert verifyObject(IBatchProcessor, self.processor) is True
1013        assert verifyClass(
1014            IBatchProcessor, StudentOnlinePaymentProcessor) is True
1015
1016    def test_getEntry(self):
1017        assert self.processor.getEntry(
1018            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
1019        assert self.processor.getEntry(
1020            dict(student_id=self.student.student_id, p_id='p120'),
1021            self.app) is self.student['payments']['p120']
1022        assert self.processor.getEntry(
1023            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
1024            self.app) is self.student['payments']['p120']
1025
1026    def test_delEntry(self):
1027        assert self.processor.getEntry(
1028            dict(student_id=self.student.student_id, p_id='p120'),
1029            self.app) is self.student['payments']['p120']
1030        self.assertEqual(len(self.student['payments'].keys()),1)
1031        self.processor.delEntry(
1032            dict(student_id=self.student.student_id, p_id='p120'),
1033            self.app)
1034        assert self.processor.getEntry(
1035            dict(student_id=self.student.student_id, p_id='p120'),
1036            self.app) is None
1037        self.assertEqual(len(self.student['payments'].keys()),0)
1038
1039    def test_addEntry(self):
1040        self.assertEqual(len(self.student['payments'].keys()),1)
1041        payment1 = createObject(u'waeup.StudentOnlinePayment')
1042        payment1.p_id = 'p234'
1043        self.processor.addEntry(
1044            payment1, dict(student_id=self.student.student_id, p_id='p234'),
1045            self.app)
1046        self.assertEqual(len(self.student['payments'].keys()),2)
1047        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
1048        payment2 = createObject(u'waeup.StudentOnlinePayment')
1049        payment1.p_id = 'nonsense'
1050        # payment1.p_id will be replaced if p_id doesn't start with 'p'
1051        # and is not an old PIN
1052        self.processor.addEntry(
1053            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
1054            self.app)
1055        self.assertEqual(len(self.student['payments'].keys()),3)
1056        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
1057        # Requirement added on 19/02/2015: same payment must not exist.
1058        payment3 = createObject(u'waeup.StudentOnlinePayment')
1059        payment3.p_id = 'p456'
1060        payment3.p_session = 2012
1061        payment3.p_category = 'schoolfee'
1062        self.assertRaises(
1063            DuplicationError, self.processor.addEntry, payment3,
1064            dict(student_id=self.student.student_id, p_id='p456'), self.app)
1065        logcontent = open(self.logfile).read()
1066        self.assertTrue(
1067            'INFO - system - StudentOnlinePayment Processor - K1000000 - '
1068            'previous update cancelled'
1069            in logcontent)
1070
1071    def test_checkConversion(self):
1072        errs, inv_errs, conv_dict = self.processor.checkConversion(
1073            dict(p_id='<IGNORE>'), mode='create')
1074        self.assertEqual(len(errs),0)
1075        errs, inv_errs, conv_dict = self.processor.checkConversion(
1076            dict(p_id='<IGNORE>'), mode='update')
1077        self.assertEqual(len(errs),1)
1078        self.assertEqual(errs[0], ('p_id', u'missing'))
1079        errs, inv_errs, conv_dict = self.processor.checkConversion(
1080            dict(p_id='3816951266236341955'))
1081        self.assertEqual(len(errs),0)
1082        errs, inv_errs, conv_dict = self.processor.checkConversion(
1083            dict(p_id='p1266236341955'))
1084        self.assertEqual(len(errs),0)
1085        errs, inv_errs, conv_dict = self.processor.checkConversion(
1086            dict(p_id='ABC-11-1234567890'))
1087        self.assertEqual(len(errs),0)
1088        errs, inv_errs, conv_dict = self.processor.checkConversion(
1089            dict(p_id='nonsense'))
1090        self.assertEqual(len(errs),1)
1091        self.assertEqual(errs[0], ('p_id', u'invalid format'))
1092        timestamp = ("%d" % int(time()*10000))[1:]
1093        p_id = "p%s" % timestamp
1094        errs, inv_errs, conv_dict = self.processor.checkConversion(
1095            dict(p_id=p_id))
1096        self.assertEqual(len(errs),0)
1097        dup_payment = createObject(u'waeup.StudentOnlinePayment')
1098        dup_payment.p_id = 'XYZ-99-1234567890'
1099        self.student['payments'][dup_payment.p_id] = dup_payment
1100        errs, inv_errs, conv_dict = self.processor.checkConversion(
1101            dict(p_id='XYZ-99-1234567890'), mode='create')
1102        self.assertEqual(len(errs),1)
1103        self.assertEqual(errs[0], ('p_id', u'p_id exists in K1000000 '))
1104
1105    def test_import(self):
1106        num, num_warns, fin_file, fail_file = self.processor.doImport(
1107            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1108        self.assertEqual(num_warns,0)
1109
1110        payment = self.processor.getEntry(dict(reg_number='1',
1111            p_id='p2907979737440'), self.app)
1112        self.assertEqual(payment.p_id, 'p2907979737440')
1113        self.assertTrue(payment.p_current)
1114        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1115        self.assertEqual(cdate, "2010-11-26 18:59:33")
1116        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1117
1118        payment = self.processor.getEntry(dict(matric_number='100001',
1119            p_id='p2907125937570'), self.app)
1120        self.assertEqual(payment.p_id, 'p2907125937570')
1121        self.assertEqual(payment.amount_auth, 19500.1)
1122        self.assertFalse(payment.p_current)
1123        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1124        # Ooooh, still the old problem, see
1125        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
1126        # WAT is interpreted as GMT-1 and not GMT+1
1127        self.assertEqual(cdate, "2010-11-25 21:16:33")
1128        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1129
1130        payment = self.processor.getEntry(dict(reg_number='3',
1131            p_id='ABC-11-1234567890'), self.app)
1132        self.assertEqual(payment.amount_auth, 19500.6)
1133
1134        shutil.rmtree(os.path.dirname(fin_file))
1135        logcontent = open(self.logfile).read()
1136        # Logging message from updateEntry
1137        self.assertTrue(
1138            'INFO - system - StudentOnlinePayment Processor - '
1139            'sample_payment_data - K1000001 - updated: '
1140            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1141            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
1142            'p_session=2009, '
1143            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
1144            'p_state=paid'
1145            in logcontent)
1146        self.assertTrue(
1147            'INFO - system - StudentOnlinePayment Processor - '
1148            'sample_payment_data - K1000001 - updated: '
1149            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1150            'p_category=schoolfee, amount_auth=19500.6, p_current=True, '
1151            'p_session=2011, '
1152            'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, '
1153            'p_state=paid'
1154            in logcontent)
1155
1156    def test_import_update(self):
1157        # We perform the same import twice,
1158        # the second time in update mode. The number
1159        # of warnings increases becaus one p_id is missing.
1160        num, num_warns, fin_file, fail_file = self.processor.doImport(
1161            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1162        shutil.rmtree(os.path.dirname(fin_file))
1163        num, num_warns, fin_file, fail_file = self.processor.doImport(
1164            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
1165        self.assertEqual(num_warns,1)
1166        content = open(fail_file).read()
1167        shutil.rmtree(os.path.dirname(fin_file))
1168        self.assertTrue('p_id: missing' in content)
1169
1170    def test_import_remove(self):
1171        # We perform the same import twice,
1172        # the second time in remove mode. The number
1173        # of warnings increases becaus one p_id is missing.
1174        num, num_warns, fin_file, fail_file = self.processor.doImport(
1175            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1176        shutil.rmtree(os.path.dirname(fin_file))
1177        num, num_warns, fin_file, fail_file = self.processor.doImport(
1178            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
1179        self.assertEqual(num_warns,1)
1180        content = open(fail_file).read()
1181        self.assertTrue('p_id: missing' in content)
1182        shutil.rmtree(os.path.dirname(fin_file))
1183        logcontent = open(self.logfile).read()
1184        self.assertTrue(
1185            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
1186            in logcontent)
1187
1188    def test_import_same_payment_exists(self):
1189        num, num_warns, fin_file, fail_file = self.processor.doImport(
1190            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
1191        # One payment with same session and category exists
1192        self.assertEqual(num_warns,1)
1193        content = open(fail_file).read()
1194        self.assertTrue(
1195            '1,942,online,BTECHBDT,2010/11/26 19:59:33.744 GMT+1,0,'
1196            '19500,schoolfee,19500,2015,paid,'
1197            'Same payment has already been made.'
1198            in content)
1199        shutil.rmtree(os.path.dirname(fin_file))
1200        self.assertEqual(len(self.app['students']['X666666']['payments']), 13)
1201
1202class StudentVerdictProcessorTest(StudentImportExportSetup):
1203
1204    def setUp(self):
1205        super(StudentVerdictProcessorTest, self).setUp()
1206
1207        # Import students with subobjects
1208        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
1209        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
1210        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
1211            student_file, STUDENT_HEADER_FIELDS)
1212        shutil.rmtree(os.path.dirname(fin_file))
1213
1214        # Update study courses
1215        studycourse_file = os.path.join(
1216            self.workdir, 'sample_studycourse_data.csv')
1217        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
1218        processor = StudentStudyCourseProcessor()
1219        num, num_warns, fin_file, fail_file = processor.doImport(
1220            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
1221        shutil.rmtree(os.path.dirname(fin_file))
1222        # Import study levels
1223        self.csv_file = os.path.join(
1224            self.workdir, 'sample_studylevel_data.csv')
1225        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
1226        processor = StudentStudyLevelProcessor()
1227        num, num_warns, fin_file, fail_file = processor.doImport(
1228            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
1229        content = open(fail_file).read()
1230        shutil.rmtree(os.path.dirname(fin_file))
1231
1232        self.processor = StudentVerdictProcessor()
1233        self.csv_file = os.path.join(
1234            self.workdir, 'sample_verdict_data.csv')
1235        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
1236        return
1237
1238    def test_import(self):
1239        studycourse = self.processor.getEntry(dict(matric_number='100000'),
1240                                              self.app)
1241        self.assertEqual(studycourse['200'].level_verdict, '0')
1242        student = self.processor.getParent(
1243            dict(matric_number='100000'), self.app)
1244        num, num_warns, fin_file, fail_file = self.processor.doImport(
1245            self.csv_file, VERDICT_HEADER_FIELDS,'update')
1246        self.assertEqual(num_warns,5)
1247        self.assertEqual(studycourse.current_verdict, '0')
1248        self.assertEqual(student.state, 'returning')
1249        self.assertEqual(studycourse.current_level, 200)
1250        self.assertEqual(studycourse['200'].level_verdict, '0')
1251        student = self.processor.getParent(
1252            dict(matric_number='100005'), self.app)
1253        self.assertEqual(student.state, 'returning')
1254        self.assertEqual(student['studycourse'].current_verdict, 'A')
1255        self.assertEqual(studycourse.current_level, 200)
1256        self.assertEqual(student['studycourse']['200'].validated_by, 'System')
1257        self.assertTrue(isinstance(
1258            student['studycourse']['200'].validation_date, datetime.datetime))
1259        student = self.processor.getParent(
1260            dict(matric_number='100008'), self.app)
1261        self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana')
1262        content = open(fail_file).read()
1263        self.assertEqual(
1264            content,
1265            'current_session,current_level,bypass_validation,current_verdict,'
1266            'matric_number,validated_by,--ERRORS--\r\n'
1267            '2008,100,False,B,100001,<IGNORE>,Current level does not correspond.\r\n'
1268            '2007,200,<IGNORE>,C,100002,<IGNORE>,Current session does not correspond.\r\n'
1269            '2008,200,<IGNORE>,A,100003,<IGNORE>,Student in wrong state.\r\n'
1270            '2008,200,<IGNORE>,<IGNORE>,100004,<IGNORE>,No verdict in import file.\r\n'
1271            '2008,200,True,A,100007,<IGNORE>,Study level object is missing.\r\n'
1272            )
1273        logcontent = open(self.logfile).read()
1274        self.assertMatches(
1275            '... INFO - system - Verdict Processor (special processor, '
1276            'update only) - sample_verdict_data - X666666 - '
1277            'updated: current_verdict=0...',
1278            logcontent)
1279        self.assertMatches(
1280            '... INFO - system - X666666 - Returned...',
1281            logcontent)
1282
1283        shutil.rmtree(os.path.dirname(fin_file))
1284
1285def test_suite():
1286    suite = unittest.TestSuite()
1287    for testcase in [
1288        StudentProcessorTest,StudentStudyCourseProcessorTest,
1289        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
1290        PaymentProcessorTest,StudentVerdictProcessorTest]:
1291        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
1292                testcase
1293                )
1294        )
1295    return suite
1296
1297
Note: See TracBrowser for help on using the repository browser.