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

Last change on this file since 12777 was 12623, checked in by Henrik Bettermann, 10 years ago

StudentOnlinePaymentProcessor?: Raise DuplicationError? if payment
(in state paid) with same p_session and p_category exists.

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