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

Last change on this file since 8490 was 8490, checked in by uli, 12 years ago

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