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

Last change on this file since 15253 was 15203, checked in by Henrik Bettermann, 6 years ago

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