source: main/waeup.sirp/trunk/src/waeup/sirp/students/tests/test_batching.py @ 7616

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

Implement course ticket importer.

Add checkConversion tests.
  • Property svn:keywords set to Id
File size: 21.7 KB
Line 
1## $Id: test_batching.py 7548 2012-02-01 11:19:56Z 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 importers.
19"""
20import os
21import shutil
22import tempfile
23import unittest
24import datetime
25from zope.component import createObject
26from zope.component.hooks import setSite, clearSite
27from zope.interface.verify import verifyClass, verifyObject
28
29from waeup.sirp.app import University
30from waeup.sirp.university.faculty import Faculty
31from waeup.sirp.university.department import Department
32from waeup.sirp.students.batching import (
33    StudentProcessor, StudentStudyCourseProcessor,
34    StudentStudyLevelProcessor, CourseTicketProcessor)
35from waeup.sirp.students.student import Student
36from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
37from waeup.sirp.interfaces import IBatchProcessor
38
39STUDENT_SAMPLE_DATA = open(
40    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
41    'rb').read()
42
43STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
44    '\n')[0].split(',')
45
46STUDENT_SAMPLE_DATA_UPDATE = open(
47    os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'),
48    'rb').read()
49
50STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split(
51    '\n')[0].split(',')
52
53STUDENT_SAMPLE_DATA_UPDATE2 = open(
54    os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'),
55    'rb').read()
56
57STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split(
58    '\n')[0].split(',')
59
60STUDYCOURSE_SAMPLE_DATA = open(
61    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
62    'rb').read()
63
64STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
65    '\n')[0].split(',')
66
67STUDENT_SAMPLE_DATA_MIGRATION = open(
68    os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'),
69    'rb').read()
70
71STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split(
72    '\n')[0].split(',')
73
74STUDYLEVEL_SAMPLE_DATA = open(
75    os.path.join(os.path.dirname(__file__), 'sample_studylevel_data.csv'),
76    'rb').read()
77
78STUDYLEVEL_HEADER_FIELDS = STUDYLEVEL_SAMPLE_DATA.split(
79    '\n')[0].split(',')
80
81COURSETICKET_SAMPLE_DATA = open(
82    os.path.join(os.path.dirname(__file__), 'sample_courseticket_data.csv'),
83    'rb').read()
84
85COURSETICKET_HEADER_FIELDS = COURSETICKET_SAMPLE_DATA.split(
86    '\n')[0].split(',')
87
88class StudentImporterTest(FunctionalTestCase):
89
90    layer = FunctionalLayer
91
92    def setUp(self):
93        super(StudentImporterTest, self).setUp()
94        # Setup a sample site for each test
95        app = University()
96        self.dc_root = tempfile.mkdtemp()
97        app['datacenter'].setStoragePath(self.dc_root)
98
99        # Prepopulate the ZODB...
100        self.getRootFolder()['app'] = app
101        # we add the site immediately after creation to the
102        # ZODB. Catalogs and other local utilities are not setup
103        # before that step.
104        self.app = self.getRootFolder()['app']
105        # Set site here. Some of the following setup code might need
106        # to access grok.getSite() and should get our new app then
107        setSite(app)
108
109        # Add student with subobjects
110        student = Student()
111        student.firstname = u'Anna'
112        student.lastname = u'Tester'
113        student.reg_number = u'123'
114        student.matric_number = u'234'
115        self.app['students'].addStudent(student)
116        self.student = self.app['students'][student.student_id]
117        self.importer = StudentProcessor()
118        self.workdir = tempfile.mkdtemp()
119        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
120        self.csv_file_update = os.path.join(
121            self.workdir, 'sample_student_data_update.csv')
122        self.csv_file_update2 = os.path.join(
123            self.workdir, 'sample_student_data_update2.csv')
124        self.csv_file_migration = os.path.join(
125            self.workdir, 'sample_student_data_migration.csv')
126        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
127        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
128        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
129        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
130
131    def tearDown(self):
132        super(StudentImporterTest, self).tearDown()
133        shutil.rmtree(self.workdir)
134        shutil.rmtree(self.dc_root)
135        clearSite()
136        return
137
138    def test_interface(self):
139        # Make sure we fulfill the interface contracts.
140        assert verifyObject(IBatchProcessor, self.importer) is True
141        assert verifyClass(
142            IBatchProcessor, StudentProcessor) is True
143
144    def test_parentsExist(self):
145        self.assertFalse(self.importer.parentsExist(None, dict()))
146        self.assertTrue(self.importer.parentsExist(None, self.app))
147
148    def test_entryExists(self):
149        assert self.importer.entryExists(
150            dict(student_id='ID_NONE'), self.app) is False
151        assert self.importer.entryExists(
152            dict(reg_number='123'), self.app) is True
153
154    def test_getParent(self):
155        parent = self.importer.getParent(None, self.app)
156        assert parent is self.app['students']
157
158    def test_getEntry(self):
159        assert self.importer.getEntry(
160            dict(student_id='ID_NONE'), self.app) is None
161        assert self.importer.getEntry(
162            dict(student_id=self.student.student_id), self.app) is self.student
163
164    def test_addEntry(self):
165        new_student = Student()
166        self.importer.addEntry(
167            new_student, dict(), self.app)
168        assert len(self.app['students'].keys()) == 2
169
170    def test_checkConversion(self):
171        errs, inv_errs, conv_dict = self.importer.checkConversion(
172            dict(reg_number='1', reg_state='admitted'))
173        self.assertEqual(len(errs),0)
174        errs, inv_errs, conv_dict = self.importer.checkConversion(
175            dict(reg_number='1', reg_state=''))
176        self.assertEqual(len(errs),1)
177        self.assertTrue(('reg_state', 'no value provided') in errs)
178        errs, inv_errs, conv_dict = self.importer.checkConversion(
179            dict(reg_number='1', reg_state='nonsense'))
180        self.assertEqual(len(errs),1)
181        self.assertTrue(('reg_state', 'not allowed') in errs)
182
183    def test_delEntry(self):
184        assert self.student.student_id in self.app['students'].keys()
185        self.importer.delEntry(
186            dict(reg_number=self.student.reg_number), self.app)
187        assert self.student.student_id not in self.app['students'].keys()
188
189    def test_import(self):
190        num, num_warns, fin_file, fail_file = self.importer.doImport(
191            self.csv_file, STUDENT_HEADER_FIELDS)
192        self.assertEqual(num_warns,0)
193        assert len(self.app['students'].keys()) == 4
194        shutil.rmtree(os.path.dirname(fin_file))
195
196    def test_import_update(self):
197        num, num_warns, fin_file, fail_file = self.importer.doImport(
198            self.csv_file, STUDENT_HEADER_FIELDS)
199        shutil.rmtree(os.path.dirname(fin_file))
200        num, num_warns, fin_file, fail_file = self.importer.doImport(
201            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
202        self.assertEqual(num_warns,0)
203        shutil.rmtree(os.path.dirname(fin_file))
204
205    def test_import_update2(self):
206        num, num_warns, fin_file, fail_file = self.importer.doImport(
207            self.csv_file, STUDENT_HEADER_FIELDS)
208        shutil.rmtree(os.path.dirname(fin_file))
209        num, num_warns, fin_file, fail_file = self.importer.doImport(
210            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
211        self.assertEqual(num_warns,0)
212        shutil.rmtree(os.path.dirname(fin_file))
213
214    def test_import_remove(self):
215        num, num_warns, fin_file, fail_file = self.importer.doImport(
216            self.csv_file, STUDENT_HEADER_FIELDS)
217        shutil.rmtree(os.path.dirname(fin_file))
218        num, num_warns, fin_file, fail_file = self.importer.doImport(
219            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
220        self.assertEqual(num_warns,0)
221        shutil.rmtree(os.path.dirname(fin_file))
222
223    def test_import_migration_data(self):
224        num, num_warns, fin_file, fail_file = self.importer.doImport(
225            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
226        self.assertEqual(num_warns,2)
227        assert len(self.app['students'].keys()) == 4
228        self.assertTrue('A123456' in self.app['students'].keys())
229        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
230        self.assertEqual(self.app['students']['A123456'].date_of_birth,
231            datetime.date(1990, 1, 2))
232        self.assertFalse(self.app['students']['A123456'].clearance_locked)
233        self.assertEqual(self.app['students']['B123456'].state,'cleared')
234        self.assertEqual(self.app['students']['B123456'].date_of_birth,
235            datetime.date(1990, 1, 3))
236        self.assertTrue(self.app['students']['B123456'].clearance_locked)
237        history = ' '.join(self.app['students']['A123456'].history.messages)
238        self.assertTrue(
239            "State 'clearance started' set by system" in history)
240        shutil.rmtree(os.path.dirname(fin_file))
241
242
243class StudentStudyCourseImporterTest(FunctionalTestCase):
244
245    layer = FunctionalLayer
246
247    def setUp(self):
248        super(StudentStudyCourseImporterTest, self).setUp()
249        self.dc_root = tempfile.mkdtemp()
250        self.workdir = tempfile.mkdtemp()
251        app = University()
252        app['datacenter'].setStoragePath(self.dc_root)
253        self.getRootFolder()['app'] = app
254        self.app = self.getRootFolder()['app']
255        setSite(app)
256
257        # Import students with subobjects
258        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
259        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
260        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
261            student_file, STUDENT_HEADER_FIELDS)
262        shutil.rmtree(os.path.dirname(fin_file))
263
264        # Populate university
265        self.certificate = createObject('waeup.Certificate')
266        self.certificate.code = 'CERT1'
267        self.certificate.application_category = 'basic'
268        self.certificate.start_level = 200
269        self.certificate.end_level = 500
270        self.app['faculties']['fac1'] = Faculty()
271        self.app['faculties']['fac1']['dep1'] = Department()
272        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
273            self.certificate)
274
275        self.importer = StudentStudyCourseProcessor()
276        self.csv_file = os.path.join(
277            self.workdir, 'sample_studycourse_data.csv')
278        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
279        return
280
281    def tearDown(self):
282        super(StudentStudyCourseImporterTest, self).tearDown()
283        shutil.rmtree(self.workdir)
284        shutil.rmtree(self.dc_root)
285        clearSite()
286        return
287
288    def test_interface(self):
289        # Make sure we fulfill the interface contracts.
290        assert verifyObject(IBatchProcessor, self.importer) is True
291        assert verifyClass(
292            IBatchProcessor, StudentStudyCourseProcessor) is True
293
294    def test_entryExists(self):
295        assert self.importer.entryExists(
296            dict(reg_number='REG_NONE'), self.app) is False
297        assert self.importer.entryExists(
298            dict(reg_number='1'), self.app) is True
299
300    def test_getEntry(self):
301        student = self.importer.getEntry(
302            dict(reg_number='1'), self.app).__parent__
303        self.assertEqual(student.reg_number,'1')
304
305    def test_checkConversion(self):
306        errs, inv_errs, conv_dict = self.importer.checkConversion(
307            dict(reg_number='1', certificate='CERT1', current_level='200'))
308        self.assertEqual(len(errs),0)
309        errs, inv_errs, conv_dict = self.importer.checkConversion(
310            dict(reg_number='1', certificate='CERT999'))
311        self.assertEqual(len(errs),1)
312        self.assertTrue(('certificate', u'Invalid value') in errs)
313        errs, inv_errs, conv_dict = self.importer.checkConversion(
314            dict(reg_number='1', certificate='CERT1', current_level='100'))
315        self.assertEqual(len(errs),1)
316        self.assertTrue(('current_level','not in range') in errs)
317        # If we import only current_level, no conversion checking is done.
318        errs, inv_errs, conv_dict = self.importer.checkConversion(
319            dict(reg_number='1', current_level='100'))
320        self.assertEqual(len(errs),0)
321
322    def test_import(self):
323        num, num_warns, fin_file, fail_file = self.importer.doImport(
324            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
325        self.assertEqual(num_warns,1)
326        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
327        self.assertEqual(studycourse.certificate.code, u'CERT1')
328        shutil.rmtree(os.path.dirname(fin_file))
329
330class StudentStudyLevelImporterTest(FunctionalTestCase):
331
332    layer = FunctionalLayer
333
334    def setUp(self):
335        super(StudentStudyLevelImporterTest, self).setUp()
336        self.dc_root = tempfile.mkdtemp()
337        self.workdir = tempfile.mkdtemp()
338        app = University()
339        app['datacenter'].setStoragePath(self.dc_root)
340        self.getRootFolder()['app'] = app
341        self.app = self.getRootFolder()['app']
342        setSite(app)
343
344        # Import students with subobjects
345        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
346        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
347        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
348            student_file, STUDENT_HEADER_FIELDS)
349        shutil.rmtree(os.path.dirname(fin_file))
350
351        # Populate university
352        self.certificate = createObject('waeup.Certificate')
353        self.certificate.code = 'CERT1'
354        self.certificate.application_category = 'basic'
355        self.certificate.start_level = 200
356        self.certificate.end_level = 500
357        self.app['faculties']['fac1'] = Faculty()
358        self.app['faculties']['fac1']['dep1'] = Department()
359        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
360            self.certificate)
361
362        # Update study courses
363        studycourse_file = os.path.join(
364            self.workdir, 'sample_studycourse_data.csv')
365        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
366        importer = StudentStudyCourseProcessor()
367        num, num_warns, fin_file, fail_file = importer.doImport(
368            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
369        shutil.rmtree(os.path.dirname(fin_file))
370
371        self.importer = StudentStudyLevelProcessor()
372        self.csv_file = os.path.join(
373            self.workdir, 'sample_studylevel_data.csv')
374        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
375
376    def tearDown(self):
377        super(StudentStudyLevelImporterTest, self).tearDown()
378        shutil.rmtree(self.workdir)
379        shutil.rmtree(self.dc_root)
380        clearSite()
381        return
382
383    def test_interface(self):
384        # Make sure we fulfill the interface contracts.
385        assert verifyObject(IBatchProcessor, self.importer) is True
386        assert verifyClass(
387            IBatchProcessor, StudentStudyLevelProcessor) is True
388
389    def test_checkConversion(self):
390        errs, inv_errs, conv_dict = self.importer.checkConversion(
391            dict(reg_number='1', level='220'))
392        self.assertEqual(len(errs),0)
393        errs, inv_errs, conv_dict = self.importer.checkConversion(
394            dict(reg_number='1', level='900'))
395        self.assertEqual(len(errs),1)
396        self.assertTrue(('level','no valid integer') in errs)
397        errs, inv_errs, conv_dict = self.importer.checkConversion(
398            dict(reg_number='1', level='xyz'))
399        self.assertEqual(len(errs),1)
400        self.assertTrue(('level','no integer') in errs)
401
402    def test_import(self):
403        num, num_warns, fin_file, fail_file = self.importer.doImport(
404            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
405        self.assertEqual(num_warns,2)
406        assert self.importer.entryExists(
407            dict(reg_number='1', level='100'), self.app) is True
408        studylevel = self.importer.getEntry(
409            dict(reg_number='1', level='100'), self.app)
410        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
411        self.assertEqual(studylevel.level_session, 2008)
412        self.assertEqual(studylevel.level_verdict, 'A')
413        self.assertEqual(studylevel.level, 100)
414        shutil.rmtree(os.path.dirname(fin_file))
415       
416
417class CourseTicketImporterTest(FunctionalTestCase):
418
419    layer = FunctionalLayer
420
421    def setUp(self):
422        super(CourseTicketImporterTest, self).setUp()
423        self.dc_root = tempfile.mkdtemp()
424        self.workdir = tempfile.mkdtemp()
425        app = University()
426        app['datacenter'].setStoragePath(self.dc_root)
427        self.getRootFolder()['app'] = app
428        self.app = self.getRootFolder()['app']
429        setSite(app)
430
431        # Import students with subobjects
432        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
433        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
434        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
435            student_file, STUDENT_HEADER_FIELDS)
436        shutil.rmtree(os.path.dirname(fin_file))
437
438        # Populate university
439        self.certificate = createObject('waeup.Certificate')
440        self.certificate.code = 'CERT1'
441        self.certificate.application_category = 'basic'
442        self.certificate.start_level = 200
443        self.certificate.end_level = 500
444        self.app['faculties']['fac1'] = Faculty()
445        self.app['faculties']['fac1']['dep1'] = Department()
446        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
447            self.certificate)
448        self.course = createObject('waeup.Course')
449        self.course.code = 'COURSE1'
450        self.course.semester = 1
451        self.course.credits = 10
452        self.course.passmark = 40
453        self.app['faculties']['fac1']['dep1'].courses.addCourse(
454            self.course)
455        self.app['faculties']['fac1']['dep1'].certificates['CERT1'].addCourseRef(
456            self.course, level=100)
457
458        # Update study courses
459        studycourse_file = os.path.join(
460            self.workdir, 'sample_studycourse_data.csv')
461        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
462        importer = StudentStudyCourseProcessor()
463        num, num_warns, fin_file, fail_file = importer.doImport(
464            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
465        shutil.rmtree(os.path.dirname(fin_file))
466
467        # Import study levels
468        importer = StudentStudyLevelProcessor()
469        studylevel_file = os.path.join(
470            self.workdir, 'sample_studylevel_data.csv')
471        open(studylevel_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
472        num, num_warns, fin_file, fail_file = importer.doImport(
473            studylevel_file, STUDYLEVEL_HEADER_FIELDS,'create')
474        shutil.rmtree(os.path.dirname(fin_file))
475
476        self.importer = CourseTicketProcessor()
477        self.csv_file = os.path.join(
478            self.workdir, 'sample_courseticket_data.csv')
479        open(self.csv_file, 'wb').write(COURSETICKET_SAMPLE_DATA)
480
481    def tearDown(self):
482        super(CourseTicketImporterTest, self).tearDown()
483        shutil.rmtree(self.workdir)
484        shutil.rmtree(self.dc_root)
485        clearSite()
486        return
487
488    def test_interface(self):
489        # Make sure we fulfill the interface contracts.
490        assert verifyObject(IBatchProcessor, self.importer) is True
491        assert verifyClass(
492            IBatchProcessor, CourseTicketProcessor) is True
493
494    def test_checkConversion(self):
495        errs, inv_errs, conv_dict = self.importer.checkConversion(
496            dict(reg_number='1', code='COURSE1', level='220'))
497        self.assertEqual(len(errs),0)
498        errs, inv_errs, conv_dict = self.importer.checkConversion(
499            dict(reg_number='1', code='COURSE2', level='220'))
500        self.assertEqual(len(errs),1)
501        self.assertTrue(('code','non-existent') in errs)
502
503    def test_import(self):
504
505        num, num_warns, fin_file, fail_file = self.importer.doImport(
506            self.csv_file, COURSETICKET_HEADER_FIELDS,'create')
507
508        self.assertEqual(num_warns,2)
509        assert self.importer.entryExists(
510            dict(reg_number='1', level='100', code='COURSE1'), self.app) is True
511        courseticket = self.importer.getEntry(
512            dict(reg_number='1', level='100', code='COURSE1'), self.app)
513        self.assertEqual(courseticket.__parent__.__parent__.certificate.code, u'CERT1')
514        self.assertEqual(courseticket.score, 1)
515        self.assertEqual(courseticket.core_or_elective, True)
516        self.assertEqual(courseticket.fcode, 'NA')
517        self.assertEqual(courseticket.dcode, 'NA')
518        self.assertEqual(courseticket.code, 'COURSE1')
519        self.assertEqual(courseticket.title, 'Unnamed Course')
520        self.assertEqual(courseticket.credits, 10)
521        self.assertEqual(courseticket.passmark, 40)
522        self.assertEqual(courseticket.semester, 1)
523        #import pdb; pdb.set_trace()
524        shutil.rmtree(os.path.dirname(fin_file))
525
526
527def test_suite():
528    suite = unittest.TestSuite()
529    for testcase in [
530        StudentImporterTest,StudentStudyCourseImporterTest,
531        StudentStudyLevelImporterTest,CourseTicketImporterTest,]:
532        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
533                testcase
534                )
535        )
536    return suite
537
538
Note: See TracBrowser for help on using the repository browser.