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

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

Implement study level importer.
  • Property svn:keywords set to Id
File size: 14.8 KB
Line 
1## $Id: test_batching.py 7536 2012-01-30 07:41:17Z 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)
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
81class StudentImporterTest(FunctionalTestCase):
82
83    layer = FunctionalLayer
84
85    def setUp(self):
86        super(StudentImporterTest, self).setUp()
87        # Setup a sample site for each test
88        app = University()
89        self.dc_root = tempfile.mkdtemp()
90        app['datacenter'].setStoragePath(self.dc_root)
91
92        # Prepopulate the ZODB...
93        self.getRootFolder()['app'] = app
94        # we add the site immediately after creation to the
95        # ZODB. Catalogs and other local utilities are not setup
96        # before that step.
97        self.app = self.getRootFolder()['app']
98        # Set site here. Some of the following setup code might need
99        # to access grok.getSite() and should get our new app then
100        setSite(app)
101
102        # Add student with subobjects
103        student = Student()
104        student.firstname = u'Anna'
105        student.lastname = u'Tester'
106        student.reg_number = u'123'
107        student.matric_number = u'234'
108        self.app['students'].addStudent(student)
109        self.student = self.app['students'][student.student_id]
110        self.importer = StudentProcessor()
111        self.workdir = tempfile.mkdtemp()
112        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
113        self.csv_file_update = os.path.join(
114            self.workdir, 'sample_student_data_update.csv')
115        self.csv_file_update2 = os.path.join(
116            self.workdir, 'sample_student_data_update2.csv')
117        self.csv_file_migration = os.path.join(
118            self.workdir, 'sample_student_data_migration.csv')
119        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
120        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
121        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
122        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
123
124    def tearDown(self):
125        super(StudentImporterTest, self).tearDown()
126        shutil.rmtree(self.workdir)
127        shutil.rmtree(self.dc_root)
128        clearSite()
129        return
130
131    def test_interface(self):
132        # Make sure we fulfill the interface contracts.
133        assert verifyObject(IBatchProcessor, self.importer) is True
134        assert verifyClass(
135            IBatchProcessor, StudentProcessor) is True
136
137    def test_parentsExist(self):
138        self.assertFalse(self.importer.parentsExist(None, dict()))
139        self.assertTrue(self.importer.parentsExist(None, self.app))
140
141    def test_entryExists(self):
142        assert self.importer.entryExists(
143            dict(student_id='ID_NONE'), self.app) is False
144        assert self.importer.entryExists(
145            dict(reg_number='123'), self.app) is True
146
147    def test_getParent(self):
148        parent = self.importer.getParent(None, self.app)
149        assert parent is self.app['students']
150
151    def test_getEntry(self):
152        assert self.importer.getEntry(
153            dict(student_id='ID_NONE'), self.app) is None
154        assert self.importer.getEntry(
155            dict(student_id=self.student.student_id), self.app) is self.student
156
157    def test_addEntry(self):
158        new_student = Student()
159        self.importer.addEntry(
160            new_student, dict(), self.app)
161        assert len(self.app['students'].keys()) == 2
162
163    def test_delEntry(self):
164        assert self.student.student_id in self.app['students'].keys()
165        self.importer.delEntry(
166            dict(reg_number=self.student.reg_number), self.app)
167        assert self.student.student_id not in self.app['students'].keys()
168
169    def test_import(self):
170        num, num_warns, fin_file, fail_file = self.importer.doImport(
171            self.csv_file, STUDENT_HEADER_FIELDS)
172        self.assertEqual(num_warns,0)
173        assert len(self.app['students'].keys()) == 4
174        shutil.rmtree(os.path.dirname(fin_file))
175
176    def test_import_update(self):
177        num, num_warns, fin_file, fail_file = self.importer.doImport(
178            self.csv_file, STUDENT_HEADER_FIELDS)
179        shutil.rmtree(os.path.dirname(fin_file))
180        num, num_warns, fin_file, fail_file = self.importer.doImport(
181            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
182        self.assertEqual(num_warns,0)
183        shutil.rmtree(os.path.dirname(fin_file))
184
185    def test_import_update2(self):
186        num, num_warns, fin_file, fail_file = self.importer.doImport(
187            self.csv_file, STUDENT_HEADER_FIELDS)
188        shutil.rmtree(os.path.dirname(fin_file))
189        num, num_warns, fin_file, fail_file = self.importer.doImport(
190            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
191        self.assertEqual(num_warns,0)
192        shutil.rmtree(os.path.dirname(fin_file))
193
194    def test_import_remove(self):
195        num, num_warns, fin_file, fail_file = self.importer.doImport(
196            self.csv_file, STUDENT_HEADER_FIELDS)
197        shutil.rmtree(os.path.dirname(fin_file))
198        num, num_warns, fin_file, fail_file = self.importer.doImport(
199            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
200        self.assertEqual(num_warns,0)
201        shutil.rmtree(os.path.dirname(fin_file))
202
203    def test_import_migration_data(self):
204        num, num_warns, fin_file, fail_file = self.importer.doImport(
205            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
206        self.assertEqual(num_warns,2)
207        assert len(self.app['students'].keys()) == 4
208        self.assertTrue('A123456' in self.app['students'].keys())
209        self.assertEqual(self.app['students']['A123456'].state,'clearance started')
210        self.assertEqual(self.app['students']['A123456'].date_of_birth,
211            datetime.date(1990, 1, 2))
212        self.assertFalse(self.app['students']['A123456'].clearance_locked)
213        self.assertEqual(self.app['students']['B123456'].state,'cleared')
214        self.assertEqual(self.app['students']['B123456'].date_of_birth,
215            datetime.date(1990, 1, 3))
216        self.assertTrue(self.app['students']['B123456'].clearance_locked)
217        history = ' '.join(self.app['students']['A123456'].history.messages)
218        self.assertTrue(
219            "State 'clearance started' set by system" in history)
220        shutil.rmtree(os.path.dirname(fin_file))
221
222
223class StudentStudyCourseImporterTest(FunctionalTestCase):
224
225    layer = FunctionalLayer
226
227    def setUp(self):
228        super(StudentStudyCourseImporterTest, self).setUp()
229        self.dc_root = tempfile.mkdtemp()
230        self.workdir = tempfile.mkdtemp()
231        app = University()
232        app['datacenter'].setStoragePath(self.dc_root)
233        self.getRootFolder()['app'] = app
234        self.app = self.getRootFolder()['app']
235        setSite(app)
236
237        # Import students with subobjects
238        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
239        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
240        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
241            student_file, STUDENT_HEADER_FIELDS)
242        shutil.rmtree(os.path.dirname(fin_file))
243
244        # Populate university
245        self.certificate = createObject('waeup.Certificate')
246        self.certificate.code = 'CERT1'
247        self.certificate.application_category = 'basic'
248        self.certificate.start_level = 200
249        self.certificate.end_level = 500
250        self.app['faculties']['fac1'] = Faculty()
251        self.app['faculties']['fac1']['dep1'] = Department()
252        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
253            self.certificate)
254
255        self.importer = StudentStudyCourseProcessor()
256        self.csv_file = os.path.join(
257            self.workdir, 'sample_studycourse_data.csv')
258        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
259        return
260
261    def tearDown(self):
262        super(StudentStudyCourseImporterTest, self).tearDown()
263        shutil.rmtree(self.workdir)
264        shutil.rmtree(self.dc_root)
265        clearSite()
266        return
267
268    def test_interface(self):
269        # Make sure we fulfill the interface contracts.
270        assert verifyObject(IBatchProcessor, self.importer) is True
271        assert verifyClass(
272            IBatchProcessor, StudentStudyCourseProcessor) is True
273
274    def test_entryExists(self):
275        assert self.importer.entryExists(
276            dict(reg_number='REG_NONE'), self.app) is False
277        assert self.importer.entryExists(
278            dict(reg_number='1'), self.app) is True
279
280    def test_getEntry(self):
281        student = self.importer.getEntry(
282            dict(reg_number='1'), self.app).__parent__
283        self.assertEqual(student.reg_number,'1')
284
285    def test_import(self):
286        num, num_warns, fin_file, fail_file = self.importer.doImport(
287            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
288        self.assertEqual(num_warns,1)
289        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
290        self.assertEqual(studycourse.certificate.code, u'CERT1')
291        shutil.rmtree(os.path.dirname(fin_file))
292
293class StudentStudyLevelImporterTest(FunctionalTestCase):
294
295    layer = FunctionalLayer
296
297    def setUp(self):
298        super(StudentStudyLevelImporterTest, self).setUp()
299        self.dc_root = tempfile.mkdtemp()
300        self.workdir = tempfile.mkdtemp()
301        app = University()
302        app['datacenter'].setStoragePath(self.dc_root)
303        self.getRootFolder()['app'] = app
304        self.app = self.getRootFolder()['app']
305        setSite(app)
306
307        # Import students with subobjects
308        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
309        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
310        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
311            student_file, STUDENT_HEADER_FIELDS)
312        shutil.rmtree(os.path.dirname(fin_file))
313
314        # Populate university
315        self.certificate = createObject('waeup.Certificate')
316        self.certificate.code = 'CERT1'
317        self.certificate.application_category = 'basic'
318        self.certificate.start_level = 200
319        self.certificate.end_level = 500
320        self.app['faculties']['fac1'] = Faculty()
321        self.app['faculties']['fac1']['dep1'] = Department()
322        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
323            self.certificate)
324
325        # Update study courses
326        studycourse_file = os.path.join(
327            self.workdir, 'sample_studycourse_data.csv')
328        open(studycourse_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
329        importer = StudentStudyCourseProcessor()
330        num, num_warns, fin_file, fail_file = importer.doImport(
331            studycourse_file, STUDYCOURSE_HEADER_FIELDS,'update')
332        shutil.rmtree(os.path.dirname(fin_file))
333
334        self.importer = StudentStudyLevelProcessor()
335        self.csv_file = os.path.join(
336            self.workdir, 'sample_studylevel_data.csv')
337        open(self.csv_file, 'wb').write(STUDYLEVEL_SAMPLE_DATA)
338
339    def tearDown(self):
340        super(StudentStudyLevelImporterTest, self).tearDown()
341        shutil.rmtree(self.workdir)
342        shutil.rmtree(self.dc_root)
343        clearSite()
344        return
345
346    def test_interface(self):
347        # Make sure we fulfill the interface contracts.
348        assert verifyObject(IBatchProcessor, self.importer) is True
349        assert verifyClass(
350            IBatchProcessor, StudentStudyLevelProcessor) is True
351
352    def test_import(self):
353        num, num_warns, fin_file, fail_file = self.importer.doImport(
354            self.csv_file, STUDYLEVEL_HEADER_FIELDS,'create')
355        self.assertEqual(num_warns,2)
356        assert self.importer.entryExists(
357            dict(reg_number='1', level='100'), self.app) is True
358        studylevel = self.importer.getEntry(
359            dict(reg_number='1', level='100'), self.app)
360        self.assertEqual(studylevel.__parent__.certificate.code, u'CERT1')
361        self.assertEqual(studylevel.level_session, 2008)
362        self.assertEqual(studylevel.level_verdict, 'A')
363        self.assertEqual(studylevel.level, 100)
364        shutil.rmtree(os.path.dirname(fin_file))
365       
366
367def test_suite():
368    suite = unittest.TestSuite()
369    for testcase in [
370        StudentImporterTest,StudentStudyCourseImporterTest,
371        StudentStudyLevelImporterTest,]:
372        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
373                testcase
374                )
375        )
376    return suite
377
378
Note: See TracBrowser for help on using the repository browser.