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

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

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