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

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

Test if error messages in pending data file are correct.
  • Property svn:keywords set to Id
File size: 26.7 KB
Line 
1## $Id: test_batching.py 7953 2012-03-22 07:59:14Z 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        shutil.rmtree(os.path.dirname(fin_file))
264
265    def test_import_remove(self):
266        num, num_warns, fin_file, fail_file = self.processor.doImport(
267            self.csv_file, STUDENT_HEADER_FIELDS)
268        shutil.rmtree(os.path.dirname(fin_file))
269        num, num_warns, fin_file, fail_file = self.processor.doImport(
270            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
271        self.assertEqual(num_warns,0)
272        shutil.rmtree(os.path.dirname(fin_file))
273
274    def test_import_migration_data(self):
275        num, num_warns, fin_file, fail_file = self.processor.doImport(
276            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
277        self.assertEqual(num_warns,2)
278        assert len(self.app['students'].keys()) == 4
279        self.assertTrue('A123456' in self.app['students'].keys())
280        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
281        self.assertEqual(self.app['students']['A123456'].date_of_birth,
282            datetime.date(1990, 1, 2))
283        self.assertFalse(self.app['students']['A123456'].clearance_locked)
284        self.assertEqual(self.app['students']['B123456'].state,'cleared')
285        self.assertEqual(self.app['students']['B123456'].date_of_birth,
286            datetime.date(1990, 1, 3))
287        self.assertTrue(self.app['students']['B123456'].clearance_locked)
288        history = ' '.join(self.app['students']['A123456'].history.messages)
289        self.assertTrue(
290            "State 'clearance started' set by system" in history)
291        shutil.rmtree(os.path.dirname(fin_file))
292
293
294class StudentStudyCourseProcessorTest(StudentImportExportSetup):
295
296    def setUp(self):
297        super(StudentStudyCourseProcessorTest, self).setUp()
298
299        # Import students with subobjects
300        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
301        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
302        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
303            student_file, STUDENT_HEADER_FIELDS)
304        shutil.rmtree(os.path.dirname(fin_file))
305
306        self.processor = StudentStudyCourseProcessor()
307        self.csv_file = os.path.join(
308            self.workdir, 'sample_studycourse_data.csv')
309        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
310        return
311
312    def test_interface(self):
313        # Make sure we fulfill the interface contracts.
314        assert verifyObject(IBatchProcessor, self.processor) is True
315        assert verifyClass(
316            IBatchProcessor, StudentStudyCourseProcessor) is True
317
318    def test_entryExists(self):
319        assert self.processor.entryExists(
320            dict(reg_number='REG_NONE'), self.app) is False
321        assert self.processor.entryExists(
322            dict(reg_number='1'), self.app) is True
323
324    def test_getEntry(self):
325        student = self.processor.getEntry(
326            dict(reg_number='1'), self.app).__parent__
327        self.assertEqual(student.reg_number,'1')
328
329    def test_checkConversion(self):
330        errs, inv_errs, conv_dict = self.processor.checkConversion(
331            dict(reg_number='1', certificate='CERT1', current_level='200'))
332        self.assertEqual(len(errs),0)
333        errs, inv_errs, conv_dict = self.processor.checkConversion(
334            dict(reg_number='1', certificate='CERT999'))
335        self.assertEqual(len(errs),1)
336        self.assertTrue(('certificate', u'Invalid value') in errs)
337        errs, inv_errs, conv_dict = self.processor.checkConversion(
338            dict(reg_number='1', certificate='CERT1', current_level='100'))
339        self.assertEqual(len(errs),1)
340        self.assertTrue(('current_level','not in range') in errs)
341        # If we import only current_level, no conversion checking is done.
342        errs, inv_errs, conv_dict = self.processor.checkConversion(
343            dict(reg_number='1', current_level='100'))
344        self.assertEqual(len(errs),0)
345
346    def test_import(self):
347        num, num_warns, fin_file, fail_file = self.processor.doImport(
348            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
349        self.assertEqual(num_warns,1)
350        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
351        self.assertEqual(studycourse.certificate.code, u'CERT1')
352        shutil.rmtree(os.path.dirname(fin_file))
353
354class StudentVerdictProcessorTest(StudentImportExportSetup):
355
356
357    def setUp(self):
358        super(StudentVerdictProcessorTest, self).setUp()
359
360        # Import students with subobjects
361        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
362        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
363        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
364            student_file, STUDENT_HEADER_FIELDS)
365        shutil.rmtree(os.path.dirname(fin_file))
366
367        # Update study courses
368        studycourse_file = os.path.join(
369            self.workdir, 'sample_studycourse_data.csv')
370        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
371        processor = StudentStudyCourseProcessor()
372        num, num_warns, fin_file, fail_file = processor.doImport(
373            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
374        shutil.rmtree(os.path.dirname(fin_file))
375
376        self.processor = StudentVerdictProcessor()
377        self.csv_file = os.path.join(
378            self.workdir, 'sample_verdict_data.csv')
379        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
380        return
381
382    def test_import(self):
383        num, num_warns, fin_file, fail_file = self.processor.doImport(
384            self.csv_file, VERDICT_HEADER_FIELDS,'update')
385        self.assertEqual(num_warns,3)
386        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
387        student = self.processor.getParent(dict(matric_number='100000'), self.app)
388        self.assertEqual(studycourse.current_verdict, 'A')
389        self.assertEqual(student.state, 'returning')
390        pending_file = fail_file
391        content = open(fail_file).read()
392        self.assertEqual(
393            content,
394            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
395            '2008,B,100001,100,Current level does not correspond.\r\n'
396            '2007,C,100002,200,Current session does not correspond.\r\n'
397            '2008,A,100003,200,Student in wrong state.\r\n'
398            )
399        shutil.rmtree(os.path.dirname(fin_file))
400
401
402class StudentStudyLevelProcessorTest(StudentImportExportSetup):
403
404    def setUp(self):
405        super(StudentStudyLevelProcessorTest, self).setUp()
406
407        # Import students with subobjects
408        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
409        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
410        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
411            student_file, STUDENT_HEADER_FIELDS)
412        shutil.rmtree(os.path.dirname(fin_file))
413
414        # Update study courses
415        studycourse_file = os.path.join(
416            self.workdir, 'sample_studycourse_data.csv')
417        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
418        processor = StudentStudyCourseProcessor()
419        num, num_warns, fin_file, fail_file = processor.doImport(
420            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
421        shutil.rmtree(os.path.dirname(fin_file))
422
423        self.processor = StudentStudyLevelProcessor()
424        self.csv_file = os.path.join(
425            self.workdir, 'sample_studylevel_data.csv')
426        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
427
428    def test_interface(self):
429        # Make sure we fulfill the interface contracts.
430        assert verifyObject(IBatchProcessor, self.processor) is True
431        assert verifyClass(
432            IBatchProcessor, StudentStudyLevelProcessor) is True
433
434    def test_checkConversion(self):
435        errs, inv_errs, conv_dict = self.processor.checkConversion(
436            dict(reg_number='1', level='220'))
437        self.assertEqual(len(errs),0)
438        errs, inv_errs, conv_dict = self.processor.checkConversion(
439            dict(reg_number='1', level='900'))
440        self.assertEqual(len(errs),1)
441        self.assertTrue(('level','no valid integer') in errs)
442        errs, inv_errs, conv_dict = self.processor.checkConversion(
443            dict(reg_number='1', level='xyz'))
444        self.assertEqual(len(errs),1)
445        self.assertTrue(('level','no integer') in errs)
446
447    def test_import(self):
448        num, num_warns, fin_file, fail_file = self.processor.doImport(
449            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
450        self.assertEqual(num_warns,2)
451        assert self.processor.entryExists(
452            dict(reg_number='1', level='100'), self.app) is True
453        studylevel = self.processor.getEntry(
454            dict(reg_number='1', level='100'), self.app)
455        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
456        self.assertEqual(studylevel.level_session, 2008)
457        self.assertEqual(studylevel.level_verdict, 'A')
458        self.assertEqual(studylevel.level, 100)
459        shutil.rmtree(os.path.dirname(fin_file))
460       
461
462class CourseTicketProcessorTest(StudentImportExportSetup):
463
464    def setUp(self):
465        super(CourseTicketProcessorTest, self).setUp()
466
467        # Import students with subobjects
468        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
469        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
470        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
471            student_file, STUDENT_HEADER_FIELDS)
472        shutil.rmtree(os.path.dirname(fin_file))
473
474        # Add course and course referrer
475        self.course = createObject('waeup.Course')
476        self.course.code = 'COURSE1'
477        self.course.semester = 1
478        self.course.credits = 10
479        self.course.passmark = 40
480        self.app['faculties']['fac1']['dep1'].courses.addCourse(
481            self.course)
482        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
483            self.course, level=100)
484
485        # Update study courses
486        studycourse_file = os.path.join(
487            self.workdir, 'sample_studycourse_data.csv')
488        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
489        processor = StudentStudyCourseProcessor()
490        num, num_warns, fin_file, fail_file = processor.doImport(
491            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
492        shutil.rmtree(os.path.dirname(fin_file))
493
494        # Import study levels
495        processor = StudentStudyLevelProcessor()
496        studylevel_file = os.path.join(
497            self.workdir, 'sample_studylevel_data.csv')
498        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
499        num, num_warns, fin_file, fail_file = processor.doImport(
500            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
501        shutil.rmtree(os.path.dirname(fin_file))
502
503        self.processor = CourseTicketProcessor()
504        self.csv_file = os.path.join(
505            self.workdir, 'sample_courseticket_data.csv')
506        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
507
508    def test_interface(self):
509        # Make sure we fulfill the interface contracts.
510        assert verifyObject(IBatchProcessor, self.processor) is True
511        assert verifyClass(
512            IBatchProcessor, CourseTicketProcessor) is True
513
514    def test_checkConversion(self):
515        errs, inv_errs, conv_dict = self.processor.checkConversion(
516            dict(reg_number='1', code='COURSE1', level='220'))
517        self.assertEqual(len(errs),0)
518        errs, inv_errs, conv_dict = self.processor.checkConversion(
519            dict(reg_number='1', code='COURSE2', level='220'))
520        self.assertEqual(len(errs),1)
521        self.assertTrue(('code','non-existent') in errs)
522
523    def test_import(self):
524
525        num, num_warns, fin_file, fail_file = self.processor.doImport(
526            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
527
528        self.assertEqual(num_warns,2)
529        assert self.processor.entryExists(
530            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
531        courseticket = self.processor.getEntry(
532            dict(reg_number='1', level='100', code='COURSE1'), self.app)
533        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
534        self.assertEqual(courseticket.score, 1)
535        self.assertEqual(courseticket.mandatory, True)
536        self.assertEqual(courseticket.fcode, 'NA')
537        self.assertEqual(courseticket.dcode, 'NA')
538        self.assertEqual(courseticket.code, 'COURSE1')
539        self.assertEqual(courseticket.title, 'Unnamed Course')
540        self.assertEqual(courseticket.credits, 10)
541        self.assertEqual(courseticket.passmark, 40)
542        self.assertEqual(courseticket.semester, 1)
543        shutil.rmtree(os.path.dirname(fin_file))
544
545class PaymentProcessorTest(StudentImportExportSetup):
546
547    def setUp(self):
548        super(PaymentProcessorTest, self).setUp()
549
550        # Add student with payment
551        student = Student()
552        student.firstname = u'Anna'
553        student.lastname = u'Tester'
554        student.reg_number = u'123'
555        student.matric_number = u'234'
556        self.app['students'].addStudent(student)
557        self.student = self.app['students'][student.student_id]
558        payment = createObject(u'waeup.StudentOnlinePayment')
559        payment.p_id = 'p123'
560        self.student['payments'][payment.p_id] = payment
561
562        # Import students with subobjects
563        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
564        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
565        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
566            student_file, STUDENT_HEADER_FIELDS)
567        shutil.rmtree(os.path.dirname(fin_file))
568
569        self.processor = StudentOnlinePaymentProcessor()
570        self.csv_file = os.path.join(
571            self.workdir, 'sample_payment_data.csv')
572        open(self.csv_file, 'wb').write(PAYMENT_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, StudentOnlinePaymentProcessor) is True
579
580    def test_getEntry(self):
581        assert self.processor.getEntry(
582            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
583        assert self.processor.getEntry(
584            dict(student_id=self.student.student_id, p_id='p123'),
585            self.app) is self.student['payments']['p123']
586        assert self.processor.getEntry(
587            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
588            self.app) is self.student['payments']['p123']
589
590    def test_addEntry(self):
591        self.assertEqual(len(self.student['payments'].keys()),1)
592        payment1 = createObject(u'waeup.StudentOnlinePayment')
593        payment1.p_id = 'p234'
594        self.processor.addEntry(
595            payment1, dict(student_id=self.student.student_id, p_id='p234'),
596            self.app)
597        self.assertEqual(len(self.student['payments'].keys()),2)
598        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
599        payment2 = createObject(u'waeup.StudentOnlinePayment')
600        payment1.p_id = 'nonsense'
601        # payment1.p_id will be replaced if p_id doesn't start with 'p'
602        self.processor.addEntry(
603            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
604            self.app)
605        self.assertEqual(len(self.student['payments'].keys()),3)
606        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
607
608    def test_checkConversion(self):
609        errs, inv_errs, conv_dict = self.processor.checkConversion(
610            dict(reg_number='1', p_id='3816951266236341955'))
611        self.assertEqual(len(errs),0)
612        errs, inv_errs, conv_dict = self.processor.checkConversion(
613            dict(reg_number='1', p_id='p1266236341955'))
614        self.assertEqual(len(errs),0)
615        errs, inv_errs, conv_dict = self.processor.checkConversion(
616            dict(reg_number='1', p_id='nonsense'))
617        self.assertEqual(len(errs),1)
618        timestamp = "%d" % int(time()*1000)
619        p_id = "p%s" % timestamp
620        errs, inv_errs, conv_dict = self.processor.checkConversion(
621            dict(reg_number='1', p_id=p_id))
622        self.assertEqual(len(errs),0)
623
624    def test_import(self):
625        num, num_warns, fin_file, fail_file = self.processor.doImport(
626            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
627        self.assertEqual(num_warns,0)
628        payment = self.processor.getEntry(dict(reg_number='1',
629            p_id='p1290797973744'), self.app)
630        self.assertEqual(payment.p_id, 'p1290797973744')
631        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
632        self.assertEqual(cdate, "2010-11-26 19:59:33")
633        shutil.rmtree(os.path.dirname(fin_file))
634
635def test_suite():
636    suite = unittest.TestSuite()
637    for testcase in [
638        StudentProcessorTest,StudentStudyCourseProcessorTest,
639        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
640        PaymentProcessorTest,StudentVerdictProcessorTest]:
641        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
642                testcase
643                )
644        )
645    return suite
646
647
Note: See TracBrowser for help on using the repository browser.