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

Last change on this file since 13216 was 13073, checked in by Henrik Bettermann, 10 years ago

Rename 'application section' for the sake of consistency.
  • Property svn:keywords set to Id
File size: 17.4 KB
Line 
1## $Id: test_batching.py 13073 2015-06-18 08:13:49Z 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('app2014', None)
154        container2 = self.app['applicants'].get('app2015', 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'app2014')
160        self.assertEqual(container.title, u'General Studies')
161        self.assertEqual(container.prefix, u'app')
162        self.assertEqual(container.year, 2014)
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 Processor - sample_applicant_data - imported: '
307            'applicant_id=dp2011_1234, password=mypwd1, '
308            'reg_number=1001, firstname=Aaren, middlename=Peter, lastname=Pieri, '
309            'sex=m, course1=CERT1, date_of_birth=1990-01-02, email=xx@yy.zz' in
310            logcontent)
311        # create applicant with random application_number which is
312        # not shown in the log file
313        self.assertTrue(
314            'Applicant Processor - sample_applicant_data - imported: '
315            'reg_number=1003, firstname=Aaren, '
316            'middlename=Alfons, lastname=Berson, sex=m, course1=CERT1, '
317            'date_of_birth=1990-01-04, email=xx@yy.zz' in
318            logcontent)
319        # Logging message from handle_applicant_transition_event
320        self.assertTrue(
321            'dp2011_1234 - Application initialized' in
322            logcontent)
323        shutil.rmtree(os.path.dirname(fin_file))
324
325    def test_import_faulty(self):
326        num, num_warns, fin_file, fail_file = self.processor.doImport(
327            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
328        # we cannot import data with faulty dates. A date is faulty
329        # when in format xx/yy/zzzz as we cannot say whether it is
330        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
331        for applicant in self.app['applicants']['dp2011'].values():
332            if applicant.date_of_birth == datetime.date(1990, 1, 2):
333                self.fail(
334                    'Wrong birthdate of imported applicant '
335                    '(1990-01-02, should be: 1990-02-01)')
336        self.assertEqual(num_warns,4)
337        fail_contents = open(fail_file, 'rb').read()
338        # CERT2 is in wrong category ...
339        self.assertTrue('course1: wrong application category' in fail_contents)
340        # ... and CERT3 does not exist.
341        self.assertTrue('course1: Invalid value' in fail_contents)
342        # Conversion checking already fails because Mister or Miss No's
343        # container does not exist.
344        self.assertTrue('no@yy.zz,container: not found' in fail_contents)
345        self.assertTrue('nobody@yy.zz,container: not found' in fail_contents)
346        shutil.rmtree(os.path.dirname(fail_file))
347        return
348
349    def test_import_update(self):
350        num, num_warns, fin_file, fail_file = self.processor.doImport(
351            self.csv_file, APPLICANT_HEADER_FIELDS)
352        shutil.rmtree(os.path.dirname(fin_file))
353        num, num_warns, fin_file, fail_file = self.processor.doImport(
354            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
355        self.assertEqual(num_warns,0)
356        # The middlename import value was None.
357        # Confirm that middlename has not been deleted.
358        container = self.app['applicants']['dp2011']
359        self.assertEqual(container['1234'].middlename, 'Peter')
360        # state of Pieri has not changed
361        self.assertEqual(container['1234'].state,'initialized')
362        # state of Finau has changed
363        self.assertEqual(container['2345'].state,'admitted')
364        # password of Pieri has been set
365        self.assertTrue(IUserAccount(container['1234']).checkPassword('mypwd1'))
366        # password of Finau is still unset
367        self.assertEqual(IUserAccount(container['2345']).password,None)
368        # password of Simon was encrypted already
369        self.assertTrue(
370            IUserAccount(container['4567']).checkPassword('mypwd1'))
371        # reg_number of Finau has changed
372        self.assertEqual(container['2345'].reg_number, '6666')
373        logcontent = open(self.logfile).read()
374        # Logging message from updateEntry,
375        # reg_number is locator
376        self.assertTrue(
377            'Applicant Processor - sample_applicant_data_update - updated: '
378            'reg_number=1001, firstname=Aaren' in
379            logcontent)
380        # applicant_id is locator
381        self.assertTrue(
382            'Applicant Processor - sample_applicant_data_update - updated: '
383            'state=admitted, reg_number=6666, '
384            'firstname=Alfons, applicant_id=dp2011_2345' in
385            logcontent)
386        shutil.rmtree(os.path.dirname(fin_file))
387
388        # Now we import another file which clears all middlename attributes
389        # and uses the new reg_number as locator. This test also checks
390        # if the catalog has been informed about the reg_no change and if
391        # applicants in state created are really blocked.
392        IWorkflowState(container['4567']).setState(CREATED)
393        num, num_warns, fin_file, fail_file = self.processor.doImport(
394            self.csv_file_update2, APPLICANT_HEADER_FIELDS_UPDATE2, 'update')
395        failcontent = open(fail_file).read()
396        self.assertTrue('Applicant is blocked' in failcontent)
397        self.assertEqual(num_warns,1)
398        # Middlename is cleared.
399        assert container['1234'].middlename is None
400        # Firstname of applicant in state created isn't changed.
401        self.assertEqual(container['4567'].firstname, 'Simon')
402        shutil.rmtree(os.path.dirname(fin_file))
403
404    def test_import_remove(self):
405        num, num_warns, fin_file, fail_file = self.processor.doImport(
406            self.csv_file, APPLICANT_HEADER_FIELDS)
407        shutil.rmtree(os.path.dirname(fin_file))
408        num, num_warns, fin_file, fail_file = self.processor.doImport(
409            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
410        self.assertEqual(num_warns,0)
411        logcontent = open(self.logfile).read()
412        # Logging message from handle_applicant_transition_event
413        self.assertTrue(
414            'dp2011_1234 - Application record removed' in
415            logcontent)
416        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.