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

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

Remove all ApplicantsContainerProvider? components. Experience has shown that we only need one type of ApplicantsContainers? and one type of Applicants but with different interfaces for form generation.
  • Property svn:keywords set to Id
File size: 12.4 KB
Line 
1## $Id: test_batching.py 8008 2012-03-30 10:23:32Z 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 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    ApplicantsContainerProcessor, ApplicantProcessor)
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 ApplicantsContainerProcessorTest(FunctionalTestCase):
69
70    layer = FunctionalLayer
71
72    def setUp(self):
73        super(ApplicantsContainerProcessorTest, 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.processor = ApplicantsContainerProcessor()
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(ApplicantsContainerProcessorTest, 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.processor) is True
104        assert verifyClass(
105            IBatchProcessor, ApplicantsContainerProcessor) is True
106
107    def test_parentsExist(self):
108        assert self.processor.parentsExist(None, dict()) is False
109        assert self.processor.parentsExist(None, self.app) is True
110
111    def test_entryExists(self):
112        assert self.processor.entryExists(
113            dict(code='REG_NONE'), self.app) is False
114        assert self.processor.entryExists(
115            dict(code='dp2011'), self.app) is True
116
117    def test_getParent(self):
118        parent = self.processor.getParent(None, self.app)
119        assert parent is self.app['applicants']
120
121    def test_getEntry(self):
122        assert self.processor.getEntry(
123            dict(code='REG_NONE'), self.app) is None
124        assert self.processor.getEntry(
125            dict(code='dp2011'), self.app) is self.container
126
127    def test_addEntry(self):
128        self.processor.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.processor.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.processor.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(container.application_category, 'basic')
154        self.assertEqual(
155            container.description,
156            u'This text can been seen by anonymous users.\n'
157            u'>>de<<\nDieser Text kann von anonymen Benutzern '
158            u'gelesen werden.')
159        self.assertEqual(container.startdate, datetime.date(2012, 3, 1))
160        self.assertEqual(container.enddate, datetime.date(2012, 4, 25))
161        shutil.rmtree(os.path.dirname(fin_file))
162
163class ApplicantImportExportSetup(FunctionalTestCase):
164
165    layer = FunctionalLayer
166
167    def setUp(self):
168        super(ApplicantImportExportSetup, self).setUp()
169        # Setup a sample site for each test
170        app = University()
171        self.dc_root = tempfile.mkdtemp()
172        app['datacenter'].setStoragePath(self.dc_root)
173
174        # Prepopulate the ZODB...
175        self.getRootFolder()['app'] = app
176        # we add the site immediately after creation to the
177        # ZODB. Catalogs and other local utilities are not setup
178        # before that step.
179        self.app = self.getRootFolder()['app']
180        # Set site here. Some of the following setup code might need
181        # to access grok.getSite() and should get our new app then
182        setSite(app)
183
184        # Add an applicants container
185        self.container = ApplicantsContainer()
186        self.container.code = u'dp2011'
187        self.app['applicants']['dp2011'] = self.container
188
189        # Populate university
190        self.certificate = createObject('waeup.Certificate')
191        self.certificate.code = 'CERT1'
192        self.certificate.application_category = 'basic'
193        self.certificate.start_level = 100
194        self.certificate.end_level = 500
195        self.app['faculties']['fac1'] = Faculty()
196        self.app['faculties']['fac1']['dep1'] = Department()
197        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
198            self.certificate)
199
200        # Add applicant with subobjects
201        applicant = Applicant()
202        applicant.firstname = u'Anna'
203        applicant.lastname = u'Tester'
204        self.app['applicants']['dp2011'].addApplicant(applicant)
205        self.application_number = applicant.application_number
206        self.applicant = self.app['applicants']['dp2011'][
207            self.application_number]
208        self.workdir = tempfile.mkdtemp()
209        return
210
211    def tearDown(self):
212        super(ApplicantImportExportSetup, self).tearDown()
213        shutil.rmtree(self.workdir)
214        shutil.rmtree(self.dc_root)
215        clearSite()
216        return
217
218class ApplicantProcessorTest(ApplicantImportExportSetup):
219
220    layer = FunctionalLayer
221
222    def setUp(self):
223        super(ApplicantProcessorTest, self).setUp()
224        self.processor = ApplicantProcessor()
225        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
226        self.csv_file_faulty = os.path.join(self.workdir,
227                                            'faulty_applicant_data.csv')
228        self.csv_file_update = os.path.join(
229            self.workdir, 'sample_applicant_data_update.csv')
230        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
231        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
232        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
233
234    def test_interface(self):
235        # Make sure we fulfill the interface contracts.
236        assert verifyObject(IBatchProcessor, self.processor) is True
237        assert verifyClass(
238            IBatchProcessor, ApplicantProcessor) is True
239
240    def test_entryExists(self):
241        assert self.processor.entryExists(
242            dict(container_code='dp2011', application_number='999'),
243            self.app) is False
244
245    def test_getEntry(self):
246        applicant = self.processor.getEntry(
247            dict(container_code='dp2011',
248                 application_number=self.application_number), self.app)
249        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
250
251    def test_addEntry(self):
252        new_applicant = Applicant()
253        self.processor.addEntry(
254            new_applicant, dict(container_code='dp2011'), self.app)
255        assert len(self.app['applicants']['dp2011'].keys()) == 2
256
257    def test_delEntry(self):
258        assert self.application_number in self.app[
259            'applicants']['dp2011'].keys()
260        self.processor.delEntry(
261            dict(container_code='dp2011',
262                application_number=self.application_number), self.app)
263        assert self.application_number not in self.app[
264            'applicants']['dp2011'].keys()
265
266    def test_import(self):
267        num, num_warns, fin_file, fail_file = self.processor.doImport(
268            self.csv_file, APPLICANT_HEADER_FIELDS)
269        self.assertEqual(num_warns,0)
270        keys = self.app['applicants']['dp2011'].keys()
271        assert len(keys) == 4
272        container = self.app['applicants']['dp2011']
273        assert  container.__implemented__.__name__ == (
274            'waeup.kofa.applicants.container.ApplicantsContainer')
275        applicant = container[keys[0]]
276        assert applicant.__implemented__.__name__ == (
277            'waeup.kofa.applicants.applicant.Applicant')
278        shutil.rmtree(os.path.dirname(fin_file))
279
280    def test_import_faulty(self):
281        # we cannot import data with faulty dates. A date is faulty
282        # when in format xx/yy/zzzz as we cannot say whether it is
283        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
284        num, num_warns, fin_file, fail_file = self.processor.doImport(
285            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
286        if fail_file is not None:
287            fail_contents = open(fail_file, 'rb').read()
288            shutil.rmtree(os.path.dirname(fail_file))
289        else:
290            shutil.rmtree(os.path.dirname(fin_file))
291        for applicant in self.app['applicants']['dp2011'].values():
292            if applicant.date_of_birth == datetime.date(1990, 1, 2):
293                self.fail(
294                    'Wrong birthdate of imported applicant '
295                    '(1990-01-02, should be: 1990-02-01)')
296        return
297
298    def test_import_update(self):
299        num, num_warns, fin_file, fail_file = self.processor.doImport(
300            self.csv_file, APPLICANT_HEADER_FIELDS)
301        shutil.rmtree(os.path.dirname(fin_file))
302        num, num_warns, fin_file, fail_file = self.processor.doImport(
303            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
304        self.assertEqual(num_warns,0)
305        shutil.rmtree(os.path.dirname(fin_file))
306
307    def test_import_remove(self):
308        num, num_warns, fin_file, fail_file = self.processor.doImport(
309            self.csv_file, APPLICANT_HEADER_FIELDS)
310        shutil.rmtree(os.path.dirname(fin_file))
311        num, num_warns, fin_file, fail_file = self.processor.doImport(
312            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
313        self.assertEqual(num_warns,0)
314        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.