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

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

Dirty fix for problem with export of student subobjects.
  • Property svn:keywords set to Id
File size: 35.9 KB
Line 
1## $Id: test_batching.py 8411 2012-05-10 16:38:15Z 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        errs, inv_errs, conv_dict = self.processor.checkConversion(
325            dict(reg_number='1', state='admitted'))
326        self.assertEqual(len(errs),0)
327        # Empty state is allowed
328        errs, inv_errs, conv_dict = self.processor.checkConversion(
329            dict(reg_number='1', state=''))
330        self.assertEqual(len(errs),0)
331        #self.assertTrue(('state', 'no value provided') in errs)
332        errs, inv_errs, conv_dict = self.processor.checkConversion(
333            dict(reg_number='1', state='nonsense'))
334        self.assertEqual(len(errs),1)
335        self.assertTrue(('state', 'not allowed') in errs)
336
337    def test_delEntry(self):
338        assert self.student.student_id in self.app['students'].keys()
339        self.processor.delEntry(
340            dict(reg_number=self.student.reg_number), self.app)
341        assert self.student.student_id not in self.app['students'].keys()
342
343    def test_import(self):
344        num, num_warns, fin_file, fail_file = self.processor.doImport(
345            self.csv_file, STUDENT_HEADER_FIELDS)
346        self.assertEqual(num_warns,0)
347        assert len(self.app['students'].keys()) == 5
348        self.assertEqual(self.app['students']['X666666'].reg_number,'1')
349        self.assertEqual(
350            self.app['students']['X666666'].state, 'courses validated')
351        shutil.rmtree(os.path.dirname(fin_file))
352
353    def test_import_update(self):
354        num, num_warns, fin_file, fail_file = self.processor.doImport(
355            self.csv_file, STUDENT_HEADER_FIELDS)
356        shutil.rmtree(os.path.dirname(fin_file))
357        num, num_warns, fin_file, fail_file = self.processor.doImport(
358            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
359        self.assertEqual(num_warns,0)
360        # state has changed
361        self.assertEqual(self.app['students']['X666666'].state,'admitted')
362        # state has not changed
363        self.assertEqual(self.app['students']['Y777777'].state,'courses validated')
364        shutil.rmtree(os.path.dirname(fin_file))
365
366    def test_import_update2(self):
367        num, num_warns, fin_file, fail_file = self.processor.doImport(
368            self.csv_file, STUDENT_HEADER_FIELDS)
369        shutil.rmtree(os.path.dirname(fin_file))
370        num, num_warns, fin_file, fail_file = self.processor.doImport(
371            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
372        self.assertEqual(num_warns,0)
373        # The phone import value of Pieri was None.
374        # Confirm that phone has not been cleared.
375        container = self.app['students']
376        for key in container.keys():
377            if container[key].firstname == 'Aaren':
378                aaren = container[key]
379                break
380        self.assertEqual(aaren.phone, '--1234')
381        # The phone import value of Claus was a deletion marker.
382        # Confirm that phone has been cleared.
383        for key in container.keys():
384            if container[key].firstname == 'Claus':
385                claus = container[key]
386                break
387        assert claus.phone is None
388        shutil.rmtree(os.path.dirname(fin_file))
389
390    def test_import_update3(self):
391        num, num_warns, fin_file, fail_file = self.processor.doImport(
392            self.csv_file, STUDENT_HEADER_FIELDS)
393        shutil.rmtree(os.path.dirname(fin_file))
394        num, num_warns, fin_file, fail_file = self.processor.doImport(
395            self.csv_file_update3, STUDENT_HEADER_FIELDS_UPDATE3, 'update')
396        content = open(fail_file).read()
397        self.assertEqual(
398            content,
399            'reg_number,student_id,transition,--ERRORS--\r\n'
400            '<IGNORE>,X666666,request_clearance,Transition not allowed.\r\n'
401            )
402        self.assertEqual(num_warns,1)
403        self.assertEqual(self.app['students']['Y777777'].state,'returning')
404
405    def test_import_update4(self):
406        num, num_warns, fin_file, fail_file = self.processor.doImport(
407            self.csv_file, STUDENT_HEADER_FIELDS)
408        shutil.rmtree(os.path.dirname(fin_file))
409        self.assertRaises(
410            FatalCSVError, self.processor.doImport, self.csv_file_update4,
411            STUDENT_HEADER_FIELDS_UPDATE4, 'update')
412
413    def test_import_remove(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        num, num_warns, fin_file, fail_file = self.processor.doImport(
418            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
419        self.assertEqual(num_warns,0)
420        shutil.rmtree(os.path.dirname(fin_file))
421
422    def test_import_migration_data(self):
423        num, num_warns, fin_file, fail_file = self.processor.doImport(
424            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
425        content = open(fail_file).read()
426        self.assertEqual(num_warns,2)
427        assert len(self.app['students'].keys()) == 5
428        self.assertEqual(
429            content,
430            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
431            '4,John,D123456,m,aa@aa.ng,1234,nonsense,1990-01-05,Wolter,mypw1,100003,state: not allowed\r\n'
432            '5,John,E123456,x,aa@aa.ng,1234,,1990-01-06,Kennedy,,100004,sex: Invalid value\r\n'
433            )
434        students = self.app['students']
435        self.assertTrue('A123456' in students.keys())
436        self.assertEqual(students['A123456'].state, 'clearance started')
437        self.assertEqual(students['A123456'].date_of_birth,
438                         datetime.date(1990, 1, 2))
439        self.assertFalse(students['A123456'].clearance_locked)
440        self.assertEqual(students['B123456'].state, 'cleared')
441        self.assertEqual(students['B123456'].date_of_birth,
442                         datetime.date(1990, 1, 3))
443        self.assertTrue(students['B123456'].clearance_locked)
444        history = ' '.join(students['A123456'].history.messages)
445        self.assertTrue(
446            "State 'clearance started' set by system" in history)
447        # state was empty and student is thus in state created
448        self.assertEqual(students['F123456'].state,'created')
449        # passwords were set correctly
450        self.assertEqual(
451            IUserAccount(students['A123456']).checkPassword('mypw1'), True)
452        self.assertEqual(
453            IUserAccount(students['C123456']).checkPassword('mypw1'), True)
454        shutil.rmtree(os.path.dirname(fin_file))
455
456    def test_import_duplicate_data(self):
457        num, num_warns, fin_file, fail_file = self.processor.doImport(
458            self.csv_file_duplicates, STUDENT_HEADER_FIELDS_DUPLICATES)
459        content = open(fail_file).read()
460        self.assertEqual(num_warns,3)
461        self.assertEqual(
462            content,
463            'reg_number,firstname,student_id,sex,email,phone,state,date_of_birth,lastname,password,matric_number,--ERRORS--\r\n'
464            '1,Aaren,B123456,m,aa@aa.ng,1234,cleared,1990-01-03,Finau,mypw1,100001,reg_number: reg_number\r\n'
465            '2,Aaren,C123456,m,aa@aa.ng,1234,admitted,1990-01-04,Berson,mypw1,100000,matric_number: matric_number\r\n'
466            '1,Frank,F123456,m,aa@aa.ng,1234,,1990-01-06,Meyer,,100000,reg_number: reg_number; matric_number: matric_number\r\n'
467            )
468        shutil.rmtree(os.path.dirname(fin_file))
469
470class StudentStudyCourseProcessorTest(StudentImportExportSetup):
471
472    def setUp(self):
473        super(StudentStudyCourseProcessorTest, self).setUp()
474
475        # Import students with subobjects
476        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
477        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
478        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
479            student_file, STUDENT_HEADER_FIELDS)
480        shutil.rmtree(os.path.dirname(fin_file))
481
482        self.processor = StudentStudyCourseProcessor()
483        self.csv_file = os.path.join(
484            self.workdir, 'sample_studycourse_data.csv')
485        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
486        return
487
488    def test_interface(self):
489        # Make sure we fulfill the interface contracts.
490        assert verifyObject(IBatchProcessor, self.processor) is True
491        assert verifyClass(
492            IBatchProcessor, StudentStudyCourseProcessor) is True
493
494    def test_entryExists(self):
495        assert self.processor.entryExists(
496            dict(reg_number='REG_NONE'), self.app) is False
497        assert self.processor.entryExists(
498            dict(reg_number='1'), self.app) is True
499
500    def test_getEntry(self):
501        student = self.processor.getEntry(
502            dict(reg_number='1'), self.app).__parent__
503        self.assertEqual(student.reg_number,'1')
504
505    def test_checkConversion(self):
506        errs, inv_errs, conv_dict = self.processor.checkConversion(
507            dict(reg_number='1', certificate='CERT1', current_level='200'))
508        self.assertEqual(len(errs),0)
509        errs, inv_errs, conv_dict = self.processor.checkConversion(
510            dict(reg_number='1', certificate='CERT999'))
511        self.assertEqual(len(errs),1)
512        self.assertTrue(('certificate', u'Invalid value') in errs)
513        errs, inv_errs, conv_dict = self.processor.checkConversion(
514            dict(reg_number='1', certificate='CERT1', current_level='100'))
515        self.assertEqual(len(errs),1)
516        self.assertTrue(('current_level','not in range') in errs)
517        # If we import only current_level, no conversion checking is done.
518        errs, inv_errs, conv_dict = self.processor.checkConversion(
519            dict(reg_number='1', current_level='100'))
520        self.assertEqual(len(errs),0)
521
522    def test_import(self):
523        num, num_warns, fin_file, fail_file = self.processor.doImport(
524            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
525        self.assertEqual(num_warns,1)
526        content = open(fail_file).read()
527        self.assertTrue('current_level: not in range' in content)
528        studycourse = self.processor.getEntry(dict(reg_number='1'), self.app)
529        self.assertEqual(studycourse.certificate.code, u'CERT1')
530        shutil.rmtree(os.path.dirname(fin_file))
531
532class StudentVerdictProcessorTest(StudentImportExportSetup):
533
534
535    def setUp(self):
536        super(StudentVerdictProcessorTest, self).setUp()
537
538        # Import students with subobjects
539        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
540        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
541        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
542            student_file, STUDENT_HEADER_FIELDS)
543        shutil.rmtree(os.path.dirname(fin_file))
544
545        # Update study courses
546        studycourse_file = os.path.join(
547            self.workdir, 'sample_studycourse_data.csv')
548        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
549        processor = StudentStudyCourseProcessor()
550        num, num_warns, fin_file, fail_file = processor.doImport(
551            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
552        shutil.rmtree(os.path.dirname(fin_file))
553
554        self.processor = StudentVerdictProcessor()
555        self.csv_file = os.path.join(
556            self.workdir, 'sample_verdict_data.csv')
557        open(self.csv_file, 'wb').write(VERDICT_SAMPLE_DATA)
558        return
559
560    def test_import(self):
561        num, num_warns, fin_file, fail_file = self.processor.doImport(
562            self.csv_file, VERDICT_HEADER_FIELDS,'update')
563        self.assertEqual(num_warns,3)
564        studycourse = self.processor.getEntry(dict(matric_number='100000'), self.app)
565        student = self.processor.getParent(dict(matric_number='100000'), self.app)
566        self.assertEqual(studycourse.current_verdict, 'A')
567        self.assertEqual(student.state, 'returning')
568        self.assertEqual(studycourse.current_level, 200)
569        content = open(fail_file).read()
570        self.assertEqual(
571            content,
572            'current_session,current_verdict,matric_number,current_level,--ERRORS--\r\n'
573            '2008,B,100001,100,Current level does not correspond.\r\n'
574            '2007,C,100002,200,Current session does not correspond.\r\n'
575            '2008,A,100003,200,Student in wrong state.\r\n'
576            )
577        shutil.rmtree(os.path.dirname(fin_file))
578
579
580class StudentStudyLevelProcessorTest(StudentImportExportSetup):
581
582    def setUp(self):
583        super(StudentStudyLevelProcessorTest, self).setUp()
584
585        # Import students with subobjects
586        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
587        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
588        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
589            student_file, STUDENT_HEADER_FIELDS)
590        shutil.rmtree(os.path.dirname(fin_file))
591
592        # Update study courses
593        studycourse_file = os.path.join(
594            self.workdir, 'sample_studycourse_data.csv')
595        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
596        processor = StudentStudyCourseProcessor()
597        num, num_warns, fin_file, fail_file = processor.doImport(
598            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
599        shutil.rmtree(os.path.dirname(fin_file))
600
601        self.processor = StudentStudyLevelProcessor()
602        self.csv_file = os.path.join(
603            self.workdir, 'sample_studylevel_data.csv')
604        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
605
606    def test_interface(self):
607        # Make sure we fulfill the interface contracts.
608        assert verifyObject(IBatchProcessor, self.processor) is True
609        assert verifyClass(
610            IBatchProcessor, StudentStudyLevelProcessor) is True
611
612    def test_checkConversion(self):
613        errs, inv_errs, conv_dict = self.processor.checkConversion(
614            dict(reg_number='1', level='220'))
615        self.assertEqual(len(errs),0)
616        errs, inv_errs, conv_dict = self.processor.checkConversion(
617            dict(reg_number='1', level='900'))
618        self.assertEqual(len(errs),1)
619        self.assertTrue(('level','no valid integer') in errs)
620        errs, inv_errs, conv_dict = self.processor.checkConversion(
621            dict(reg_number='1', level='xyz'))
622        self.assertEqual(len(errs),1)
623        self.assertTrue(('level','no integer') in errs)
624
625    def test_import(self):
626        num, num_warns, fin_file, fail_file = self.processor.doImport(
627            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
628        self.assertEqual(num_warns,2)
629        assert self.processor.entryExists(
630            dict(reg_number='1', level='100'), self.app) is True
631        studylevel = self.processor.getEntry(
632            dict(reg_number='1', level='100'), self.app)
633        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
634        self.assertEqual(studylevel.level_session, 2008)
635        self.assertEqual(studylevel.level_verdict, 'A')
636        self.assertEqual(studylevel.level, 100)
637        shutil.rmtree(os.path.dirname(fin_file))
638
639    def test_import_update(self):
640        # We perform the same import twice,
641        # the second time in update mode. The number
642        # of warnings must be the same.
643        num, num_warns, fin_file, fail_file = self.processor.doImport(
644            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
645        num, num_warns, fin_file, fail_file = self.processor.doImport(
646            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'update')
647        self.assertEqual(num_warns,2)
648        shutil.rmtree(os.path.dirname(fin_file))
649       
650
651class CourseTicketProcessorTest(StudentImportExportSetup):
652
653    def setUp(self):
654        super(CourseTicketProcessorTest, self).setUp()
655
656        # Import students with subobjects
657        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
658        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
659        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
660            student_file, STUDENT_HEADER_FIELDS)
661        shutil.rmtree(os.path.dirname(fin_file))
662
663        # Add course and course referrer
664        self.course = createObject('waeup.Course')
665        self.course.code = 'COURSE1'
666        self.course.semester = 1
667        self.course.credits = 10
668        self.course.passmark = 40
669        self.app['faculties']['fac1']['dep1'].courses.addCourse(
670            self.course)
671        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
672            self.course, level=100)
673
674        # Update study courses
675        studycourse_file = os.path.join(
676            self.workdir, 'sample_studycourse_data.csv')
677        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
678        processor = StudentStudyCourseProcessor()
679        num, num_warns, fin_file, fail_file = processor.doImport(
680            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
681        shutil.rmtree(os.path.dirname(fin_file))
682
683        # Import study levels
684        processor = StudentStudyLevelProcessor()
685        studylevel_file = os.path.join(
686            self.workdir, 'sample_studylevel_data.csv')
687        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
688        num, num_warns, fin_file, fail_file = processor.doImport(
689            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
690        shutil.rmtree(os.path.dirname(fin_file))
691
692        self.processor = CourseTicketProcessor()
693        self.csv_file = os.path.join(
694            self.workdir, 'sample_courseticket_data.csv')
695        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
696
697    def test_interface(self):
698        # Make sure we fulfill the interface contracts.
699        assert verifyObject(IBatchProcessor, self.processor) is True
700        assert verifyClass(
701            IBatchProcessor, CourseTicketProcessor) is True
702
703    def test_checkConversion(self):
704        errs, inv_errs, conv_dict = self.processor.checkConversion(
705            dict(reg_number='1', code='COURSE1', level='220'))
706        self.assertEqual(len(errs),0)
707        errs, inv_errs, conv_dict = self.processor.checkConversion(
708            dict(reg_number='1', code='COURSE2', level='220'))
709        self.assertEqual(len(errs),1)
710        self.assertTrue(('code','non-existent') in errs)
711
712    def test_import(self):
713
714        num, num_warns, fin_file, fail_file = self.processor.doImport(
715            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
716
717        self.assertEqual(num_warns,2)
718        assert self.processor.entryExists(
719            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
720        courseticket = self.processor.getEntry(
721            dict(reg_number='1', level='100', code='COURSE1'), self.app)
722        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
723        self.assertEqual(courseticket.score, 1)
724        self.assertEqual(courseticket.mandatory, True)
725        self.assertEqual(courseticket.fcode, 'NA')
726        self.assertEqual(courseticket.dcode, 'NA')
727        self.assertEqual(courseticket.code, 'COURSE1')
728        self.assertEqual(courseticket.title, 'Unnamed Course')
729        self.assertEqual(courseticket.credits, 10)
730        self.assertEqual(courseticket.passmark, 40)
731        self.assertEqual(courseticket.semester, 1)
732        shutil.rmtree(os.path.dirname(fin_file))
733
734    def test_import_update(self):
735        # We perform the same import twice,
736        # the second time in update mode. The number
737        # of warnings must be the same.
738        num, num_warns, fin_file, fail_file = self.processor.doImport(
739            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
740        num, num_warns, fin_file, fail_file = self.processor.doImport(
741            self.csv_file, COURSETICKET_HEADER_FIELDS,'update')
742        self.assertEqual(num_warns,2)
743        shutil.rmtree(os.path.dirname(fin_file))
744
745class PaymentProcessorTest(StudentImportExportSetup):
746
747    def setUp(self):
748        super(PaymentProcessorTest, self).setUp()
749
750        # Add student with payment
751        student = Student()
752        student.firstname = u'Anna'
753        student.lastname = u'Tester'
754        student.reg_number = u'123'
755        student.matric_number = u'234'
756        self.app['students'].addStudent(student)
757        self.student = self.app['students'][student.student_id]
758        payment = createObject(u'waeup.StudentOnlinePayment')
759        payment.p_id = 'p123'
760        self.student['payments'][payment.p_id] = payment
761
762        # Import students with subobjects
763        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
764        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
765        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
766            student_file, STUDENT_HEADER_FIELDS)
767        shutil.rmtree(os.path.dirname(fin_file))
768
769        self.processor = StudentOnlinePaymentProcessor()
770        self.csv_file = os.path.join(
771            self.workdir, 'sample_payment_data.csv')
772        open(self.csv_file, 'wb').write(PAYMENT_SAMPLE_DATA)
773
774    def test_interface(self):
775        # Make sure we fulfill the interface contracts.
776        assert verifyObject(IBatchProcessor, self.processor) is True
777        assert verifyClass(
778            IBatchProcessor, StudentOnlinePaymentProcessor) is True
779
780    def test_getEntry(self):
781        assert self.processor.getEntry(
782            dict(student_id='ID_NONE', p_id='nonsense'), self.app) is None
783        assert self.processor.getEntry(
784            dict(student_id=self.student.student_id, p_id='p123'),
785            self.app) is self.student['payments']['p123']
786        assert self.processor.getEntry(
787            dict(student_id=self.student.student_id, p_id='XXXXXX123'),
788            self.app) is self.student['payments']['p123']
789
790    def test_addEntry(self):
791        self.assertEqual(len(self.student['payments'].keys()),1)
792        payment1 = createObject(u'waeup.StudentOnlinePayment')
793        payment1.p_id = 'p234'
794        self.processor.addEntry(
795            payment1, dict(student_id=self.student.student_id, p_id='p234'),
796            self.app)
797        self.assertEqual(len(self.student['payments'].keys()),2)
798        self.assertEqual(self.student['payments']['p234'].p_id, 'p234')
799        payment2 = createObject(u'waeup.StudentOnlinePayment')
800        payment1.p_id = 'nonsense'
801        # payment1.p_id will be replaced if p_id doesn't start with 'p'
802        self.processor.addEntry(
803            payment2, dict(student_id=self.student.student_id, p_id='XXXXXX456'),
804            self.app)
805        self.assertEqual(len(self.student['payments'].keys()),3)
806        self.assertEqual(self.student['payments']['p456'].p_id, 'p456')
807
808    def test_checkConversion(self):
809        errs, inv_errs, conv_dict = self.processor.checkConversion(
810            dict(reg_number='1', p_id='3816951266236341955'))
811        self.assertEqual(len(errs),0)
812        errs, inv_errs, conv_dict = self.processor.checkConversion(
813            dict(reg_number='1', p_id='p1266236341955'))
814        self.assertEqual(len(errs),0)
815        errs, inv_errs, conv_dict = self.processor.checkConversion(
816            dict(reg_number='1', p_id='nonsense'))
817        self.assertEqual(len(errs),1)
818        timestamp = "%d" % int(time()*1000)
819        p_id = "p%s" % timestamp
820        errs, inv_errs, conv_dict = self.processor.checkConversion(
821            dict(reg_number='1', p_id=p_id))
822        self.assertEqual(len(errs),0)
823
824    def test_import(self):
825        num, num_warns, fin_file, fail_file = self.processor.doImport(
826            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
827        self.assertEqual(num_warns,0)
828        payment = self.processor.getEntry(dict(reg_number='1',
829            p_id='p1290797973744'), self.app)
830        self.assertEqual(payment.p_id, 'p1290797973744')
831        cdate = payment.creation_date.strftime("%Y-%m-%d %H:%M:%S")
832        self.assertEqual(cdate, "2010-11-26 18:59:33")
833        self.assertEqual(str(payment.creation_date.tzinfo),'UTC')
834        shutil.rmtree(os.path.dirname(fin_file))
835
836    def test_import_update(self):
837        # We perform the same import twice,
838        # the second time in update mode. The number
839        # of warnings must be the same.
840        num, num_warns, fin_file, fail_file = self.processor.doImport(
841            self.csv_file, PAYMENT_HEADER_FIELDS,'create')
842        num, num_warns, fin_file, fail_file = self.processor.doImport(
843            self.csv_file, PAYMENT_HEADER_FIELDS,'update')
844        self.assertEqual(num_warns,0)
845        shutil.rmtree(os.path.dirname(fin_file))
846
847def test_suite():
848    suite = unittest.TestSuite()
849    for testcase in [
850        StudentProcessorTest,StudentStudyCourseProcessorTest,
851        StudentStudyLevelProcessorTest,CourseTicketProcessorTest,
852        PaymentProcessorTest,StudentVerdictProcessorTest]:
853        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
854                testcase
855                )
856        )
857    return suite
858
859
Note: See TracBrowser for help on using the repository browser.