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

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

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