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

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

Add missing sample data import file.
  • Property svn:keywords set to Id
File size: 53.8 KB
Line 
1# -*- coding: utf-8 -*-
2## $Id: test_batching.py 9961 2013-02-16 12:25:42Z 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']), 2)
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        shutil.rmtree(os.path.dirname(fin_file))
798
799    def test_import_remove(self):
800        # We perform the same import twice,
801        # the second time in remove mode. The number
802        # of warnings must be the same.
803        num, num_warns, fin_file, fail_file = self.processor.doImport(
804            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
805        shutil.rmtree(os.path.dirname(fin_file))
806        num, num_warns, fin_file, fail_file = self.processor.doImport(
807            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'remove')
808        assert self.processor.entryExists(
809            dict(reg_number='1', level='100'), self.app) is False
810        self.assertEqual(num_warns,2)
811
812        shutil.rmtree(os.path.dirname(fin_file))
813
814class CourseTicketProcessorTest(StudentImportExportSetup):
815
816    def setUp(self):
817        super(CourseTicketProcessorTest, self).setUp()
818
819        # Import students with subobjects
820        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
821        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
822        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
823            student_file, STUDENT_HEADER_FIELDS)
824        shutil.rmtree(os.path.dirname(fin_file))
825
826        # Add course and certificate course
827        self.course = createObject('waeup.Course')
828        self.course.code = 'COURSE1'
829        self.course.semester = 1
830        self.course.credits = 10
831        self.course.passmark = 40
832        self.app['faculties']['fac1']['dep1'].courses.addCourse(
833            self.course)
834        self.app['faculties']['fac1']['dep1'].certificates[
835            'CERT1'].addCertCourse(
836            self.course, level=100)
837
838        # Update study courses
839        studycourse_file = os.path.join(
840            self.workdir, 'sample_studycourse_data.csv')
841        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
842        processor = StudentStudyCourseProcessor()
843        num, num_warns, fin_file, fail_file = processor.doImport(
844            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
845        shutil.rmtree(os.path.dirname(fin_file))
846
847        # Import study levels
848        processor = StudentStudyLevelProcessor()
849        studylevel_file = os.path.join(
850            self.workdir, 'sample_studylevel_data.csv')
851        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
852        num, num_warns, fin_file, fail_file = processor.doImport(
853            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
854        shutil.rmtree(os.path.dirname(fin_file))
855
856        self.processor = CourseTicketProcessor()
857        self.csv_file = os.path.join(
858            self.workdir, 'sample_courseticket_data.csv')
859        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
860
861    def test_interface(self):
862        # Make sure we fulfill the interface contracts.
863        assert verifyObject(IBatchProcessor, self.processor) is True
864        assert verifyClass(
865            IBatchProcessor, CourseTicketProcessor) is True
866
867    def test_checkConversion(self):
868        errs, inv_errs, conv_dict = self.processor.checkConversion(
869            dict(reg_number='1', code='COURSE1', level='220'))
870        self.assertEqual(len(errs),0)
871        errs, inv_errs, conv_dict = self.processor.checkConversion(
872            dict(reg_number='1', code='COURSE2', level='220'))
873        self.assertEqual(len(errs),1)
874        self.assertTrue(('code','non-existent') in errs)
875
876    def test_import(self):
877        num, num_warns, fin_file, fail_file = self.processor.doImport(
878            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
879        fail_file = open(fail_file).read()
880        self.assertEqual(num_warns,5)
881        self.assertEqual(fail_file,
882            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
883            '1,COURSE1,,nonsense,,5,,Not all parents do exist yet. Skipping\r\n'
884            '1,NONSENSE,,100,,5,,code: non-existent\r\n'
885            '1,COURSE1,,200,2004,6,,level_session: does not match 2008\r\n'
886            '1,COURSE1,,300,2008,6,,level: does not exist\r\n'
887            '1,COURSE1,,300,2008X,6,,level_session: Invalid value\r\n')
888        assert self.processor.entryExists(
889            dict(reg_number='1', level='100', code='COURSE1'),
890            self.app) is True
891        courseticket = self.processor.getEntry(
892            dict(reg_number='1', level='100', code='COURSE1'), self.app)
893        self.assertEqual(courseticket.__parent__.__parent__.certificate.code,
894                         u'CERT1')
895        self.assertEqual(courseticket.score, 1)
896        self.assertEqual(courseticket.mandatory, True)
897        self.assertEqual(courseticket.fcode, 'NA')
898        self.assertEqual(courseticket.dcode, 'NA')
899        self.assertEqual(courseticket.code, 'COURSE1')
900        self.assertEqual(courseticket.title, 'Unnamed Course')
901        self.assertEqual(courseticket.credits, 10)
902        self.assertEqual(courseticket.passmark, 40)
903        self.assertEqual(courseticket.semester, 1)
904        shutil.rmtree(os.path.dirname(fin_file))
905        logcontent = open(self.logfile).read()
906        # Logging message from updateEntry,
907        self.assertTrue(
908            'INFO - system - CourseTicket Processor - '
909            'sample_courseticket_data - K1000000 - 100 - '
910            'updated: code=COURSE1, '
911            'mandatory=False, score=3'
912            in logcontent)
913
914    def test_import_update(self):
915        # We perform the same import twice,
916        # the second time in update mode. The number
917        # of warnings must be the same.
918        num, num_warns, fin_file, fail_file = self.processor.doImport(
919            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
920        shutil.rmtree(os.path.dirname(fin_file))
921        num, num_warns, fin_file, fail_file = self.processor.doImport(
922            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
923        fail_file = open(fail_file).read()
924        self.assertEqual(num_warns,5)
925        self.assertEqual(fail_file,
926            'reg_number,code,mandatory,level,level_session,score,matric_number,--ERRORS--\r\n'
927            '1,COURSE1,<IGNORE>,nonsense,<IGNORE>,5,<IGNORE>,Cannot update: no such entry\r\n'
928            '1,NONSENSE,<IGNORE>,100,<IGNORE>,5,<IGNORE>,code: non-existent\r\n'
929            '1,COURSE1,<IGNORE>,200,2004,6,<IGNORE>,level_session: does not match 2008\r\n'
930            '1,COURSE1,<IGNORE>,300,2008,6,<IGNORE>,level: does not exist\r\n'
931            '1,COURSE1,<IGNORE>,300,2008X,6,<IGNORE>,level_session: Invalid value\r\n')
932        shutil.rmtree(os.path.dirname(fin_file))
933
934    def test_import_remove(self):
935        # We perform the same import twice,
936        # the second time in remove mode. The number
937        # of warnings must be the same.
938        num, num_warns, fin_file, fail_file = self.processor.doImport(
939            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
940        shutil.rmtree(os.path.dirname(fin_file))
941        assert self.processor.entryExists(
942            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
943        num, num_warns, fin_file, fail_file = self.processor.doImport(
944            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
945        self.assertEqual(num_warns,5)
946        assert self.processor.entryExists(
947            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
948        shutil.rmtree(os.path.dirname(fin_file))
949        logcontent = open(self.logfile).read()
950        self.assertTrue(
951            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
952            in logcontent)
953
954class PaymentProcessorTest(StudentImportExportSetup):
955
956    def setUp(self):
957        super(PaymentProcessorTest, self).setUp()
958
959        # Add student with payment
960        student = Student()
961        student.firstname = u'Anna'
962        student.lastname = u'Tester'
963        student.reg_number = u'123'
964        student.matric_number = u'234'
965        self.app['students'].addStudent(student)
966        self.student = self.app['students'][student.student_id]
967        payment = createObject(u'waeup.StudentOnlinePayment')
968        payment.p_id = 'p120'
969        self.student['payments'][payment.p_id] = payment
970
971        # Import students with subobjects
972        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
973        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
974        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
975            student_file, STUDENT_HEADER_FIELDS)
976        shutil.rmtree(os.path.dirname(fin_file))
977
978        self.processor = StudentOnlinePaymentProcessor()
979        self.csv_file = os.path.join(
980            self.workdir, 'sample_payment_data.csv')
981        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
982        self.csv_file2 = os.path.join(
983            self.workdir, 'sample_create_payment_data.csv')
984        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
985
986    def test_interface(self):
987        # Make sure we fulfill the interface contracts.
988        assert verifyObject(IBatchProcessor, self.processor) is True
989        assert verifyClass(
990            IBatchProcessor, StudentOnlinePaymentProcessor) is True
991
992    def test_getEntry(self):
993        assert self.processor.getEntry(
994            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
995        assert self.processor.getEntry(
996            dict(student_id=self.student.student_id, p_id='p120'),
997            self.app) is self.student['payments']['p120']
998        assert self.processor.getEntry(
999            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
1000            self.app) is self.student['payments']['p120']
1001
1002    def test_delEntry(self):
1003        assert self.processor.getEntry(
1004            dict(student_id=self.student.student_id, p_id='p120'),
1005            self.app) is self.student['payments']['p120']
1006        self.assertEqual(len(self.student['payments'].keys()),1)
1007        self.processor.delEntry(
1008            dict(student_id=self.student.student_id, p_id='p120'),
1009            self.app)
1010        assert self.processor.getEntry(
1011            dict(student_id=self.student.student_id, p_id='p120'),
1012            self.app) is None
1013        self.assertEqual(len(self.student['payments'].keys()),0)
1014
1015    def test_addEntry(self):
1016        self.assertEqual(len(self.student['payments'].keys()),1)
1017        payment1 = createObject(u'waeup.StudentOnlinePayment')
1018        payment1.p_id = 'p234'
1019        self.processor.addEntry(
1020            payment1, dict(student_id=self.student.student_id, p_id='p234'),
1021            self.app)
1022        self.assertEqual(len(self.student['payments'].keys()),2)
1023        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
1024        payment2 = createObject(u'waeup.StudentOnlinePayment')
1025        payment1.p_id = 'nonsense'
1026        # payment1.p_id will be replaced if p_id doesn't start with 'p'
1027        # and is not an old PIN
1028        self.processor.addEntry(
1029            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
1030            self.app)
1031        self.assertEqual(len(self.student['payments'].keys()),3)
1032        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
1033
1034    def test_checkConversion(self):
1035        errs, inv_errs, conv_dict = self.processor.checkConversion(
1036            dict(p_id='3816951266236341955'))
1037        self.assertEqual(len(errs),0)
1038        errs, inv_errs, conv_dict = self.processor.checkConversion(
1039            dict(p_id='p1266236341955'))
1040        self.assertEqual(len(errs),0)
1041        errs, inv_errs, conv_dict = self.processor.checkConversion(
1042            dict(p_id='ABC-11-1234567890'))
1043        self.assertEqual(len(errs),0)
1044        errs, inv_errs, conv_dict = self.processor.checkConversion(
1045            dict(p_id='nonsense'))
1046        self.assertEqual(len(errs),1)
1047        timestamp = ("%d" % int(time()*10000))[1:]
1048        p_id = "p%s" % timestamp
1049        errs, inv_errs, conv_dict = self.processor.checkConversion(
1050            dict(p_id=p_id))
1051        self.assertEqual(len(errs),0)
1052
1053    def test_import(self):
1054        num, num_warns, fin_file, fail_file = self.processor.doImport(
1055            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1056        self.assertEqual(num_warns,0)
1057
1058        payment = self.processor.getEntry(dict(reg_number='1',
1059            p_id='p2907979737440'), self.app)
1060        self.assertEqual(payment.p_id, 'p2907979737440')
1061        self.assertTrue(payment.p_current)
1062        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1063        self.assertEqual(cdate, "2010-11-26 18:59:33")
1064        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1065
1066        payment = self.processor.getEntry(dict(matric_number='100001',
1067            p_id='p2907125937570'), self.app)
1068        self.assertEqual(payment.p_id, 'p2907125937570')
1069        self.assertEqual(payment.amount_auth, 19500.1)
1070        self.assertFalse(payment.p_current)
1071        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
1072        # Ooooh, still the old problem, see
1073        # http://mail.dzug.org/mailman/archives/zope/2006-August/001153.html.
1074        # WAT is interpreted as GMT-1 and not GMT+1
1075        self.assertEqual(cdate, "2010-11-25 21:16:33")
1076        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
1077
1078        payment = self.processor.getEntry(dict(reg_number='3',
1079            p_id='ABC-11-1234567890'), self.app)
1080        self.assertEqual(payment.amount_auth, 19500.6)
1081
1082        shutil.rmtree(os.path.dirname(fin_file))
1083        logcontent = open(self.logfile).read()
1084        # Logging message from updateEntry
1085        self.assertTrue(
1086            'INFO - system - Student Payment Processor - '
1087            'sample_payment_data - K1000001 - updated: '
1088            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1089            'p_category=schoolfee, amount_auth=19500.0, p_current=True, '
1090            'p_id=p1266236341955, r_code=00, r_amount_approved=19500.0, '
1091            'p_state=paid'
1092            in logcontent)
1093        self.assertTrue(
1094            'INFO - system - Student Payment Processor - '
1095            'sample_payment_data - K1000001 - updated: '
1096            'p_item=BTECHBDT, creation_date=2010-02-15 13:19:01+00:00, '
1097            'p_category=schoolfee, amount_auth=19500.6, p_current=True, '
1098            'p_id=ABC-11-1234567890, r_code=SC, r_amount_approved=19500.0, '
1099            'p_state=paid'
1100            in logcontent)
1101
1102    def test_import_update(self):
1103        # We perform the same import twice,
1104        # the second time in update mode. The number
1105        # of warnings must be the same.
1106        num, num_warns, fin_file, fail_file = self.processor.doImport(
1107            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1108        shutil.rmtree(os.path.dirname(fin_file))
1109        num, num_warns, fin_file, fail_file = self.processor.doImport(
1110            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
1111        self.assertEqual(num_warns,0)
1112        shutil.rmtree(os.path.dirname(fin_file))
1113
1114    def test_import_remove(self):
1115        num, num_warns, fin_file, fail_file = self.processor.doImport(
1116            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
1117        shutil.rmtree(os.path.dirname(fin_file))
1118        num, num_warns, fin_file, fail_file = self.processor.doImport(
1119            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
1120        self.assertEqual(num_warns,0)
1121        shutil.rmtree(os.path.dirname(fin_file))
1122        logcontent = open(self.logfile).read()
1123        self.assertTrue(
1124            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
1125            in logcontent)
1126
1127    def test_import_wo_pid(self):
1128        num, num_warns, fin_file, fail_file = self.processor.doImport(
1129            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
1130        self.assertEqual(num_warns,0)
1131        shutil.rmtree(os.path.dirname(fin_file))
1132        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
1133
1134class StudentVerdictProcessorTest(StudentImportExportSetup):
1135
1136    def setUp(self):
1137        super(StudentVerdictProcessorTest, self).setUp()
1138
1139        # Import students with subobjects
1140        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
1141        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
1142        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
1143            student_file, STUDENT_HEADER_FIELDS)
1144        shutil.rmtree(os.path.dirname(fin_file))
1145
1146        # Update study courses
1147        studycourse_file = os.path.join(
1148            self.workdir, 'sample_studycourse_data.csv')
1149        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
1150        processor = StudentStudyCourseProcessor()
1151        num, num_warns, fin_file, fail_file = processor.doImport(
1152            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
1153        shutil.rmtree(os.path.dirname(fin_file))
1154        # Import study levels
1155        self.csv_file = os.path.join(
1156            self.workdir, 'sample_studylevel_data.csv')
1157        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
1158        processor = StudentStudyLevelProcessor()
1159        num, num_warns, fin_file, fail_file = processor.doImport(
1160            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
1161        content = open(fail_file).read()
1162        shutil.rmtree(os.path.dirname(fin_file))
1163
1164        self.processor = StudentVerdictProcessor()
1165        self.csv_file = os.path.join(
1166            self.workdir, 'sample_verdict_data.csv')
1167        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
1168        return
1169
1170    def test_import(self):
1171        studycourse = self.processor.getEntry(dict(matric_number='100000'),
1172                                              self.app)
1173        self.assertEqual(studycourse['200'].level_verdict, None)
1174        student = self.processor.getParent(
1175            dict(matric_number='100000'), self.app)
1176        num, num_warns, fin_file, fail_file = self.processor.doImport(
1177            self.csv_file, VERDICT_HEADER_FIELDS,'update')
1178        #content = open(fail_file).read()
1179        #import pdb; pdb.set_trace()
1180        self.assertEqual(num_warns,5)
1181        self.assertEqual(studycourse.current_verdict, '0')
1182        self.assertEqual(student.state, 'returning')
1183        self.assertEqual(studycourse.current_level, 200)
1184        self.assertEqual(studycourse['200'].level_verdict, '0')
1185        student = self.processor.getParent(
1186            dict(matric_number='100005'), self.app)
1187        self.assertEqual(student.state, 'returning')
1188        self.assertEqual(student['studycourse'].current_verdict, 'A')
1189        self.assertEqual(studycourse.current_level, 200)
1190        self.assertEqual(student['studycourse']['200'].validated_by, 'System')
1191        self.assertTrue(isinstance(
1192            student['studycourse']['200'].validation_date, datetime.datetime))
1193        student = self.processor.getParent(
1194            dict(matric_number='100008'), self.app)
1195        self.assertEqual(student['studycourse']['200'].validated_by, 'Juliana')
1196        content = open(fail_file).read()
1197        self.assertEqual(
1198            content,
1199            'current_session,current_level,bypass_validation,current_verdict,'
1200            'matric_number,validated_by,--ERRORS--\r\n'
1201            '2008,100,False,B,100001,<IGNORE>,Current level does not correspond.\r\n'
1202            '2007,200,<IGNORE>,C,100002,<IGNORE>,Current session does not correspond.\r\n'
1203            '2008,200,<IGNORE>,A,100003,<IGNORE>,Student in wrong state.\r\n'
1204            '2008,200,<IGNORE>,<IGNORE>,100004,<IGNORE>,No verdict in import file.\r\n'
1205            '2008,200,True,A,100007,<IGNORE>,Study level object is missing.\r\n'
1206            )
1207        logcontent = open(self.logfile).read()
1208        self.assertMatches(
1209            '... INFO - system - Verdict Processor (special processor, '
1210            'update only) - sample_verdict_data - X666666 - '
1211            'updated: current_verdict=0...',
1212            logcontent)
1213        self.assertMatches(
1214            '... INFO - system - X666666 - Returned...',
1215            logcontent)
1216
1217        shutil.rmtree(os.path.dirname(fin_file))
1218
1219def test_suite():
1220    suite = unittest.TestSuite()
1221    for testcase in [
1222        StudentProcessorTest,StudentStudyCourseProcessorTest,
1223        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
1224        PaymentProcessorTest,StudentVerdictProcessorTest]:
1225        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
1226                testcase
1227                )
1228        )
1229    return suite
1230
1231
Note: See TracBrowser for help on using the repository browser.