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

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

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