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

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

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