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

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

Implement BatchProcessor.checkCreateRequirements and
BatchProcessor.checkUpdateRequirements methods mainly
to protect course result lists of graduated student.
  • Property svn:keywords set to Id
File size: 65.6 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 16012 2020-02-24 21:26:35Z 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        self.assertRaises(
499            FatalCSVError, self.processor.doImport, self.csv_file_update4,
500            STUDENT_HEADER_FIELDS_UPDATE4, 'update')
501        shutil.rmtree(os.path.dirname(fin_file))
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        shutil.rmtree(os.path.dirname(fin_file))
894
895class CourseTicketProcessorTest(StudentImportExportSetup):
896
897    def setUp(self):
898        super(CourseTicketProcessorTest, self).setUp()
899
900        # Import students with subobjects
901        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
902        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
903        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
904            student_file, STUDENT_HEADER_FIELDS)
905        shutil.rmtree(os.path.dirname(fin_file))
906
907        # Add course and certificate course
908        self.course = createObject('waeup.Course')
909        self.course.code = 'COURSE1'
910        self.course.semester = 1
911        self.course.credits = 10
912        self.course.passmark = 40
913        self.app['faculties']['fac1']['dep1'].courses.addCourse(
914            self.course)
915        self.app['faculties']['fac1']['dep1'].certificates[
916            'CERT1'].addCertCourse(
917            self.course, level=100)
918
919        # Update study courses
920        studycourse_file = os.path.join(
921            self.workdir, 'sample_studycourse_data.csv')
922        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
923        processor = StudentStudyCourseProcessor()
924        num, num_warns, fin_file, fail_file = processor.doImport(
925            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
926        shutil.rmtree(os.path.dirname(fin_file))
927
928        # Import study levels
929        processor = StudentStudyLevelProcessor()
930        studylevel_file = os.path.join(
931            self.workdir, 'sample_studylevel_data.csv')
932        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
933        num, num_warns, fin_file, fail_file = processor.doImport(
934            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
935        shutil.rmtree(os.path.dirname(fin_file))
936
937        self.processor = CourseTicketProcessor()
938        self.csv_file = os.path.join(
939            self.workdir, 'sample_courseticket_data.csv')
940        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
941
942    def test_interface(self):
943        # Make sure we fulfill the interface contracts.
944        assert verifyObject(IBatchProcessor, self.processor) is True
945        assert verifyClass(
946            IBatchProcessor, CourseTicketProcessor) is True
947
948    def test_checkConversion(self):
949        errs, inv_errs, conv_dict = self.processor.checkConversion(
950            dict(reg_number='1', code='COURSE1', level='220'))
951        self.assertEqual(len(errs),0)
952        errs, inv_errs, conv_dict = self.processor.checkConversion(
953            dict(reg_number='1', code='COURSE2', level='220'))
954        self.assertEqual(len(errs),1)
955        self.assertTrue(('code','non-existent') in errs)
956
957    def test_import(self):
958        num, num_warns, fin_file, fail_file = self.processor.doImport(
959            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
960        fail_file = open(fail_file).read()
961        self.assertEqual(num_warns,5)
962        self.assertEqual(fail_file,
963            'reg_number,code,mandatory,level,level_session,ticket_session,score,unlock_score,matric_number,--ERRORS--\r\n'
964            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,Not all parents do exist yet.\r\n'
965            '1,NONSENSE,<IGNORE>,100,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,code: non-existent\r\n'
966            '1,COURSE1,<IGNORE>,200,2004,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: does not match 2008\r\n'
967            '1,COURSE1,<IGNORE>,300,2008,<IGNORE>,6,<IGNORE>,<IGNORE>,level object: does not exist\r\n'
968            '1,COURSE1,<IGNORE>,300,2008X,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: Invalid value\r\n')
969        assert self.processor.entryExists(
970            dict(reg_number='1', level='100', code='COURSE1'),
971            self.app) is True
972        courseticket = self.processor.getEntry(
973            dict(reg_number='1', level='100', code='COURSE1'), self.app)
974        self.assertEqual(courseticket.__parent__.__parent__.certificate.code,
975                         u'CERT1')
976        self.assertEqual(courseticket.score, 1)
977        self.assertEqual(courseticket.mandatory, True)
978        self.assertEqual(courseticket.fcode, 'NA')
979        self.assertEqual(courseticket.dcode, 'NA')
980        self.assertEqual(courseticket.code, 'COURSE1')
981        self.assertEqual(courseticket.title, 'Unnamed Course')
982        self.assertEqual(courseticket.credits, 10)
983        self.assertEqual(courseticket.passmark, 40)
984        self.assertEqual(courseticket.semester, 1)
985        self.assertEqual(courseticket.level, 100)
986        self.assertEqual(courseticket.level_session, 2008)
987        shutil.rmtree(os.path.dirname(fin_file))
988        logcontent = open(self.logfile).read()
989        # Logging message from updateEntry,
990        self.assertTrue(
991            'INFO - system - CourseTicket Processor - '
992            'sample_courseticket_data - K1000000 - 100 - '
993            'updated: code=COURSE1, '
994            'mandatory=False, score=3'
995            in logcontent)
996
997        # The catalog has been updated
998        cat = queryUtility(ICatalog, name='coursetickets_catalog')
999        results = list(
1000            cat.searchResults(
1001            level=(100, 100)))
1002        self.assertEqual(len(results),3)
1003
1004    def test_import_update(self):
1005        # We perform the same import twice,
1006        # the second time in update mode. The number
1007        # of warnings must be the same.
1008        num, num_warns, fin_file, fail_file = self.processor.doImport(
1009            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
1010        num, num_warns, fin_file, fail_file = self.processor.doImport(
1011            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
1012        fail_file = open(fail_file).read()
1013        self.assertEqual(num_warns,5)
1014        self.assertEqual(fail_file,
1015            'reg_number,code,mandatory,level,level_session,ticket_session,score,unlock_score,matric_number,--ERRORS--\r\n'
1016            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,Cannot update: no such entry\r\n'
1017            '1,NONSENSE,<IGNORE>,100,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,code: non-existent\r\n'
1018            '1,COURSE1,<IGNORE>,200,2004,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: does not match 2008\r\n'
1019            '1,COURSE1,<IGNORE>,300,2008,<IGNORE>,6,<IGNORE>,<IGNORE>,level object: does not exist\r\n'
1020            '1,COURSE1,<IGNORE>,300,2008X,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: Invalid value\r\n')
1021        shutil.rmtree(os.path.dirname(fin_file))
1022
1023    def test_import_remove(self):
1024        # We perform the same import twice,
1025        # the second time in remove mode. The number
1026        # of warnings must be the same.
1027        num, num_warns, fin_file, fail_file = self.processor.doImport(
1028            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
1029        shutil.rmtree(os.path.dirname(fin_file))
1030        assert self.processor.entryExists(
1031            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
1032        num, num_warns, fin_file, fail_file = self.processor.doImport(
1033            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
1034        self.assertEqual(num_warns,5)
1035        assert self.processor.entryExists(
1036            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
1037        shutil.rmtree(os.path.dirname(fin_file))
1038        logcontent = open(self.logfile).read()
1039        self.assertTrue(
1040            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
1041            in logcontent)
1042
1043    def test_import_update_locked(self):
1044        num, num_warns, fin_file, fail_file = self.processor.doImport(
1045            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
1046        shutil.rmtree(os.path.dirname(fin_file))
1047        # In state 'transcript validated' course tickets can't edited
1048        student = self.app['students']['X666666']
1049        IWorkflowState(student).setState('transcript validated')
1050        num, num_warns, fin_file, fail_file = self.processor.doImport(
1051            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
1052        fail_file = open(fail_file).read()
1053        # Two more records could't be imported because course tickets
1054        # of X666666 are locked
1055        self.assertEqual(num_warns,7)
1056        self.assertEqual(fail_file,
1057            'reg_number,code,mandatory,level,level_session,ticket_session,score,unlock_score,matric_number,--ERRORS--\r\n'
1058            '1,COURSE1,True,100,<IGNORE>,<IGNORE>,1,<IGNORE>,<IGNORE>,Studycourse is locked.\r\n'
1059            '1,COURSE1,True,200,2008,<IGNORE>,1,<IGNORE>,<IGNORE>,Studycourse is locked.\r\n'
1060            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,Cannot update: no such entry\r\n'
1061            '1,NONSENSE,<IGNORE>,100,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,code: non-existent\r\n'
1062            '1,COURSE1,<IGNORE>,200,2004,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: does not match 2008\r\n'
1063            '1,COURSE1,<IGNORE>,300,2008,<IGNORE>,6,<IGNORE>,<IGNORE>,level object: does not exist\r\n'
1064            '1,COURSE1,<IGNORE>,300,2008X,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: Invalid value\r\n')
1065        shutil.rmtree(os.path.dirname(fin_file))
1066
1067    def test_import_create_locked(self):
1068        # In state 'transcript validated' course tickets can't be created
1069        student = self.app['students']['X666666']
1070        IWorkflowState(student).setState('transcript validated')
1071        num, num_warns, fin_file, fail_file = self.processor.doImport(
1072            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
1073        fail_file = open(fail_file).read()
1074        self.assertEqual(num_warns,7)
1075        self.assertEqual(fail_file,
1076            'reg_number,code,mandatory,level,level_session,ticket_session,score,unlock_score,matric_number,--ERRORS--\r\n'
1077            '1,COURSE1,True,100,<IGNORE>,<IGNORE>,1,<IGNORE>,<IGNORE>,Studycourse is locked.\r\n'
1078            '1,COURSE1,True,200,2008,<IGNORE>,1,<IGNORE>,<IGNORE>,Studycourse is locked.\r\n'
1079            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,Not all parents do exist yet.\r\n'
1080            '1,NONSENSE,<IGNORE>,100,<IGNORE>,<IGNORE>,5,<IGNORE>,<IGNORE>,code: non-existent\r\n'
1081            '1,COURSE1,<IGNORE>,200,2004,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: does not match 2008\r\n'
1082            '1,COURSE1,<IGNORE>,300,2008,<IGNORE>,6,<IGNORE>,<IGNORE>,level object: does not exist\r\n'
1083            '1,COURSE1,<IGNORE>,300,2008X,<IGNORE>,6,<IGNORE>,<IGNORE>,level_session: Invalid value\r\n')
1084        self.assertFalse(self.processor.entryExists(
1085            dict(reg_number='1', level='100', code='COURSE1'),
1086            self.app))
1087        shutil.rmtree(os.path.dirname(fin_file))
1088
1089    def test_import_remove_locked(self):
1090        # We perform the same import twice,
1091        # the second time in remove mode.
1092        num, num_warns, fin_file, fail_file = self.processor.doImport(
1093            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
1094        shutil.rmtree(os.path.dirname(fin_file))
1095        assert self.processor.entryExists(
1096            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
1097        # In state 'transcript validated' course tickets can't be removed
1098        student = self.app['students']['X666666']
1099        IWorkflowState(student).setState('transcript validated')
1100        num, num_warns, fin_file, fail_file = self.processor.doImport(
1101            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
1102        self.assertEqual(num_warns,7)
1103        assert self.processor.entryExists(
1104            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
1105        logcontent = open(self.logfile).read()
1106        self.assertFalse(
1107            'INFO - system - X666666 - Course ticket in 100 removed: COURSE1'
1108            in logcontent)
1109        fail_file = open(fail_file).read()
1110        self.assertEqual(fail_file,
1111            'reg_number,code,matric_number,level,--ERRORS--\r\n'
1112            '1,COURSE1,<IGNORE>,100,Studycourse is locked.\r\n'
1113            '1,COURSE1,<IGNORE>,200,Studycourse is locked.\r\n'
1114            '1,COURSE1,<IGNORE>,nonsense,Cannot remove: no such entry\r\n'
1115            '1,NONSENSE,<IGNORE>,100,Cannot remove: no such entry\r\n'
1116            '1,COURSE1,<IGNORE>,200,Studycourse is locked.\r\n'
1117            '1,COURSE1,<IGNORE>,300,Cannot remove: no such entry\r\n'
1118            '1,COURSE1,<IGNORE>,300,Cannot remove: no such entry\r\n'
1119            )
1120        shutil.rmtree(os.path.dirname(fin_file))
1121
1122class PaymentProcessorTest(StudentImportExportSetup):
1123
1124    def setUp(self):
1125        super(PaymentProcessorTest, self).setUp()
1126
1127        # Add student with payment
1128        student = Student()
1129        student.firstname = u'Anna'
1130        student.lastname = u'Tester'
1131        student.reg_number = u'123'
1132        student.matric_number = u'234'
1133        self.app['students'].addStudent(student)
1134        self.student = self.app['students'][student.student_id]
1135        payment = createObject(u'waeup.StudentOnlinePayment')
1136        payment.p_id = 'p120'
1137        payment.p_session = 2012
1138        payment.p_category = 'schoolfee'
1139        payment.p_state = 'paid'
1140        self.student['payments'][payment.p_id] = payment
1141
1142        # Import students with subobjects
1143        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
1144        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
1145        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
1146            student_file, STUDENT_HEADER_FIELDS)
1147        shutil.rmtree(os.path.dirname(fin_file))
1148
1149        self.processor = StudentOnlinePaymentProcessor()
1150        self.csv_file = os.path.join(
1151            self.workdir, 'sample_payment_data.csv')
1152        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
1153        self.csv_file2 = os.path.join(
1154            self.workdir, 'sample_create_payment_data.csv')
1155        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
1156
1157    def test_interface(self):
1158        # Make sure we fulfill the interface contracts.
1159        assert verifyObject(IBatchProcessor, self.processor) is True
1160        assert verifyClass(
1161            IBatchProcessor, StudentOnlinePaymentProcessor) is True
1162
1163    def test_getEntry(self):
1164        assert self.processor.getEntry(
1165            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
1166        assert self.processor.getEntry(
1167            dict(student_id=self.student.student_id, p_id='p120'),
1168            self.app) is self.student['payments']['p120']
1169        assert self.processor.getEntry(
1170            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
1171            self.app) is self.student['payments']['p120']
1172
1173    def test_delEntry(self):
1174        assert self.processor.getEntry(
1175            dict(student_id=self.student.student_id, p_id='p120'),
1176            self.app) is self.student['payments']['p120']
1177        self.assertEqual(len(self.student['payments'].keys()),1)
1178        self.processor.delEntry(
1179            dict(student_id=self.student.student_id, p_id='p120'),
1180            self.app)
1181        assert self.processor.getEntry(
1182            dict(student_id=self.student.student_id, p_id='p120'),
1183            self.app) is None
1184        self.assertEqual(len(self.student['payments'].keys()),0)
1185
1186    def test_addEntry(self):
1187        self.assertEqual(len(self.student['payments'].keys()),1)
1188        payment1 = createObject(u'waeup.StudentOnlinePayment')
1189        payment1.p_id = 'p234'
1190        self.processor.addEntry(
1191            payment1, dict(student_id=self.student.student_id, p_id='p234'),
1192            self.app)
1193        self.assertEqual(len(self.student['payments'].keys()),2)
1194        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
1195        payment2 = createObject(u'waeup.StudentOnlinePayment')
1196        payment1.p_id = 'nonsense'
1197        # payment1.p_id will be replaced if p_id doesn't start with 'p'
1198        # and is not an old PIN
1199        self.processor.addEntry(
1200            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
1201            self.app)
1202        self.assertEqual(len(self.student['payments'].keys()),3)
1203        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
1204        # Requirement added on 19/02/2015: same payment must not exist.
1205        payment3 = createObject(u'waeup.StudentOnlinePayment')
1206        payment3.p_id = 'p456'
1207        payment3.p_session = 2012
1208        payment3.p_category = 'schoolfee'
1209        self.assertRaises(
1210            DuplicationError, self.processor.addEntry, payment3,
1211            dict(student_id=self.student.student_id, p_id='p456'), self.app)
1212        logcontent = open(self.logfile).read()
1213        self.assertTrue(
1214            'INFO - system - StudentOnlinePayment Processor - K1000000 - '
1215            'previous update cancelled'
1216            in logcontent)
1217
1218    def test_checkConversion(self):
1219        errs, inv_errs, conv_dict = self.processor.checkConversion(
1220            dict(p_id='<IGNORE>'), mode='create')
1221        self.assertEqual(len(errs),0)
1222        errs, inv_errs, conv_dict = self.processor.checkConversion(
1223            dict(p_id='<IGNORE>'), mode='update')
1224        self.assertEqual(len(errs),1)
1225        self.assertEqual(errs[0], ('p_id', u'missing'))
1226        errs, inv_errs, conv_dict = self.processor.checkConversion(
1227            dict(p_id='3816951266236341955'))
1228        self.assertEqual(len(errs),0)
1229        errs, inv_errs, conv_dict = self.processor.checkConversion(
1230            dict(p_id='p1266236341955'))
1231        self.assertEqual(len(errs),0)
1232        errs, inv_errs, conv_dict = self.processor.checkConversion(
1233            dict(p_id='ABC-11-1234567890'))
1234        self.assertEqual(len(errs),0)
1235        errs, inv_errs, conv_dict = self.processor.checkConversion(
1236            dict(p_id='nonsense'))
1237        self.assertEqual(len(errs),1)
1238        self.assertEqual(errs[0], ('p_id', u'invalid format'))
1239        timestamp = ("%d" % int(time()*10000))[1:]
1240        p_id = "p%s" % timestamp
1241        errs, inv_errs, conv_dict = self.processor.checkConversion(
1242            dict(p_id=p_id))
1243        self.assertEqual(len(errs),0)
1244        dup_payment = createObject(u'waeup.StudentOnlinePayment')
1245        dup_payment.p_id = 'XYZ-99-1234567890'
1246        self.student['payments'][dup_payment.p_id] = dup_payment
1247        errs, inv_errs, conv_dict = self.processor.checkConversion(
1248            dict(p_id='XYZ-99-1234567890'), mode='create')
1249        self.assertEqual(len(errs),1)
1250        self.assertEqual(errs[0], ('p_id', u'p_id exists in K1000000 '))
1251
1252    def test_import(self):
1253        num, num_warns, fin_file, fail_file = self.processor.doImport(
1254            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1255        self.assertEqual(num_warns,0)
1256
1257        payment = self.processor.getEntry(dict(reg_number='1',
1258            p_id='p2907979737440'), self.app)
1259        self.assertEqual(payment.p_id, 'p2907979737440')
1260        self.assertTrue(payment.p_current)
1261        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1262        self.assertEqual(cdate, "2010-11-26 18:59:33")
1263        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1264
1265        payment = self.processor.getEntry(dict(matric_number='100001',
1266            p_id='p2907125937570'), self.app)
1267        self.assertEqual(payment.p_id, 'p2907125937570')
1268        self.assertEqual(payment.amount_auth, 19500.1)
1269        self.assertFalse(payment.p_current)
1270        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1271        # Ooooh, still the old problem, see
1272        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
1273        # WAT is interpreted as GMT-1 and not GMT+1
1274        self.assertEqual(cdate, "2010-11-25 21:16:33")
1275        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1276
1277        payment = self.processor.getEntry(dict(reg_number='3',
1278            p_id='ABC-11-1234567890'), self.app)
1279        self.assertEqual(payment.amount_auth, 19500.6)
1280
1281        shutil.rmtree(os.path.dirname(fin_file))
1282        logcontent = open(self.logfile).read()
1283        # Logging message from updateEntry
1284        self.assertTrue(
1285            'INFO - system - StudentOnlinePayment Processor - '
1286            'sample_payment_data - K1000001 - updated: '
1287            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1288            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
1289            'p_session=2009, '
1290            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
1291            'p_state=paid'
1292            in logcontent)
1293        self.assertTrue(
1294            'INFO - system - StudentOnlinePayment Processor - '
1295            'sample_payment_data - K1000001 - updated: '
1296            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1297            'p_category=schoolfee, amount_auth=19500.6, p_current=True, '
1298            'p_session=2011, '
1299            'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, '
1300            'p_state=paid'
1301            in logcontent)
1302
1303    def test_import_update(self):
1304        # We perform the same import twice,
1305        # the second time in update mode. The number
1306        # of warnings increases becaus one p_id is missing.
1307        num, num_warns, fin_file, fail_file = self.processor.doImport(
1308            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1309        shutil.rmtree(os.path.dirname(fin_file))
1310        num, num_warns, fin_file, fail_file = self.processor.doImport(
1311            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
1312        self.assertEqual(num_warns,1)
1313        content = open(fail_file).read()
1314        shutil.rmtree(os.path.dirname(fin_file))
1315        self.assertTrue('p_id: missing' in content)
1316
1317    def test_import_remove(self):
1318        # We perform the same import twice,
1319        # the second time in remove mode. The number
1320        # of warnings increases becaus one p_id is missing.
1321        num, num_warns, fin_file, fail_file = self.processor.doImport(
1322            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1323        shutil.rmtree(os.path.dirname(fin_file))
1324        num, num_warns, fin_file, fail_file = self.processor.doImport(
1325            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
1326        self.assertEqual(num_warns,1)
1327        content = open(fail_file).read()
1328        self.assertTrue('p_id: missing' in content)
1329        shutil.rmtree(os.path.dirname(fin_file))
1330        logcontent = open(self.logfile).read()
1331        self.assertTrue(
1332            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
1333            in logcontent)
1334
1335    def test_import_same_payment_exists(self):
1336        num, num_warns, fin_file, fail_file = self.processor.doImport(
1337            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
1338        # One payment with same session and category exists
1339        self.assertEqual(num_warns,1)
1340        content = open(fail_file).read()
1341        self.assertTrue(
1342            '1,942,online,BTECHBDT,2010/11/26 19:59:33.744 GMT+1,0,'
1343            '19500,schoolfee,19500,2015,paid,'
1344            'Same payment has already been made.'
1345            in content)
1346        shutil.rmtree(os.path.dirname(fin_file))
1347        self.assertEqual(len(self.app['students']['X666666']['payments']), 13)
1348
1349class StudentVerdictProcessorTest(StudentImportExportSetup):
1350
1351    def setUp(self):
1352        super(StudentVerdictProcessorTest, self).setUp()
1353
1354        # Import students with subobjects
1355        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
1356        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
1357        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
1358            student_file, STUDENT_HEADER_FIELDS)
1359        shutil.rmtree(os.path.dirname(fin_file))
1360
1361        # Update study courses
1362        studycourse_file = os.path.join(
1363            self.workdir, 'sample_studycourse_data.csv')
1364        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
1365        processor = StudentStudyCourseProcessor()
1366        num, num_warns, fin_file, fail_file = processor.doImport(
1367            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
1368        shutil.rmtree(os.path.dirname(fin_file))
1369        # Import study levels
1370        self.csv_file = os.path.join(
1371            self.workdir, 'sample_studylevel_data.csv')
1372        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
1373        processor = StudentStudyLevelProcessor()
1374        num, num_warns, fin_file, fail_file = processor.doImport(
1375            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
1376        content = open(fail_file).read()
1377        shutil.rmtree(os.path.dirname(fin_file))
1378
1379        self.processor = StudentVerdictProcessor()
1380        self.csv_file = os.path.join(
1381            self.workdir, 'sample_verdict_data.csv')
1382        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
1383        return
1384
1385    def test_import(self):
1386        studycourse = self.processor.getEntry(dict(matric_number='100000'),
1387                                              self.app)
1388        self.assertEqual(studycourse['200'].level_verdict, '0')
1389        student = self.processor.getParent(
1390            dict(matric_number='100000'), self.app)
1391        num, num_warns, fin_file, fail_file = self.processor.doImport(
1392            self.csv_file, VERDICT_HEADER_FIELDS,'update')
1393        self.assertEqual(num_warns,5)
1394        self.assertEqual(studycourse.current_verdict, '0')
1395        self.assertEqual(student.state, 'returning')
1396        self.assertEqual(studycourse.current_level, 200)
1397        self.assertEqual(studycourse['200'].level_verdict, '0')
1398        student = self.processor.getParent(
1399            dict(matric_number='100005'), self.app)
1400        self.assertEqual(student.state, 'returning')
1401        self.assertEqual(student['studycourse'].current_verdict, 'A')
1402        self.assertEqual(studycourse.current_level, 200)
1403        self.assertEqual(student['studycourse']['200'].validated_by, 'System')
1404        self.assertTrue(isinstance(
1405            student['studycourse']['200'].validation_date, datetime.datetime))
1406        student = self.processor.getParent(
1407            dict(matric_number='100008'), self.app)
1408        self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana')
1409        content = open(fail_file).read()
1410        self.assertEqual(
1411            content,
1412            'current_session,current_level,bypass_validation,current_verdict,'
1413            'matric_number,validated_by,--ERRORS--\r\n'
1414            '2008,100,False,B,100001,<IGNORE>,Current level does not correspond.\r\n'
1415            '2007,200,<IGNORE>,C,100002,<IGNORE>,Current session does not correspond.\r\n'
1416            '2008,200,<IGNORE>,A,100003,<IGNORE>,Student in wrong state.\r\n'
1417            '2008,200,<IGNORE>,<IGNORE>,100004,<IGNORE>,No verdict in import file.\r\n'
1418            '2008,200,True,A,100007,<IGNORE>,Study level object is missing.\r\n'
1419            )
1420        logcontent = open(self.logfile).read()
1421        self.assertMatches(
1422            '... INFO - system - Verdict Processor (special processor, '
1423            'update only) - sample_verdict_data - X666666 - '
1424            'updated: current_verdict=0...',
1425            logcontent)
1426        self.assertMatches(
1427            '... INFO - system - X666666 - Returned...',
1428            logcontent)
1429        shutil.rmtree(os.path.dirname(fin_file))
1430
1431def test_suite():
1432    suite = unittest.TestSuite()
1433    for testcase in [
1434        StudentProcessorTest,StudentStudyCourseProcessorTest,
1435        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
1436        PaymentProcessorTest,StudentVerdictProcessorTest]:
1437        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
1438                testcase
1439                )
1440        )
1441    return suite
1442
1443
Note: See TracBrowser for help on using the repository browser.