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

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

Rename reg_state to state to be in accordance with applicants.

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