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

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

Enable import of student registration states which are in IMPORTABLE_STATES.

Set workflow state directly rather than firing transitions.

  • Property svn:keywords set to Id
File size: 11.1 KB
Line 
1## $Id: test_batching.py 7513 2012-01-25 17:47:23Z 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.firstname = u'Anna'
96        student.lastname = u'Tester'
97        student.reg_number = u'123'
98        student.matric_number = u'234'
99        self.app['students'].addStudent(student)
100        self.student = self.app['students'][student.student_id]
101        self.importer = StudentProcessor()
102        self.workdir = tempfile.mkdtemp()
103        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
104        self.csv_file_update = os.path.join(
105            self.workdir, 'sample_student_data_update.csv')
106        self.csv_file_update2 = os.path.join(
107            self.workdir, 'sample_student_data_update2.csv')
108        self.csv_file_migration = os.path.join(
109            self.workdir, 'sample_student_data_migration.csv')
110        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
111        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
112        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
113        open(self.csv_file_migration, 'wb').write(STUDENT_SAMPLE_DATA_MIGRATION)
114
115    def tearDown(self):
116        super(StudentImporterTest, self).tearDown()
117        shutil.rmtree(self.workdir)
118        shutil.rmtree(self.dc_root)
119        clearSite()
120        return
121
122    def test_interface(self):
123        # Make sure we fulfill the interface contracts.
124        assert verifyObject(IBatchProcessor, self.importer) is True
125        assert verifyClass(
126            IBatchProcessor, StudentProcessor) is True
127
128    def test_parentsExist(self):
129        self.assertFalse(self.importer.parentsExist(None, dict()))
130        self.assertTrue(self.importer.parentsExist(None, self.app))
131
132    def test_entryExists(self):
133        self.assertFalse(self.importer.entryExists(
134            dict(student_id='ID_NONE'), self.app))
135        student = self.importer.getEntry(
136            dict(student_id=self.student.student_id), self.app)
137        self.assertEqual(student.reg_number, u'123')
138
139    def test_getParent(self):
140        parent = self.importer.getParent(None, self.app)
141        assert parent is self.app['students']
142
143    def test_getEntry(self):
144        assert self.importer.getEntry(
145            dict(student_id='ID_NONE'), self.app) is None
146        assert self.importer.getEntry(
147            dict(student_id=self.student.student_id), self.app) is self.student
148
149    def test_addEntry(self):
150        new_student = Student()
151        self.importer.addEntry(
152            new_student, dict(), self.app)
153        assert len(self.app['students'].keys()) == 2
154
155    def test_delEntry(self):
156        assert self.student.student_id in self.app['students'].keys()
157        self.importer.delEntry(
158            dict(reg_number=self.student.reg_number), self.app)
159        assert self.student.student_id not in self.app['students'].keys()
160
161    def test_import(self):
162        num, num_warns, fin_file, fail_file = self.importer.doImport(
163            self.csv_file, STUDENT_HEADER_FIELDS)
164        self.assertEqual(num_warns,1)
165        assert len(self.app['students'].keys()) == 4
166        shutil.rmtree(os.path.dirname(fin_file))
167
168    def test_import_update(self):
169        num, num_warns, fin_file, fail_file = self.importer.doImport(
170            self.csv_file, STUDENT_HEADER_FIELDS)
171        shutil.rmtree(os.path.dirname(fin_file))
172        num, num_warns, fin_file, fail_file = self.importer.doImport(
173            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
174        self.assertEqual(num_warns,0)
175        shutil.rmtree(os.path.dirname(fin_file))
176
177    def test_import_update2(self):
178        num, num_warns, fin_file, fail_file = self.importer.doImport(
179            self.csv_file, STUDENT_HEADER_FIELDS)
180        shutil.rmtree(os.path.dirname(fin_file))
181        num, num_warns, fin_file, fail_file = self.importer.doImport(
182            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
183        self.assertEqual(num_warns,0)
184        shutil.rmtree(os.path.dirname(fin_file))
185
186    def test_import_remove(self):
187        num, num_warns, fin_file, fail_file = self.importer.doImport(
188            self.csv_file, STUDENT_HEADER_FIELDS)
189        shutil.rmtree(os.path.dirname(fin_file))
190        num, num_warns, fin_file, fail_file = self.importer.doImport(
191            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
192        self.assertEqual(num_warns,0)
193        shutil.rmtree(os.path.dirname(fin_file))
194
195    def test_import_migration_data(self):
196        num, num_warns, fin_file, fail_file = self.importer.doImport(
197            self.csv_file_migration, STUDENT_HEADER_FIELDS_MIGRATION)
198        self.assertEqual(num_warns,1)
199        assert len(self.app['students'].keys()) == 4
200        self.assertTrue('A123456' in self.app['students'].keys())
201        self.assertEqual(self.app['students']['A123456'].state,'school fee paid')
202        shutil.rmtree(os.path.dirname(fin_file))
203
204class StudentStudyCourseImporterTest(FunctionalTestCase):
205
206    layer = FunctionalLayer
207
208    def setUp(self):
209        super(StudentStudyCourseImporterTest, self).setUp()
210        app = University()
211        self.dc_root = tempfile.mkdtemp()
212        app['datacenter'].setStoragePath(self.dc_root)
213
214        self.getRootFolder()['app'] = app
215        self.app = self.getRootFolder()['app']
216        setSite(app)
217
218        self.workdir = tempfile.mkdtemp()
219        self.importer = StudentStudyCourseProcessor()
220        self.csv_file = os.path.join(
221            self.workdir, 'sample_studycourse_data.csv')
222        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
223
224        # Import students with subobjects
225        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
226        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
227        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
228            student_file, STUDENT_HEADER_FIELDS)
229        shutil.rmtree(os.path.dirname(fin_file))
230
231        # Populate university
232        self.certificate = createObject('waeup.Certificate')
233        self.certificate.code = 'CERT1'
234        self.certificate.application_category = 'basic'
235        self.certificate.start_level = 100
236        self.certificate.end_level = 500
237        self.app['faculties']['fac1'] = Faculty()
238        self.app['faculties']['fac1']['dep1'] = Department()
239        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
240            self.certificate)
241        return
242
243    def tearDown(self):
244        super(StudentStudyCourseImporterTest, self).tearDown()
245        shutil.rmtree(self.workdir)
246        shutil.rmtree(self.dc_root)
247        clearSite()
248        return
249
250    def test_interface(self):
251        # Make sure we fulfill the interface contracts.
252        assert verifyObject(IBatchProcessor, self.importer) is True
253        assert verifyClass(
254            IBatchProcessor, StudentStudyCourseProcessor) is True
255
256    def test_entryExists(self):
257        assert self.importer.entryExists(
258            dict(reg_number='REG_NONE'), self.app) is None
259        student = self.importer.entryExists(dict(reg_number='1'), self.app)
260        self.assertEqual(student.reg_number, u'1')
261
262    def test_getEntry(self):
263        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
264        student = studycourse.__parent__
265        s_id = student.student_id
266        assert studycourse is self.app['students'][s_id]['studycourse']
267
268    def test_import(self):
269        num, num_warns, fin_file, fail_file = self.importer.doImport(
270            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
271        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
272        self.assertEqual(num_warns,0)
273        self.assertEqual(studycourse.certificate.code, u'CERT1')
274        shutil.rmtree(os.path.dirname(fin_file))
275
276def test_suite():
277    suite = unittest.TestSuite()
278    for testcase in [
279        StudentImporterTest,StudentStudyCourseImporterTest,
280        ]:
281        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
282                testcase
283                )
284        )
285    return suite
Note: See TracBrowser for help on using the repository browser.