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

Last change on this file since 8305 was 8289, checked in by Henrik Bettermann, 13 years ago

Add test for duplicate reg_numbers and matric_numbers.
  • Property svn:keywords set to Id
File size: 31.1 KB
Line 
1## $Id: test_batching.py 8289 2012-04-26 14:30:24Z 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.university.faculty import Faculty
32from waeup.kofa.university.department import Department
33from waeup.kofa.students.batching import (
34    StudentProcessor, StudentStudyCourseProcessor,
35    StudentStudyLevelProcessor, CourseTicketProcessor,
36    StudentOnlinePaymentProcessor, StudentVerdictProcessor)
37from waeup.kofa.students.student import Student
38from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
39from waeup.kofa.interfaces import IBatchProcessor
40
41STUDENT_SAMPLE_DATA = open(
42    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
43    'rb').read()
44
45STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
46    '\n')[0].split(',')
47
48STUDENT_SAMPLE_DATA_UPDATE = open(
49    os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'),
50    'rb').read()
51
52STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split(
53    '\n')[0].split(',')
54
55STUDENT_SAMPLE_DATA_UPDATE2 = open(
56    os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'),
57    'rb').read()
58
59STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split(
60    '\n')[0].split(',')
61
62STUDYCOURSE_SAMPLE_DATA = open(
63    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
64    'rb').read()
65
66STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
67    '\n')[0].split(',')
68
69VERDICT_SAMPLE_DATA = open(
70    os.path.join(os.path.dirname(__file__), 'sample_verdict_data.csv'),
71    'rb').read()
72
73VERDICT_HEADER_FIELDS = VERDICT_SAMPLE_DATA.split(
74    '\n')[0].split(',')
75
76STUDENT_SAMPLE_DATA_MIGRATION = open(
77    os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'),
78    'rb').read()
79
80STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split(
81    '\n')[0].split(',')
82
83STUDENT_SAMPLE_DATA_DUPLICATES = open(
84    os.path.join(os.path.dirname(__file__), 'sample_student_data_duplicates.csv'),
85    'rb').read()
86
87STUDENT_HEADER_FIELDS_DUPLICATES = STUDENT_SAMPLE_DATA_DUPLICATES.split(
88    '\n')[0].split(',')
89
90STUDYLEVEL_SAMPLE_DATA = open(
91    os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'),
92    'rb').read()
93
94STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split(
95    '\n')[0].split(',')
96
97COURSETICKET_SAMPLE_DATA = open(
98    os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'),
99    'rb').read()
100
101COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
102    '\n')[0].split(',')
103
104PAYMENT_SAMPLE_DATA = open(
105    os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'),
106    'rb').read()
107
108PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
109    '\n')[0].split(',')
110
111class StudentImportExportSetup(FunctionalTestCase):
112
113    layer = FunctionalLayer
114
115    def setUp(self):
116        super(StudentImportExportSetup, self).setUp()
117        self.dc_root = tempfile.mkdtemp()
118        self.workdir = tempfile.mkdtemp()
119        app = University()
120        app['datacenter'].setStoragePath(self.dc_root)
121        self.getRootFolder()['app'] = app
122        self.app = self.getRootFolder()['app']
123        setSite(app)
124
125        # Populate university
126        self.certificate = createObject('waeup.Certificate')
127        self.certificate.code = 'CERT1'
128        self.certificate.application_category = 'basic'
129        self.certificate.start_level = 200
130        self.certificate.end_level = 500
131        self.app['faculties']['fac1'] = Faculty()
132        self.app['faculties']['fac1']['dep1'] = Department()
133        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
134            self.certificate)
135        return
136
137    def tearDown(self):
138        super(StudentImportExportSetup, self).tearDown()
139        shutil.rmtree(self.workdir)
140        shutil.rmtree(self.dc_root)
141        clearSite()
142        return
143
144
145class StudentProcessorTest(FunctionalTestCase):
146
147    layer = FunctionalLayer
148
149    def setUp(self):
150        super(StudentProcessorTest, self).setUp()
151        # Setup a sample site for each test
152        app = University()
153        self.dc_root = tempfile.mkdtemp()
154        app['datacenter'].setStoragePath(self.dc_root)
155
156        # Prepopulate the ZODB...
157        self.getRootFolder()['app'] = app
158        # we add the site immediately after creation to the
159        # ZODB. Catalogs and other local utilities are not setup
160        # before that step.
161        self.app = self.getRootFolder()['app']
162        # Set site here. Some of the following setup code might need
163        # to access grok.getSite() and should get our new app then
164        setSite(app)
165
166        # Add student with subobjects
167        student = Student()
168        student.firstname = u'Anna'
169        student.lastname = u'Tester'
170        student.reg_number = u'123'
171        student.matric_number = u'234'
172        self.app['students'].addStudent(student)
173        self.student = self.app['students'][student.student_id]
174        self.processor = StudentProcessor()
175        self.workdir = tempfile.mkdtemp()
176        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
177        self.csv_file_update = os.path.join(
178            self.workdir, 'sample_student_data_update.csv')
179        self.csv_file_update2 = os.path.join(
180            self.workdir, 'sample_student_data_update2.csv')
181        self.csv_file_migration = os.path.join(
182            self.workdir, 'sample_student_data_migration.csv')
183        self.csv_file_duplicates = os.path.join(
184            self.workdir, 'sample_student_data_duplicates.csv')
185        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
186        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
187        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
188        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
189        open(self.csv_file_duplicates, 'wb').write(STUDENT_SAMPLE_DATA_DUPLICATES)
190
191    def tearDown(self):
192        super(StudentProcessorTest, self).tearDown()
193        shutil.rmtree(self.workdir)
194        shutil.rmtree(self.dc_root)
195        clearSite()
196        return
197
198    def test_interface(self):
199        # Make sure we fulfill the interface contracts.
200        assert verifyObject(IBatchProcessor, self.processor) is True
201        assert verifyClass(
202            IBatchProcessor, StudentProcessor) is True
203
204    def test_parentsExist(self):
205        self.assertFalse(self.processor.parentsExist(None, dict()))
206        self.assertTrue(self.processor.parentsExist(None, self.app))
207
208    def test_entryExists(self):
209        assert self.processor.entryExists(
210            dict(student_id='ID_NONE'), self.app) is False
211        assert self.processor.entryExists(
212            dict(reg_number='123'), self.app) is True
213
214    def test_getParent(self):
215        parent = self.processor.getParent(None, self.app)
216        assert parent is self.app['students']
217
218    def test_getEntry(self):
219        assert self.processor.getEntry(
220            dict(student_id='ID_NONE'), self.app) is None
221        assert self.processor.getEntry(
222            dict(student_id=self.student.student_id), self.app) is self.student
223
224    def test_addEntry(self):
225        new_student = Student()
226        self.processor.addEntry(
227            new_student, dict(), self.app)
228        assert len(self.app['students'].keys()) == 2
229
230    def test_checkConversion(self):
231        errs, inv_errs, conv_dict = self.processor.checkConversion(
232            dict(reg_number='1', state='admitted'))
233        self.assertEqual(len(errs),0)
234        # Empty state is allowed
235        errs, inv_errs, conv_dict = self.processor.checkConversion(
236            dict(reg_number='1', state=''))
237        self.assertEqual(len(errs),0)
238        #self.assertTrue(('state', 'no value provided') in errs)
239        errs, inv_errs, conv_dict = self.processor.checkConversion(
240            dict(reg_number='1', state='nonsense'))
241        self.assertEqual(len(errs),1)
242        self.assertTrue(('state', 'not allowed') in errs)
243
244    def test_delEntry(self):
245        assert self.student.student_id in self.app['students'].keys()
246        self.processor.delEntry(
247            dict(reg_number=self.student.reg_number), self.app)
248        assert self.student.student_id not in self.app['students'].keys()
249
250    def test_import(self):
251        num, num_warns, fin_file, fail_file = self.processor.doImport(
252            self.csv_file, STUDENT_HEADER_FIELDS)
253        self.assertEqual(num_warns,0)
254        assert len(self.app['students'].keys()) == 5
255        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
256        self.assertEqual(self.app['students']['X666666'].state,'courses validated')
257        shutil.rmtree(os.path.dirname(fin_file))
258
259    def test_import_update(self):
260        num, num_warns, fin_file, fail_file = self.processor.doImport(
261            self.csv_file, STUDENT_HEADER_FIELDS)
262        shutil.rmtree(os.path.dirname(fin_file))
263        num, num_warns, fin_file, fail_file = self.processor.doImport(
264            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
265        self.assertEqual(num_warns,0)
266        # state has changed
267        self.assertEqual(self.app['students']['X666666'].state,'admitted')
268        # state has not changed
269        self.assertEqual(self.app['students']['Y777777'].state,'courses validated')
270        shutil.rmtree(os.path.dirname(fin_file))
271
272    def test_import_update2(self):
273        num, num_warns, fin_file, fail_file = self.processor.doImport(
274            self.csv_file, STUDENT_HEADER_FIELDS)
275        shutil.rmtree(os.path.dirname(fin_file))
276        num, num_warns, fin_file, fail_file = self.processor.doImport(
277            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
278        self.assertEqual(num_warns,0)
279        # The phone import value of Pieri was None.
280        # Confirm that phone has not been cleared.
281        container = self.app['students']
282        for key in container.keys():
283            if container[key].firstname == 'Aaren':
284                aaren = container[key]
285                break
286        self.assertEqual(aaren.phone, '--1234')
287        # The phone import value of Claus was a deletion marker.
288        # Confirm that phone has been cleared.
289        for key in container.keys():
290            if container[key].firstname == 'Claus':
291                claus = container[key]
292                break
293        assert claus.phone is None
294        shutil.rmtree(os.path.dirname(fin_file))
295
296    def test_import_remove(self):
297        num, num_warns, fin_file, fail_file = self.processor.doImport(
298            self.csv_file, STUDENT_HEADER_FIELDS)
299        shutil.rmtree(os.path.dirname(fin_file))
300        num, num_warns, fin_file, fail_file = self.processor.doImport(
301            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
302        self.assertEqual(num_warns,0)
303        shutil.rmtree(os.path.dirname(fin_file))
304
305    def test_import_migration_data(self):
306        num, num_warns, fin_file, fail_file = self.processor.doImport(
307            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
308        content = open(fail_file).read()
309        self.assertEqual(num_warns,2)
310        assert len(self.app['students'].keys()) == 5
311        self.assertEqual(
312            content,
313            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
314            '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n'
315            '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n'
316            )
317        self.assertTrue('A123456' in self.app['students'].keys())
318        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
319        self.assertEqual(self.app['students']['A123456'].date_of_birth,
320            datetime.date(1990, 1, 2))
321        self.assertFalse(self.app['students']['A123456'].clearance_locked)
322        self.assertEqual(self.app['students']['B123456'].state,'cleared')
323        self.assertEqual(self.app['students']['B123456'].date_of_birth,
324            datetime.date(1990, 1, 3))
325        self.assertTrue(self.app['students']['B123456'].clearance_locked)
326        history = ' '.join(self.app['students']['A123456'].history.messages)
327        self.assertTrue(
328            "State 'clearance started' set by system" in history)
329        # state was empty and student is thus in state created
330        self.assertEqual(self.app['students']['F123456'].state,'created')
331        shutil.rmtree(os.path.dirname(fin_file))
332
333    def test_import_duplicate_data(self):
334        num, num_warns, fin_file, fail_file = self.processor.doImport(
335            self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES)
336        content = open(fail_file).read()
337        self.assertEqual(num_warns,3)
338        self.assertEqual(
339            content,
340            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
341            '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n'
342            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n'
343            '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n'
344            )
345        shutil.rmtree(os.path.dirname(fin_file))
346
347class StudentStudyCourseProcessorTest(StudentImportExportSetup):
348
349    def setUp(self):
350        super(StudentStudyCourseProcessorTest, self).setUp()
351
352        # Import students with subobjects
353        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
354        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
355        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
356            student_file, STUDENT_HEADER_FIELDS)
357        shutil.rmtree(os.path.dirname(fin_file))
358
359        self.processor = StudentStudyCourseProcessor()
360        self.csv_file = os.path.join(
361            self.workdir, 'sample_studycourse_data.csv')
362        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
363        return
364
365    def test_interface(self):
366        # Make sure we fulfill the interface contracts.
367        assert verifyObject(IBatchProcessor, self.processor) is True
368        assert verifyClass(
369            IBatchProcessor, StudentStudyCourseProcessor) is True
370
371    def test_entryExists(self):
372        assert self.processor.entryExists(
373            dict(reg_number='REG_NONE'), self.app) is False
374        assert self.processor.entryExists(
375            dict(reg_number='1'), self.app) is True
376
377    def test_getEntry(self):
378        student = self.processor.getEntry(
379            dict(reg_number='1'), self.app).__parent__
380        self.assertEqual(student.reg_number,'1')
381
382    def test_checkConversion(self):
383        errs, inv_errs, conv_dict = self.processor.checkConversion(
384            dict(reg_number='1', certificate='CERT1', current_level='200'))
385        self.assertEqual(len(errs),0)
386        errs, inv_errs, conv_dict = self.processor.checkConversion(
387            dict(reg_number='1', certificate='CERT999'))
388        self.assertEqual(len(errs),1)
389        self.assertTrue(('certificate', u'Invalid value') in errs)
390        errs, inv_errs, conv_dict = self.processor.checkConversion(
391            dict(reg_number='1', certificate='CERT1', current_level='100'))
392        self.assertEqual(len(errs),1)
393        self.assertTrue(('current_level','not in range') in errs)
394        # If we import only current_level, no conversion checking is done.
395        errs, inv_errs, conv_dict = self.processor.checkConversion(
396            dict(reg_number='1', current_level='100'))
397        self.assertEqual(len(errs),0)
398
399    def test_import(self):
400        num, num_warns, fin_file, fail_file = self.processor.doImport(
401            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
402        self.assertEqual(num_warns,1)
403        content = open(fail_file).read()
404        self.assertTrue('current_level: not in range' in content)
405        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
406        self.assertEqual(studycourse.certificate.code, u'CERT1')
407        shutil.rmtree(os.path.dirname(fin_file))
408
409class StudentVerdictProcessorTest(StudentImportExportSetup):
410
411
412    def setUp(self):
413        super(StudentVerdictProcessorTest, self).setUp()
414
415        # Import students with subobjects
416        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
417        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
418        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
419            student_file, STUDENT_HEADER_FIELDS)
420        shutil.rmtree(os.path.dirname(fin_file))
421
422        # Update study courses
423        studycourse_file = os.path.join(
424            self.workdir, 'sample_studycourse_data.csv')
425        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
426        processor = StudentStudyCourseProcessor()
427        num, num_warns, fin_file, fail_file = processor.doImport(
428            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
429        shutil.rmtree(os.path.dirname(fin_file))
430
431        self.processor = StudentVerdictProcessor()
432        self.csv_file = os.path.join(
433            self.workdir, 'sample_verdict_data.csv')
434        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
435        return
436
437    def test_import(self):
438        num, num_warns, fin_file, fail_file = self.processor.doImport(
439            self.csv_file, VERDICT_HEADER_FIELDS,'update')
440        self.assertEqual(num_warns,3)
441        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
442        student = self.processor.getParent(dict(matric_number='100000'), self.app)
443        self.assertEqual(studycourse.current_verdict, 'A')
444        self.assertEqual(student.state, 'returning')
445        self.assertEqual(studycourse.current_level, 200)
446        content = open(fail_file).read()
447        self.assertEqual(
448            content,
449            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
450            '2008,B,100001,100,Current level does not correspond.\r\n'
451            '2007,C,100002,200,Current session does not correspond.\r\n'
452            '2008,A,100003,200,Student in wrong state.\r\n'
453            )
454        shutil.rmtree(os.path.dirname(fin_file))
455
456
457class StudentStudyLevelProcessorTest(StudentImportExportSetup):
458
459    def setUp(self):
460        super(StudentStudyLevelProcessorTest, self).setUp()
461
462        # Import students with subobjects
463        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
464        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
465        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
466            student_file, STUDENT_HEADER_FIELDS)
467        shutil.rmtree(os.path.dirname(fin_file))
468
469        # Update study courses
470        studycourse_file = os.path.join(
471            self.workdir, 'sample_studycourse_data.csv')
472        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
473        processor = StudentStudyCourseProcessor()
474        num, num_warns, fin_file, fail_file = processor.doImport(
475            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
476        shutil.rmtree(os.path.dirname(fin_file))
477
478        self.processor = StudentStudyLevelProcessor()
479        self.csv_file = os.path.join(
480            self.workdir, 'sample_studylevel_data.csv')
481        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
482
483    def test_interface(self):
484        # Make sure we fulfill the interface contracts.
485        assert verifyObject(IBatchProcessor, self.processor) is True
486        assert verifyClass(
487            IBatchProcessor, StudentStudyLevelProcessor) is True
488
489    def test_checkConversion(self):
490        errs, inv_errs, conv_dict = self.processor.checkConversion(
491            dict(reg_number='1', level='220'))
492        self.assertEqual(len(errs),0)
493        errs, inv_errs, conv_dict = self.processor.checkConversion(
494            dict(reg_number='1', level='900'))
495        self.assertEqual(len(errs),1)
496        self.assertTrue(('level','no valid integer') in errs)
497        errs, inv_errs, conv_dict = self.processor.checkConversion(
498            dict(reg_number='1', level='xyz'))
499        self.assertEqual(len(errs),1)
500        self.assertTrue(('level','no integer') in errs)
501
502    def test_import(self):
503        num, num_warns, fin_file, fail_file = self.processor.doImport(
504            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
505        self.assertEqual(num_warns,2)
506        assert self.processor.entryExists(
507            dict(reg_number='1', level='100'), self.app) is True
508        studylevel = self.processor.getEntry(
509            dict(reg_number='1', level='100'), self.app)
510        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
511        self.assertEqual(studylevel.level_session, 2008)
512        self.assertEqual(studylevel.level_verdict, 'A')
513        self.assertEqual(studylevel.level, 100)
514        shutil.rmtree(os.path.dirname(fin_file))
515
516    def test_import_update(self):
517        # We perform the same import twice,
518        # the second time in update mode. The number
519        # of warnings must be the same.
520        num, num_warns, fin_file, fail_file = self.processor.doImport(
521            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
522        num, num_warns, fin_file, fail_file = self.processor.doImport(
523            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
524        self.assertEqual(num_warns,2)
525        shutil.rmtree(os.path.dirname(fin_file))
526       
527
528class CourseTicketProcessorTest(StudentImportExportSetup):
529
530    def setUp(self):
531        super(CourseTicketProcessorTest, self).setUp()
532
533        # Import students with subobjects
534        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
535        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
536        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
537            student_file, STUDENT_HEADER_FIELDS)
538        shutil.rmtree(os.path.dirname(fin_file))
539
540        # Add course and course referrer
541        self.course = createObject('waeup.Course')
542        self.course.code = 'COURSE1'
543        self.course.semester = 1
544        self.course.credits = 10
545        self.course.passmark = 40
546        self.app['faculties']['fac1']['dep1'].courses.addCourse(
547            self.course)
548        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
549            self.course, level=100)
550
551        # Update study courses
552        studycourse_file = os.path.join(
553            self.workdir, 'sample_studycourse_data.csv')
554        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
555        processor = StudentStudyCourseProcessor()
556        num, num_warns, fin_file, fail_file = processor.doImport(
557            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
558        shutil.rmtree(os.path.dirname(fin_file))
559
560        # Import study levels
561        processor = StudentStudyLevelProcessor()
562        studylevel_file = os.path.join(
563            self.workdir, 'sample_studylevel_data.csv')
564        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
565        num, num_warns, fin_file, fail_file = processor.doImport(
566            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
567        shutil.rmtree(os.path.dirname(fin_file))
568
569        self.processor = CourseTicketProcessor()
570        self.csv_file = os.path.join(
571            self.workdir, 'sample_courseticket_data.csv')
572        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
573
574    def test_interface(self):
575        # Make sure we fulfill the interface contracts.
576        assert verifyObject(IBatchProcessor, self.processor) is True
577        assert verifyClass(
578            IBatchProcessor, CourseTicketProcessor) is True
579
580    def test_checkConversion(self):
581        errs, inv_errs, conv_dict = self.processor.checkConversion(
582            dict(reg_number='1', code='COURSE1', level='220'))
583        self.assertEqual(len(errs),0)
584        errs, inv_errs, conv_dict = self.processor.checkConversion(
585            dict(reg_number='1', code='COURSE2', level='220'))
586        self.assertEqual(len(errs),1)
587        self.assertTrue(('code','non-existent') in errs)
588
589    def test_import(self):
590
591        num, num_warns, fin_file, fail_file = self.processor.doImport(
592            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
593
594        self.assertEqual(num_warns,2)
595        assert self.processor.entryExists(
596            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
597        courseticket = self.processor.getEntry(
598            dict(reg_number='1', level='100', code='COURSE1'), self.app)
599        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
600        self.assertEqual(courseticket.score, 1)
601        self.assertEqual(courseticket.mandatory, True)
602        self.assertEqual(courseticket.fcode, 'NA')
603        self.assertEqual(courseticket.dcode, 'NA')
604        self.assertEqual(courseticket.code, 'COURSE1')
605        self.assertEqual(courseticket.title, 'Unnamed Course')
606        self.assertEqual(courseticket.credits, 10)
607        self.assertEqual(courseticket.passmark, 40)
608        self.assertEqual(courseticket.semester, 1)
609        shutil.rmtree(os.path.dirname(fin_file))
610
611    def test_import_update(self):
612        # We perform the same import twice,
613        # the second time in update mode. The number
614        # of warnings must be the same.
615        num, num_warns, fin_file, fail_file = self.processor.doImport(
616            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
617        num, num_warns, fin_file, fail_file = self.processor.doImport(
618            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
619        self.assertEqual(num_warns,2)
620        shutil.rmtree(os.path.dirname(fin_file))
621
622class PaymentProcessorTest(StudentImportExportSetup):
623
624    def setUp(self):
625        super(PaymentProcessorTest, self).setUp()
626
627        # Add student with payment
628        student = Student()
629        student.firstname = u'Anna'
630        student.lastname = u'Tester'
631        student.reg_number = u'123'
632        student.matric_number = u'234'
633        self.app['students'].addStudent(student)
634        self.student = self.app['students'][student.student_id]
635        payment = createObject(u'waeup.StudentOnlinePayment')
636        payment.p_id = 'p123'
637        self.student['payments'][payment.p_id] = payment
638
639        # Import students with subobjects
640        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
641        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
642        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
643            student_file, STUDENT_HEADER_FIELDS)
644        shutil.rmtree(os.path.dirname(fin_file))
645
646        self.processor = StudentOnlinePaymentProcessor()
647        self.csv_file = os.path.join(
648            self.workdir, 'sample_payment_data.csv')
649        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
650
651    def test_interface(self):
652        # Make sure we fulfill the interface contracts.
653        assert verifyObject(IBatchProcessor, self.processor) is True
654        assert verifyClass(
655            IBatchProcessor, StudentOnlinePaymentProcessor) is True
656
657    def test_getEntry(self):
658        assert self.processor.getEntry(
659            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
660        assert self.processor.getEntry(
661            dict(student_id=self.student.student_id, p_id='p123'),
662            self.app) is self.student['payments']['p123']
663        assert self.processor.getEntry(
664            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
665            self.app) is self.student['payments']['p123']
666
667    def test_addEntry(self):
668        self.assertEqual(len(self.student['payments'].keys()),1)
669        payment1 = createObject(u'waeup.StudentOnlinePayment')
670        payment1.p_id = 'p234'
671        self.processor.addEntry(
672            payment1, dict(student_id=self.student.student_id, p_id='p234'),
673            self.app)
674        self.assertEqual(len(self.student['payments'].keys()),2)
675        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
676        payment2 = createObject(u'waeup.StudentOnlinePayment')
677        payment1.p_id = 'nonsense'
678        # payment1.p_id will be replaced if p_id doesn't start with 'p'
679        self.processor.addEntry(
680            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
681            self.app)
682        self.assertEqual(len(self.student['payments'].keys()),3)
683        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
684
685    def test_checkConversion(self):
686        errs, inv_errs, conv_dict = self.processor.checkConversion(
687            dict(reg_number='1', p_id='3816951266236341955'))
688        self.assertEqual(len(errs),0)
689        errs, inv_errs, conv_dict = self.processor.checkConversion(
690            dict(reg_number='1', p_id='p1266236341955'))
691        self.assertEqual(len(errs),0)
692        errs, inv_errs, conv_dict = self.processor.checkConversion(
693            dict(reg_number='1', p_id='nonsense'))
694        self.assertEqual(len(errs),1)
695        timestamp = "%d" % int(time()*1000)
696        p_id = "p%s" % timestamp
697        errs, inv_errs, conv_dict = self.processor.checkConversion(
698            dict(reg_number='1', p_id=p_id))
699        self.assertEqual(len(errs),0)
700
701    def test_import(self):
702        num, num_warns, fin_file, fail_file = self.processor.doImport(
703            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
704        self.assertEqual(num_warns,0)
705        payment = self.processor.getEntry(dict(reg_number='1',
706            p_id='p1290797973744'), self.app)
707        self.assertEqual(payment.p_id, 'p1290797973744')
708        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
709        self.assertEqual(cdate, "2010-11-26 18:59:33")
710        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
711        shutil.rmtree(os.path.dirname(fin_file))
712
713    def test_import_update(self):
714        # We perform the same import twice,
715        # the second time in update mode. The number
716        # of warnings must be the same.
717        num, num_warns, fin_file, fail_file = self.processor.doImport(
718            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
719        num, num_warns, fin_file, fail_file = self.processor.doImport(
720            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
721        self.assertEqual(num_warns,0)
722        shutil.rmtree(os.path.dirname(fin_file))
723
724def test_suite():
725    suite = unittest.TestSuite()
726    for testcase in [
727        StudentProcessorTest,StudentStudyCourseProcessorTest,
728        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
729        PaymentProcessorTest,StudentVerdictProcessorTest]:
730        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
731                testcase
732                )
733        )
734    return suite
735
736
Note: See TracBrowser for help on using the repository browser.