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

Last change on this file since 16824 was 16821, checked in by Henrik Bettermann, 3 years ago

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