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

Last change on this file since 8610 was 8610, checked in by Henrik Bettermann, 12 years ago

Adding another certificate with code CERT2 but in another application category now leads to an assertion error, i.e. test_import_faulty fails because the second record is being imported although the application category is not correct.
  • Property svn:keywords set to Id
File size: 16.8 KB
Line 
1## $Id: test_batching.py 8610 2012-06-03 09:07:07Z 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.app['applicants']['dp2011'] = self.container
200
201        # Populate university
202        self.certificate = createObject('waeup.Certificate')
203        self.certificate.code = 'CERT1'
204        self.certificate.application_category = 'basic'
205        self.certificate.start_level = 100
206        self.certificate.end_level = 500
207        self.app['faculties']['fac1'] = Faculty()
208        self.app['faculties']['fac1']['dep1'] = Department()
209        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
210            self.certificate)
211        self.certificate2 = createObject('waeup.Certificate')
212        self.certificate2.code = 'CERT2'
213        self.certificate2.application_category = 'xyz'
214        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
215            self.certificate2)
216
217        # Add applicant with subobjects
218        applicant = Applicant()
219        applicant.firstname = u'Anna'
220        applicant.lastname = u'Tester'
221        self.app['applicants']['dp2011'].addApplicant(applicant)
222        self.application_number = applicant.application_number
223        self.applicant = self.app['applicants']['dp2011'][
224            self.application_number]
225        self.workdir = tempfile.mkdtemp()
226        return
227
228    def tearDown(self):
229        super(ApplicantImportExportSetup, self).tearDown()
230        shutil.rmtree(self.workdir)
231        shutil.rmtree(self.dc_root)
232        clearSite()
233        return
234
235class ApplicantProcessorTest(ApplicantImportExportSetup):
236
237    layer = FunctionalLayer
238
239    def setUp(self):
240        super(ApplicantProcessorTest, self).setUp()
241        self.processor = ApplicantProcessor()
242        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
243        self.csv_file_faulty = os.path.join(self.workdir,
244                                            'faulty_applicant_data.csv')
245        self.csv_file_update = os.path.join(
246            self.workdir, 'sample_applicant_data_update.csv')
247        self.csv_file_update2 = os.path.join(
248            self.workdir, 'sample_applicant_data_update2.csv')
249        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
250        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
251        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
252        open(self.csv_file_update2, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE2)
253
254        self.logfile = os.path.join(
255            self.app['datacenter'].storage, 'logs', 'applicants.log')
256
257    def test_interface(self):
258        # Make sure we fulfill the interface contracts.
259        assert verifyObject(IBatchProcessor, self.processor) is True
260        assert verifyClass(
261            IBatchProcessor, ApplicantProcessor) is True
262
263    def test_entryExists(self):
264        assert self.processor.entryExists(
265            dict(container_code='dp2011', application_number='999'),
266            self.app) is False
267
268    def test_getEntry(self):
269        applicant = self.processor.getEntry(
270            dict(container_code='dp2011',
271                 application_number=self.application_number), self.app)
272        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
273
274    def test_addEntry(self):
275        new_applicant = Applicant()
276        self.processor.addEntry(
277            new_applicant, dict(container_code='dp2011'), self.app)
278        assert len(self.app['applicants']['dp2011'].keys()) == 2
279
280    def test_delEntry(self):
281        assert self.application_number in self.app[
282            'applicants']['dp2011'].keys()
283        self.processor.delEntry(
284            dict(container_code='dp2011',
285                application_number=self.application_number), self.app)
286        assert self.application_number not in self.app[
287            'applicants']['dp2011'].keys()
288
289    def test_import(self):
290        num, num_warns, fin_file, fail_file = self.processor.doImport(
291            self.csv_file, APPLICANT_HEADER_FIELDS)
292        self.assertEqual(num_warns,0)
293        keys = self.app['applicants']['dp2011'].keys()
294        assert len(keys) == 5
295        container = self.app['applicants']['dp2011']
296        assert  container.__implemented__.__name__ == (
297            'waeup.kofa.applicants.container.ApplicantsContainer')
298        applicant = container[keys[0]]
299        assert applicant.__implemented__.__name__ == (
300            'waeup.kofa.applicants.applicant.Applicant')
301        logcontent = open(self.logfile).read()
302        # Logging message from updateEntry,
303        # create applicant with given application_number
304        self.assertTrue(
305            'Applicant imported: applicant_id=dp2011_1234, password=mypwd1, '
306            'reg_number=1001, firstname=Aaren, middlename=Peter, lastname=Pieri, '
307            'sex=m, course1=CERT1, date_of_birth=1990-01-02, email=xx@yy.zz' in
308            logcontent)
309        # create applicant with random application_number which is
310        # not shown in the log file
311        self.assertTrue(
312            'Applicant imported: reg_number=1003, firstname=Aaren, '
313            'middlename=Alfons, lastname=Berson, sex=m, course1=CERT1, '
314            'date_of_birth=1990-01-04, email=xx@yy.zz' in
315            logcontent)
316        # Logging message from handle_applicant_transition_event
317        self.assertTrue(
318            'dp2011_1234 - Application initialized' in
319            logcontent)
320        shutil.rmtree(os.path.dirname(fin_file))
321
322    def test_import_faulty(self):
323        num, num_warns, fin_file, fail_file = self.processor.doImport(
324            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
325        # we cannot import data with faulty dates. A date is faulty
326        # when in format xx/yy/zzzz as we cannot say whether it is
327        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
328        for applicant in self.app['applicants']['dp2011'].values():
329            if applicant.date_of_birth == datetime.date(1990, 1, 2):
330                self.fail(
331                    'Wrong birthdate of imported applicant '
332                    '(1990-01-02, should be: 1990-02-01)')
333        # CERT2 does not exist.
334        self.assertEqual(num_warns,1)
335        fail_contents = open(fail_file, 'rb').read()
336        shutil.rmtree(os.path.dirname(fail_file))
337        self.assertTrue('course1: Invalid value' in fail_contents)
338        return
339
340    def test_import_update(self):
341        num, num_warns, fin_file, fail_file = self.processor.doImport(
342            self.csv_file, APPLICANT_HEADER_FIELDS)
343        shutil.rmtree(os.path.dirname(fin_file))
344        num, num_warns, fin_file, fail_file = self.processor.doImport(
345            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
346        self.assertEqual(num_warns,0)
347        # The middlename import value was None.
348        # Confirm that middlename has not been deleted.
349        container = self.app['applicants']['dp2011']
350        self.assertEqual(container['1234'].middlename, 'Peter')
351        # state of Pieri has not changed
352        self.assertEqual(container['1234'].state,'initialized')
353        # state of Finau has changed
354        self.assertEqual(container['2345'].state,'admitted')
355        # password of Pieri has been set
356        self.assertTrue(IUserAccount(container['1234']).checkPassword('mypwd1'))
357        # password of Finau is still unset
358        self.assertEqual(IUserAccount(container['2345']).password,None)
359        # password of Simon was encrypted already
360        self.assertTrue(
361            IUserAccount(container['4567']).checkPassword('mypwd1'))
362        # reg_number of Finau has changed
363        self.assertEqual(container['2345'].reg_number, '6666')
364        logcontent = open(self.logfile).read()
365        # Logging message from updateEntry,
366        # reg_number is locator
367        self.assertTrue(
368            'Applicant updated: reg_number=1001, firstname=Aaren' in
369            logcontent)
370        # applicant_id is locator
371        self.assertTrue(
372            'Applicant updated: state=admitted, reg_number=6666, '
373            'firstname=Alfons, applicant_id=dp2011_2345' in
374            logcontent)
375        shutil.rmtree(os.path.dirname(fin_file))
376
377        # Now we import another file which clears all middlename attributes
378        # and uses the new reg_number as locator. This test also checks
379        # if the catalog has been informed about the reg_no change and if
380        # applicants in state created are really blocked.
381        IWorkflowState(container['4567']).setState(CREATED)
382        num, num_warns, fin_file, fail_file = self.processor.doImport(
383            self.csv_file_update2, APPLICANT_HEADER_FIELDS_UPDATE2, 'update')
384        failcontent = open(fail_file).read()
385        self.assertTrue('Applicant is blocked' in failcontent)
386        self.assertEqual(num_warns,1)
387        # Middlename is cleared.
388        assert container['1234'].middlename is None
389        # Firstname of applicant in state created isn't changed.
390        self.assertEqual(container['4567'].firstname, 'Simon')
391        shutil.rmtree(os.path.dirname(fin_file))
392
393    def test_import_remove(self):
394        num, num_warns, fin_file, fail_file = self.processor.doImport(
395            self.csv_file, APPLICANT_HEADER_FIELDS)
396        shutil.rmtree(os.path.dirname(fin_file))
397        num, num_warns, fin_file, fail_file = self.processor.doImport(
398            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
399        self.assertEqual(num_warns,0)
400        logcontent = open(self.logfile).read()
401        # Logging message from handle_applicant_transition_event
402        self.assertTrue(
403            'dp2011_1234 - Applicant record removed' in
404            logcontent)
405        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.