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

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

Add second student data update test with matric_numbers as locators.
  • Property svn:keywords set to Id
File size: 10.7 KB
Line 
1##
2## test_batching.py
3## Login : <uli@pu.smp.net>
4## Started on  Tue Aug 24 02:04:44 2010 Uli Fouquet
5## $Id: test_batching.py 6851 2011-10-03 11:27:15Z henrik $
6##
7## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann
8## This program is free software; you can redistribute it and/or modify
9## it under the terms of the GNU General Public License as published by
10## the Free Software Foundation; either version 2 of the License, or
11## (at your option) any later version.
12##
13## This program is distributed in the hope that it will be useful,
14## but WITHOUT ANY WARRANTY; without even the implied warranty of
15## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16## GNU General Public License for more details.
17##
18## You should have received a copy of the GNU General Public License
19## along with this program; if not, write to the Free Software
20## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
21##
22"""Unit tests for students-related data importers.
23"""
24import os
25import shutil
26import tempfile
27import unittest
28from zope.component import createObject
29from zope.component.hooks import setSite, clearSite
30from zope.interface.verify import verifyClass, verifyObject
31
32from waeup.sirp.app import University
33from waeup.sirp.university.faculty import Faculty
34from waeup.sirp.university.department import Department
35from waeup.sirp.students.batching import (
36    StudentProcessor, StudentStudyCourseProcessor)
37from waeup.sirp.students.student import Student
38from waeup.sirp.students.studycourse import StudentStudyCourse
39from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
40from waeup.sirp.interfaces import IBatchProcessor
41
42STUDENT_SAMPLE_DATA = open(
43    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
44    'rb').read()
45
46STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
47    '\n')[0].split(',')
48
49STUDENT_SAMPLE_DATA_UPDATE = open(
50    os.path.join(os.path.dirname(__file__), 'sample_student_data_update.csv'),
51    'rb').read()
52
53STUDENT_HEADER_FIELDS_UPDATE = STUDENT_SAMPLE_DATA_UPDATE.split(
54    '\n')[0].split(',')
55
56STUDENT_SAMPLE_DATA_UPDATE2 = open(
57    os.path.join(os.path.dirname(__file__), 'sample_student_data_update2.csv'),
58    'rb').read()
59
60STUDENT_HEADER_FIELDS_UPDATE2 = STUDENT_SAMPLE_DATA_UPDATE2.split(
61    '\n')[0].split(',')
62
63STUDENT_SAMPLE_DATA = open(
64    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
65    'rb').read()
66
67STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
68    '\n')[0].split(',')
69
70STUDYCOURSE_SAMPLE_DATA = open(
71    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
72    'rb').read()
73
74STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
75    '\n')[0].split(',')
76
77class StudentImporterTest(FunctionalTestCase):
78
79    layer = FunctionalLayer
80
81    def setUp(self):
82        super(StudentImporterTest, self).setUp()
83        # Setup a sample site for each test
84        app = University()
85        self.dc_root = tempfile.mkdtemp()
86        app['datacenter'].setStoragePath(self.dc_root)
87
88        # Prepopulate the ZODB...
89        self.getRootFolder()['app'] = app
90        # we add the site immediately after creation to the
91        # ZODB. Catalogs and other local utilities are not setup
92        # before that step.
93        self.app = self.getRootFolder()['app']
94        # Set site here. Some of the following setup code might need
95        # to access grok.getSite() and should get our new app then
96        setSite(app)
97
98        # Add student with subobjects
99        student = Student()
100        student.fullname = u'Anna Tester'
101        student.reg_number = u'123'
102        student.matric_number = u'234'
103        self.app['students'].addStudent(student)
104        self.student = self.app['students'][student.student_id]
105        self.importer = StudentProcessor()
106        self.workdir = tempfile.mkdtemp()
107        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
108        self.csv_file_update = os.path.join(
109            self.workdir, 'sample_student_data_update.csv')
110        self.csv_file_update2 = os.path.join(
111            self.workdir, 'sample_student_data_update2.csv')
112        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
113        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
114        open(self.csv_file_update2, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE2)
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        assert self.importer.parentsExist(None, dict()) is False
131        assert self.importer.parentsExist(None, self.app) is True
132
133    def test_entryExists(self):
134        assert self.importer.entryExists(
135            dict(student_id='ID_NONE'), self.app) is None
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        self.importer.delEntry(
158            dict(student_id=self.student.student_id), self.app)
159        assert self.student.student_id not in self.app['students'].keys()
160
161    def test_delEntry(self):
162        assert self.student.student_id in self.app['students'].keys()
163        self.importer.delEntry(
164            dict(reg_number=self.student.reg_number), self.app)
165        assert self.student.student_id not in self.app['students'].keys()
166
167    def test_import(self):
168        num, num_warns, fin_file, fail_file = self.importer.doImport(
169            self.csv_file, STUDENT_HEADER_FIELDS)
170        self.assertEqual(num_warns,0)
171        assert len(self.app['students'].keys()) == 4
172        shutil.rmtree(os.path.dirname(fin_file))
173
174    def test_import_update(self):
175        num, num_warns, fin_file, fail_file = self.importer.doImport(
176            self.csv_file, STUDENT_HEADER_FIELDS)
177        shutil.rmtree(os.path.dirname(fin_file))
178        num, num_warns, fin_file, fail_file = self.importer.doImport(
179            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
180        self.assertEqual(num_warns,0)
181        shutil.rmtree(os.path.dirname(fin_file))
182
183    def test_import_update2(self):
184        num, num_warns, fin_file, fail_file = self.importer.doImport(
185            self.csv_file, STUDENT_HEADER_FIELDS)
186        shutil.rmtree(os.path.dirname(fin_file))
187        num, num_warns, fin_file, fail_file = self.importer.doImport(
188            self.csv_file_update2, STUDENT_HEADER_FIELDS_UPDATE2, 'update')
189        self.assertEqual(num_warns,0)
190        shutil.rmtree(os.path.dirname(fin_file))
191
192    def test_import_remove(self):
193        num, num_warns, fin_file, fail_file = self.importer.doImport(
194            self.csv_file, STUDENT_HEADER_FIELDS)
195        shutil.rmtree(os.path.dirname(fin_file))
196        num, num_warns, fin_file, fail_file = self.importer.doImport(
197            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
198        self.assertEqual(num_warns,0)
199        shutil.rmtree(os.path.dirname(fin_file))
200
201class StudentStudyCourseImporterTest(FunctionalTestCase):
202
203    layer = FunctionalLayer
204
205    def setUp(self):
206        super(StudentStudyCourseImporterTest, self).setUp()
207        app = University()
208        self.dc_root = tempfile.mkdtemp()
209        app['datacenter'].setStoragePath(self.dc_root)
210
211        self.getRootFolder()['app'] = app
212        self.app = self.getRootFolder()['app']
213        setSite(app)
214
215        self.workdir = tempfile.mkdtemp()
216        self.importer = StudentStudyCourseProcessor()
217        self.csv_file = os.path.join(
218            self.workdir, 'sample_studycourse_data.csv')
219        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
220
221        # Import students with subobjects
222        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
223        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
224        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
225            student_file, STUDENT_HEADER_FIELDS)
226        shutil.rmtree(os.path.dirname(fin_file))
227
228        # Populate university
229        self.certificate = createObject('waeup.Certificate')
230        self.certificate.code = 'CERT1'
231        self.certificate.application_category = 'basic'
232        self.certificate.start_level = 100
233        self.certificate.end_level = 500
234        self.app['faculties']['fac1'] = Faculty()
235        self.app['faculties']['fac1']['dep1'] = Department()
236        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
237            self.certificate)
238        return
239
240    def tearDown(self):
241        super(StudentStudyCourseImporterTest, self).tearDown()
242        shutil.rmtree(self.workdir)
243        shutil.rmtree(self.dc_root)
244        clearSite()
245        return
246
247    def test_interface(self):
248        # Make sure we fulfill the interface contracts.
249        assert verifyObject(IBatchProcessor, self.importer) is True
250        assert verifyClass(
251            IBatchProcessor, StudentStudyCourseProcessor) is True
252
253    def test_entryExists(self):
254        assert self.importer.entryExists(
255            dict(reg_number='REG_NONE'), self.app) is None
256        student = self.importer.entryExists(dict(reg_number='1'), self.app)
257        self.assertEqual(student.reg_number, u'1')
258
259    def test_getEntry(self):
260        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
261        student = studycourse.__parent__
262        s_id = student.student_id
263        assert studycourse is self.app['students'][s_id]['studycourse']
264
265    def test_import(self):
266        num, num_warns, fin_file, fail_file = self.importer.doImport(
267            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
268        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
269        self.assertEqual(num_warns,0)
270        self.assertEqual(studycourse.certificate.code, u'CERT1')
271        shutil.rmtree(os.path.dirname(fin_file))
272
273def test_suite():
274    suite = unittest.TestSuite()
275    for testcase in [
276        StudentImporterTest,StudentStudyCourseImporterTest,
277        ]:
278        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
279                testcase
280                )
281        )
282    return suite
Note: See TracBrowser for help on using the repository browser.