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

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

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