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

Last change on this file since 13225 was 13103, checked in by Henrik Bettermann, 9 years ago

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