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

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

Some minor test improvements.
  • Property svn:keywords set to Id
File size: 26.8 KB
Line 
1## $Id: test_batching.py 7958 2012-03-22 14:49:20Z 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        content = open(fail_file).read()
351        self.assertTrue('current_level: not in range' in content)
352        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
353        self.assertEqual(studycourse.certificate.code, u'CERT1')
354        shutil.rmtree(os.path.dirname(fin_file))
355
356class StudentVerdictProcessorTest(StudentImportExportSetup):
357
358
359    def setUp(self):
360        super(StudentVerdictProcessorTest, self).setUp()
361
362        # Import students with subobjects
363        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
364        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
365        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
366            student_file, STUDENT_HEADER_FIELDS)
367        shutil.rmtree(os.path.dirname(fin_file))
368
369        # Update study courses
370        studycourse_file = os.path.join(
371            self.workdir, 'sample_studycourse_data.csv')
372        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
373        processor = StudentStudyCourseProcessor()
374        num, num_warns, fin_file, fail_file = processor.doImport(
375            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
376        shutil.rmtree(os.path.dirname(fin_file))
377
378        self.processor = StudentVerdictProcessor()
379        self.csv_file = os.path.join(
380            self.workdir, 'sample_verdict_data.csv')
381        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
382        return
383
384    def test_import(self):
385        num, num_warns, fin_file, fail_file = self.processor.doImport(
386            self.csv_file, VERDICT_HEADER_FIELDS,'update')
387        self.assertEqual(num_warns,3)
388        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
389        student = self.processor.getParent(dict(matric_number='100000'), self.app)
390        self.assertEqual(studycourse.current_verdict, 'A')
391        self.assertEqual(student.state, 'returning')
392        self.assertEqual(studycourse.current_level, 200)
393        content = open(fail_file).read()
394        self.assertEqual(
395            content,
396            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
397            '2008,B,100001,100,Current level does not correspond.\r\n'
398            '2007,C,100002,200,Current session does not correspond.\r\n'
399            '2008,A,100003,200,Student in wrong state.\r\n'
400            )
401        shutil.rmtree(os.path.dirname(fin_file))
402
403
404class StudentStudyLevelProcessorTest(StudentImportExportSetup):
405
406    def setUp(self):
407        super(StudentStudyLevelProcessorTest, self).setUp()
408
409        # Import students with subobjects
410        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
411        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
412        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
413            student_file, STUDENT_HEADER_FIELDS)
414        shutil.rmtree(os.path.dirname(fin_file))
415
416        # Update study courses
417        studycourse_file = os.path.join(
418            self.workdir, 'sample_studycourse_data.csv')
419        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
420        processor = StudentStudyCourseProcessor()
421        num, num_warns, fin_file, fail_file = processor.doImport(
422            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
423        shutil.rmtree(os.path.dirname(fin_file))
424
425        self.processor = StudentStudyLevelProcessor()
426        self.csv_file = os.path.join(
427            self.workdir, 'sample_studylevel_data.csv')
428        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
429
430    def test_interface(self):
431        # Make sure we fulfill the interface contracts.
432        assert verifyObject(IBatchProcessor, self.processor) is True
433        assert verifyClass(
434            IBatchProcessor, StudentStudyLevelProcessor) is True
435
436    def test_checkConversion(self):
437        errs, inv_errs, conv_dict = self.processor.checkConversion(
438            dict(reg_number='1', level='220'))
439        self.assertEqual(len(errs),0)
440        errs, inv_errs, conv_dict = self.processor.checkConversion(
441            dict(reg_number='1', level='900'))
442        self.assertEqual(len(errs),1)
443        self.assertTrue(('level','no valid integer') in errs)
444        errs, inv_errs, conv_dict = self.processor.checkConversion(
445            dict(reg_number='1', level='xyz'))
446        self.assertEqual(len(errs),1)
447        self.assertTrue(('level','no integer') in errs)
448
449    def test_import(self):
450        num, num_warns, fin_file, fail_file = self.processor.doImport(
451            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
452        self.assertEqual(num_warns,2)
453        assert self.processor.entryExists(
454            dict(reg_number='1', level='100'), self.app) is True
455        studylevel = self.processor.getEntry(
456            dict(reg_number='1', level='100'), self.app)
457        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
458        self.assertEqual(studylevel.level_session, 2008)
459        self.assertEqual(studylevel.level_verdict, 'A')
460        self.assertEqual(studylevel.level, 100)
461        shutil.rmtree(os.path.dirname(fin_file))
462       
463
464class CourseTicketProcessorTest(StudentImportExportSetup):
465
466    def setUp(self):
467        super(CourseTicketProcessorTest, self).setUp()
468
469        # Import students with subobjects
470        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
471        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
472        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
473            student_file, STUDENT_HEADER_FIELDS)
474        shutil.rmtree(os.path.dirname(fin_file))
475
476        # Add course and course referrer
477        self.course = createObject('waeup.Course')
478        self.course.code = 'COURSE1'
479        self.course.semester = 1
480        self.course.credits = 10
481        self.course.passmark = 40
482        self.app['faculties']['fac1']['dep1'].courses.addCourse(
483            self.course)
484        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
485            self.course, level=100)
486
487        # Update study courses
488        studycourse_file = os.path.join(
489            self.workdir, 'sample_studycourse_data.csv')
490        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
491        processor = StudentStudyCourseProcessor()
492        num, num_warns, fin_file, fail_file = processor.doImport(
493            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
494        shutil.rmtree(os.path.dirname(fin_file))
495
496        # Import study levels
497        processor = StudentStudyLevelProcessor()
498        studylevel_file = os.path.join(
499            self.workdir, 'sample_studylevel_data.csv')
500        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
501        num, num_warns, fin_file, fail_file = processor.doImport(
502            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
503        shutil.rmtree(os.path.dirname(fin_file))
504
505        self.processor = CourseTicketProcessor()
506        self.csv_file = os.path.join(
507            self.workdir, 'sample_courseticket_data.csv')
508        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
509
510    def test_interface(self):
511        # Make sure we fulfill the interface contracts.
512        assert verifyObject(IBatchProcessor, self.processor) is True
513        assert verifyClass(
514            IBatchProcessor, CourseTicketProcessor) is True
515
516    def test_checkConversion(self):
517        errs, inv_errs, conv_dict = self.processor.checkConversion(
518            dict(reg_number='1', code='COURSE1', level='220'))
519        self.assertEqual(len(errs),0)
520        errs, inv_errs, conv_dict = self.processor.checkConversion(
521            dict(reg_number='1', code='COURSE2', level='220'))
522        self.assertEqual(len(errs),1)
523        self.assertTrue(('code','non-existent') in errs)
524
525    def test_import(self):
526
527        num, num_warns, fin_file, fail_file = self.processor.doImport(
528            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
529
530        self.assertEqual(num_warns,2)
531        assert self.processor.entryExists(
532            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
533        courseticket = self.processor.getEntry(
534            dict(reg_number='1', level='100', code='COURSE1'), self.app)
535        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
536        self.assertEqual(courseticket.score, 1)
537        self.assertEqual(courseticket.mandatory, True)
538        self.assertEqual(courseticket.fcode, 'NA')
539        self.assertEqual(courseticket.dcode, 'NA')
540        self.assertEqual(courseticket.code, 'COURSE1')
541        self.assertEqual(courseticket.title, 'Unnamed Course')
542        self.assertEqual(courseticket.credits, 10)
543        self.assertEqual(courseticket.passmark, 40)
544        self.assertEqual(courseticket.semester, 1)
545        shutil.rmtree(os.path.dirname(fin_file))
546
547class PaymentProcessorTest(StudentImportExportSetup):
548
549    def setUp(self):
550        super(PaymentProcessorTest, self).setUp()
551
552        # Add student with payment
553        student = Student()
554        student.firstname = u'Anna'
555        student.lastname = u'Tester'
556        student.reg_number = u'123'
557        student.matric_number = u'234'
558        self.app['students'].addStudent(student)
559        self.student = self.app['students'][student.student_id]
560        payment = createObject(u'waeup.StudentOnlinePayment')
561        payment.p_id = 'p123'
562        self.student['payments'][payment.p_id] = payment
563
564        # Import students with subobjects
565        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
566        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
567        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
568            student_file, STUDENT_HEADER_FIELDS)
569        shutil.rmtree(os.path.dirname(fin_file))
570
571        self.processor = StudentOnlinePaymentProcessor()
572        self.csv_file = os.path.join(
573            self.workdir, 'sample_payment_data.csv')
574        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
575
576    def test_interface(self):
577        # Make sure we fulfill the interface contracts.
578        assert verifyObject(IBatchProcessor, self.processor) is True
579        assert verifyClass(
580            IBatchProcessor, StudentOnlinePaymentProcessor) is True
581
582    def test_getEntry(self):
583        assert self.processor.getEntry(
584            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
585        assert self.processor.getEntry(
586            dict(student_id=self.student.student_id, p_id='p123'),
587            self.app) is self.student['payments']['p123']
588        assert self.processor.getEntry(
589            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
590            self.app) is self.student['payments']['p123']
591
592    def test_addEntry(self):
593        self.assertEqual(len(self.student['payments'].keys()),1)
594        payment1 = createObject(u'waeup.StudentOnlinePayment')
595        payment1.p_id = 'p234'
596        self.processor.addEntry(
597            payment1, dict(student_id=self.student.student_id, p_id='p234'),
598            self.app)
599        self.assertEqual(len(self.student['payments'].keys()),2)
600        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
601        payment2 = createObject(u'waeup.StudentOnlinePayment')
602        payment1.p_id = 'nonsense'
603        # payment1.p_id will be replaced if p_id doesn't start with 'p'
604        self.processor.addEntry(
605            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
606            self.app)
607        self.assertEqual(len(self.student['payments'].keys()),3)
608        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
609
610    def test_checkConversion(self):
611        errs, inv_errs, conv_dict = self.processor.checkConversion(
612            dict(reg_number='1', p_id='3816951266236341955'))
613        self.assertEqual(len(errs),0)
614        errs, inv_errs, conv_dict = self.processor.checkConversion(
615            dict(reg_number='1', p_id='p1266236341955'))
616        self.assertEqual(len(errs),0)
617        errs, inv_errs, conv_dict = self.processor.checkConversion(
618            dict(reg_number='1', p_id='nonsense'))
619        self.assertEqual(len(errs),1)
620        timestamp = "%d" % int(time()*1000)
621        p_id = "p%s" % timestamp
622        errs, inv_errs, conv_dict = self.processor.checkConversion(
623            dict(reg_number='1', p_id=p_id))
624        self.assertEqual(len(errs),0)
625
626    def test_import(self):
627        num, num_warns, fin_file, fail_file = self.processor.doImport(
628            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
629        self.assertEqual(num_warns,0)
630        payment = self.processor.getEntry(dict(reg_number='1',
631            p_id='p1290797973744'), self.app)
632        self.assertEqual(payment.p_id, 'p1290797973744')
633        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
634        self.assertEqual(cdate, "2010-11-26 19:59:33")
635        shutil.rmtree(os.path.dirname(fin_file))
636
637def test_suite():
638    suite = unittest.TestSuite()
639    for testcase in [
640        StudentProcessorTest,StudentStudyCourseProcessorTest,
641        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
642        PaymentProcessorTest,StudentVerdictProcessorTest]:
643        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
644                testcase
645                )
646        )
647    return suite
648
649
Note: See TracBrowser for help on using the repository browser.