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

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

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