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

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

Add history messages and log file entries when importing students.

Do not accept empty reg_state fields. If the the reg_state column exists a valid value must be provided.
  • Property svn:keywords set to Id
File size: 11.5 KB
Line 
1## $Id: test_batching.py 7522 2012-01-27 16:33:02Z 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,'school fee paid')
203        self.assertEqual(self.app['students']['A123456'].date_of_birth,
204            datetime.date(1990, 1, 2))
205        self.assertEqual(self.app['students']['B123456'].state,'cleared')
206        self.assertEqual(self.app['students']['B123456'].date_of_birth,
207            datetime.date(1990, 1, 3))
208        history = ' '.join(self.app['students']['A123456'].history.messages)
209        self.assertTrue(
210            "State 'school fee paid' set by system" in history)
211        shutil.rmtree(os.path.dirname(fin_file))
212
213class StudentStudyCourseImporterTest(FunctionalTestCase):
214
215    layer = FunctionalLayer
216
217    def setUp(self):
218        super(StudentStudyCourseImporterTest, self).setUp()
219        app = University()
220        self.dc_root = tempfile.mkdtemp()
221        app['datacenter'].setStoragePath(self.dc_root)
222
223        self.getRootFolder()['app'] = app
224        self.app = self.getRootFolder()['app']
225        setSite(app)
226
227        self.workdir = tempfile.mkdtemp()
228        self.importer = StudentStudyCourseProcessor()
229        self.csv_file = os.path.join(
230            self.workdir, 'sample_studycourse_data.csv')
231        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
232
233        # Import students with subobjects
234        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
235        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
236        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
237            student_file, STUDENT_HEADER_FIELDS)
238        shutil.rmtree(os.path.dirname(fin_file))
239
240        # Populate university
241        self.certificate = createObject('waeup.Certificate')
242        self.certificate.code = 'CERT1'
243        self.certificate.application_category = 'basic'
244        self.certificate.start_level = 100
245        self.certificate.end_level = 500
246        self.app['faculties']['fac1'] = Faculty()
247        self.app['faculties']['fac1']['dep1'] = Department()
248        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
249            self.certificate)
250        return
251
252    def tearDown(self):
253        super(StudentStudyCourseImporterTest, self).tearDown()
254        shutil.rmtree(self.workdir)
255        shutil.rmtree(self.dc_root)
256        clearSite()
257        return
258
259    def test_interface(self):
260        # Make sure we fulfill the interface contracts.
261        assert verifyObject(IBatchProcessor, self.importer) is True
262        assert verifyClass(
263            IBatchProcessor, StudentStudyCourseProcessor) is True
264
265    def test_entryExists(self):
266        assert self.importer.entryExists(
267            dict(reg_number='REG_NONE'), self.app) is None
268        student = self.importer.entryExists(dict(reg_number='1'), self.app)
269        self.assertEqual(student.reg_number, u'1')
270
271    def test_getEntry(self):
272        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
273        student = studycourse.__parent__
274        s_id = student.student_id
275        assert studycourse is self.app['students'][s_id]['studycourse']
276
277    def test_import(self):
278        num, num_warns, fin_file, fail_file = self.importer.doImport(
279            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
280        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
281        self.assertEqual(num_warns,0)
282        self.assertEqual(studycourse.certificate.code, u'CERT1')
283        shutil.rmtree(os.path.dirname(fin_file))
284
285def test_suite():
286    suite = unittest.TestSuite()
287    for testcase in [
288        StudentImporterTest,StudentStudyCourseImporterTest,
289        ]:
290        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
291                testcase
292                )
293        )
294    return suite
Note: See TracBrowser for help on using the repository browser.