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

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

Empty If no value is provided in import files, attributes must not be cleared. Clear attribute only if value == DELETIONMARKER.
  • Property svn:keywords set to Id
File size: 13.6 KB
Line 
1## $Id: test_batching.py 8202 2012-04-18 05:12:32Z 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
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.entry_level, 100)
156        self.assertEqual(container.year, 2012)
157        self.assertEqual(container.application_category, 'basic')
158        self.assertEqual(
159            container.description,
160            u'This text can been seen by anonymous users.\n'
161            u'>>de<<\nDieser Text kann von anonymen Benutzern '
162            u'gelesen werden.')
163        self.assertEqual(container.startdate,
164                         datetime.datetime(2012, 3, 1, 0, 0, tzinfo=pytz.utc))
165        self.assertEqual(container.enddate,
166                         datetime.datetime(2012, 4, 25, 0, 0, tzinfo=pytz.utc))
167        shutil.rmtree(os.path.dirname(fin_file))
168
169class ApplicantImportExportSetup(FunctionalTestCase):
170
171    layer = FunctionalLayer
172
173    def setUp(self):
174        super(ApplicantImportExportSetup, self).setUp()
175        # Setup a sample site for each test
176        app = University()
177        self.dc_root = tempfile.mkdtemp()
178        app['datacenter'].setStoragePath(self.dc_root)
179
180        # Prepopulate the ZODB...
181        self.getRootFolder()['app'] = app
182        # we add the site immediately after creation to the
183        # ZODB. Catalogs and other local utilities are not setup
184        # before that step.
185        self.app = self.getRootFolder()['app']
186        # Set site here. Some of the following setup code might need
187        # to access grok.getSite() and should get our new app then
188        setSite(app)
189
190        # Add an applicants container
191        self.container = ApplicantsContainer()
192        self.container.code = u'dp2011'
193        self.app['applicants']['dp2011'] = self.container
194
195        # Populate university
196        self.certificate = createObject('waeup.Certificate')
197        self.certificate.code = 'CERT1'
198        self.certificate.application_category = 'basic'
199        self.certificate.start_level = 100
200        self.certificate.end_level = 500
201        self.app['faculties']['fac1'] = Faculty()
202        self.app['faculties']['fac1']['dep1'] = Department()
203        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
204            self.certificate)
205
206        # Add applicant with subobjects
207        applicant = Applicant()
208        applicant.firstname = u'Anna'
209        applicant.lastname = u'Tester'
210        self.app['applicants']['dp2011'].addApplicant(applicant)
211        self.application_number = applicant.application_number
212        self.applicant = self.app['applicants']['dp2011'][
213            self.application_number]
214        self.workdir = tempfile.mkdtemp()
215        return
216
217    def tearDown(self):
218        super(ApplicantImportExportSetup, self).tearDown()
219        shutil.rmtree(self.workdir)
220        shutil.rmtree(self.dc_root)
221        clearSite()
222        return
223
224class ApplicantProcessorTest(ApplicantImportExportSetup):
225
226    layer = FunctionalLayer
227
228    def setUp(self):
229        super(ApplicantProcessorTest, self).setUp()
230        self.processor = ApplicantProcessor()
231        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
232        self.csv_file_faulty = os.path.join(self.workdir,
233                                            'faulty_applicant_data.csv')
234        self.csv_file_update = os.path.join(
235            self.workdir, 'sample_applicant_data_update.csv')
236        self.csv_file_update2 = os.path.join(
237            self.workdir, 'sample_applicant_data_update2.csv')
238        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
239        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
240        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
241        open(self.csv_file_update2, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE2)
242
243    def test_interface(self):
244        # Make sure we fulfill the interface contracts.
245        assert verifyObject(IBatchProcessor, self.processor) is True
246        assert verifyClass(
247            IBatchProcessor, ApplicantProcessor) is True
248
249    def test_entryExists(self):
250        assert self.processor.entryExists(
251            dict(container_code='dp2011', application_number='999'),
252            self.app) is False
253
254    def test_getEntry(self):
255        applicant = self.processor.getEntry(
256            dict(container_code='dp2011',
257                 application_number=self.application_number), self.app)
258        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
259
260    def test_addEntry(self):
261        new_applicant = Applicant()
262        self.processor.addEntry(
263            new_applicant, dict(container_code='dp2011'), self.app)
264        assert len(self.app['applicants']['dp2011'].keys()) == 2
265
266    def test_delEntry(self):
267        assert self.application_number in self.app[
268            'applicants']['dp2011'].keys()
269        self.processor.delEntry(
270            dict(container_code='dp2011',
271                application_number=self.application_number), self.app)
272        assert self.application_number not in self.app[
273            'applicants']['dp2011'].keys()
274
275    def test_import(self):
276        num, num_warns, fin_file, fail_file = self.processor.doImport(
277            self.csv_file, APPLICANT_HEADER_FIELDS)
278        self.assertEqual(num_warns,0)
279        keys = self.app['applicants']['dp2011'].keys()
280        assert len(keys) == 4
281        container = self.app['applicants']['dp2011']
282        assert  container.__implemented__.__name__ == (
283            'waeup.kofa.applicants.container.ApplicantsContainer')
284        applicant = container[keys[0]]
285        assert applicant.__implemented__.__name__ == (
286            'waeup.kofa.applicants.applicant.Applicant')
287        shutil.rmtree(os.path.dirname(fin_file))
288
289    def test_import_faulty(self):
290        # we cannot import data with faulty dates. A date is faulty
291        # when in format xx/yy/zzzz as we cannot say whether it is
292        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
293        num, num_warns, fin_file, fail_file = self.processor.doImport(
294            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
295        if fail_file is not None:
296            fail_contents = open(fail_file, 'rb').read()
297            shutil.rmtree(os.path.dirname(fail_file))
298        else:
299            shutil.rmtree(os.path.dirname(fin_file))
300        for applicant in self.app['applicants']['dp2011'].values():
301            if applicant.date_of_birth == datetime.date(1990, 1, 2):
302                self.fail(
303                    'Wrong birthdate of imported applicant '
304                    '(1990-01-02, should be: 1990-02-01)')
305        return
306
307    def test_import_update(self):
308        num, num_warns, fin_file, fail_file = self.processor.doImport(
309            self.csv_file, APPLICANT_HEADER_FIELDS)
310        shutil.rmtree(os.path.dirname(fin_file))
311        num, num_warns, fin_file, fail_file = self.processor.doImport(
312            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
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        for key in container.keys():
318            if container[key].lastname == 'Pieri':
319                applicant = container[key]
320                break
321        self.assertEqual(applicant.middlename, 'Peter')
322        shutil.rmtree(os.path.dirname(fin_file))
323        # Now we import another file which clears all middlename attributes.
324        num, num_warns, fin_file, fail_file = self.processor.doImport(
325            self.csv_file_update2, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
326        self.assertEqual(num_warns,0)
327        assert applicant.middlename is None
328        shutil.rmtree(os.path.dirname(fin_file))
329
330    def test_import_remove(self):
331        num, num_warns, fin_file, fail_file = self.processor.doImport(
332            self.csv_file, APPLICANT_HEADER_FIELDS)
333        shutil.rmtree(os.path.dirname(fin_file))
334        num, num_warns, fin_file, fail_file = self.processor.doImport(
335            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
336        self.assertEqual(num_warns,0)
337        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.