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

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

DataForBursaryExporter?: Export only paid payment tickets.

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