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

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