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

Last change on this file since 7932 was 7912, checked in by uli, 13 years ago
  • Make batched applicant setup an own test case for reimport in other tests.
  • Fix little typo.
  • Property svn:keywords set to Id
File size: 12.5 KB
Line 
1## $Id: test_batching.py 7912 2012-03-19 01:55:47Z 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 importers.
19"""
20import datetime
21import os
22import shutil
23import tempfile
24import unittest
25from zope.component.hooks import setSite, clearSite
26from zope.component import createObject
27from zope.interface.verify import verifyClass, verifyObject
28
29from waeup.kofa.app import University
30from waeup.kofa.applicants.batching import (
31    ApplicantsContainerImporter, ApplicantImporter)
32from waeup.kofa.applicants.container import ApplicantsContainer
33from waeup.kofa.applicants.applicant import Applicant
34from waeup.kofa.university.faculty import Faculty
35from waeup.kofa.university.department import Department
36from waeup.kofa.testing import FunctionalLayer, FunctionalTestCase
37from waeup.kofa.interfaces import IBatchProcessor
38
39
40# Sample data we can use in tests...
41APPS_CONTAINER_SAMPLE_DATA = open(
42    os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'),
43    'rb').read()
44
45# The header fields of the above CSV snippet
46APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split(
47    '\n')[0].split(',')
48
49# The same for students
50APPLICANT_SAMPLE_DATA = open(
51    os.path.join(os.path.dirname(__file__), 'sample_applicant_data.csv'),
52    'rb').read()
53FAULTY_APPLICANT_SAMPLE_DATA = open(
54    os.path.join(os.path.dirname(__file__),
55                 'sample_faulty_applicant_data.csv'), 'rb').read()
56
57
58APPLICANT_HEADER_FIELDS = APPLICANT_SAMPLE_DATA.split(
59    '\n')[0].split(',')
60
61APPLICANT_SAMPLE_DATA_UPDATE = open(
62    os.path.join(os.path.dirname(__file__),
63                 'sample_applicant_data_update.csv'), 'rb').read()
64
65APPLICANT_HEADER_FIELDS_UPDATE = APPLICANT_SAMPLE_DATA_UPDATE.split(
66    '\n')[0].split(',')
67
68class ApplicantsContainerImporterTest(FunctionalTestCase):
69
70    layer = FunctionalLayer
71
72    def setUp(self):
73        super(ApplicantsContainerImporterTest, self).setUp()
74
75        # Setup a sample site for each test
76        app = University()
77        self.dc_root = tempfile.mkdtemp()
78        app['datacenter'].setStoragePath(self.dc_root)
79
80        # Prepopulate the ZODB...
81        self.getRootFolder()['app'] = app
82        self.app = self.getRootFolder()['app']
83        self.container = ApplicantsContainer()
84        self.container.code = u'dp2011'
85        self.app['applicants']['dp2011'] = self.container
86
87        self.importer = ApplicantsContainerImporter()
88        self.workdir = tempfile.mkdtemp()
89        self.csv_file = os.path.join(self.workdir, 'sampledata.csv')
90        open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA)
91        setSite(self.app)
92        return
93
94    def tearDown(self):
95        super(ApplicantsContainerImporterTest, self).tearDown()
96        shutil.rmtree(self.workdir)
97        shutil.rmtree(self.dc_root)
98        clearSite()
99        return
100
101    def test_interface(self):
102        # Make sure we fulfill the interface contracts.
103        assert verifyObject(IBatchProcessor, self.importer) is True
104        assert verifyClass(
105            IBatchProcessor, ApplicantsContainerImporter) is True
106
107    def test_parentsExist(self):
108        assert self.importer.parentsExist(None, dict()) is False
109        assert self.importer.parentsExist(None, self.app) is True
110
111    def test_entryExists(self):
112        assert self.importer.entryExists(
113            dict(code='REG_NONE'), self.app) is False
114        assert self.importer.entryExists(
115            dict(code='dp2011'), self.app) is True
116
117    def test_getParent(self):
118        parent = self.importer.getParent(None, self.app)
119        assert parent is self.app['applicants']
120
121    def test_getEntry(self):
122        assert self.importer.getEntry(
123            dict(code='REG_NONE'), self.app) is None
124        assert self.importer.getEntry(
125            dict(code='dp2011'), self.app) is self.container
126
127    def test_addEntry(self):
128        self.importer.addEntry(
129            'New application', dict(code='dp2012'), self.app)
130        assert self.app['applicants']['dp2012'] == 'New application'
131
132    def test_delEntry(self):
133        self.importer.delEntry(dict(code='dp2011'), self.app)
134        assert 'dp2011' not in self.app['applicants'].keys()
135
136    def test_import(self):
137        # Do a real import
138        # see local sample_container.csv file for input
139        num, num_warns, fin_file, fail_file = self.importer.doImport(
140            self.csv_file, APPS_CONTAINER_HEADER_FIELDS)
141        avail_containers = [x for x in self.app['applicants'].keys()]
142        container = self.app['applicants'].get('app2012', None)
143        container2 = self.app['applicants'].get('app2013', None)
144        self.assertTrue(container is not None)
145        self.assertTrue(container2 is not None)
146
147        # check attributes
148        self.assertEqual(container.code, u'app2012')
149        self.assertEqual(container.title, u'General Studies 2012/2013')
150        self.assertEqual(container.prefix, u'app')
151        self.assertEqual(container.entry_level, 100)
152        self.assertEqual(container.year, 2012)
153        self.assertEqual(
154            container.provider[0],
155            u'waeup.kofa.applicants.ApplicantsContainer')
156        self.assertEqual(container.application_category, 'basic')
157        self.assertEqual(
158            container.description,
159            u'This text can been seen by anonymous users.\n'
160            u'>>de<<\nDieser Text kann von anonymen Benutzern '
161            u'gelesen werden.')
162        self.assertEqual(container.startdate, datetime.date(2012, 3, 1))
163        self.assertEqual(container.enddate, datetime.date(2012, 4, 25))
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 ApplicantImporterTest(ApplicantImportExportSetup):
222
223    layer = FunctionalLayer
224
225    def setUp(self):
226        super(ApplicantImporterTest, self).setUp()
227        self.importer = ApplicantImporter()
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.importer) is True
240        assert verifyClass(
241            IBatchProcessor, ApplicantImporter) is True
242
243    def test_entryExists(self):
244        assert self.importer.entryExists(
245            dict(container_code='dp2011', application_number='999'),
246            self.app) is False
247
248    def test_getEntry(self):
249        applicant = self.importer.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.importer.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.importer.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.importer.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.importer.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.importer.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.importer.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.importer.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.importer.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.