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

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

Add test for student data migration to be sure that student_ids provided in import files are correctly produced and that the random id generator is thus neutralized.
  • Property svn:keywords set to Id
File size: 11.0 KB
Line 
1## $Id: test_batching.py 7273 2011-12-04 21:08:53Z 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
24from zope.component import createObject
25from zope.component.hooks import setSite, clearSite
26from zope.interface.verify import verifyClass, verifyObject
27
28from waeup.sirp.app import University
29from waeup.sirp.university.faculty import Faculty
30from waeup.sirp.university.department import Department
31from waeup.sirp.students.batching import (
32    StudentProcessor, StudentStudyCourseProcessor)
33from waeup.sirp.students.student import Student
34from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
35from waeup.sirp.interfaces import IBatchProcessor
36
37STUDENT_SAMPLE_DATA = open(
38    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
39    'rb').read()
40
41STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
42    '\n')[0].split(',')
43
44STUDENT_SAMPLE_DATA_UPDATE = open(
45    os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'),
46    'rb').read()
47
48STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split(
49    '\n')[0].split(',')
50
51STUDENT_SAMPLE_DATA_UPDATE2 = open(
52    os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'),
53    'rb').read()
54
55STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split(
56    '\n')[0].split(',')
57
58STUDYCOURSE_SAMPLE_DATA = open(
59    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
60    'rb').read()
61
62STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
63    '\n')[0].split(',')
64
65STUDENT_SAMPLE_DATA_MIGRATION = open(
66    os.path.join(os.path.dirname(__file__), 'sample_student_data_migration.csv'),
67    'rb').read()
68
69STUDENT_HEADER_FIELDS_MIGRATION = STUDENT_SAMPLE_DATA_MIGRATION.split(
70    '\n')[0].split(',')
71
72class StudentImporterTest(FunctionalTestCase):
73
74    layer = FunctionalLayer
75
76    def setUp(self):
77        super(StudentImporterTest, self).setUp()
78        # Setup a sample site for each test
79        app = University()
80        self.dc_root = tempfile.mkdtemp()
81        app['datacenter'].setStoragePath(self.dc_root)
82
83        # Prepopulate the ZODB...
84        self.getRootFolder()['app'] = app
85        # we add the site immediately after creation to the
86        # ZODB. Catalogs and other local utilities are not setup
87        # before that step.
88        self.app = self.getRootFolder()['app']
89        # Set site here. Some of the following setup code might need
90        # to access grok.getSite() and should get our new app then
91        setSite(app)
92
93        # Add student with subobjects
94        student = Student()
95        student.fullname = u'Anna Tester'
96        student.reg_number = u'123'
97        student.matric_number = u'234'
98        self.app['students'].addStudent(student)
99        self.student = self.app['students'][student.student_id]
100        self.importer = StudentProcessor()
101        self.workdir = tempfile.mkdtemp()
102        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
103        self.csv_file_update = os.path.join(
104            self.workdir, 'sample_student_data_update.csv')
105        self.csv_file_update2 = os.path.join(
106            self.workdir, 'sample_student_data_update2.csv')
107        self.csv_file_migration = os.path.join(
108            self.workdir, 'sample_student_data_migration.csv')
109        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
110        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
111        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
112        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
113
114    def tearDown(self):
115        super(StudentImporterTest, self).tearDown()
116        shutil.rmtree(self.workdir)
117        shutil.rmtree(self.dc_root)
118        clearSite()
119        return
120
121    def test_interface(self):
122        # Make sure we fulfill the interface contracts.
123        assert verifyObject(IBatchProcessor, self.importer) is True
124        assert verifyClass(
125            IBatchProcessor, StudentProcessor) is True
126
127    def test_parentsExist(self):
128        self.assertFalse(self.importer.parentsExist(None, dict()))
129        self.assertTrue(self.importer.parentsExist(None, self.app))
130
131    def test_entryExists(self):
132        self.assertFalse(self.importer.entryExists(
133            dict(student_id='ID_NONE'), self.app))
134        student = self.importer.getEntry(
135            dict(student_id=self.student.student_id), self.app)
136        self.assertEqual(student.reg_number, u'123')
137
138    def test_getParent(self):
139        parent = self.importer.getParent(None, self.app)
140        assert parent is self.app['students']
141
142    def test_getEntry(self):
143        assert self.importer.getEntry(
144            dict(student_id='ID_NONE'), self.app) is None
145        assert self.importer.getEntry(
146            dict(student_id=self.student.student_id), self.app) is self.student
147
148    def test_addEntry(self):
149        new_student = Student()
150        self.importer.addEntry(
151            new_student, dict(), self.app)
152        assert len(self.app['students'].keys()) == 2
153
154    def test_delEntry(self):
155        assert self.student.student_id in self.app['students'].keys()
156        self.importer.delEntry(
157            dict(reg_number=self.student.reg_number), self.app)
158        assert self.student.student_id not in self.app['students'].keys()
159
160    def test_import(self):
161        num, num_warns, fin_file, fail_file = self.importer.doImport(
162            self.csv_file, STUDENT_HEADER_FIELDS)
163        self.assertEqual(num_warns,0)
164        assert len(self.app['students'].keys()) == 4
165        shutil.rmtree(os.path.dirname(fin_file))
166
167    def test_import_update(self):
168        num, num_warns, fin_file, fail_file = self.importer.doImport(
169            self.csv_file, STUDENT_HEADER_FIELDS)
170        shutil.rmtree(os.path.dirname(fin_file))
171        num, num_warns, fin_file, fail_file = self.importer.doImport(
172            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
173        self.assertEqual(num_warns,0)
174        shutil.rmtree(os.path.dirname(fin_file))
175
176    def test_import_update2(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_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
182        self.assertEqual(num_warns,0)
183        shutil.rmtree(os.path.dirname(fin_file))
184
185    def test_import_remove(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_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
191        self.assertEqual(num_warns,0)
192        shutil.rmtree(os.path.dirname(fin_file))
193
194    def test_import_migration_data(self):
195        num, num_warns, fin_file, fail_file = self.importer.doImport(
196            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
197        self.assertEqual(num_warns,0)
198        assert len(self.app['students'].keys()) == 4
199        self.assertTrue('A123456' in self.app['students'].keys())
200        shutil.rmtree(os.path.dirname(fin_file))
201
202class StudentStudyCourseImporterTest(FunctionalTestCase):
203
204    layer = FunctionalLayer
205
206    def setUp(self):
207        super(StudentStudyCourseImporterTest, self).setUp()
208        app = University()
209        self.dc_root = tempfile.mkdtemp()
210        app['datacenter'].setStoragePath(self.dc_root)
211
212        self.getRootFolder()['app'] = app
213        self.app = self.getRootFolder()['app']
214        setSite(app)
215
216        self.workdir = tempfile.mkdtemp()
217        self.importer = StudentStudyCourseProcessor()
218        self.csv_file = os.path.join(
219            self.workdir, 'sample_studycourse_data.csv')
220        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
221
222        # Import students with subobjects
223        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
224        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
225        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
226            student_file, STUDENT_HEADER_FIELDS)
227        shutil.rmtree(os.path.dirname(fin_file))
228
229        # Populate university
230        self.certificate = createObject('waeup.Certificate')
231        self.certificate.code = 'CERT1'
232        self.certificate.application_category = 'basic'
233        self.certificate.start_level = 100
234        self.certificate.end_level = 500
235        self.app['faculties']['fac1'] = Faculty()
236        self.app['faculties']['fac1']['dep1'] = Department()
237        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
238            self.certificate)
239        return
240
241    def tearDown(self):
242        super(StudentStudyCourseImporterTest, self).tearDown()
243        shutil.rmtree(self.workdir)
244        shutil.rmtree(self.dc_root)
245        clearSite()
246        return
247
248    def test_interface(self):
249        # Make sure we fulfill the interface contracts.
250        assert verifyObject(IBatchProcessor, self.importer) is True
251        assert verifyClass(
252            IBatchProcessor, StudentStudyCourseProcessor) is True
253
254    def test_entryExists(self):
255        assert self.importer.entryExists(
256            dict(reg_number='REG_NONE'), self.app) is None
257        student = self.importer.entryExists(dict(reg_number='1'), self.app)
258        self.assertEqual(student.reg_number, u'1')
259
260    def test_getEntry(self):
261        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
262        student = studycourse.__parent__
263        s_id = student.student_id
264        assert studycourse is self.app['students'][s_id]['studycourse']
265
266    def test_import(self):
267        num, num_warns, fin_file, fail_file = self.importer.doImport(
268            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
269        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
270        self.assertEqual(num_warns,0)
271        self.assertEqual(studycourse.certificate.code, u'CERT1')
272        shutil.rmtree(os.path.dirname(fin_file))
273
274def test_suite():
275    suite = unittest.TestSuite()
276    for testcase in [
277        StudentImporterTest,StudentStudyCourseImporterTest,
278        ]:
279        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
280                testcase
281                )
282        )
283    return suite
Note: See TracBrowser for help on using the repository browser.