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

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

Extend import-export-setup to avoid repetition.
  • Property svn:keywords set to Id
File size: 35.0 KB
Line 
1## $Id: test_batching.py 8340 2012-05-04 11:49:25Z 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.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.students.studylevel import StudentStudyLevel, CourseTicket
39from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
40from waeup.kofa.interfaces import IBatchProcessor, FatalCSVError
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
63STUDENT_SAMPLE_DATA_UPDATE3 = open(
64    os.path.join(os.path.dirname(__file__), 'sample_student_data_update3.csv'),
65    'rb').read()
66
67STUDENT_HEADER_FIELDS_UPDATE3 = STUDENT_SAMPLE_DATA_UPDATE3.split(
68    '\n')[0].split(',')
69
70STUDENT_SAMPLE_DATA_UPDATE4 = open(
71    os.path.join(os.path.dirname(__file__), 'sample_student_data_update4.csv'),
72    'rb').read()
73
74STUDENT_HEADER_FIELDS_UPDATE4 = STUDENT_SAMPLE_DATA_UPDATE4.split(
75    '\n')[0].split(',')
76
77STUDYCOURSE_SAMPLE_DATA = open(
78    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
79    'rb').read()
80
81STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
82    '\n')[0].split(',')
83
84VERDICT_SAMPLE_DATA = open(
85    os.path.join(os.path.dirname(__file__), 'sample_verdict_data.csv'),
86    'rb').read()
87
88VERDICT_HEADER_FIELDS = VERDICT_SAMPLE_DATA.split(
89    '\n')[0].split(',')
90
91STUDENT_SAMPLE_DATA_MIGRATION = open(
92    os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'),
93    'rb').read()
94
95STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split(
96    '\n')[0].split(',')
97
98STUDENT_SAMPLE_DATA_DUPLICATES = open(
99    os.path.join(os.path.dirname(__file__), 'sample_student_data_duplicates.csv'),
100    'rb').read()
101
102STUDENT_HEADER_FIELDS_DUPLICATES = STUDENT_SAMPLE_DATA_DUPLICATES.split(
103    '\n')[0].split(',')
104
105STUDYLEVEL_SAMPLE_DATA = open(
106    os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'),
107    'rb').read()
108
109STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split(
110    '\n')[0].split(',')
111
112COURSETICKET_SAMPLE_DATA = open(
113    os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'),
114    'rb').read()
115
116COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
117    '\n')[0].split(',')
118
119PAYMENT_SAMPLE_DATA = open(
120    os.path.join(os.path.dirname(__file__), 'sample_payment_data.csv'),
121    'rb').read()
122
123PAYMENT_HEADER_FIELDS = PAYMENT_SAMPLE_DATA.split(
124    '\n')[0].split(',')
125
126class StudentImportExportSetup(FunctionalTestCase):
127
128    layer = FunctionalLayer
129
130    def setUp(self):
131        super(StudentImportExportSetup, self).setUp()
132        self.dc_root = tempfile.mkdtemp()
133        self.workdir = tempfile.mkdtemp()
134        app = University()
135        app['datacenter'].setStoragePath(self.dc_root)
136        self.getRootFolder()['app'] = app
137        self.app = self.getRootFolder()['app']
138        setSite(app)
139
140        # Populate university
141        self.certificate = createObject('waeup.Certificate')
142        self.certificate.code = 'CERT1'
143        self.certificate.application_category = 'basic'
144        self.certificate.start_level = 200
145        self.certificate.end_level = 500
146        self.app['faculties']['fac1'] = Faculty()
147        self.app['faculties']['fac1']['dep1'] = Department()
148        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
149            self.certificate)
150        return
151
152    def tearDown(self):
153        super(StudentImportExportSetup, self).tearDown()
154        shutil.rmtree(self.workdir)
155        shutil.rmtree(self.dc_root)
156        clearSite()
157        return
158
159    def setup_for_export(self):
160        student = Student()
161        student.student_id = u'A111111'
162        self.app['students'][student.student_id] = self.student = student
163        self.outfile = os.path.join(self.workdir, 'myoutput.csv')
164        return
165
166    def setup_student(self, student):
167        # set predictable values for `student`
168        student.matric_number = u'M123456'
169        student.adm_code = u'my adm code'
170        student.clearance_locked = False
171        student.clr_code = u'my clr code'
172        student.perm_address = u'Studentroad 21\nLagos 123456\n'
173        student.reg_number = u'123456'
174        student.student_id = u'A111111'
175        student.firstname = u'Anna'
176        student.lastname = u'Tester'
177        student.middlename = u'M.'
178        student.date_of_birth = datetime.date(1981, 2, 4)
179        student.sex = 'f'
180        student.email = 'anna@sample.com'
181        student.phone = u'+234-123-12345'
182        student.notice = u'Some notice\nin lines.'
183        student.nationality = u'NG'
184
185        student['studycourse'].certificate = self.certificate
186        student['studycourse'].entry_mode = 'ug_ft'
187        student['studycourse'].entry_session = 2010
188        student['studycourse'].current_session = 2012
189        student['studycourse'].current_level = int(self.certificate.start_level)
190
191        study_level = StudentStudyLevel()
192        study_level.level_session = 2012
193        study_level.level_verdict = "A"
194        study_level.level = 100
195        student['studycourse'].addStudentStudyLevel(
196            self.certificate, study_level)
197
198        ticket = CourseTicket()
199        ticket.automatic = True
200        ticket.carry_over = True
201        ticket.code = u'CRS1'
202        ticket.title = u'Course 1'
203        ticket.fcode = u'FAC1'
204        ticket.dcode = u'DEP1'
205        ticket.credits = 100
206        ticket.passmark = 100
207        ticket.semester = 2
208        study_level.addCourseTicket(ticket)
209
210        return student
211
212
213class StudentProcessorTest(FunctionalTestCase):
214
215    layer = FunctionalLayer
216
217    def setUp(self):
218        super(StudentProcessorTest, self).setUp()
219        # Setup a sample site for each test
220        app = University()
221        self.dc_root = tempfile.mkdtemp()
222        app['datacenter'].setStoragePath(self.dc_root)
223
224        # Prepopulate the ZODB...
225        self.getRootFolder()['app'] = app
226        # we add the site immediately after creation to the
227        # ZODB. Catalogs and other local utilities are not setup
228        # before that step.
229        self.app = self.getRootFolder()['app']
230        # Set site here. Some of the following setup code might need
231        # to access grok.getSite() and should get our new app then
232        setSite(app)
233
234        # Add student with subobjects
235        student = Student()
236        student.firstname = u'Anna'
237        student.lastname = u'Tester'
238        student.reg_number = u'123'
239        student.matric_number = u'234'
240        self.app['students'].addStudent(student)
241        self.student = self.app['students'][student.student_id]
242        self.processor = StudentProcessor()
243        self.workdir = tempfile.mkdtemp()
244        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
245        self.csv_file_update = os.path.join(
246            self.workdir, 'sample_student_data_update.csv')
247        self.csv_file_update2 = os.path.join(
248            self.workdir, 'sample_student_data_update2.csv')
249        self.csv_file_update3 = os.path.join(
250            self.workdir, 'sample_student_data_update3.csv')
251        self.csv_file_update4 = os.path.join(
252            self.workdir, 'sample_student_data_update4.csv')
253        self.csv_file_migration = os.path.join(
254            self.workdir, 'sample_student_data_migration.csv')
255        self.csv_file_duplicates = os.path.join(
256            self.workdir, 'sample_student_data_duplicates.csv')
257        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
258        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
259        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
260        open(self.csv_file_update3, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE3)
261        open(self.csv_file_update4, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE4)
262        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
263        open(self.csv_file_duplicates, 'wb').write(STUDENT_SAMPLE_DATA_DUPLICATES)
264
265    def tearDown(self):
266        super(StudentProcessorTest, self).tearDown()
267        shutil.rmtree(self.workdir)
268        shutil.rmtree(self.dc_root)
269        clearSite()
270        return
271
272    def test_interface(self):
273        # Make sure we fulfill the interface contracts.
274        assert verifyObject(IBatchProcessor, self.processor) is True
275        assert verifyClass(
276            IBatchProcessor, StudentProcessor) is True
277
278    def test_parentsExist(self):
279        self.assertFalse(self.processor.parentsExist(None, dict()))
280        self.assertTrue(self.processor.parentsExist(None, self.app))
281
282    def test_entryExists(self):
283        assert self.processor.entryExists(
284            dict(student_id='ID_NONE'), self.app) is False
285        assert self.processor.entryExists(
286            dict(reg_number='123'), self.app) is True
287
288    def test_getParent(self):
289        parent = self.processor.getParent(None, self.app)
290        assert parent is self.app['students']
291
292    def test_getEntry(self):
293        assert self.processor.getEntry(
294            dict(student_id='ID_NONE'), self.app) is None
295        assert self.processor.getEntry(
296            dict(student_id=self.student.student_id), self.app) is self.student
297
298    def test_addEntry(self):
299        new_student = Student()
300        self.processor.addEntry(
301            new_student, dict(), self.app)
302        assert len(self.app['students'].keys()) == 2
303
304    def test_checkConversion(self):
305        errs, inv_errs, conv_dict = self.processor.checkConversion(
306            dict(reg_number='1', state='admitted'))
307        self.assertEqual(len(errs),0)
308        # Empty state is allowed
309        errs, inv_errs, conv_dict = self.processor.checkConversion(
310            dict(reg_number='1', state=''))
311        self.assertEqual(len(errs),0)
312        #self.assertTrue(('state', 'no value provided') in errs)
313        errs, inv_errs, conv_dict = self.processor.checkConversion(
314            dict(reg_number='1', state='nonsense'))
315        self.assertEqual(len(errs),1)
316        self.assertTrue(('state', 'not allowed') in errs)
317
318    def test_delEntry(self):
319        assert self.student.student_id in self.app['students'].keys()
320        self.processor.delEntry(
321            dict(reg_number=self.student.reg_number), self.app)
322        assert self.student.student_id not in self.app['students'].keys()
323
324    def test_import(self):
325        num, num_warns, fin_file, fail_file = self.processor.doImport(
326            self.csv_file, STUDENT_HEADER_FIELDS)
327        self.assertEqual(num_warns,0)
328        assert len(self.app['students'].keys()) == 5
329        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
330        self.assertEqual(self.app['students']['X666666'].state,'courses validated')
331        shutil.rmtree(os.path.dirname(fin_file))
332
333    def test_import_update(self):
334        num, num_warns, fin_file, fail_file = self.processor.doImport(
335            self.csv_file, STUDENT_HEADER_FIELDS)
336        shutil.rmtree(os.path.dirname(fin_file))
337        num, num_warns, fin_file, fail_file = self.processor.doImport(
338            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
339        self.assertEqual(num_warns,0)
340        # state has changed
341        self.assertEqual(self.app['students']['X666666'].state,'admitted')
342        # state has not changed
343        self.assertEqual(self.app['students']['Y777777'].state,'courses validated')
344        shutil.rmtree(os.path.dirname(fin_file))
345
346    def test_import_update2(self):
347        num, num_warns, fin_file, fail_file = self.processor.doImport(
348            self.csv_file, STUDENT_HEADER_FIELDS)
349        shutil.rmtree(os.path.dirname(fin_file))
350        num, num_warns, fin_file, fail_file = self.processor.doImport(
351            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
352        self.assertEqual(num_warns,0)
353        # The phone import value of Pieri was None.
354        # Confirm that phone has not been cleared.
355        container = self.app['students']
356        for key in container.keys():
357            if container[key].firstname == 'Aaren':
358                aaren = container[key]
359                break
360        self.assertEqual(aaren.phone, '--1234')
361        # The phone import value of Claus was a deletion marker.
362        # Confirm that phone has been cleared.
363        for key in container.keys():
364            if container[key].firstname == 'Claus':
365                claus = container[key]
366                break
367        assert claus.phone is None
368        shutil.rmtree(os.path.dirname(fin_file))
369
370    def test_import_update3(self):
371        num, num_warns, fin_file, fail_file = self.processor.doImport(
372            self.csv_file, STUDENT_HEADER_FIELDS)
373        shutil.rmtree(os.path.dirname(fin_file))
374        num, num_warns, fin_file, fail_file = self.processor.doImport(
375            self.csv_file_update3, STUDENT_HEADER_FIELDS_UPDATE3, 'update')
376        content = open(fail_file).read()
377        self.assertEqual(
378            content,
379            'reg_number,student_id,transition,--ERRORS--\r\n'
380            '<IGNORE>,X666666,request_clearance,Transition not allowed.\r\n'
381            )
382        self.assertEqual(num_warns,1)
383        self.assertEqual(self.app['students']['Y777777'].state,'returning')
384
385    def test_import_update4(self):
386        num, num_warns, fin_file, fail_file = self.processor.doImport(
387            self.csv_file, STUDENT_HEADER_FIELDS)
388        shutil.rmtree(os.path.dirname(fin_file))
389        self.assertRaises(
390            FatalCSVError, self.processor.doImport, self.csv_file_update4,
391            STUDENT_HEADER_FIELDS_UPDATE4, 'update')
392
393    def test_import_remove(self):
394        num, num_warns, fin_file, fail_file = self.processor.doImport(
395            self.csv_file, STUDENT_HEADER_FIELDS)
396        shutil.rmtree(os.path.dirname(fin_file))
397        num, num_warns, fin_file, fail_file = self.processor.doImport(
398            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
399        self.assertEqual(num_warns,0)
400        shutil.rmtree(os.path.dirname(fin_file))
401
402    def test_import_migration_data(self):
403        num, num_warns, fin_file, fail_file = self.processor.doImport(
404            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
405        content = open(fail_file).read()
406        self.assertEqual(num_warns,2)
407        assert len(self.app['students'].keys()) == 5
408        self.assertEqual(
409            content,
410            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
411            '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n'
412            '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n'
413            )
414        self.assertTrue('A123456' in self.app['students'].keys())
415        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
416        self.assertEqual(self.app['students']['A123456'].date_of_birth,
417            datetime.date(1990, 1, 2))
418        self.assertFalse(self.app['students']['A123456'].clearance_locked)
419        self.assertEqual(self.app['students']['B123456'].state,'cleared')
420        self.assertEqual(self.app['students']['B123456'].date_of_birth,
421            datetime.date(1990, 1, 3))
422        self.assertTrue(self.app['students']['B123456'].clearance_locked)
423        history = ' '.join(self.app['students']['A123456'].history.messages)
424        self.assertTrue(
425            "State 'clearance started' set by system" in history)
426        # state was empty and student is thus in state created
427        self.assertEqual(self.app['students']['F123456'].state,'created')
428        shutil.rmtree(os.path.dirname(fin_file))
429
430    def test_import_duplicate_data(self):
431        num, num_warns, fin_file, fail_file = self.processor.doImport(
432            self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES)
433        content = open(fail_file).read()
434        self.assertEqual(num_warns,3)
435        self.assertEqual(
436            content,
437            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
438            '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n'
439            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n'
440            '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n'
441            )
442        shutil.rmtree(os.path.dirname(fin_file))
443
444class StudentStudyCourseProcessorTest(StudentImportExportSetup):
445
446    def setUp(self):
447        super(StudentStudyCourseProcessorTest, self).setUp()
448
449        # Import students with subobjects
450        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
451        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
452        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
453            student_file, STUDENT_HEADER_FIELDS)
454        shutil.rmtree(os.path.dirname(fin_file))
455
456        self.processor = StudentStudyCourseProcessor()
457        self.csv_file = os.path.join(
458            self.workdir, 'sample_studycourse_data.csv')
459        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
460        return
461
462    def test_interface(self):
463        # Make sure we fulfill the interface contracts.
464        assert verifyObject(IBatchProcessor, self.processor) is True
465        assert verifyClass(
466            IBatchProcessor, StudentStudyCourseProcessor) is True
467
468    def test_entryExists(self):
469        assert self.processor.entryExists(
470            dict(reg_number='REG_NONE'), self.app) is False
471        assert self.processor.entryExists(
472            dict(reg_number='1'), self.app) is True
473
474    def test_getEntry(self):
475        student = self.processor.getEntry(
476            dict(reg_number='1'), self.app).__parent__
477        self.assertEqual(student.reg_number,'1')
478
479    def test_checkConversion(self):
480        errs, inv_errs, conv_dict = self.processor.checkConversion(
481            dict(reg_number='1', certificate='CERT1', current_level='200'))
482        self.assertEqual(len(errs),0)
483        errs, inv_errs, conv_dict = self.processor.checkConversion(
484            dict(reg_number='1', certificate='CERT999'))
485        self.assertEqual(len(errs),1)
486        self.assertTrue(('certificate', u'Invalid value') in errs)
487        errs, inv_errs, conv_dict = self.processor.checkConversion(
488            dict(reg_number='1', certificate='CERT1', current_level='100'))
489        self.assertEqual(len(errs),1)
490        self.assertTrue(('current_level','not in range') in errs)
491        # If we import only current_level, no conversion checking is done.
492        errs, inv_errs, conv_dict = self.processor.checkConversion(
493            dict(reg_number='1', current_level='100'))
494        self.assertEqual(len(errs),0)
495
496    def test_import(self):
497        num, num_warns, fin_file, fail_file = self.processor.doImport(
498            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
499        self.assertEqual(num_warns,1)
500        content = open(fail_file).read()
501        self.assertTrue('current_level: not in range' in content)
502        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
503        self.assertEqual(studycourse.certificate.code, u'CERT1')
504        shutil.rmtree(os.path.dirname(fin_file))
505
506class StudentVerdictProcessorTest(StudentImportExportSetup):
507
508
509    def setUp(self):
510        super(StudentVerdictProcessorTest, self).setUp()
511
512        # Import students with subobjects
513        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
514        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
515        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
516            student_file, STUDENT_HEADER_FIELDS)
517        shutil.rmtree(os.path.dirname(fin_file))
518
519        # Update study courses
520        studycourse_file = os.path.join(
521            self.workdir, 'sample_studycourse_data.csv')
522        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
523        processor = StudentStudyCourseProcessor()
524        num, num_warns, fin_file, fail_file = processor.doImport(
525            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
526        shutil.rmtree(os.path.dirname(fin_file))
527
528        self.processor = StudentVerdictProcessor()
529        self.csv_file = os.path.join(
530            self.workdir, 'sample_verdict_data.csv')
531        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
532        return
533
534    def test_import(self):
535        num, num_warns, fin_file, fail_file = self.processor.doImport(
536            self.csv_file, VERDICT_HEADER_FIELDS,'update')
537        self.assertEqual(num_warns,3)
538        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
539        student = self.processor.getParent(dict(matric_number='100000'), self.app)
540        self.assertEqual(studycourse.current_verdict, 'A')
541        self.assertEqual(student.state, 'returning')
542        self.assertEqual(studycourse.current_level, 200)
543        content = open(fail_file).read()
544        self.assertEqual(
545            content,
546            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
547            '2008,B,100001,100,Current level does not correspond.\r\n'
548            '2007,C,100002,200,Current session does not correspond.\r\n'
549            '2008,A,100003,200,Student in wrong state.\r\n'
550            )
551        shutil.rmtree(os.path.dirname(fin_file))
552
553
554class StudentStudyLevelProcessorTest(StudentImportExportSetup):
555
556    def setUp(self):
557        super(StudentStudyLevelProcessorTest, self).setUp()
558
559        # Import students with subobjects
560        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
561        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
562        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
563            student_file, STUDENT_HEADER_FIELDS)
564        shutil.rmtree(os.path.dirname(fin_file))
565
566        # Update study courses
567        studycourse_file = os.path.join(
568            self.workdir, 'sample_studycourse_data.csv')
569        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
570        processor = StudentStudyCourseProcessor()
571        num, num_warns, fin_file, fail_file = processor.doImport(
572            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
573        shutil.rmtree(os.path.dirname(fin_file))
574
575        self.processor = StudentStudyLevelProcessor()
576        self.csv_file = os.path.join(
577            self.workdir, 'sample_studylevel_data.csv')
578        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
579
580    def test_interface(self):
581        # Make sure we fulfill the interface contracts.
582        assert verifyObject(IBatchProcessor, self.processor) is True
583        assert verifyClass(
584            IBatchProcessor, StudentStudyLevelProcessor) is True
585
586    def test_checkConversion(self):
587        errs, inv_errs, conv_dict = self.processor.checkConversion(
588            dict(reg_number='1', level='220'))
589        self.assertEqual(len(errs),0)
590        errs, inv_errs, conv_dict = self.processor.checkConversion(
591            dict(reg_number='1', level='900'))
592        self.assertEqual(len(errs),1)
593        self.assertTrue(('level','no valid integer') in errs)
594        errs, inv_errs, conv_dict = self.processor.checkConversion(
595            dict(reg_number='1', level='xyz'))
596        self.assertEqual(len(errs),1)
597        self.assertTrue(('level','no integer') in errs)
598
599    def test_import(self):
600        num, num_warns, fin_file, fail_file = self.processor.doImport(
601            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
602        self.assertEqual(num_warns,2)
603        assert self.processor.entryExists(
604            dict(reg_number='1', level='100'), self.app) is True
605        studylevel = self.processor.getEntry(
606            dict(reg_number='1', level='100'), self.app)
607        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
608        self.assertEqual(studylevel.level_session, 2008)
609        self.assertEqual(studylevel.level_verdict, 'A')
610        self.assertEqual(studylevel.level, 100)
611        shutil.rmtree(os.path.dirname(fin_file))
612
613    def test_import_update(self):
614        # We perform the same import twice,
615        # the second time in update mode. The number
616        # of warnings must be the same.
617        num, num_warns, fin_file, fail_file = self.processor.doImport(
618            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
619        num, num_warns, fin_file, fail_file = self.processor.doImport(
620            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
621        self.assertEqual(num_warns,2)
622        shutil.rmtree(os.path.dirname(fin_file))
623       
624
625class CourseTicketProcessorTest(StudentImportExportSetup):
626
627    def setUp(self):
628        super(CourseTicketProcessorTest, self).setUp()
629
630        # Import students with subobjects
631        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
632        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
633        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
634            student_file, STUDENT_HEADER_FIELDS)
635        shutil.rmtree(os.path.dirname(fin_file))
636
637        # Add course and course referrer
638        self.course = createObject('waeup.Course')
639        self.course.code = 'COURSE1'
640        self.course.semester = 1
641        self.course.credits = 10
642        self.course.passmark = 40
643        self.app['faculties']['fac1']['dep1'].courses.addCourse(
644            self.course)
645        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
646            self.course, level=100)
647
648        # Update study courses
649        studycourse_file = os.path.join(
650            self.workdir, 'sample_studycourse_data.csv')
651        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
652        processor = StudentStudyCourseProcessor()
653        num, num_warns, fin_file, fail_file = processor.doImport(
654            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
655        shutil.rmtree(os.path.dirname(fin_file))
656
657        # Import study levels
658        processor = StudentStudyLevelProcessor()
659        studylevel_file = os.path.join(
660            self.workdir, 'sample_studylevel_data.csv')
661        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
662        num, num_warns, fin_file, fail_file = processor.doImport(
663            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
664        shutil.rmtree(os.path.dirname(fin_file))
665
666        self.processor = CourseTicketProcessor()
667        self.csv_file = os.path.join(
668            self.workdir, 'sample_courseticket_data.csv')
669        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
670
671    def test_interface(self):
672        # Make sure we fulfill the interface contracts.
673        assert verifyObject(IBatchProcessor, self.processor) is True
674        assert verifyClass(
675            IBatchProcessor, CourseTicketProcessor) is True
676
677    def test_checkConversion(self):
678        errs, inv_errs, conv_dict = self.processor.checkConversion(
679            dict(reg_number='1', code='COURSE1', level='220'))
680        self.assertEqual(len(errs),0)
681        errs, inv_errs, conv_dict = self.processor.checkConversion(
682            dict(reg_number='1', code='COURSE2', level='220'))
683        self.assertEqual(len(errs),1)
684        self.assertTrue(('code','non-existent') in errs)
685
686    def test_import(self):
687
688        num, num_warns, fin_file, fail_file = self.processor.doImport(
689            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
690
691        self.assertEqual(num_warns,2)
692        assert self.processor.entryExists(
693            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
694        courseticket = self.processor.getEntry(
695            dict(reg_number='1', level='100', code='COURSE1'), self.app)
696        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
697        self.assertEqual(courseticket.score, 1)
698        self.assertEqual(courseticket.mandatory, True)
699        self.assertEqual(courseticket.fcode, 'NA')
700        self.assertEqual(courseticket.dcode, 'NA')
701        self.assertEqual(courseticket.code, 'COURSE1')
702        self.assertEqual(courseticket.title, 'Unnamed Course')
703        self.assertEqual(courseticket.credits, 10)
704        self.assertEqual(courseticket.passmark, 40)
705        self.assertEqual(courseticket.semester, 1)
706        shutil.rmtree(os.path.dirname(fin_file))
707
708    def test_import_update(self):
709        # We perform the same import twice,
710        # the second time in update mode. The number
711        # of warnings must be the same.
712        num, num_warns, fin_file, fail_file = self.processor.doImport(
713            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
714        num, num_warns, fin_file, fail_file = self.processor.doImport(
715            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
716        self.assertEqual(num_warns,2)
717        shutil.rmtree(os.path.dirname(fin_file))
718
719class PaymentProcessorTest(StudentImportExportSetup):
720
721    def setUp(self):
722        super(PaymentProcessorTest, self).setUp()
723
724        # Add student with payment
725        student = Student()
726        student.firstname = u'Anna'
727        student.lastname = u'Tester'
728        student.reg_number = u'123'
729        student.matric_number = u'234'
730        self.app['students'].addStudent(student)
731        self.student = self.app['students'][student.student_id]
732        payment = createObject(u'waeup.StudentOnlinePayment')
733        payment.p_id = 'p123'
734        self.student['payments'][payment.p_id] = payment
735
736        # Import students with subobjects
737        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
738        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
739        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
740            student_file, STUDENT_HEADER_FIELDS)
741        shutil.rmtree(os.path.dirname(fin_file))
742
743        self.processor = StudentOnlinePaymentProcessor()
744        self.csv_file = os.path.join(
745            self.workdir, 'sample_payment_data.csv')
746        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
747
748    def test_interface(self):
749        # Make sure we fulfill the interface contracts.
750        assert verifyObject(IBatchProcessor, self.processor) is True
751        assert verifyClass(
752            IBatchProcessor, StudentOnlinePaymentProcessor) is True
753
754    def test_getEntry(self):
755        assert self.processor.getEntry(
756            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
757        assert self.processor.getEntry(
758            dict(student_id=self.student.student_id, p_id='p123'),
759            self.app) is self.student['payments']['p123']
760        assert self.processor.getEntry(
761            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
762            self.app) is self.student['payments']['p123']
763
764    def test_addEntry(self):
765        self.assertEqual(len(self.student['payments'].keys()),1)
766        payment1 = createObject(u'waeup.StudentOnlinePayment')
767        payment1.p_id = 'p234'
768        self.processor.addEntry(
769            payment1, dict(student_id=self.student.student_id, p_id='p234'),
770            self.app)
771        self.assertEqual(len(self.student['payments'].keys()),2)
772        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
773        payment2 = createObject(u'waeup.StudentOnlinePayment')
774        payment1.p_id = 'nonsense'
775        # payment1.p_id will be replaced if p_id doesn't start with 'p'
776        self.processor.addEntry(
777            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
778            self.app)
779        self.assertEqual(len(self.student['payments'].keys()),3)
780        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
781
782    def test_checkConversion(self):
783        errs, inv_errs, conv_dict = self.processor.checkConversion(
784            dict(reg_number='1', p_id='3816951266236341955'))
785        self.assertEqual(len(errs),0)
786        errs, inv_errs, conv_dict = self.processor.checkConversion(
787            dict(reg_number='1', p_id='p1266236341955'))
788        self.assertEqual(len(errs),0)
789        errs, inv_errs, conv_dict = self.processor.checkConversion(
790            dict(reg_number='1', p_id='nonsense'))
791        self.assertEqual(len(errs),1)
792        timestamp = "%d" % int(time()*1000)
793        p_id = "p%s" % timestamp
794        errs, inv_errs, conv_dict = self.processor.checkConversion(
795            dict(reg_number='1', p_id=p_id))
796        self.assertEqual(len(errs),0)
797
798    def test_import(self):
799        num, num_warns, fin_file, fail_file = self.processor.doImport(
800            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
801        self.assertEqual(num_warns,0)
802        payment = self.processor.getEntry(dict(reg_number='1',
803            p_id='p1290797973744'), self.app)
804        self.assertEqual(payment.p_id, 'p1290797973744')
805        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
806        self.assertEqual(cdate, "2010-11-26 18:59:33")
807        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
808        shutil.rmtree(os.path.dirname(fin_file))
809
810    def test_import_update(self):
811        # We perform the same import twice,
812        # the second time in update mode. The number
813        # of warnings must be the same.
814        num, num_warns, fin_file, fail_file = self.processor.doImport(
815            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
816        num, num_warns, fin_file, fail_file = self.processor.doImport(
817            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
818        self.assertEqual(num_warns,0)
819        shutil.rmtree(os.path.dirname(fin_file))
820
821def test_suite():
822    suite = unittest.TestSuite()
823    for testcase in [
824        StudentProcessorTest,StudentStudyCourseProcessorTest,
825        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
826        PaymentProcessorTest,StudentVerdictProcessorTest]:
827        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
828                testcase
829                )
830        )
831    return suite
832
833
Note: See TracBrowser for help on using the repository browser.