source: main/waeup.kofa/branches/0.2/src/waeup/kofa/students/tests/test_batching.py @ 11476

Last change on this file since 11476 was 10842, checked in by Henrik Bettermann, 11 years ago

We do not need a default value for p_category.
  • Property svn:keywords set to Id
File size: 54.4 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 10842 2013-12-11 13:21:37Z 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        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: reg_number\r\n'
544            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n'
545            '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\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
1068    def test_import(self):
1069        num, num_warns, fin_file, fail_file = self.processor.doImport(
1070            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1071        self.assertEqual(num_warns,0)
1072
1073        payment = self.processor.getEntry(dict(reg_number='1',
1074            p_id='p2907979737440'), self.app)
1075        self.assertEqual(payment.p_id, 'p2907979737440')
1076        self.assertTrue(payment.p_current)
1077        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1078        self.assertEqual(cdate, "2010-11-26 18:59:33")
1079        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1080
1081        payment = self.processor.getEntry(dict(matric_number='100001',
1082            p_id='p2907125937570'), self.app)
1083        self.assertEqual(payment.p_id, 'p2907125937570')
1084        self.assertEqual(payment.amount_auth, 19500.1)
1085        self.assertFalse(payment.p_current)
1086        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1087        # Ooooh, still the old problem, see
1088        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
1089        # WAT is interpreted as GMT-1 and not GMT+1
1090        self.assertEqual(cdate, "2010-11-25 21:16:33")
1091        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1092
1093        payment = self.processor.getEntry(dict(reg_number='3',
1094            p_id='ABC-11-1234567890'), self.app)
1095        self.assertEqual(payment.amount_auth, 19500.6)
1096
1097        shutil.rmtree(os.path.dirname(fin_file))
1098        logcontent = open(self.logfile).read()
1099        # Logging message from updateEntry
1100        self.assertTrue(
1101            'INFO - system - Student Payment Processor - '
1102            'sample_payment_data - K1000001 - updated: '
1103            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1104            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
1105            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
1106            'p_state=paid'
1107            in logcontent)
1108        self.assertTrue(
1109            'INFO - system - Student Payment Processor - '
1110            'sample_payment_data - K1000001 - updated: '
1111            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1112            'p_category=schoolfee, amount_auth=19500.6, p_current=True, '
1113            'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, '
1114            'p_state=paid'
1115            in logcontent)
1116
1117    def test_import_update(self):
1118        # We perform the same import twice,
1119        # the second time in update mode. The number
1120        # of warnings must be the same.
1121        num, num_warns, fin_file, fail_file = self.processor.doImport(
1122            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1123        shutil.rmtree(os.path.dirname(fin_file))
1124        num, num_warns, fin_file, fail_file = self.processor.doImport(
1125            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
1126        self.assertEqual(num_warns,0)
1127        shutil.rmtree(os.path.dirname(fin_file))
1128
1129    def test_import_remove(self):
1130        num, num_warns, fin_file, fail_file = self.processor.doImport(
1131            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1132        shutil.rmtree(os.path.dirname(fin_file))
1133        num, num_warns, fin_file, fail_file = self.processor.doImport(
1134            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
1135        self.assertEqual(num_warns,0)
1136        shutil.rmtree(os.path.dirname(fin_file))
1137        logcontent = open(self.logfile).read()
1138        self.assertTrue(
1139            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
1140            in logcontent)
1141
1142    def test_import_wo_pid(self):
1143        num, num_warns, fin_file, fail_file = self.processor.doImport(
1144            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
1145        self.assertEqual(num_warns,0)
1146        shutil.rmtree(os.path.dirname(fin_file))
1147        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
1148
1149class StudentVerdictProcessorTest(StudentImportExportSetup):
1150
1151    def setUp(self):
1152        super(StudentVerdictProcessorTest, self).setUp()
1153
1154        # Import students with subobjects
1155        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
1156        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
1157        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
1158            student_file, STUDENT_HEADER_FIELDS)
1159        shutil.rmtree(os.path.dirname(fin_file))
1160
1161        # Update study courses
1162        studycourse_file = os.path.join(
1163            self.workdir, 'sample_studycourse_data.csv')
1164        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
1165        processor = StudentStudyCourseProcessor()
1166        num, num_warns, fin_file, fail_file = processor.doImport(
1167            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
1168        shutil.rmtree(os.path.dirname(fin_file))
1169        # Import study levels
1170        self.csv_file = os.path.join(
1171            self.workdir, 'sample_studylevel_data.csv')
1172        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
1173        processor = StudentStudyLevelProcessor()
1174        num, num_warns, fin_file, fail_file = processor.doImport(
1175            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
1176        content = open(fail_file).read()
1177        shutil.rmtree(os.path.dirname(fin_file))
1178
1179        self.processor = StudentVerdictProcessor()
1180        self.csv_file = os.path.join(
1181            self.workdir, 'sample_verdict_data.csv')
1182        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
1183        return
1184
1185    def test_import(self):
1186        studycourse = self.processor.getEntry(dict(matric_number='100000'),
1187                                              self.app)
1188        self.assertEqual(studycourse['200'].level_verdict, None)
1189        student = self.processor.getParent(
1190            dict(matric_number='100000'), self.app)
1191        num, num_warns, fin_file, fail_file = self.processor.doImport(
1192            self.csv_file, VERDICT_HEADER_FIELDS,'update')
1193        #content = open(fail_file).read()
1194        #import pdb; pdb.set_trace()
1195        self.assertEqual(num_warns,5)
1196        self.assertEqual(studycourse.current_verdict, '0')
1197        self.assertEqual(student.state, 'returning')
1198        self.assertEqual(studycourse.current_level, 200)
1199        self.assertEqual(studycourse['200'].level_verdict, '0')
1200        student = self.processor.getParent(
1201            dict(matric_number='100005'), self.app)
1202        self.assertEqual(student.state, 'returning')
1203        self.assertEqual(student['studycourse'].current_verdict, 'A')
1204        self.assertEqual(studycourse.current_level, 200)
1205        self.assertEqual(student['studycourse']['200'].validated_by, 'System')
1206        self.assertTrue(isinstance(
1207            student['studycourse']['200'].validation_date, datetime.datetime))
1208        student = self.processor.getParent(
1209            dict(matric_number='100008'), self.app)
1210        self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana')
1211        content = open(fail_file).read()
1212        self.assertEqual(
1213            content,
1214            'current_session,current_level,bypass_validation,current_verdict,'
1215            'matric_number,validated_by,--ERRORS--\r\n'
1216            '2008,100,False,B,100001,<IGNORE>,Current level does not correspond.\r\n'
1217            '2007,200,<IGNORE>,C,100002,<IGNORE>,Current session does not correspond.\r\n'
1218            '2008,200,<IGNORE>,A,100003,<IGNORE>,Student in wrong state.\r\n'
1219            '2008,200,<IGNORE>,<IGNORE>,100004,<IGNORE>,No verdict in import file.\r\n'
1220            '2008,200,True,A,100007,<IGNORE>,Study level object is missing.\r\n'
1221            )
1222        logcontent = open(self.logfile).read()
1223        self.assertMatches(
1224            '... INFO - system - Verdict Processor (special processor, '
1225            'update only) - sample_verdict_data - X666666 - '
1226            'updated: current_verdict=0...',
1227            logcontent)
1228        self.assertMatches(
1229            '... INFO - system - X666666 - Returned...',
1230            logcontent)
1231
1232        shutil.rmtree(os.path.dirname(fin_file))
1233
1234def test_suite():
1235    suite = unittest.TestSuite()
1236    for testcase in [
1237        StudentProcessorTest,StudentStudyCourseProcessorTest,
1238        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
1239        PaymentProcessorTest,StudentVerdictProcessorTest]:
1240        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
1241                testcase
1242                )
1243        )
1244    return suite
1245
1246
Note: See TracBrowser for help on using the repository browser.