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

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

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