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

Last change on this file since 8163 was 8163, checked in by uli, 13 years ago

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