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

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

Rename course referrers to certificate courses (according to the discussion long time ago).

Edit some button labels and titles.

Move code from view to content components.
  • Property svn:keywords set to Id
File size: 40.3 KB
Line 
1## $Id: test_batching.py 8920 2012-07-05 14:48:51Z henrik $
2##
3## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
4## This program is free software; you can redistribute it and/or modify
5## it under the terms of the GNU General Public License as published by
6## the Free Software Foundation; either version 2 of the License, or
7## (at your option) any later version.
8##
9## This program is distributed in the hope that it will be useful,
10## but WITHOUT ANY WARRANTY; without even the implied warranty of
11## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12## GNU General Public License for more details.
13##
14## You should have received a copy of the GNU General Public License
15## along with this program; if not, write to the Free Software
16## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17##
18"""Unit tests for students-related data processors.
19"""
20import os
21import shutil
22import tempfile
23import unittest
24import datetime
25from time import time
26from zope.component import createObject
27from zope.component.hooks import setSite, clearSite
28from zope.interface.verify import verifyClass, verifyObject
29
30from waeup.kofa.app import University
31from waeup.kofa.interfaces import IBatchProcessor, FatalCSVError, IUserAccount
32from waeup.kofa.students.batching import (
33    StudentProcessor, StudentStudyCourseProcessor,
34    StudentStudyLevelProcessor, CourseTicketProcessor,
35    StudentOnlinePaymentProcessor, StudentVerdictProcessor)
36from waeup.kofa.students.payments import StudentOnlinePayment
37from waeup.kofa.students.student import Student
38from waeup.kofa.students.studylevel import StudentStudyLevel, CourseTicket
39from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
40from waeup.kofa.university.faculty import Faculty
41from waeup.kofa.university.department import Department
42
43
44STUDENT_SAMPLE_DATA = open(
45    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
46    'rb').read()
47
48STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
49    '\n')[0].split(',')
50
51STUDENT_SAMPLE_DATA_UPDATE = open(
52    os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'),
53    'rb').read()
54
55STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split(
56    '\n')[0].split(',')
57
58STUDENT_SAMPLE_DATA_UPDATE2 = open(
59    os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'),
60    'rb').read()
61
62STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split(
63    '\n')[0].split(',')
64
65STUDENT_SAMPLE_DATA_UPDATE3 = open(
66    os.path.join(os.path.dirname(__file__), 'sample_student_data_update3.csv'),
67    'rb').read()
68
69STUDENT_HEADER_FIELDS_UPDATE3 = STUDENT_SAMPLE_DATA_UPDATE3.split(
70    '\n')[0].split(',')
71
72STUDENT_SAMPLE_DATA_UPDATE4 = open(
73    os.path.join(os.path.dirname(__file__), 'sample_student_data_update4.csv'),
74    'rb').read()
75
76STUDENT_HEADER_FIELDS_UPDATE4 = STUDENT_SAMPLE_DATA_UPDATE4.split(
77    '\n')[0].split(',')
78
79STUDYCOURSE_SAMPLE_DATA = open(
80    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
81    'rb').read()
82
83STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
84    '\n')[0].split(',')
85
86VERDICT_SAMPLE_DATA = open(
87    os.path.join(os.path.dirname(__file__), 'sample_verdict_data.csv'),
88    'rb').read()
89
90VERDICT_HEADER_FIELDS = VERDICT_SAMPLE_DATA.split(
91    '\n')[0].split(',')
92
93STUDENT_SAMPLE_DATA_MIGRATION = open(
94    os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'),
95    'rb').read()
96
97STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split(
98    '\n')[0].split(',')
99
100STUDENT_SAMPLE_DATA_DUPLICATES = open(
101    os.path.join(os.path.dirname(__file__), 'sample_student_data_duplicates.csv'),
102    'rb').read()
103
104STUDENT_HEADER_FIELDS_DUPLICATES = STUDENT_SAMPLE_DATA_DUPLICATES.split(
105    '\n')[0].split(',')
106
107STUDYLEVEL_SAMPLE_DATA = open(
108    os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'),
109    'rb').read()
110
111STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split(
112    '\n')[0].split(',')
113
114COURSETICKET_SAMPLE_DATA = open(
115    os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'),
116    'rb').read()
117
118COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
119    '\n')[0].split(',')
120
121PAYMENT_SAMPLE_DATA = open(
122    os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'),
123    'rb').read()
124
125PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
126    '\n')[0].split(',')
127
128PAYMENT_CREATE_SAMPLE_DATA = open(
129    os.path.join(os.path.dirname(__file__), 'sample_create_payment_data.csv'),
130    'rb').read()
131
132PAYMENT_CREATE_HEADER_FIELDS = PAYMENT_CREATE_SAMPLE_DATA.split(
133    '\n')[0].split(',')
134
135class StudentImportExportSetup(FunctionalTestCase):
136
137    layer = FunctionalLayer
138
139    def setUp(self):
140        super(StudentImportExportSetup, self).setUp()
141        self.dc_root = tempfile.mkdtemp()
142        self.workdir = tempfile.mkdtemp()
143        app = University()
144        app['datacenter'].setStoragePath(self.dc_root)
145        self.getRootFolder()['app'] = app
146        self.app = self.getRootFolder()['app']
147        setSite(app)
148
149        # Populate university
150        self.certificate = createObject('waeup.Certificate')
151        self.certificate.code = 'CERT1'
152        self.certificate.application_category = 'basic'
153        self.certificate.start_level = 200
154        self.certificate.end_level = 500
155        self.app['faculties']['fac1'] = Faculty()
156        self.app['faculties']['fac1']['dep1'] = Department()
157        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
158            self.certificate)
159
160        self.logfile = os.path.join(
161            self.app['datacenter'].storage, 'logs', 'students.log')
162        return
163
164    def tearDown(self):
165        super(StudentImportExportSetup, self).tearDown()
166        shutil.rmtree(self.workdir)
167        shutil.rmtree(self.dc_root)
168        clearSite()
169        return
170
171    def setup_for_export(self):
172        student = Student()
173        student.student_id = u'A111111'
174        self.app['students'][student.student_id] = self.student = student
175        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
176        return
177
178    def setup_student(self, student):
179        # set predictable values for `student`
180        student.matric_number = u'M123456'
181        student.adm_code = u'my adm code'
182        student.clearance_locked = False
183        student.clr_code = u'my clr code'
184        student.perm_address = u'Studentroad 21\nLagos 123456\n'
185        student.reg_number = u'123456'
186        student.student_id = u'A111111'
187        student.firstname = u'Anna'
188        student.lastname = u'Tester'
189        student.middlename = u'M.'
190        student.date_of_birth = datetime.date(1981, 2, 4)
191        student.sex = 'f'
192        student.email = 'anna@sample.com'
193        student.phone = u'+234-123-12345'
194        student.notice = u'Some notice\nin lines.'
195        student.nationality = u'NG'
196
197        student['studycourse'].certificate = self.certificate
198        student['studycourse'].entry_mode = 'ug_ft'
199        student['studycourse'].entry_session = 2010
200        student['studycourse'].current_session = 2012
201        student['studycourse'].current_level = int(self.certificate.start_level)
202
203        study_level = StudentStudyLevel()
204        study_level.level_session = 2012
205        study_level.level_verdict = "A"
206        study_level.level = 100
207        student['studycourse'].addStudentStudyLevel(
208            self.certificate, study_level)
209
210        ticket = CourseTicket()
211        ticket.automatic = True
212        ticket.carry_over = True
213        ticket.code = u'CRS1'
214        ticket.title = u'Course 1'
215        ticket.fcode = u'FAC1'
216        ticket.dcode = u'DEP1'
217        ticket.credits = 100
218        ticket.passmark = 100
219        ticket.semester = 2
220        study_level[ticket.code] = ticket
221        self.add_payment(student)
222        return student
223
224    def add_payment(self, student):
225        # get a payment with all fields set
226        payment = StudentOnlinePayment()
227        payment.creation_date = datetime.datetime(2012, 4, 1, 13, 12, 1)
228        payment.p_id = 'my-id'
229        payment.ac = u'666'
230        payment.p_item = u'p-item'
231        payment.p_level = 100
232        payment.p_session = 2012
233        payment.payment_date = datetime.datetime(2012, 4, 1, 14, 12, 1)
234        payment.r_amount_approved = 12.12
235        payment.r_code = u'r-code'
236        # XXX: there is no addPayment method to give predictable names
237        student['payments']['my-payment'] = payment
238        return payment
239
240
241
242class StudentProcessorTest(FunctionalTestCase):
243
244    layer = FunctionalLayer
245
246    def setUp(self):
247        super(StudentProcessorTest, self).setUp()
248        # Setup a sample site for each test
249        app = University()
250        self.dc_root = tempfile.mkdtemp()
251        app['datacenter'].setStoragePath(self.dc_root)
252
253        # Prepopulate the ZODB...
254        self.getRootFolder()['app'] = app
255        # we add the site immediately after creation to the
256        # ZODB. Catalogs and other local utilities are not setup
257        # before that step.
258        self.app = self.getRootFolder()['app']
259        # Set site here. Some of the following setup code might need
260        # to access grok.getSite() and should get our new app then
261        setSite(app)
262
263        # Add student with subobjects
264        student = Student()
265        student.firstname = u'Anna'
266        student.lastname = u'Tester'
267        student.reg_number = u'123'
268        student.matric_number = u'234'
269        self.app['students'].addStudent(student)
270        self.student = self.app['students'][student.student_id]
271        self.processor = StudentProcessor()
272        self.workdir = tempfile.mkdtemp()
273        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
274        self.csv_file_update = os.path.join(
275            self.workdir, 'sample_student_data_update.csv')
276        self.csv_file_update2 = os.path.join(
277            self.workdir, 'sample_student_data_update2.csv')
278        self.csv_file_update3 = os.path.join(
279            self.workdir, 'sample_student_data_update3.csv')
280        self.csv_file_update4 = os.path.join(
281            self.workdir, 'sample_student_data_update4.csv')
282        self.csv_file_migration = os.path.join(
283            self.workdir, 'sample_student_data_migration.csv')
284        self.csv_file_duplicates = os.path.join(
285            self.workdir, 'sample_student_data_duplicates.csv')
286        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
287        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
288        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
289        open(self.csv_file_update3, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE3)
290        open(self.csv_file_update4, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE4)
291        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
292        open(self.csv_file_duplicates, 'wb').write(STUDENT_SAMPLE_DATA_DUPLICATES)
293
294    def tearDown(self):
295        super(StudentProcessorTest, self).tearDown()
296        shutil.rmtree(self.workdir)
297        shutil.rmtree(self.dc_root)
298        clearSite()
299        return
300
301    def test_interface(self):
302        # Make sure we fulfill the interface contracts.
303        assert verifyObject(IBatchProcessor, self.processor) is True
304        assert verifyClass(
305            IBatchProcessor, StudentProcessor) is True
306
307    def test_parentsExist(self):
308        self.assertFalse(self.processor.parentsExist(None, dict()))
309        self.assertTrue(self.processor.parentsExist(None, self.app))
310
311    def test_entryExists(self):
312        assert self.processor.entryExists(
313            dict(student_id='ID_NONE'), self.app) is False
314        assert self.processor.entryExists(
315            dict(reg_number='123'), self.app) is True
316
317    def test_getParent(self):
318        parent = self.processor.getParent(None, self.app)
319        assert parent is self.app['students']
320
321    def test_getEntry(self):
322        assert self.processor.getEntry(
323            dict(student_id='ID_NONE'), self.app) is None
324        assert self.processor.getEntry(
325            dict(student_id=self.student.student_id), self.app) is self.student
326
327    def test_addEntry(self):
328        new_student = Student()
329        self.processor.addEntry(
330            new_student, dict(), self.app)
331        assert len(self.app['students'].keys()) == 2
332
333    def test_checkConversion(self):
334        # Make sure we can check conversions and that the stud_id
335        # counter is not raised during such checks.
336        initial_stud_id = self.app['students']._curr_stud_id
337        errs, inv_errs, conv_dict = self.processor.checkConversion(
338            dict(reg_number='1', state='admitted'))
339        self.assertEqual(len(errs),0)
340        # Empty state is allowed
341        errs, inv_errs, conv_dict = self.processor.checkConversion(
342            dict(reg_number='1', state=''))
343        self.assertEqual(len(errs),0)
344        #self.assertTrue(('state', 'no value provided') in errs)
345        errs, inv_errs, conv_dict = self.processor.checkConversion(
346            dict(reg_number='1', state='nonsense'))
347        self.assertEqual(len(errs),1)
348        self.assertTrue(('state', 'not allowed') in errs)
349        new_stud_id = self.app['students']._curr_stud_id
350        self.assertEqual(initial_stud_id, new_stud_id)
351        return
352
353    def test_delEntry(self):
354        assert self.student.student_id in self.app['students'].keys()
355        self.processor.delEntry(
356            dict(reg_number=self.student.reg_number), self.app)
357        assert self.student.student_id not in self.app['students'].keys()
358
359    def test_import(self):
360        self.assertEqual(self.app['students']._curr_stud_id, 1000001)
361        num, num_warns, fin_file, fail_file = self.processor.doImport(
362            self.csv_file, STUDENT_HEADER_FIELDS)
363        self.assertEqual(num_warns,0)
364        assert len(self.app['students'].keys()) == 5
365        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
366        self.assertEqual(
367            self.app['students']['X666666'].state, 'courses validated')
368        # Two new student_ids have been created.
369        self.assertEqual(self.app['students']._curr_stud_id, 1000003)
370        shutil.rmtree(os.path.dirname(fin_file))
371
372    def test_import_update(self):
373        num, num_warns, fin_file, fail_file = self.processor.doImport(
374            self.csv_file, STUDENT_HEADER_FIELDS)
375        shutil.rmtree(os.path.dirname(fin_file))
376        num, num_warns, fin_file, fail_file = self.processor.doImport(
377            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
378        self.assertEqual(num_warns,0)
379        # state has changed
380        self.assertEqual(self.app['students']['X666666'].state,'admitted')
381        # state has not changed
382        self.assertEqual(self.app['students']['Y777777'].state,'courses validated')
383        shutil.rmtree(os.path.dirname(fin_file))
384
385    def test_import_update2(self):
386        num, num_warns, fin_file, fail_file = self.processor.doImport(
387            self.csv_file, STUDENT_HEADER_FIELDS)
388        shutil.rmtree(os.path.dirname(fin_file))
389        num, num_warns, fin_file, fail_file = self.processor.doImport(
390            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
391        self.assertEqual(num_warns,0)
392        # The phone import value of Pieri was None.
393        # Confirm that phone has not been cleared.
394        container = self.app['students']
395        for key in container.keys():
396            if container[key].firstname == 'Aaren':
397                aaren = container[key]
398                break
399        self.assertEqual(aaren.phone, '--1234')
400        # The phone import value of Claus was a deletion marker.
401        # Confirm that phone has been cleared.
402        for key in container.keys():
403            if container[key].firstname == 'Claus':
404                claus = container[key]
405                break
406        assert claus.phone is None
407        shutil.rmtree(os.path.dirname(fin_file))
408
409    def test_import_update3(self):
410        num, num_warns, fin_file, fail_file = self.processor.doImport(
411            self.csv_file, STUDENT_HEADER_FIELDS)
412        shutil.rmtree(os.path.dirname(fin_file))
413        num, num_warns, fin_file, fail_file = self.processor.doImport(
414            self.csv_file_update3, STUDENT_HEADER_FIELDS_UPDATE3, 'update')
415        content = open(fail_file).read()
416        self.assertEqual(
417            content,
418            'reg_number,student_id,transition,--ERRORS--\r\n'
419            '<IGNORE>,X666666,request_clearance,Transition not allowed.\r\n'
420            )
421        self.assertEqual(num_warns,1)
422        self.assertEqual(self.app['students']['Y777777'].state,'returning')
423
424    def test_import_update4(self):
425        num, num_warns, fin_file, fail_file = self.processor.doImport(
426            self.csv_file, STUDENT_HEADER_FIELDS)
427        shutil.rmtree(os.path.dirname(fin_file))
428        self.assertRaises(
429            FatalCSVError, self.processor.doImport, self.csv_file_update4,
430            STUDENT_HEADER_FIELDS_UPDATE4, 'update')
431
432    def test_import_remove(self):
433        num, num_warns, fin_file, fail_file = self.processor.doImport(
434            self.csv_file, STUDENT_HEADER_FIELDS)
435        shutil.rmtree(os.path.dirname(fin_file))
436        num, num_warns, fin_file, fail_file = self.processor.doImport(
437            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
438        self.assertEqual(num_warns,0)
439        shutil.rmtree(os.path.dirname(fin_file))
440
441    def test_import_migration_data(self):
442        num, num_warns, fin_file, fail_file = self.processor.doImport(
443            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
444        content = open(fail_file).read()
445        self.assertEqual(num_warns,2)
446        assert len(self.app['students'].keys()) == 5
447        self.assertEqual(
448            content,
449            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
450            '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n'
451            '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n'
452            )
453        students = self.app['students']
454        self.assertTrue('A123456' in students.keys())
455        self.assertEqual(students['A123456'].state, 'clearance started')
456        self.assertEqual(students['A123456'].date_of_birth,
457                         datetime.date(1990, 1, 2))
458        self.assertFalse(students['A123456'].clearance_locked)
459        self.assertEqual(students['B123456'].state, 'cleared')
460        self.assertEqual(students['B123456'].date_of_birth,
461                         datetime.date(1990, 1, 3))
462        self.assertTrue(students['B123456'].clearance_locked)
463        history = ' '.join(students['A123456'].history.messages)
464        self.assertTrue(
465            "State 'clearance started' set by system" in history)
466        # state was empty and student is thus in state created
467        self.assertEqual(students['F123456'].state,'created')
468        # passwords were set correctly
469        self.assertEqual(
470            IUserAccount(students['A123456']).checkPassword('mypw1'), True)
471        self.assertEqual(
472            IUserAccount(students['C123456']).checkPassword('mypw1'), True)
473        shutil.rmtree(os.path.dirname(fin_file))
474
475    def test_import_duplicate_data(self):
476        num, num_warns, fin_file, fail_file = self.processor.doImport(
477            self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES)
478        content = open(fail_file).read()
479        self.assertEqual(num_warns,4)
480        self.assertEqual(
481            content,
482            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
483            '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n'
484            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n'
485            '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n'
486            '3,Uli,A123456,m,aa@aa.ng,1234,,1990-01-07,Schulz,,100002,This object already exists. Skipping.\r\n'
487            )
488        shutil.rmtree(os.path.dirname(fin_file))
489
490class StudentStudyCourseProcessorTest(StudentImportExportSetup):
491
492    def setUp(self):
493        super(StudentStudyCourseProcessorTest, self).setUp()
494
495        # Import students with subobjects
496        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
497        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
498        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
499            student_file, STUDENT_HEADER_FIELDS)
500        shutil.rmtree(os.path.dirname(fin_file))
501
502        self.processor = StudentStudyCourseProcessor()
503        self.csv_file = os.path.join(
504            self.workdir, 'sample_studycourse_data.csv')
505        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
506        return
507
508    def test_interface(self):
509        # Make sure we fulfill the interface contracts.
510        assert verifyObject(IBatchProcessor, self.processor) is True
511        assert verifyClass(
512            IBatchProcessor, StudentStudyCourseProcessor) is True
513
514    def test_entryExists(self):
515        assert self.processor.entryExists(
516            dict(reg_number='REG_NONE'), self.app) is False
517        assert self.processor.entryExists(
518            dict(reg_number='1'), self.app) is True
519
520    def test_getEntry(self):
521        student = self.processor.getEntry(
522            dict(reg_number='1'), self.app).__parent__
523        self.assertEqual(student.reg_number,'1')
524
525    def test_checkConversion(self):
526        errs, inv_errs, conv_dict = self.processor.checkConversion(
527            dict(reg_number='1', certificate='CERT1', current_level='200'))
528        self.assertEqual(len(errs),0)
529        errs, inv_errs, conv_dict = self.processor.checkConversion(
530            dict(reg_number='1', certificate='CERT999'))
531        self.assertEqual(len(errs),1)
532        self.assertTrue(('certificate', u'Invalid value') in errs)
533        errs, inv_errs, conv_dict = self.processor.checkConversion(
534            dict(reg_number='1', certificate='CERT1', current_level='100'))
535        self.assertEqual(len(errs),1)
536        self.assertTrue(('current_level','not in range') in errs)
537        # If we import only current_level, no conversion checking is done.
538        errs, inv_errs, conv_dict = self.processor.checkConversion(
539            dict(reg_number='1', current_level='100'))
540        self.assertEqual(len(errs),0)
541
542    def test_import(self):
543        num, num_warns, fin_file, fail_file = self.processor.doImport(
544            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
545        self.assertEqual(num_warns,1)
546        content = open(fail_file).read()
547        self.assertTrue('current_level: not in range' in content)
548        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
549        self.assertEqual(studycourse.certificate.code, u'CERT1')
550        shutil.rmtree(os.path.dirname(fin_file))
551
552class StudentVerdictProcessorTest(StudentImportExportSetup):
553
554
555    def setUp(self):
556        super(StudentVerdictProcessorTest, self).setUp()
557
558        # Import students with subobjects
559        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
560        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
561        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
562            student_file, STUDENT_HEADER_FIELDS)
563        shutil.rmtree(os.path.dirname(fin_file))
564
565        # Update study courses
566        studycourse_file = os.path.join(
567            self.workdir, 'sample_studycourse_data.csv')
568        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
569        processor = StudentStudyCourseProcessor()
570        num, num_warns, fin_file, fail_file = processor.doImport(
571            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
572        shutil.rmtree(os.path.dirname(fin_file))
573
574        self.processor = StudentVerdictProcessor()
575        self.csv_file = os.path.join(
576            self.workdir, 'sample_verdict_data.csv')
577        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
578        return
579
580    def test_import(self):
581        num, num_warns, fin_file, fail_file = self.processor.doImport(
582            self.csv_file, VERDICT_HEADER_FIELDS,'update')
583        self.assertEqual(num_warns,3)
584        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
585        student = self.processor.getParent(dict(matric_number='100000'), self.app)
586        self.assertEqual(studycourse.current_verdict, 'A')
587        self.assertEqual(student.state, 'returning')
588        self.assertEqual(studycourse.current_level, 200)
589        content = open(fail_file).read()
590        self.assertEqual(
591            content,
592            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
593            '2008,B,100001,100,Current level does not correspond.\r\n'
594            '2007,C,100002,200,Current session does not correspond.\r\n'
595            '2008,A,100003,200,Student in wrong state.\r\n'
596            )
597        shutil.rmtree(os.path.dirname(fin_file))
598
599
600class StudentStudyLevelProcessorTest(StudentImportExportSetup):
601
602    def setUp(self):
603        super(StudentStudyLevelProcessorTest, self).setUp()
604
605        # Import students with subobjects
606        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
607        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
608        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
609            student_file, STUDENT_HEADER_FIELDS)
610        shutil.rmtree(os.path.dirname(fin_file))
611
612        # Update study courses
613        studycourse_file = os.path.join(
614            self.workdir, 'sample_studycourse_data.csv')
615        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
616        processor = StudentStudyCourseProcessor()
617        num, num_warns, fin_file, fail_file = processor.doImport(
618            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
619        shutil.rmtree(os.path.dirname(fin_file))
620
621        self.processor = StudentStudyLevelProcessor()
622        self.csv_file = os.path.join(
623            self.workdir, 'sample_studylevel_data.csv')
624        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
625
626    def test_interface(self):
627        # Make sure we fulfill the interface contracts.
628        assert verifyObject(IBatchProcessor, self.processor) is True
629        assert verifyClass(
630            IBatchProcessor, StudentStudyLevelProcessor) is True
631
632    def test_checkConversion(self):
633        errs, inv_errs, conv_dict = self.processor.checkConversion(
634            dict(reg_number='1', level='220'))
635        self.assertEqual(len(errs),0)
636        errs, inv_errs, conv_dict = self.processor.checkConversion(
637            dict(reg_number='1', level='900'))
638        self.assertEqual(len(errs),1)
639        self.assertTrue(('level','no valid integer') in errs)
640        errs, inv_errs, conv_dict = self.processor.checkConversion(
641            dict(reg_number='1', level='xyz'))
642        self.assertEqual(len(errs),1)
643        self.assertTrue(('level','no integer') in errs)
644
645    def test_import(self):
646        num, num_warns, fin_file, fail_file = self.processor.doImport(
647            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
648        self.assertEqual(num_warns,2)
649        assert self.processor.entryExists(
650            dict(reg_number='1', level='100'), self.app) is True
651        studylevel = self.processor.getEntry(
652            dict(reg_number='1', level='100'), self.app)
653        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
654        self.assertEqual(studylevel.level_session, 2008)
655        self.assertEqual(studylevel.level_verdict, 'A')
656        self.assertEqual(studylevel.level, 100)
657        shutil.rmtree(os.path.dirname(fin_file))
658
659        logcontent = open(self.logfile).read()
660        # Logging message from updateEntry,
661        self.assertTrue(
662            'INFO - system - K1000000 - Study level '
663            'updated: level=100, level_verdict=C, level_session=2009'
664            in logcontent)
665
666    def test_import_update(self):
667        # We perform the same import twice,
668        # the second time in update mode. The number
669        # of warnings must be the same.
670        num, num_warns, fin_file, fail_file = self.processor.doImport(
671            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
672        num, num_warns, fin_file, fail_file = self.processor.doImport(
673            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
674        self.assertEqual(num_warns,2)
675        shutil.rmtree(os.path.dirname(fin_file))
676       
677
678class CourseTicketProcessorTest(StudentImportExportSetup):
679
680    def setUp(self):
681        super(CourseTicketProcessorTest, self).setUp()
682
683        # Import students with subobjects
684        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
685        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
686        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
687            student_file, STUDENT_HEADER_FIELDS)
688        shutil.rmtree(os.path.dirname(fin_file))
689
690        # Add course and certificate course
691        self.course = createObject('waeup.Course')
692        self.course.code = 'COURSE1'
693        self.course.semester = 1
694        self.course.credits = 10
695        self.course.passmark = 40
696        self.app['faculties']['fac1']['dep1'].courses.addCourse(
697            self.course)
698        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCertCourse(
699            self.course, level=100)
700
701        # Update study courses
702        studycourse_file = os.path.join(
703            self.workdir, 'sample_studycourse_data.csv')
704        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
705        processor = StudentStudyCourseProcessor()
706        num, num_warns, fin_file, fail_file = processor.doImport(
707            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
708        shutil.rmtree(os.path.dirname(fin_file))
709
710        # Import study levels
711        processor = StudentStudyLevelProcessor()
712        studylevel_file = os.path.join(
713            self.workdir, 'sample_studylevel_data.csv')
714        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
715        num, num_warns, fin_file, fail_file = processor.doImport(
716            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
717        shutil.rmtree(os.path.dirname(fin_file))
718
719        self.processor = CourseTicketProcessor()
720        self.csv_file = os.path.join(
721            self.workdir, 'sample_courseticket_data.csv')
722        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
723
724    def test_interface(self):
725        # Make sure we fulfill the interface contracts.
726        assert verifyObject(IBatchProcessor, self.processor) is True
727        assert verifyClass(
728            IBatchProcessor, CourseTicketProcessor) is True
729
730    def test_checkConversion(self):
731        errs, inv_errs, conv_dict = self.processor.checkConversion(
732            dict(reg_number='1', code='COURSE1', level='220'))
733        self.assertEqual(len(errs),0)
734        errs, inv_errs, conv_dict = self.processor.checkConversion(
735            dict(reg_number='1', code='COURSE2', level='220'))
736        self.assertEqual(len(errs),1)
737        self.assertTrue(('code','non-existent') in errs)
738
739    def test_import(self):
740
741        num, num_warns, fin_file, fail_file = self.processor.doImport(
742            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
743
744        self.assertEqual(num_warns,2)
745        assert self.processor.entryExists(
746            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
747        courseticket = self.processor.getEntry(
748            dict(reg_number='1', level='100', code='COURSE1'), self.app)
749        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
750        self.assertEqual(courseticket.score, 1)
751        self.assertEqual(courseticket.mandatory, True)
752        self.assertEqual(courseticket.fcode, 'NA')
753        self.assertEqual(courseticket.dcode, 'NA')
754        self.assertEqual(courseticket.code, 'COURSE1')
755        self.assertEqual(courseticket.title, 'Unnamed Course')
756        self.assertEqual(courseticket.credits, 10)
757        self.assertEqual(courseticket.passmark, 40)
758        self.assertEqual(courseticket.semester, 1)
759        shutil.rmtree(os.path.dirname(fin_file))
760
761        logcontent = open(self.logfile).read()
762        # Logging message from updateEntry,
763        self.assertTrue(
764            'INFO - system - K1000000 - Course ticket in 100 updated: code=COURSE1, '
765            'mandatory=False, score=3'
766            in logcontent)
767
768    def test_import_update(self):
769        # We perform the same import twice,
770        # the second time in update mode. The number
771        # of warnings must be the same.
772        num, num_warns, fin_file, fail_file = self.processor.doImport(
773            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
774        num, num_warns, fin_file, fail_file = self.processor.doImport(
775            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
776        self.assertEqual(num_warns,2)
777        shutil.rmtree(os.path.dirname(fin_file))
778
779    def test_import_remove(self):
780        # We perform the same import twice,
781        # the second time in remove mode. The number
782        # of warnings must be the same.
783        num, num_warns, fin_file, fail_file = self.processor.doImport(
784            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
785        assert self.processor.entryExists(
786            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
787        num, num_warns, fin_file, fail_file = self.processor.doImport(
788            self.csv_file, COURSETICKET_HEADER_FIELDS,'remove')
789        self.assertEqual(num_warns,2)
790        assert self.processor.entryExists(
791            dict(reg_number='1', level='100', code='COURSE1'), self.app) is False
792        shutil.rmtree(os.path.dirname(fin_file))
793        logcontent = open(self.logfile).read()
794        self.assertTrue(
795            'INFO - system - K1000000 - Course ticket in 100 removed: COURSE1'
796            in logcontent)
797
798class PaymentProcessorTest(StudentImportExportSetup):
799
800    def setUp(self):
801        super(PaymentProcessorTest, self).setUp()
802
803        # Add student with payment
804        student = Student()
805        student.firstname = u'Anna'
806        student.lastname = u'Tester'
807        student.reg_number = u'123'
808        student.matric_number = u'234'
809        self.app['students'].addStudent(student)
810        self.student = self.app['students'][student.student_id]
811        payment = createObject(u'waeup.StudentOnlinePayment')
812        payment.p_id = 'p120'
813        self.student['payments'][payment.p_id] = payment
814
815        # Import students with subobjects
816        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
817        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
818        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
819            student_file, STUDENT_HEADER_FIELDS)
820        shutil.rmtree(os.path.dirname(fin_file))
821
822        self.processor = StudentOnlinePaymentProcessor()
823        self.csv_file = os.path.join(
824            self.workdir, 'sample_payment_data.csv')
825        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
826        self.csv_file2 = os.path.join(
827            self.workdir, 'sample_create_payment_data.csv')
828        open(self.csv_file2, 'wb').write(PAYMENT_CREATE_SAMPLE_DATA)
829
830    def test_interface(self):
831        # Make sure we fulfill the interface contracts.
832        assert verifyObject(IBatchProcessor, self.processor) is True
833        assert verifyClass(
834            IBatchProcessor, StudentOnlinePaymentProcessor) is True
835
836    def test_getEntry(self):
837        assert self.processor.getEntry(
838            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
839        assert self.processor.getEntry(
840            dict(student_id=self.student.student_id, p_id='p120'),
841            self.app) is self.student['payments']['p120']
842        assert self.processor.getEntry(
843            dict(student_id=self.student.student_id, p_id='XXXXXX112'),
844            self.app) is self.student['payments']['p120']
845
846    def test_delEntry(self):
847        assert self.processor.getEntry(
848            dict(student_id=self.student.student_id, p_id='p120'),
849            self.app) is self.student['payments']['p120']
850        self.assertEqual(len(self.student['payments'].keys()),1)
851        self.processor.delEntry(
852            dict(student_id=self.student.student_id, p_id='p120'),
853            self.app)
854        assert self.processor.getEntry(
855            dict(student_id=self.student.student_id, p_id='p120'),
856            self.app) is None
857        self.assertEqual(len(self.student['payments'].keys()),0)
858
859    def test_addEntry(self):
860        self.assertEqual(len(self.student['payments'].keys()),1)
861        payment1 = createObject(u'waeup.StudentOnlinePayment')
862        payment1.p_id = 'p234'
863        self.processor.addEntry(
864            payment1, dict(student_id=self.student.student_id, p_id='p234'),
865            self.app)
866        self.assertEqual(len(self.student['payments'].keys()),2)
867        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
868        payment2 = createObject(u'waeup.StudentOnlinePayment')
869        payment1.p_id = 'nonsense'
870        # payment1.p_id will be replaced if p_id doesn't start with 'p'
871        self.processor.addEntry(
872            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
873            self.app)
874        self.assertEqual(len(self.student['payments'].keys()),3)
875        self.assertEqual(self.student['payments']['p560'].p_id, 'p560')
876
877    def test_checkConversion(self):
878        errs, inv_errs, conv_dict = self.processor.checkConversion(
879            dict(p_id='3816951266236341955'))
880        self.assertEqual(len(errs),0)
881        errs, inv_errs, conv_dict = self.processor.checkConversion(
882            dict(p_id='p1266236341955'))
883        self.assertEqual(len(errs),0)
884        errs, inv_errs, conv_dict = self.processor.checkConversion(
885            dict(p_id='nonsense'))
886        self.assertEqual(len(errs),1)
887        timestamp = ("%d" % int(time()*10000))[1:]
888        p_id = "p%s" % timestamp
889        errs, inv_errs, conv_dict = self.processor.checkConversion(
890            dict(p_id=p_id))
891        self.assertEqual(len(errs),0)
892
893    def test_import(self):
894        num, num_warns, fin_file, fail_file = self.processor.doImport(
895            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
896        self.assertEqual(num_warns,0)
897        payment = self.processor.getEntry(dict(reg_number='1',
898            p_id='p2907979737440'), self.app)
899        self.assertEqual(payment.p_id, 'p2907979737440')
900        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
901        self.assertEqual(cdate, "2010-11-26 18:59:33")
902        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
903        shutil.rmtree(os.path.dirname(fin_file))
904        logcontent = open(self.logfile).read()
905        # Logging message from updateEntry
906        self.assertTrue(
907            'INFO - system - K1000001 - Payment ticket updated: '
908            'p_id=p1266236341955, p_item=BTECHBDT, '
909            'creation_date=2010-02-15 13:19:01+00:00, r_code=00, '
910            'r_amount_approved=19500.0, p_category=schoolfee, '
911            'amount_auth=19500.0, p_state=paid'
912            in logcontent)
913
914    def test_import_update(self):
915        # We perform the same import twice,
916        # the second time in update mode. The number
917        # of warnings must be the same.
918        num, num_warns, fin_file, fail_file = self.processor.doImport(
919            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
920        num, num_warns, fin_file, fail_file = self.processor.doImport(
921            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
922        self.assertEqual(num_warns,0)
923        shutil.rmtree(os.path.dirname(fin_file))
924
925    def test_import_remove(self):
926        num, num_warns, fin_file, fail_file = self.processor.doImport(
927            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
928        num, num_warns, fin_file, fail_file = self.processor.doImport(
929            self.csv_file, PAYMENT_HEADER_FIELDS,'remove')
930        self.assertEqual(num_warns,0)
931        shutil.rmtree(os.path.dirname(fin_file))
932        logcontent = open(self.logfile).read()
933        self.assertTrue(
934            'INFO - system - K1000001 - Payment ticket removed: p1266236341955'
935            in logcontent)
936
937    def test_import_wo_pid(self):
938        num, num_warns, fin_file, fail_file = self.processor.doImport(
939            self.csv_file2, PAYMENT_CREATE_HEADER_FIELDS,'create')
940        self.assertEqual(num_warns,0)
941        shutil.rmtree(os.path.dirname(fin_file))
942        self.assertEqual(len(self.app['students']['X666666']['payments']), 50)
943
944
945def test_suite():
946    suite = unittest.TestSuite()
947    for testcase in [
948        StudentProcessorTest,StudentStudyCourseProcessorTest,
949        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
950        PaymentProcessorTest,StudentVerdictProcessorTest]:
951        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
952                testcase
953                )
954        )
955    return suite
956
957
Note: See TracBrowser for help on using the repository browser.