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

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

Now we have two different error messages which is great.
  • Property svn:keywords set to Id
File size: 17.0 KB
Line 
1## $Id: test_batching.py 8616 2012-06-03 13:50:22Z 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
26import grok
27from hurry.workflow.interfaces import IWorkflowState
28from zope.component.hooks import setSite, clearSite
29from zope.component import createObject
30from zope.interface.verify import verifyClass, verifyObject
31from zope.event import notify
32
33from waeup.kofa.app import University
34from waeup.kofa.applicants.batching import (
35    ApplicantsContainerProcessor, ApplicantProcessor)
36from waeup.kofa.applicants.container import ApplicantsContainer
37from waeup.kofa.applicants.applicant import Applicant
38from waeup.kofa.university.faculty import Faculty
39from waeup.kofa.university.department import Department
40from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
41from waeup.kofa.interfaces import IBatchProcessor, IUserAccount
42from waeup.kofa.applicants.workflow import CREATED
43
44
45# Sample data we can use in tests...
46APPS_CONTAINER_SAMPLE_DATA = open(
47    os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'),
48    'rb').read()
49
50# The header fields of the above CSV snippet
51APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split(
52    '\n')[0].split(',')
53
54# The same for applicants
55APPLICANT_SAMPLE_DATA = open(
56    os.path.join(os.path.dirname(__file__), 'sample_applicant_data.csv'),
57    'rb').read()
58FAULTY_APPLICANT_SAMPLE_DATA = open(
59    os.path.join(os.path.dirname(__file__),
60                 'sample_faulty_applicant_data.csv'), 'rb').read()
61
62APPLICANT_HEADER_FIELDS = APPLICANT_SAMPLE_DATA.split(
63    '\n')[0].split(',')
64
65APPLICANT_SAMPLE_DATA_UPDATE = open(
66    os.path.join(os.path.dirname(__file__),
67                 'sample_applicant_data_update.csv'), 'rb').read()
68
69APPLICANT_SAMPLE_DATA_UPDATE2 = open(
70    os.path.join(os.path.dirname(__file__),
71                 'sample_applicant_data_update2.csv'), 'rb').read()
72
73APPLICANT_HEADER_FIELDS_UPDATE = APPLICANT_SAMPLE_DATA_UPDATE.split(
74    '\n')[0].split(',')
75
76APPLICANT_HEADER_FIELDS_UPDATE2 = APPLICANT_SAMPLE_DATA_UPDATE2.split(
77    '\n')[0].split(',')
78
79class ApplicantsContainerProcessorTest(FunctionalTestCase):
80
81    layer = FunctionalLayer
82
83    def setUp(self):
84        super(ApplicantsContainerProcessorTest, self).setUp()
85
86        # Setup a sample site for each test
87        app = University()
88        self.dc_root = tempfile.mkdtemp()
89        app['datacenter'].setStoragePath(self.dc_root)
90
91        # Prepopulate the ZODB...
92        self.getRootFolder()['app'] = app
93        self.app = self.getRootFolder()['app']
94        self.container = ApplicantsContainer()
95        self.container.code = u'dp2011'
96        self.app['applicants']['dp2011'] = self.container
97
98        self.processor = ApplicantsContainerProcessor()
99        self.workdir = tempfile.mkdtemp()
100        self.csv_file = os.path.join(self.workdir, 'sampledata.csv')
101        open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA)
102        setSite(self.app)
103        return
104
105    def tearDown(self):
106        super(ApplicantsContainerProcessorTest, 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.processor) is True
115        assert verifyClass(
116            IBatchProcessor, ApplicantsContainerProcessor) is True
117
118    def test_parentsExist(self):
119        assert self.processor.parentsExist(None, dict()) is False
120        assert self.processor.parentsExist(None, self.app) is True
121
122    def test_entryExists(self):
123        assert self.processor.entryExists(
124            dict(code='REG_NONE'), self.app) is False
125        assert self.processor.entryExists(
126            dict(code='dp2011'), self.app) is True
127
128    def test_getParent(self):
129        parent = self.processor.getParent(None, self.app)
130        assert parent is self.app['applicants']
131
132    def test_getEntry(self):
133        assert self.processor.getEntry(
134            dict(code='REG_NONE'), self.app) is None
135        assert self.processor.getEntry(
136            dict(code='dp2011'), self.app) is self.container
137
138    def test_addEntry(self):
139        self.processor.addEntry(
140            'New application', dict(code='dp2012'), self.app)
141        assert self.app['applicants']['dp2012'] == 'New application'
142
143    def test_delEntry(self):
144        self.processor.delEntry(dict(code='dp2011'), self.app)
145        assert 'dp2011' not in self.app['applicants'].keys()
146
147    def test_import(self):
148        # Do a real import
149        # see local sample_container.csv file for input
150        num, num_warns, fin_file, fail_file = self.processor.doImport(
151            self.csv_file, APPS_CONTAINER_HEADER_FIELDS)
152        avail_containers = [x for x in self.app['applicants'].keys()]
153        container = self.app['applicants'].get('app2012', None)
154        container2 = self.app['applicants'].get('app2013', None)
155        self.assertTrue(container is not None)
156        self.assertTrue(container2 is not None)
157
158        # check attributes
159        self.assertEqual(container.code, u'app2012')
160        self.assertEqual(container.title, u'General Studies 2012/2013')
161        self.assertEqual(container.prefix, u'app')
162        self.assertEqual(container.year, 2012)
163        self.assertEqual(container.application_category, 'basic')
164        self.assertEqual(
165            container.description,
166            u'This text can been seen by anonymous users.\n'
167            u'>>de<<\nDieser Text kann von anonymen Benutzern '
168            u'gelesen werden.')
169        self.assertEqual(container.startdate,
170                         datetime.datetime(2012, 3, 1, 0, 0, tzinfo=pytz.utc))
171        self.assertEqual(container.enddate,
172                         datetime.datetime(2012, 4, 25, 0, 0, tzinfo=pytz.utc))
173        shutil.rmtree(os.path.dirname(fin_file))
174
175class ApplicantImportExportSetup(FunctionalTestCase):
176
177    layer = FunctionalLayer
178
179    def setUp(self):
180        super(ApplicantImportExportSetup, self).setUp()
181        # Setup a sample site for each test
182        app = University()
183        self.dc_root = tempfile.mkdtemp()
184        app['datacenter'].setStoragePath(self.dc_root)
185
186        # Prepopulate the ZODB...
187        self.getRootFolder()['app'] = app
188        # we add the site immediately after creation to the
189        # ZODB. Catalogs and other local utilities are not setup
190        # before that step.
191        self.app = self.getRootFolder()['app']
192        # Set site here. Some of the following setup code might need
193        # to access grok.getSite() and should get our new app then
194        setSite(app)
195
196        # Add an applicants container
197        self.container = ApplicantsContainer()
198        self.container.code = u'dp2011'
199        self.container.application_category = u'basic'
200        self.app['applicants']['dp2011'] = self.container
201
202        # Populate university
203        self.certificate = createObject('waeup.Certificate')
204        self.certificate.code = 'CERT1'
205        self.certificate.application_category = 'basic'
206        self.certificate.start_level = 100
207        self.certificate.end_level = 500
208        self.app['faculties']['fac1'] = Faculty()
209        self.app['faculties']['fac1']['dep1'] = Department()
210        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
211            self.certificate)
212        self.certificate2 = createObject('waeup.Certificate')
213        self.certificate2.code = 'CERT2'
214        self.certificate2.application_category = 'xyz'
215        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
216            self.certificate2)
217
218        # Add applicant with subobjects
219        applicant = Applicant()
220        applicant.firstname = u'Anna'
221        applicant.lastname = u'Tester'
222        self.app['applicants']['dp2011'].addApplicant(applicant)
223        self.application_number = applicant.application_number
224        self.applicant = self.app['applicants']['dp2011'][
225            self.application_number]
226        self.workdir = tempfile.mkdtemp()
227        return
228
229    def tearDown(self):
230        super(ApplicantImportExportSetup, self).tearDown()
231        shutil.rmtree(self.workdir)
232        shutil.rmtree(self.dc_root)
233        clearSite()
234        return
235
236class ApplicantProcessorTest(ApplicantImportExportSetup):
237
238    layer = FunctionalLayer
239
240    def setUp(self):
241        super(ApplicantProcessorTest, self).setUp()
242        self.processor = ApplicantProcessor()
243        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
244        self.csv_file_faulty = os.path.join(self.workdir,
245                                            'faulty_applicant_data.csv')
246        self.csv_file_update = os.path.join(
247            self.workdir, 'sample_applicant_data_update.csv')
248        self.csv_file_update2 = os.path.join(
249            self.workdir, 'sample_applicant_data_update2.csv')
250        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
251        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
252        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
253        open(self.csv_file_update2, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE2)
254
255        self.logfile = os.path.join(
256            self.app['datacenter'].storage, 'logs', 'applicants.log')
257
258    def test_interface(self):
259        # Make sure we fulfill the interface contracts.
260        assert verifyObject(IBatchProcessor, self.processor) is True
261        assert verifyClass(
262            IBatchProcessor, ApplicantProcessor) is True
263
264    def test_entryExists(self):
265        assert self.processor.entryExists(
266            dict(container_code='dp2011', application_number='999'),
267            self.app) is False
268
269    def test_getEntry(self):
270        applicant = self.processor.getEntry(
271            dict(container_code='dp2011',
272                 application_number=self.application_number), self.app)
273        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
274
275    def test_addEntry(self):
276        new_applicant = Applicant()
277        self.processor.addEntry(
278            new_applicant, dict(container_code='dp2011'), self.app)
279        assert len(self.app['applicants']['dp2011'].keys()) == 2
280
281    def test_delEntry(self):
282        assert self.application_number in self.app[
283            'applicants']['dp2011'].keys()
284        self.processor.delEntry(
285            dict(container_code='dp2011',
286                application_number=self.application_number), self.app)
287        assert self.application_number not in self.app[
288            'applicants']['dp2011'].keys()
289
290    def test_import(self):
291        num, num_warns, fin_file, fail_file = self.processor.doImport(
292            self.csv_file, APPLICANT_HEADER_FIELDS)
293        self.assertEqual(num_warns,0)
294        keys = self.app['applicants']['dp2011'].keys()
295        assert len(keys) == 5
296        container = self.app['applicants']['dp2011']
297        assert  container.__implemented__.__name__ == (
298            'waeup.kofa.applicants.container.ApplicantsContainer')
299        applicant = container[keys[0]]
300        assert applicant.__implemented__.__name__ == (
301            'waeup.kofa.applicants.applicant.Applicant')
302        logcontent = open(self.logfile).read()
303        # Logging message from updateEntry,
304        # create applicant with given application_number
305        self.assertTrue(
306            'Applicant imported: applicant_id=dp2011_1234, password=mypwd1, '
307            'reg_number=1001, firstname=Aaren, middlename=Peter, lastname=Pieri, '
308            'sex=m, course1=CERT1, date_of_birth=1990-01-02, email=xx@yy.zz' in
309            logcontent)
310        # create applicant with random application_number which is
311        # not shown in the log file
312        self.assertTrue(
313            'Applicant imported: reg_number=1003, firstname=Aaren, '
314            'middlename=Alfons, lastname=Berson, sex=m, course1=CERT1, '
315            'date_of_birth=1990-01-04, email=xx@yy.zz' in
316            logcontent)
317        # Logging message from handle_applicant_transition_event
318        self.assertTrue(
319            'dp2011_1234 - Application initialized' in
320            logcontent)
321        shutil.rmtree(os.path.dirname(fin_file))
322
323    def test_import_faulty(self):
324        num, num_warns, fin_file, fail_file = self.processor.doImport(
325            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
326        # we cannot import data with faulty dates. A date is faulty
327        # when in format xx/yy/zzzz as we cannot say whether it is
328        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
329        for applicant in self.app['applicants']['dp2011'].values():
330            if applicant.date_of_birth == datetime.date(1990, 1, 2):
331                self.fail(
332                    'Wrong birthdate of imported applicant '
333                    '(1990-01-02, should be: 1990-02-01)')
334        self.assertEqual(num_warns,2)
335        fail_contents = open(fail_file, 'rb').read()
336        shutil.rmtree(os.path.dirname(fail_file))
337        # CERT2 is in wrong category ...
338        self.assertTrue('course1: wrong application category' in fail_contents)
339        # ... and CERT3 does not exist.
340        self.assertTrue('course1: Invalid value' in fail_contents)
341        return
342
343    def test_import_update(self):
344        num, num_warns, fin_file, fail_file = self.processor.doImport(
345            self.csv_file, APPLICANT_HEADER_FIELDS)
346        shutil.rmtree(os.path.dirname(fin_file))
347        num, num_warns, fin_file, fail_file = self.processor.doImport(
348            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
349        self.assertEqual(num_warns,0)
350        # The middlename import value was None.
351        # Confirm that middlename has not been deleted.
352        container = self.app['applicants']['dp2011']
353        self.assertEqual(container['1234'].middlename, 'Peter')
354        # state of Pieri has not changed
355        self.assertEqual(container['1234'].state,'initialized')
356        # state of Finau has changed
357        self.assertEqual(container['2345'].state,'admitted')
358        # password of Pieri has been set
359        self.assertTrue(IUserAccount(container['1234']).checkPassword('mypwd1'))
360        # password of Finau is still unset
361        self.assertEqual(IUserAccount(container['2345']).password,None)
362        # password of Simon was encrypted already
363        self.assertTrue(
364            IUserAccount(container['4567']).checkPassword('mypwd1'))
365        # reg_number of Finau has changed
366        self.assertEqual(container['2345'].reg_number, '6666')
367        logcontent = open(self.logfile).read()
368        # Logging message from updateEntry,
369        # reg_number is locator
370        self.assertTrue(
371            'Applicant updated: reg_number=1001, firstname=Aaren' in
372            logcontent)
373        # applicant_id is locator
374        self.assertTrue(
375            'Applicant updated: state=admitted, reg_number=6666, '
376            'firstname=Alfons, applicant_id=dp2011_2345' in
377            logcontent)
378        shutil.rmtree(os.path.dirname(fin_file))
379
380        # Now we import another file which clears all middlename attributes
381        # and uses the new reg_number as locator. This test also checks
382        # if the catalog has been informed about the reg_no change and if
383        # applicants in state created are really blocked.
384        IWorkflowState(container['4567']).setState(CREATED)
385        num, num_warns, fin_file, fail_file = self.processor.doImport(
386            self.csv_file_update2, APPLICANT_HEADER_FIELDS_UPDATE2, 'update')
387        failcontent = open(fail_file).read()
388        self.assertTrue('Applicant is blocked' in failcontent)
389        self.assertEqual(num_warns,1)
390        # Middlename is cleared.
391        assert container['1234'].middlename is None
392        # Firstname of applicant in state created isn't changed.
393        self.assertEqual(container['4567'].firstname, 'Simon')
394        shutil.rmtree(os.path.dirname(fin_file))
395
396    def test_import_remove(self):
397        num, num_warns, fin_file, fail_file = self.processor.doImport(
398            self.csv_file, APPLICANT_HEADER_FIELDS)
399        shutil.rmtree(os.path.dirname(fin_file))
400        num, num_warns, fin_file, fail_file = self.processor.doImport(
401            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
402        self.assertEqual(num_warns,0)
403        logcontent = open(self.logfile).read()
404        # Logging message from handle_applicant_transition_event
405        self.assertTrue(
406            'dp2011_1234 - Applicant record removed' in
407            logcontent)
408        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.