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

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

Fix logic. entryExists must return True or False, nothing else.
  • Property svn:keywords set to Id
File size: 11.5 KB
Line 
1## $Id: test_batching.py 7534 2012-01-29 08:29:43Z 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        assert self.importer.entryExists(
135            dict(student_id='ID_NONE'), self.app) is False
136        assert self.importer.entryExists(
137            dict(reg_number='123'), self.app) is True
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,2)
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,'clearance started')
202        self.assertEqual(self.app['students']['A123456'].date_of_birth,
203            datetime.date(1990, 1, 2))
204        self.assertFalse(self.app['students']['A123456'].clearance_locked)
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        self.assertTrue(self.app['students']['B123456'].clearance_locked)
209        history = ' '.join(self.app['students']['A123456'].history.messages)
210        self.assertTrue(
211            "State 'clearance started' set by system" in history)
212        shutil.rmtree(os.path.dirname(fin_file))
213
214class StudentStudyCourseImporterTest(FunctionalTestCase):
215
216    layer = FunctionalLayer
217
218    def setUp(self):
219        super(StudentStudyCourseImporterTest, self).setUp()
220        app = University()
221        self.dc_root = tempfile.mkdtemp()
222        app['datacenter'].setStoragePath(self.dc_root)
223
224        self.getRootFolder()['app'] = app
225        self.app = self.getRootFolder()['app']
226        setSite(app)
227
228        self.workdir = tempfile.mkdtemp()
229        self.importer = StudentStudyCourseProcessor()
230        self.csv_file = os.path.join(
231            self.workdir, 'sample_studycourse_data.csv')
232        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
233
234        # Import students with subobjects
235        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
236        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
237        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
238            student_file, STUDENT_HEADER_FIELDS)
239        shutil.rmtree(os.path.dirname(fin_file))
240
241        # Populate university
242        self.certificate = createObject('waeup.Certificate')
243        self.certificate.code = 'CERT1'
244        self.certificate.application_category = 'basic'
245        self.certificate.start_level = 200
246        self.certificate.end_level = 500
247        self.app['faculties']['fac1'] = Faculty()
248        self.app['faculties']['fac1']['dep1'] = Department()
249        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
250            self.certificate)
251        return
252
253    def tearDown(self):
254        super(StudentStudyCourseImporterTest, self).tearDown()
255        shutil.rmtree(self.workdir)
256        shutil.rmtree(self.dc_root)
257        clearSite()
258        return
259
260    def test_interface(self):
261        # Make sure we fulfill the interface contracts.
262        assert verifyObject(IBatchProcessor, self.importer) is True
263        assert verifyClass(
264            IBatchProcessor, StudentStudyCourseProcessor) is True
265
266    def test_entryExists(self):
267        assert self.importer.entryExists(
268            dict(reg_number='REG_NONE'), self.app) is False
269        assert self.importer.entryExists(
270            dict(reg_number='1'), self.app) is True
271
272    def test_getEntry(self):
273        student = self.importer.getEntry(
274            dict(reg_number='1'), self.app).__parent__
275        self.assertEqual(student.reg_number,'1')
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        self.assertEqual(num_warns,1)
281        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
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.