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

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

Add tests for update mode.
  • Property svn:keywords set to Id
File size: 29.5 KB
Line 
1## $Id: test_batching.py 8231 2012-04-20 08:31:38Z 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', reg_state='admitted'))
223        self.assertEqual(len(errs),0)
224        errs, inv_errs, conv_dict = self.processor.checkConversion(
225            dict(reg_number='1', reg_state=''))
226        self.assertEqual(len(errs),1)
227        self.assertTrue(('reg_state', 'no value provided') in errs)
228        errs, inv_errs, conv_dict = self.processor.checkConversion(
229            dict(reg_number='1', reg_state='nonsense'))
230        self.assertEqual(len(errs),1)
231        self.assertTrue(('reg_state', 'not allowed') in errs)
232
233    def test_delEntry(self):
234        assert self.student.student_id in self.app['students'].keys()
235        self.processor.delEntry(
236            dict(reg_number=self.student.reg_number), self.app)
237        assert self.student.student_id not in self.app['students'].keys()
238
239    def test_import(self):
240        num, num_warns, fin_file, fail_file = self.processor.doImport(
241            self.csv_file, STUDENT_HEADER_FIELDS)
242        self.assertEqual(num_warns,0)
243        assert len(self.app['students'].keys()) == 5
244        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
245        shutil.rmtree(os.path.dirname(fin_file))
246
247    def test_import_update(self):
248        num, num_warns, fin_file, fail_file = self.processor.doImport(
249            self.csv_file, STUDENT_HEADER_FIELDS)
250        shutil.rmtree(os.path.dirname(fin_file))
251        num, num_warns, fin_file, fail_file = self.processor.doImport(
252            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
253        self.assertEqual(num_warns,0)
254        shutil.rmtree(os.path.dirname(fin_file))
255
256    def test_import_update2(self):
257        num, num_warns, fin_file, fail_file = self.processor.doImport(
258            self.csv_file, STUDENT_HEADER_FIELDS)
259        shutil.rmtree(os.path.dirname(fin_file))
260        num, num_warns, fin_file, fail_file = self.processor.doImport(
261            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
262        self.assertEqual(num_warns,0)
263        # The phone import value of Pieri was None.
264        # Confirm that phone has not been cleared.
265        container = self.app['students']
266        for key in container.keys():
267            if container[key].firstname == 'Aaren':
268                aaren = container[key]
269                break
270        self.assertEqual(aaren.phone, '--1234')
271        # The phone import value of Claus was a deletion marker.
272        # Confirm that phone has been cleared.
273        for key in container.keys():
274            if container[key].firstname == 'Claus':
275                claus = container[key]
276                break
277        assert claus.phone is None
278        shutil.rmtree(os.path.dirname(fin_file))
279
280    def test_import_remove(self):
281        num, num_warns, fin_file, fail_file = self.processor.doImport(
282            self.csv_file, STUDENT_HEADER_FIELDS)
283        shutil.rmtree(os.path.dirname(fin_file))
284        num, num_warns, fin_file, fail_file = self.processor.doImport(
285            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
286        self.assertEqual(num_warns,0)
287        shutil.rmtree(os.path.dirname(fin_file))
288
289    def test_import_migration_data(self):
290        num, num_warns, fin_file, fail_file = self.processor.doImport(
291            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
292        self.assertEqual(num_warns,2)
293        assert len(self.app['students'].keys()) == 4
294        content = open(fail_file).read()
295        self.assertEqual(
296            content,
297            'reg_number,firstname,student_id,sex,email,phone,date_of_birth,reg_state,lastname,password,matric_number,--ERRORS--\r\n'
298            '4,John,D123456,m,aa@aa.ng,1234,1990-01-05,nonsense,Wolter,mypw1,100003,reg_state: not allowed\r\n'
299            '5,John,E123456,x,aa@aa.ng,1234,1990-01-06,,Kennedy,,100004,sex: Invalid value; reg_state: not allowed\r\n'
300            )
301        self.assertTrue('A123456' in self.app['students'].keys())
302        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
303        self.assertEqual(self.app['students']['A123456'].date_of_birth,
304            datetime.date(1990, 1, 2))
305        self.assertFalse(self.app['students']['A123456'].clearance_locked)
306        self.assertEqual(self.app['students']['B123456'].state,'cleared')
307        self.assertEqual(self.app['students']['B123456'].date_of_birth,
308            datetime.date(1990, 1, 3))
309        self.assertTrue(self.app['students']['B123456'].clearance_locked)
310        history = ' '.join(self.app['students']['A123456'].history.messages)
311        self.assertTrue(
312            "State 'clearance started' set by system" in history)
313        shutil.rmtree(os.path.dirname(fin_file))
314
315
316class StudentStudyCourseProcessorTest(StudentImportExportSetup):
317
318    def setUp(self):
319        super(StudentStudyCourseProcessorTest, self).setUp()
320
321        # Import students with subobjects
322        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
323        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
324        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
325            student_file, STUDENT_HEADER_FIELDS)
326        shutil.rmtree(os.path.dirname(fin_file))
327
328        self.processor = StudentStudyCourseProcessor()
329        self.csv_file = os.path.join(
330            self.workdir, 'sample_studycourse_data.csv')
331        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
332        return
333
334    def test_interface(self):
335        # Make sure we fulfill the interface contracts.
336        assert verifyObject(IBatchProcessor, self.processor) is True
337        assert verifyClass(
338            IBatchProcessor, StudentStudyCourseProcessor) is True
339
340    def test_entryExists(self):
341        assert self.processor.entryExists(
342            dict(reg_number='REG_NONE'), self.app) is False
343        assert self.processor.entryExists(
344            dict(reg_number='1'), self.app) is True
345
346    def test_getEntry(self):
347        student = self.processor.getEntry(
348            dict(reg_number='1'), self.app).__parent__
349        self.assertEqual(student.reg_number,'1')
350
351    def test_checkConversion(self):
352        errs, inv_errs, conv_dict = self.processor.checkConversion(
353            dict(reg_number='1', certificate='CERT1', current_level='200'))
354        self.assertEqual(len(errs),0)
355        errs, inv_errs, conv_dict = self.processor.checkConversion(
356            dict(reg_number='1', certificate='CERT999'))
357        self.assertEqual(len(errs),1)
358        self.assertTrue(('certificate', u'Invalid value') in errs)
359        errs, inv_errs, conv_dict = self.processor.checkConversion(
360            dict(reg_number='1', certificate='CERT1', current_level='100'))
361        self.assertEqual(len(errs),1)
362        self.assertTrue(('current_level','not in range') in errs)
363        # If we import only current_level, no conversion checking is done.
364        errs, inv_errs, conv_dict = self.processor.checkConversion(
365            dict(reg_number='1', current_level='100'))
366        self.assertEqual(len(errs),0)
367
368    def test_import(self):
369        num, num_warns, fin_file, fail_file = self.processor.doImport(
370            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
371        self.assertEqual(num_warns,1)
372        content = open(fail_file).read()
373        self.assertTrue('current_level: not in range' in content)
374        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
375        self.assertEqual(studycourse.certificate.code, u'CERT1')
376        shutil.rmtree(os.path.dirname(fin_file))
377
378class StudentVerdictProcessorTest(StudentImportExportSetup):
379
380
381    def setUp(self):
382        super(StudentVerdictProcessorTest, self).setUp()
383
384        # Import students with subobjects
385        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
386        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
387        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
388            student_file, STUDENT_HEADER_FIELDS)
389        shutil.rmtree(os.path.dirname(fin_file))
390
391        # Update study courses
392        studycourse_file = os.path.join(
393            self.workdir, 'sample_studycourse_data.csv')
394        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
395        processor = StudentStudyCourseProcessor()
396        num, num_warns, fin_file, fail_file = processor.doImport(
397            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
398        shutil.rmtree(os.path.dirname(fin_file))
399
400        self.processor = StudentVerdictProcessor()
401        self.csv_file = os.path.join(
402            self.workdir, 'sample_verdict_data.csv')
403        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
404        return
405
406    def test_import(self):
407        num, num_warns, fin_file, fail_file = self.processor.doImport(
408            self.csv_file, VERDICT_HEADER_FIELDS,'update')
409        self.assertEqual(num_warns,3)
410        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
411        student = self.processor.getParent(dict(matric_number='100000'), self.app)
412        self.assertEqual(studycourse.current_verdict, 'A')
413        self.assertEqual(student.state, 'returning')
414        self.assertEqual(studycourse.current_level, 200)
415        content = open(fail_file).read()
416        self.assertEqual(
417            content,
418            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
419            '2008,B,100001,100,Current level does not correspond.\r\n'
420            '2007,C,100002,200,Current session does not correspond.\r\n'
421            '2008,A,100003,200,Student in wrong state.\r\n'
422            )
423        shutil.rmtree(os.path.dirname(fin_file))
424
425
426class StudentStudyLevelProcessorTest(StudentImportExportSetup):
427
428    def setUp(self):
429        super(StudentStudyLevelProcessorTest, self).setUp()
430
431        # Import students with subobjects
432        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
433        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
434        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
435            student_file, STUDENT_HEADER_FIELDS)
436        shutil.rmtree(os.path.dirname(fin_file))
437
438        # Update study courses
439        studycourse_file = os.path.join(
440            self.workdir, 'sample_studycourse_data.csv')
441        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
442        processor = StudentStudyCourseProcessor()
443        num, num_warns, fin_file, fail_file = processor.doImport(
444            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
445        shutil.rmtree(os.path.dirname(fin_file))
446
447        self.processor = StudentStudyLevelProcessor()
448        self.csv_file = os.path.join(
449            self.workdir, 'sample_studylevel_data.csv')
450        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
451
452    def test_interface(self):
453        # Make sure we fulfill the interface contracts.
454        assert verifyObject(IBatchProcessor, self.processor) is True
455        assert verifyClass(
456            IBatchProcessor, StudentStudyLevelProcessor) is True
457
458    def test_checkConversion(self):
459        errs, inv_errs, conv_dict = self.processor.checkConversion(
460            dict(reg_number='1', level='220'))
461        self.assertEqual(len(errs),0)
462        errs, inv_errs, conv_dict = self.processor.checkConversion(
463            dict(reg_number='1', level='900'))
464        self.assertEqual(len(errs),1)
465        self.assertTrue(('level','no valid integer') in errs)
466        errs, inv_errs, conv_dict = self.processor.checkConversion(
467            dict(reg_number='1', level='xyz'))
468        self.assertEqual(len(errs),1)
469        self.assertTrue(('level','no integer') in errs)
470
471    def test_import(self):
472        num, num_warns, fin_file, fail_file = self.processor.doImport(
473            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
474        self.assertEqual(num_warns,2)
475        assert self.processor.entryExists(
476            dict(reg_number='1', level='100'), self.app) is True
477        studylevel = self.processor.getEntry(
478            dict(reg_number='1', level='100'), self.app)
479        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
480        self.assertEqual(studylevel.level_session, 2008)
481        self.assertEqual(studylevel.level_verdict, 'A')
482        self.assertEqual(studylevel.level, 100)
483        shutil.rmtree(os.path.dirname(fin_file))
484
485    def test_import_update(self):
486        # We perform the same import twice,
487        # the second time in update mode. The number
488        # of warnings must be the same.
489        num, num_warns, fin_file, fail_file = self.processor.doImport(
490            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
491        num, num_warns, fin_file, fail_file = self.processor.doImport(
492            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
493        self.assertEqual(num_warns,2)
494        shutil.rmtree(os.path.dirname(fin_file))
495       
496
497class CourseTicketProcessorTest(StudentImportExportSetup):
498
499    def setUp(self):
500        super(CourseTicketProcessorTest, self).setUp()
501
502        # Import students with subobjects
503        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
504        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
505        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
506            student_file, STUDENT_HEADER_FIELDS)
507        shutil.rmtree(os.path.dirname(fin_file))
508
509        # Add course and course referrer
510        self.course = createObject('waeup.Course')
511        self.course.code = 'COURSE1'
512        self.course.semester = 1
513        self.course.credits = 10
514        self.course.passmark = 40
515        self.app['faculties']['fac1']['dep1'].courses.addCourse(
516            self.course)
517        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
518            self.course, level=100)
519
520        # Update study courses
521        studycourse_file = os.path.join(
522            self.workdir, 'sample_studycourse_data.csv')
523        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
524        processor = StudentStudyCourseProcessor()
525        num, num_warns, fin_file, fail_file = processor.doImport(
526            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
527        shutil.rmtree(os.path.dirname(fin_file))
528
529        # Import study levels
530        processor = StudentStudyLevelProcessor()
531        studylevel_file = os.path.join(
532            self.workdir, 'sample_studylevel_data.csv')
533        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
534        num, num_warns, fin_file, fail_file = processor.doImport(
535            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
536        shutil.rmtree(os.path.dirname(fin_file))
537
538        self.processor = CourseTicketProcessor()
539        self.csv_file = os.path.join(
540            self.workdir, 'sample_courseticket_data.csv')
541        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
542
543    def test_interface(self):
544        # Make sure we fulfill the interface contracts.
545        assert verifyObject(IBatchProcessor, self.processor) is True
546        assert verifyClass(
547            IBatchProcessor, CourseTicketProcessor) is True
548
549    def test_checkConversion(self):
550        errs, inv_errs, conv_dict = self.processor.checkConversion(
551            dict(reg_number='1', code='COURSE1', level='220'))
552        self.assertEqual(len(errs),0)
553        errs, inv_errs, conv_dict = self.processor.checkConversion(
554            dict(reg_number='1', code='COURSE2', level='220'))
555        self.assertEqual(len(errs),1)
556        self.assertTrue(('code','non-existent') in errs)
557
558    def test_import(self):
559
560        num, num_warns, fin_file, fail_file = self.processor.doImport(
561            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
562
563        self.assertEqual(num_warns,2)
564        assert self.processor.entryExists(
565            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
566        courseticket = self.processor.getEntry(
567            dict(reg_number='1', level='100', code='COURSE1'), self.app)
568        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
569        self.assertEqual(courseticket.score, 1)
570        self.assertEqual(courseticket.mandatory, True)
571        self.assertEqual(courseticket.fcode, 'NA')
572        self.assertEqual(courseticket.dcode, 'NA')
573        self.assertEqual(courseticket.code, 'COURSE1')
574        self.assertEqual(courseticket.title, 'Unnamed Course')
575        self.assertEqual(courseticket.credits, 10)
576        self.assertEqual(courseticket.passmark, 40)
577        self.assertEqual(courseticket.semester, 1)
578        shutil.rmtree(os.path.dirname(fin_file))
579
580    def test_import_update(self):
581        # We perform the same import twice,
582        # the second time in update mode. The number
583        # of warnings must be the same.
584        num, num_warns, fin_file, fail_file = self.processor.doImport(
585            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
586        num, num_warns, fin_file, fail_file = self.processor.doImport(
587            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
588        self.assertEqual(num_warns,2)
589        shutil.rmtree(os.path.dirname(fin_file))
590
591class PaymentProcessorTest(StudentImportExportSetup):
592
593    def setUp(self):
594        super(PaymentProcessorTest, self).setUp()
595
596        # Add student with payment
597        student = Student()
598        student.firstname = u'Anna'
599        student.lastname = u'Tester'
600        student.reg_number = u'123'
601        student.matric_number = u'234'
602        self.app['students'].addStudent(student)
603        self.student = self.app['students'][student.student_id]
604        payment = createObject(u'waeup.StudentOnlinePayment')
605        payment.p_id = 'p123'
606        self.student['payments'][payment.p_id] = payment
607
608        # Import students with subobjects
609        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
610        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
611        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
612            student_file, STUDENT_HEADER_FIELDS)
613        shutil.rmtree(os.path.dirname(fin_file))
614
615        self.processor = StudentOnlinePaymentProcessor()
616        self.csv_file = os.path.join(
617            self.workdir, 'sample_payment_data.csv')
618        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
619
620    def test_interface(self):
621        # Make sure we fulfill the interface contracts.
622        assert verifyObject(IBatchProcessor, self.processor) is True
623        assert verifyClass(
624            IBatchProcessor, StudentOnlinePaymentProcessor) is True
625
626    def test_getEntry(self):
627        assert self.processor.getEntry(
628            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
629        assert self.processor.getEntry(
630            dict(student_id=self.student.student_id, p_id='p123'),
631            self.app) is self.student['payments']['p123']
632        assert self.processor.getEntry(
633            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
634            self.app) is self.student['payments']['p123']
635
636    def test_addEntry(self):
637        self.assertEqual(len(self.student['payments'].keys()),1)
638        payment1 = createObject(u'waeup.StudentOnlinePayment')
639        payment1.p_id = 'p234'
640        self.processor.addEntry(
641            payment1, dict(student_id=self.student.student_id, p_id='p234'),
642            self.app)
643        self.assertEqual(len(self.student['payments'].keys()),2)
644        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
645        payment2 = createObject(u'waeup.StudentOnlinePayment')
646        payment1.p_id = 'nonsense'
647        # payment1.p_id will be replaced if p_id doesn't start with 'p'
648        self.processor.addEntry(
649            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
650            self.app)
651        self.assertEqual(len(self.student['payments'].keys()),3)
652        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
653
654    def test_checkConversion(self):
655        errs, inv_errs, conv_dict = self.processor.checkConversion(
656            dict(reg_number='1', p_id='3816951266236341955'))
657        self.assertEqual(len(errs),0)
658        errs, inv_errs, conv_dict = self.processor.checkConversion(
659            dict(reg_number='1', p_id='p1266236341955'))
660        self.assertEqual(len(errs),0)
661        errs, inv_errs, conv_dict = self.processor.checkConversion(
662            dict(reg_number='1', p_id='nonsense'))
663        self.assertEqual(len(errs),1)
664        timestamp = "%d" % int(time()*1000)
665        p_id = "p%s" % timestamp
666        errs, inv_errs, conv_dict = self.processor.checkConversion(
667            dict(reg_number='1', p_id=p_id))
668        self.assertEqual(len(errs),0)
669
670    def test_import(self):
671        num, num_warns, fin_file, fail_file = self.processor.doImport(
672            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
673        self.assertEqual(num_warns,0)
674        payment = self.processor.getEntry(dict(reg_number='1',
675            p_id='p1290797973744'), self.app)
676        self.assertEqual(payment.p_id, 'p1290797973744')
677        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
678        self.assertEqual(cdate, "2010-11-26 18:59:33")
679        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
680        shutil.rmtree(os.path.dirname(fin_file))
681
682    def test_import_update(self):
683        # We perform the same import twice,
684        # the second time in update mode. The number
685        # of warnings must be the same.
686        num, num_warns, fin_file, fail_file = self.processor.doImport(
687            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
688        num, num_warns, fin_file, fail_file = self.processor.doImport(
689            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
690        self.assertEqual(num_warns,0)
691        shutil.rmtree(os.path.dirname(fin_file))
692
693def test_suite():
694    suite = unittest.TestSuite()
695    for testcase in [
696        StudentProcessorTest,StudentStudyCourseProcessorTest,
697        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
698        PaymentProcessorTest,StudentVerdictProcessorTest]:
699        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
700                testcase
701                )
702        )
703    return suite
704
705
Note: See TracBrowser for help on using the repository browser.