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

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

In import files we can use the hash symbol at the end of a date string to
avoid annoying automatic date transformation by Excel or Calc.
  • Property svn:keywords set to Id
File size: 11.4 KB
Line 
1## $Id: test_batching.py 7515 2012-01-26 08:59:05Z 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,1)
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,1)
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        shutil.rmtree(os.path.dirname(fin_file))
209
210class StudentStudyCourseImporterTest(FunctionalTestCase):
211
212    layer = FunctionalLayer
213
214    def setUp(self):
215        super(StudentStudyCourseImporterTest, self).setUp()
216        app = University()
217        self.dc_root = tempfile.mkdtemp()
218        app['datacenter'].setStoragePath(self.dc_root)
219
220        self.getRootFolder()['app'] = app
221        self.app = self.getRootFolder()['app']
222        setSite(app)
223
224        self.workdir = tempfile.mkdtemp()
225        self.importer = StudentStudyCourseProcessor()
226        self.csv_file = os.path.join(
227            self.workdir, 'sample_studycourse_data.csv')
228        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
229
230        # Import students with subobjects
231        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
232        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
233        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
234            student_file, STUDENT_HEADER_FIELDS)
235        shutil.rmtree(os.path.dirname(fin_file))
236
237        # Populate university
238        self.certificate = createObject('waeup.Certificate')
239        self.certificate.code = 'CERT1'
240        self.certificate.application_category = 'basic'
241        self.certificate.start_level = 100
242        self.certificate.end_level = 500
243        self.app['faculties']['fac1'] = Faculty()
244        self.app['faculties']['fac1']['dep1'] = Department()
245        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
246            self.certificate)
247        return
248
249    def tearDown(self):
250        super(StudentStudyCourseImporterTest, self).tearDown()
251        shutil.rmtree(self.workdir)
252        shutil.rmtree(self.dc_root)
253        clearSite()
254        return
255
256    def test_interface(self):
257        # Make sure we fulfill the interface contracts.
258        assert verifyObject(IBatchProcessor, self.importer) is True
259        assert verifyClass(
260            IBatchProcessor, StudentStudyCourseProcessor) is True
261
262    def test_entryExists(self):
263        assert self.importer.entryExists(
264            dict(reg_number='REG_NONE'), self.app) is None
265        student = self.importer.entryExists(dict(reg_number='1'), self.app)
266        self.assertEqual(student.reg_number, u'1')
267
268    def test_getEntry(self):
269        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
270        student = studycourse.__parent__
271        s_id = student.student_id
272        assert studycourse is self.app['students'][s_id]['studycourse']
273
274    def test_import(self):
275        num, num_warns, fin_file, fail_file = self.importer.doImport(
276            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
277        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
278        self.assertEqual(num_warns,0)
279        self.assertEqual(studycourse.certificate.code, u'CERT1')
280        shutil.rmtree(os.path.dirname(fin_file))
281
282def test_suite():
283    suite = unittest.TestSuite()
284    for testcase in [
285        StudentImporterTest,StudentStudyCourseImporterTest,
286        ]:
287        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
288                testcase
289                )
290        )
291    return suite
Note: See TracBrowser for help on using the repository browser.