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

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

Include student update and removal tests.
  • Property svn:keywords set to Id
File size: 9.8 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 6848 2011-10-03 08:05:53Z 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 = open(
57    os.path.join(os.path.dirname(__file__), 'sample_student_data.csv'),
58    'rb').read()
59
60STUDENT_HEADER_FIELDS = STUDENT_SAMPLE_DATA.split(
61    '\n')[0].split(',')
62
63STUDYCOURSE_SAMPLE_DATA = open(
64    os.path.join(os.path.dirname(__file__), 'sample_studycourse_data.csv'),
65    'rb').read()
66
67STUDYCOURSE_HEADER_FIELDS = STUDYCOURSE_SAMPLE_DATA.split(
68    '\n')[0].split(',')
69
70class StudentImporterTest(FunctionalTestCase):
71
72    layer = FunctionalLayer
73
74    def setUp(self):
75        super(StudentImporterTest, self).setUp()
76        # Setup a sample site for each test
77        app = University()
78        self.dc_root = tempfile.mkdtemp()
79        app['datacenter'].setStoragePath(self.dc_root)
80
81        # Prepopulate the ZODB...
82        self.getRootFolder()['app'] = app
83        # we add the site immediately after creation to the
84        # ZODB. Catalogs and other local utilities are not setup
85        # before that step.
86        self.app = self.getRootFolder()['app']
87        # Set site here. Some of the following setup code might need
88        # to access grok.getSite() and should get our new app then
89        setSite(app)
90
91        # Add student with subobjects
92        student = Student()
93        student.fullname = u'Anna Tester'
94        student.reg_number = u'123'
95        student.matric_number = u'234'
96        self.app['students'].addStudent(student)
97        self.student = self.app['students'][student.student_id]
98        self.importer = StudentProcessor()
99        self.workdir = tempfile.mkdtemp()
100        self.csv_file = os.path.join(self.workdir, 'sample_student_data.csv')
101        self.csv_file_update = os.path.join(self.workdir, 'sample_student_data_update.csv')
102        open(self.csv_file, 'wb').write(STUDENT_SAMPLE_DATA)
103        open(self.csv_file_update, 'wb').write(STUDENT_SAMPLE_DATA_UPDATE)
104
105    def tearDown(self):
106        super(StudentImporterTest, self).tearDown()
107        shutil.rmtree(self.workdir)
108        shutil.rmtree(self.dc_root)
109        clearSite()
110        return
111
112    def test_interface(self):
113        # Make sure we fulfill the interface contracts.
114        assert verifyObject(IBatchProcessor, self.importer) is True
115        assert verifyClass(
116            IBatchProcessor, StudentProcessor) is True
117
118    def test_parentsExist(self):
119        assert self.importer.parentsExist(None, dict()) is False
120        assert self.importer.parentsExist(None, self.app) is True
121
122    def test_entryExists(self):
123        assert self.importer.entryExists(
124            dict(student_id='ID_NONE'), self.app) is None
125        student = self.importer.getEntry(
126            dict(student_id=self.student.student_id), self.app)
127        self.assertEqual(student.reg_number, u'123')
128
129    def test_getParent(self):
130        parent = self.importer.getParent(None, self.app)
131        assert parent is self.app['students']
132
133    def test_getEntry(self):
134        assert self.importer.getEntry(
135            dict(student_id='ID_NONE'), self.app) is None
136        assert self.importer.getEntry(
137            dict(student_id=self.student.student_id), self.app) is self.student
138
139    def test_addEntry(self):
140        new_student = Student()
141        self.importer.addEntry(
142            new_student, dict(), self.app)
143        assert len(self.app['students'].keys()) == 2
144
145    def test_delEntry(self):
146        self.importer.delEntry(dict(student_id=self.student.student_id), self.app)
147        assert self.student.student_id not in self.app['students'].keys()
148
149    def test_delEntry(self):
150        assert self.student.student_id in self.app['students'].keys()
151        self.importer.delEntry(dict(reg_number=self.student.reg_number), self.app)
152        assert self.student.student_id not in self.app['students'].keys()
153
154    def test_import(self):
155        num, num_warns, fin_file, fail_file = self.importer.doImport(
156            self.csv_file, STUDENT_HEADER_FIELDS)
157        self.assertEqual(num_warns,0)
158        assert len(self.app['students'].keys()) == 4
159        shutil.rmtree(os.path.dirname(fin_file))
160
161    def test_import_update(self):
162        num, num_warns, fin_file, fail_file = self.importer.doImport(
163            self.csv_file, STUDENT_HEADER_FIELDS)
164        shutil.rmtree(os.path.dirname(fin_file))
165        num, num_warns, fin_file, fail_file = self.importer.doImport(
166            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'update')
167        self.assertEqual(num_warns,0)
168        shutil.rmtree(os.path.dirname(fin_file))
169
170    def test_import_remove(self):
171        num, num_warns, fin_file, fail_file = self.importer.doImport(
172            self.csv_file, STUDENT_HEADER_FIELDS)
173        shutil.rmtree(os.path.dirname(fin_file))
174        num, num_warns, fin_file, fail_file = self.importer.doImport(
175            self.csv_file_update, STUDENT_HEADER_FIELDS_UPDATE, 'remove')
176        self.assertEqual(num_warns,0)
177        shutil.rmtree(os.path.dirname(fin_file))
178
179class StudentStudyCourseImporterTest(FunctionalTestCase):
180
181    layer = FunctionalLayer
182
183    def setUp(self):
184        super(StudentStudyCourseImporterTest, self).setUp()
185        app = University()
186        self.dc_root = tempfile.mkdtemp()
187        app['datacenter'].setStoragePath(self.dc_root)
188
189        self.getRootFolder()['app'] = app
190        self.app = self.getRootFolder()['app']
191        setSite(app)
192
193        self.workdir = tempfile.mkdtemp()
194        self.importer = StudentStudyCourseProcessor()
195        self.csv_file = os.path.join(self.workdir, 'sample_studycourse_data.csv')
196        open(self.csv_file, 'wb').write(STUDYCOURSE_SAMPLE_DATA)
197
198        # Import students with subobjects
199        student_file = os.path.join(self.workdir, 'sample_student_data.csv')
200        open(student_file, 'wb').write(STUDENT_SAMPLE_DATA)
201        num, num_warns, fin_file, fail_file = StudentProcessor().doImport(
202            student_file, STUDENT_HEADER_FIELDS)
203        shutil.rmtree(os.path.dirname(fin_file))
204
205        # Populate university
206        self.certificate = createObject('waeup.Certificate')
207        self.certificate.code = 'CERT1'
208        self.certificate.application_category = 'basic'
209        self.certificate.start_level = 100
210        self.certificate.end_level = 500
211        self.app['faculties']['fac1'] = Faculty()
212        self.app['faculties']['fac1']['dep1'] = Department()
213        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
214            self.certificate)
215        return
216
217    def tearDown(self):
218        super(StudentStudyCourseImporterTest, self).tearDown()
219        shutil.rmtree(self.workdir)
220        shutil.rmtree(self.dc_root)
221        clearSite()
222        return
223
224    def test_interface(self):
225        # Make sure we fulfill the interface contracts.
226        assert verifyObject(IBatchProcessor, self.importer) is True
227        assert verifyClass(
228            IBatchProcessor, StudentStudyCourseProcessor) is True
229
230    def test_entryExists(self):
231        assert self.importer.entryExists(
232            dict(reg_number='REG_NONE'), self.app) is None
233        student = self.importer.entryExists(dict(reg_number='1'), self.app)
234        self.assertEqual(student.reg_number, u'1')
235
236    def test_getEntry(self):
237        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
238        student = studycourse.__parent__
239        s_id = student.student_id
240        assert studycourse is self.app['students'][s_id]['studycourse']
241
242    def test_import(self):
243        num, num_warns, fin_file, fail_file = self.importer.doImport(
244            self.csv_file, STUDYCOURSE_HEADER_FIELDS,'update')
245        studycourse = self.importer.getEntry(dict(reg_number='1'), self.app)
246        self.assertEqual(num_warns,0)
247        self.assertEqual(studycourse.certificate.code, u'CERT1')
248        shutil.rmtree(os.path.dirname(fin_file))
249
250def test_suite():
251    suite = unittest.TestSuite()
252    for testcase in [
253        StudentImporterTest,StudentStudyCourseImporterTest,
254        ]:
255        suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
256                testcase
257                )
258        )
259    return suite
Note: See TracBrowser for help on using the repository browser.