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

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

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