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

Last change on this file since 15865 was 15628, checked in by Henrik Bettermann, 5 years ago

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