source: main/waeup.kofa/trunk/src/waeup/kofa/applicants/tests/test_batching.py @ 8317

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

Remove entry_level attribute. The entry level is given by the start level of the study course admitted.
  • Property svn:keywords set to Id
File size: 13.9 KB
Line 
1## $Id: test_batching.py 8311 2012-04-29 19:47:24Z 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 applicants-related data processors.
19"""
20import datetime
21import os
22import pytz
23import shutil
24import tempfile
25import unittest
26from zope.component.hooks import setSite, clearSite
27from zope.component import createObject
28from zope.interface.verify import verifyClass, verifyObject
29
30from waeup.kofa.app import University
31from waeup.kofa.applicants.batching import (
32    ApplicantsContainerProcessor, ApplicantProcessor)
33from waeup.kofa.applicants.container import ApplicantsContainer
34from waeup.kofa.applicants.applicant import Applicant
35from waeup.kofa.university.faculty import Faculty
36from waeup.kofa.university.department import Department
37from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
38from waeup.kofa.interfaces import IBatchProcessor, IUserAccount
39
40
41# Sample data we can use in tests...
42APPS_CONTAINER_SAMPLE_DATA = open(
43    os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'),
44    'rb').read()
45
46# The header fields of the above CSV snippet
47APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split(
48    '\n')[0].split(',')
49
50# The same for applicants
51APPLICANT_SAMPLE_DATA = open(
52    os.path.join(os.path.dirname(__file__), 'sample_applicant_data.csv'),
53    'rb').read()
54FAULTY_APPLICANT_SAMPLE_DATA = open(
55    os.path.join(os.path.dirname(__file__),
56                 'sample_faulty_applicant_data.csv'), 'rb').read()
57
58APPLICANT_HEADER_FIELDS = APPLICANT_SAMPLE_DATA.split(
59    '\n')[0].split(',')
60
61APPLICANT_SAMPLE_DATA_UPDATE = open(
62    os.path.join(os.path.dirname(__file__),
63                 'sample_applicant_data_update.csv'), 'rb').read()
64
65APPLICANT_SAMPLE_DATA_UPDATE2 = open(
66    os.path.join(os.path.dirname(__file__),
67                 'sample_applicant_data_update2.csv'), 'rb').read()
68
69APPLICANT_HEADER_FIELDS_UPDATE = APPLICANT_SAMPLE_DATA_UPDATE.split(
70    '\n')[0].split(',')
71
72class ApplicantsContainerProcessorTest(FunctionalTestCase):
73
74    layer = FunctionalLayer
75
76    def setUp(self):
77        super(ApplicantsContainerProcessorTest, self).setUp()
78
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        self.app = self.getRootFolder()['app']
87        self.container = ApplicantsContainer()
88        self.container.code = u'dp2011'
89        self.app['applicants']['dp2011'] = self.container
90
91        self.processor = ApplicantsContainerProcessor()
92        self.workdir = tempfile.mkdtemp()
93        self.csv_file = os.path.join(self.workdir, 'sampledata.csv')
94        open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA)
95        setSite(self.app)
96        return
97
98    def tearDown(self):
99        super(ApplicantsContainerProcessorTest, self).tearDown()
100        shutil.rmtree(self.workdir)
101        shutil.rmtree(self.dc_root)
102        clearSite()
103        return
104
105    def test_interface(self):
106        # Make sure we fulfill the interface contracts.
107        assert verifyObject(IBatchProcessor, self.processor) is True
108        assert verifyClass(
109            IBatchProcessor, ApplicantsContainerProcessor) is True
110
111    def test_parentsExist(self):
112        assert self.processor.parentsExist(None, dict()) is False
113        assert self.processor.parentsExist(None, self.app) is True
114
115    def test_entryExists(self):
116        assert self.processor.entryExists(
117            dict(code='REG_NONE'), self.app) is False
118        assert self.processor.entryExists(
119            dict(code='dp2011'), self.app) is True
120
121    def test_getParent(self):
122        parent = self.processor.getParent(None, self.app)
123        assert parent is self.app['applicants']
124
125    def test_getEntry(self):
126        assert self.processor.getEntry(
127            dict(code='REG_NONE'), self.app) is None
128        assert self.processor.getEntry(
129            dict(code='dp2011'), self.app) is self.container
130
131    def test_addEntry(self):
132        self.processor.addEntry(
133            'New application', dict(code='dp2012'), self.app)
134        assert self.app['applicants']['dp2012'] == 'New application'
135
136    def test_delEntry(self):
137        self.processor.delEntry(dict(code='dp2011'), self.app)
138        assert 'dp2011' not in self.app['applicants'].keys()
139
140    def test_import(self):
141        # Do a real import
142        # see local sample_container.csv file for input
143        num, num_warns, fin_file, fail_file = self.processor.doImport(
144            self.csv_file, APPS_CONTAINER_HEADER_FIELDS)
145        avail_containers = [x for x in self.app['applicants'].keys()]
146        container = self.app['applicants'].get('app2012', None)
147        container2 = self.app['applicants'].get('app2013', None)
148        self.assertTrue(container is not None)
149        self.assertTrue(container2 is not None)
150
151        # check attributes
152        self.assertEqual(container.code, u'app2012')
153        self.assertEqual(container.title, u'General Studies 2012/2013')
154        self.assertEqual(container.prefix, u'app')
155        self.assertEqual(container.year, 2012)
156        self.assertEqual(container.application_category, 'basic')
157        self.assertEqual(
158            container.description,
159            u'This text can been seen by anonymous users.\n'
160            u'>>de<<\nDieser Text kann von anonymen Benutzern '
161            u'gelesen werden.')
162        self.assertEqual(container.startdate,
163                         datetime.datetime(2012, 3, 1, 0, 0, tzinfo=pytz.utc))
164        self.assertEqual(container.enddate,
165                         datetime.datetime(2012, 4, 25, 0, 0, tzinfo=pytz.utc))
166        shutil.rmtree(os.path.dirname(fin_file))
167
168class ApplicantImportExportSetup(FunctionalTestCase):
169
170    layer = FunctionalLayer
171
172    def setUp(self):
173        super(ApplicantImportExportSetup, self).setUp()
174        # Setup a sample site for each test
175        app = University()
176        self.dc_root = tempfile.mkdtemp()
177        app['datacenter'].setStoragePath(self.dc_root)
178
179        # Prepopulate the ZODB...
180        self.getRootFolder()['app'] = app
181        # we add the site immediately after creation to the
182        # ZODB. Catalogs and other local utilities are not setup
183        # before that step.
184        self.app = self.getRootFolder()['app']
185        # Set site here. Some of the following setup code might need
186        # to access grok.getSite() and should get our new app then
187        setSite(app)
188
189        # Add an applicants container
190        self.container = ApplicantsContainer()
191        self.container.code = u'dp2011'
192        self.app['applicants']['dp2011'] = self.container
193
194        # Populate university
195        self.certificate = createObject('waeup.Certificate')
196        self.certificate.code = 'CERT1'
197        self.certificate.application_category = 'basic'
198        self.certificate.start_level = 100
199        self.certificate.end_level = 500
200        self.app['faculties']['fac1'] = Faculty()
201        self.app['faculties']['fac1']['dep1'] = Department()
202        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
203            self.certificate)
204
205        # Add applicant with subobjects
206        applicant = Applicant()
207        applicant.firstname = u'Anna'
208        applicant.lastname = u'Tester'
209        self.app['applicants']['dp2011'].addApplicant(applicant)
210        self.application_number = applicant.application_number
211        self.applicant = self.app['applicants']['dp2011'][
212            self.application_number]
213        self.workdir = tempfile.mkdtemp()
214        return
215
216    def tearDown(self):
217        super(ApplicantImportExportSetup, self).tearDown()
218        shutil.rmtree(self.workdir)
219        shutil.rmtree(self.dc_root)
220        clearSite()
221        return
222
223class ApplicantProcessorTest(ApplicantImportExportSetup):
224
225    layer = FunctionalLayer
226
227    def setUp(self):
228        super(ApplicantProcessorTest, self).setUp()
229        self.processor = ApplicantProcessor()
230        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
231        self.csv_file_faulty = os.path.join(self.workdir,
232                                            'faulty_applicant_data.csv')
233        self.csv_file_update = os.path.join(
234            self.workdir, 'sample_applicant_data_update.csv')
235        self.csv_file_update2 = os.path.join(
236            self.workdir, 'sample_applicant_data_update2.csv')
237        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
238        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
239        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
240        open(self.csv_file_update2, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE2)
241
242    def test_interface(self):
243        # Make sure we fulfill the interface contracts.
244        assert verifyObject(IBatchProcessor, self.processor) is True
245        assert verifyClass(
246            IBatchProcessor, ApplicantProcessor) is True
247
248    def test_entryExists(self):
249        assert self.processor.entryExists(
250            dict(container_code='dp2011', application_number='999'),
251            self.app) is False
252
253    def test_getEntry(self):
254        applicant = self.processor.getEntry(
255            dict(container_code='dp2011',
256                 application_number=self.application_number), self.app)
257        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
258
259    def test_addEntry(self):
260        new_applicant = Applicant()
261        self.processor.addEntry(
262            new_applicant, dict(container_code='dp2011'), self.app)
263        assert len(self.app['applicants']['dp2011'].keys()) == 2
264
265    def test_delEntry(self):
266        assert self.application_number in self.app[
267            'applicants']['dp2011'].keys()
268        self.processor.delEntry(
269            dict(container_code='dp2011',
270                application_number=self.application_number), self.app)
271        assert self.application_number not in self.app[
272            'applicants']['dp2011'].keys()
273
274    def test_import(self):
275        num, num_warns, fin_file, fail_file = self.processor.doImport(
276            self.csv_file, APPLICANT_HEADER_FIELDS)
277        self.assertEqual(num_warns,0)
278        keys = self.app['applicants']['dp2011'].keys()
279        assert len(keys) == 4
280        container = self.app['applicants']['dp2011']
281        assert  container.__implemented__.__name__ == (
282            'waeup.kofa.applicants.container.ApplicantsContainer')
283        applicant = container[keys[0]]
284        assert applicant.__implemented__.__name__ == (
285            'waeup.kofa.applicants.applicant.Applicant')
286        shutil.rmtree(os.path.dirname(fin_file))
287
288    def test_import_faulty(self):
289        # we cannot import data with faulty dates. A date is faulty
290        # when in format xx/yy/zzzz as we cannot say whether it is
291        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
292        num, num_warns, fin_file, fail_file = self.processor.doImport(
293            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
294        if fail_file is not None:
295            fail_contents = open(fail_file, 'rb').read()
296            shutil.rmtree(os.path.dirname(fail_file))
297        else:
298            shutil.rmtree(os.path.dirname(fin_file))
299        for applicant in self.app['applicants']['dp2011'].values():
300            if applicant.date_of_birth == datetime.date(1990, 1, 2):
301                self.fail(
302                    'Wrong birthdate of imported applicant '
303                    '(1990-01-02, should be: 1990-02-01)')
304        return
305
306    def test_import_update(self):
307        num, num_warns, fin_file, fail_file = self.processor.doImport(
308            self.csv_file, APPLICANT_HEADER_FIELDS)
309        shutil.rmtree(os.path.dirname(fin_file))
310        num, num_warns, fin_file, fail_file = self.processor.doImport(
311            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
312        #content = open(fail_file).read()
313        self.assertEqual(num_warns,0)
314        # The middlename import value was None.
315        # Confirm that middlename has not been deleted.
316        container = self.app['applicants']['dp2011']
317        self.assertEqual(container['1234'].middlename, 'Peter')
318        # state of Pieri has not changed
319        self.assertEqual(container['1234'].state,'initialized')
320        # state of Finau has changed
321        self.assertEqual(container['2345'].state,'admitted')
322        # password of Pieri has been set
323        self.assertTrue(IUserAccount(container['1234']).checkPassword('mypwd1'))
324        # password of Finau is still unset
325        self.assertEqual(IUserAccount(container['2345']).password,'')
326        shutil.rmtree(os.path.dirname(fin_file))
327        # Now we import another file which clears all middlename attributes.
328        num, num_warns, fin_file, fail_file = self.processor.doImport(
329            self.csv_file_update2, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
330        self.assertEqual(num_warns,0)
331        assert container['1234'].middlename is None
332        shutil.rmtree(os.path.dirname(fin_file))
333
334    def test_import_remove(self):
335        num, num_warns, fin_file, fail_file = self.processor.doImport(
336            self.csv_file, APPLICANT_HEADER_FIELDS)
337        shutil.rmtree(os.path.dirname(fin_file))
338        num, num_warns, fin_file, fail_file = self.processor.doImport(
339            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
340        self.assertEqual(num_warns,0)
341        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.