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

Last change on this file since 8200 was 8200, checked in by uli, 12 years ago
  • Use datetime for applicants containers and
  • Register new datetime widget as default for entering datetimes.

Fixed all tests to work with the new stuff.

As this is just a quick shot I put all changes into one commit
to ease any later rollback.
  • Property svn:keywords set to Id
File size: 12.5 KB
Line 
1## $Id: test_batching.py 8200 2012-04-17 23:35:24Z uli $
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 students
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
58
59APPLICANT_HEADER_FIELDS = APPLICANT_SAMPLE_DATA.split(
60    '\n')[0].split(',')
61
62APPLICANT_SAMPLE_DATA_UPDATE = open(
63    os.path.join(os.path.dirname(__file__),
64                 'sample_applicant_data_update.csv'), 'rb').read()
65
66APPLICANT_HEADER_FIELDS_UPDATE = APPLICANT_SAMPLE_DATA_UPDATE.split(
67    '\n')[0].split(',')
68
69class ApplicantsContainerProcessorTest(FunctionalTestCase):
70
71    layer = FunctionalLayer
72
73    def setUp(self):
74        super(ApplicantsContainerProcessorTest, self).setUp()
75
76        # Setup a sample site for each test
77        app = University()
78        self.dc_root = tempfile.mkdtemp()
79        app['datacenter'].setStoragePath(self.dc_root)
80
81        # Prepopulate the ZODB...
82        self.getRootFolder()['app'] = app
83        self.app = self.getRootFolder()['app']
84        self.container = ApplicantsContainer()
85        self.container.code = u'dp2011'
86        self.app['applicants']['dp2011'] = self.container
87
88        self.processor = ApplicantsContainerProcessor()
89        self.workdir = tempfile.mkdtemp()
90        self.csv_file = os.path.join(self.workdir, 'sampledata.csv')
91        open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA)
92        setSite(self.app)
93        return
94
95    def tearDown(self):
96        super(ApplicantsContainerProcessorTest, self).tearDown()
97        shutil.rmtree(self.workdir)
98        shutil.rmtree(self.dc_root)
99        clearSite()
100        return
101
102    def test_interface(self):
103        # Make sure we fulfill the interface contracts.
104        assert verifyObject(IBatchProcessor, self.processor) is True
105        assert verifyClass(
106            IBatchProcessor, ApplicantsContainerProcessor) is True
107
108    def test_parentsExist(self):
109        assert self.processor.parentsExist(None, dict()) is False
110        assert self.processor.parentsExist(None, self.app) is True
111
112    def test_entryExists(self):
113        assert self.processor.entryExists(
114            dict(code='REG_NONE'), self.app) is False
115        assert self.processor.entryExists(
116            dict(code='dp2011'), self.app) is True
117
118    def test_getParent(self):
119        parent = self.processor.getParent(None, self.app)
120        assert parent is self.app['applicants']
121
122    def test_getEntry(self):
123        assert self.processor.getEntry(
124            dict(code='REG_NONE'), self.app) is None
125        assert self.processor.getEntry(
126            dict(code='dp2011'), self.app) is self.container
127
128    def test_addEntry(self):
129        self.processor.addEntry(
130            'New application', dict(code='dp2012'), self.app)
131        assert self.app['applicants']['dp2012'] == 'New application'
132
133    def test_delEntry(self):
134        self.processor.delEntry(dict(code='dp2011'), self.app)
135        assert 'dp2011' not in self.app['applicants'].keys()
136
137    def test_import(self):
138        # Do a real import
139        # see local sample_container.csv file for input
140        num, num_warns, fin_file, fail_file = self.processor.doImport(
141            self.csv_file, APPS_CONTAINER_HEADER_FIELDS)
142        avail_containers = [x for x in self.app['applicants'].keys()]
143        container = self.app['applicants'].get('app2012', None)
144        container2 = self.app['applicants'].get('app2013', None)
145        self.assertTrue(container is not None)
146        self.assertTrue(container2 is not None)
147
148        # check attributes
149        self.assertEqual(container.code, u'app2012')
150        self.assertEqual(container.title, u'General Studies 2012/2013')
151        self.assertEqual(container.prefix, u'app')
152        self.assertEqual(container.entry_level, 100)
153        self.assertEqual(container.year, 2012)
154        self.assertEqual(container.application_category, 'basic')
155        self.assertEqual(
156            container.description,
157            u'This text can been seen by anonymous users.\n'
158            u'>>de<<\nDieser Text kann von anonymen Benutzern '
159            u'gelesen werden.')
160        self.assertEqual(container.startdate,
161                         datetime.datetime(2012, 3, 1, 0, 0, tzinfo=pytz.utc))
162        self.assertEqual(container.enddate,
163                         datetime.datetime(2012, 4, 25, 0, 0, tzinfo=pytz.utc))
164        shutil.rmtree(os.path.dirname(fin_file))
165
166class ApplicantImportExportSetup(FunctionalTestCase):
167
168    layer = FunctionalLayer
169
170    def setUp(self):
171        super(ApplicantImportExportSetup, self).setUp()
172        # Setup a sample site for each test
173        app = University()
174        self.dc_root = tempfile.mkdtemp()
175        app['datacenter'].setStoragePath(self.dc_root)
176
177        # Prepopulate the ZODB...
178        self.getRootFolder()['app'] = app
179        # we add the site immediately after creation to the
180        # ZODB. Catalogs and other local utilities are not setup
181        # before that step.
182        self.app = self.getRootFolder()['app']
183        # Set site here. Some of the following setup code might need
184        # to access grok.getSite() and should get our new app then
185        setSite(app)
186
187        # Add an applicants container
188        self.container = ApplicantsContainer()
189        self.container.code = u'dp2011'
190        self.app['applicants']['dp2011'] = self.container
191
192        # Populate university
193        self.certificate = createObject('waeup.Certificate')
194        self.certificate.code = 'CERT1'
195        self.certificate.application_category = 'basic'
196        self.certificate.start_level = 100
197        self.certificate.end_level = 500
198        self.app['faculties']['fac1'] = Faculty()
199        self.app['faculties']['fac1']['dep1'] = Department()
200        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
201            self.certificate)
202
203        # Add applicant with subobjects
204        applicant = Applicant()
205        applicant.firstname = u'Anna'
206        applicant.lastname = u'Tester'
207        self.app['applicants']['dp2011'].addApplicant(applicant)
208        self.application_number = applicant.application_number
209        self.applicant = self.app['applicants']['dp2011'][
210            self.application_number]
211        self.workdir = tempfile.mkdtemp()
212        return
213
214    def tearDown(self):
215        super(ApplicantImportExportSetup, self).tearDown()
216        shutil.rmtree(self.workdir)
217        shutil.rmtree(self.dc_root)
218        clearSite()
219        return
220
221class ApplicantProcessorTest(ApplicantImportExportSetup):
222
223    layer = FunctionalLayer
224
225    def setUp(self):
226        super(ApplicantProcessorTest, self).setUp()
227        self.processor = ApplicantProcessor()
228        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
229        self.csv_file_faulty = os.path.join(self.workdir,
230                                            'faulty_applicant_data.csv')
231        self.csv_file_update = os.path.join(
232            self.workdir, 'sample_applicant_data_update.csv')
233        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
234        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
235        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
236
237    def test_interface(self):
238        # Make sure we fulfill the interface contracts.
239        assert verifyObject(IBatchProcessor, self.processor) is True
240        assert verifyClass(
241            IBatchProcessor, ApplicantProcessor) is True
242
243    def test_entryExists(self):
244        assert self.processor.entryExists(
245            dict(container_code='dp2011', application_number='999'),
246            self.app) is False
247
248    def test_getEntry(self):
249        applicant = self.processor.getEntry(
250            dict(container_code='dp2011',
251                 application_number=self.application_number), self.app)
252        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
253
254    def test_addEntry(self):
255        new_applicant = Applicant()
256        self.processor.addEntry(
257            new_applicant, dict(container_code='dp2011'), self.app)
258        assert len(self.app['applicants']['dp2011'].keys()) == 2
259
260    def test_delEntry(self):
261        assert self.application_number in self.app[
262            'applicants']['dp2011'].keys()
263        self.processor.delEntry(
264            dict(container_code='dp2011',
265                application_number=self.application_number), self.app)
266        assert self.application_number not in self.app[
267            'applicants']['dp2011'].keys()
268
269    def test_import(self):
270        num, num_warns, fin_file, fail_file = self.processor.doImport(
271            self.csv_file, APPLICANT_HEADER_FIELDS)
272        self.assertEqual(num_warns,0)
273        keys = self.app['applicants']['dp2011'].keys()
274        assert len(keys) == 4
275        container = self.app['applicants']['dp2011']
276        assert  container.__implemented__.__name__ == (
277            'waeup.kofa.applicants.container.ApplicantsContainer')
278        applicant = container[keys[0]]
279        assert applicant.__implemented__.__name__ == (
280            'waeup.kofa.applicants.applicant.Applicant')
281        shutil.rmtree(os.path.dirname(fin_file))
282
283    def test_import_faulty(self):
284        # we cannot import data with faulty dates. A date is faulty
285        # when in format xx/yy/zzzz as we cannot say whether it is
286        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
287        num, num_warns, fin_file, fail_file = self.processor.doImport(
288            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
289        if fail_file is not None:
290            fail_contents = open(fail_file, 'rb').read()
291            shutil.rmtree(os.path.dirname(fail_file))
292        else:
293            shutil.rmtree(os.path.dirname(fin_file))
294        for applicant in self.app['applicants']['dp2011'].values():
295            if applicant.date_of_birth == datetime.date(1990, 1, 2):
296                self.fail(
297                    'Wrong birthdate of imported applicant '
298                    '(1990-01-02, should be: 1990-02-01)')
299        return
300
301    def test_import_update(self):
302        num, num_warns, fin_file, fail_file = self.processor.doImport(
303            self.csv_file, APPLICANT_HEADER_FIELDS)
304        shutil.rmtree(os.path.dirname(fin_file))
305        num, num_warns, fin_file, fail_file = self.processor.doImport(
306            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
307        self.assertEqual(num_warns,0)
308        shutil.rmtree(os.path.dirname(fin_file))
309
310    def test_import_remove(self):
311        num, num_warns, fin_file, fail_file = self.processor.doImport(
312            self.csv_file, APPLICANT_HEADER_FIELDS)
313        shutil.rmtree(os.path.dirname(fin_file))
314        num, num_warns, fin_file, fail_file = self.processor.doImport(
315            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
316        self.assertEqual(num_warns,0)
317        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.