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

Last change on this file since 8309 was 8309, checked in by Henrik Bettermann, 12 years ago

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