source: main/waeup.kofa/branches/uli-async-update/src/waeup/kofa/students/tests/test_batching.py @ 9168

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

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