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

Last change on this file since 8524 was 8491, checked in by Henrik Bettermann, 13 years ago

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