source: main/waeup.sirp/branches/henrik-bootstrap/src/waeup/sirp/students/tests/test_batching.py @ 7454

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

Replace fullname form field by first-, middle- and lastname form fields.

We don't need a studentaddpage.pt.
  • Property svn:keywords set to Id
File size: 11.0 KB
Line 
1## $Id: test_batching.py 7357 2011-12-16 06:40:31Z 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,0)
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,0)
199        assert len(self.app['students'].keys()) == 4
200        self.assertTrue('A123456' in self.app['students'].keys())
201        shutil.rmtree(os.path.dirname(fin_file))
202
203class StudentStudyCourseImporterTest(FunctionalTestCase):
204
205    layer = FunctionalLayer
206
207    def setUp(self):
208        super(StudentStudyCourseImporterTest, self).setUp()
209        app = University()
210        self.dc_root = tempfile.mkdtemp()
211        app['datacenter'].setStoragePath(self.dc_root)
212
213        self.getRootFolder()['app'] = app
214        self.app = self.getRootFolder()['app']
215        setSite(app)
216
217        self.workdir = tempfile.mkdtemp()
218        self.importer = StudentStudyCourseProcessor()
219        self.csv_file = os.path.join(
220            self.workdir, 'sample_studycourse_data.csv')
221        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
222
223        # Import students with subobjects
224        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
225        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
226        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
227            student_file, STUDENT_HEADER_FIELDS)
228        shutil.rmtree(os.path.dirname(fin_file))
229
230        # Populate university
231        self.certificate = createObject('waeup.Certificate')
232        self.certificate.code = 'CERT1'
233        self.certificate.application_category = 'basic'
234        self.certificate.start_level = 100
235        self.certificate.end_level = 500
236        self.app['faculties']['fac1'] = Faculty()
237        self.app['faculties']['fac1']['dep1'] = Department()
238        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
239            self.certificate)
240        return
241
242    def tearDown(self):
243        super(StudentStudyCourseImporterTest, self).tearDown()
244        shutil.rmtree(self.workdir)
245        shutil.rmtree(self.dc_root)
246        clearSite()
247        return
248
249    def test_interface(self):
250        # Make sure we fulfill the interface contracts.
251        assert verifyObject(IBatchProcessor, self.importer) is True
252        assert verifyClass(
253            IBatchProcessor, StudentStudyCourseProcessor) is True
254
255    def test_entryExists(self):
256        assert self.importer.entryExists(
257            dict(reg_number='REG_NONE'), self.app) is None
258        student = self.importer.entryExists(dict(reg_number='1'), self.app)
259        self.assertEqual(student.reg_number, u'1')
260
261    def test_getEntry(self):
262        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
263        student = studycourse.__parent__
264        s_id = student.student_id
265        assert studycourse is self.app['students'][s_id]['studycourse']
266
267    def test_import(self):
268        num, num_warns, fin_file, fail_file = self.importer.doImport(
269            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
270        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
271        self.assertEqual(num_warns,0)
272        self.assertEqual(studycourse.certificate.code, u'CERT1')
273        shutil.rmtree(os.path.dirname(fin_file))
274
275def test_suite():
276    suite = unittest.TestSuite()
277    for testcase in [
278        StudentImporterTest,StudentStudyCourseImporterTest,
279        ]:
280        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
281                testcase
282                )
283        )
284    return suite
Note: See TracBrowser for help on using the repository browser.