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

Last change on this file since 9960 was 9960, checked in by Henrik Bettermann, 12 years ago

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