source: main/waeup.sirp/trunk/src/waeup/sirp/applicants/tests/test_batching.py @ 11102

Last change on this file since 11102 was 7509, checked in by uli, 13 years ago

Be more accurate about imported date. We complain only, when a wrong
date is imported.
  • Property svn:keywords set to Id
File size: 11.0 KB
Line 
1## $Id: test_batching.py 7509 2012-01-25 12:52:43Z 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.sirp.app import University
30from waeup.sirp.applicants.batching import (
31    ApplicantsContainerImporter, ApplicantImporter)
32from waeup.sirp.applicants.container import ApplicantsContainer
33from waeup.sirp.applicants.applicant import Applicant
34from waeup.sirp.university.faculty import Faculty
35from waeup.sirp.university.department import Department
36from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
37from waeup.sirp.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        self.assertTrue(u'CODE1' in avail_containers)
143        self.assertTrue(u'CODE2' in avail_containers)
144        shutil.rmtree(os.path.dirname(fin_file))
145
146class ApplicantImporterTest(FunctionalTestCase):
147
148    layer = FunctionalLayer
149
150    def setUp(self):
151        super(ApplicantImporterTest, self).setUp()
152        # Setup a sample site for each test
153        app = University()
154        self.dc_root = tempfile.mkdtemp()
155        app['datacenter'].setStoragePath(self.dc_root)
156
157        # Prepopulate the ZODB...
158        self.getRootFolder()['app'] = app
159        # we add the site immediately after creation to the
160        # ZODB. Catalogs and other local utilities are not setup
161        # before that step.
162        self.app = self.getRootFolder()['app']
163        # Set site here. Some of the following setup code might need
164        # to access grok.getSite() and should get our new app then
165        setSite(app)
166
167        # Add an applicants container
168        self.container = ApplicantsContainer()
169        self.container.code = u'dp2011'
170        self.app['applicants']['dp2011'] = self.container
171
172        # Populate university
173        self.certificate = createObject('waeup.Certificate')
174        self.certificate.code = 'CERT1'
175        self.certificate.application_category = 'basic'
176        self.certificate.start_level = 100
177        self.certificate.end_level = 500
178        self.app['faculties']['fac1'] = Faculty()
179        self.app['faculties']['fac1']['dep1'] = Department()
180        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
181            self.certificate)
182
183        # Add applicant with subobjects
184        applicant = Applicant()
185        applicant.firstname = u'Anna'
186        applicant.firstname = u'Tester'
187        self.app['applicants']['dp2011'].addApplicant(applicant)
188        self.application_number = applicant.application_number
189        self.applicant = self.app['applicants']['dp2011'][
190            self.application_number]
191        self.importer = ApplicantImporter()
192        self.workdir = tempfile.mkdtemp()
193        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
194        self.csv_file_faulty = os.path.join(self.workdir,
195                                            'faulty_applicant_data.csv')
196        self.csv_file_update = os.path.join(
197            self.workdir, 'sample_applicant_data_update.csv')
198        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
199        open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA)
200        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
201
202    def tearDown(self):
203        super(ApplicantImporterTest, self).tearDown()
204        shutil.rmtree(self.workdir)
205        shutil.rmtree(self.dc_root)
206        clearSite()
207        return
208
209    def test_interface(self):
210        # Make sure we fulfill the interface contracts.
211        assert verifyObject(IBatchProcessor, self.importer) is True
212        assert verifyClass(
213            IBatchProcessor, ApplicantImporter) is True
214
215    def test_entryExists(self):
216        assert self.importer.entryExists(
217            dict(container_code='dp2011', application_number='999'),
218            self.app) is False
219
220    def test_getEntry(self):
221        applicant = self.importer.getEntry(
222            dict(container_code='dp2011',
223                 application_number=self.application_number), self.app)
224        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
225
226    def test_addEntry(self):
227        new_applicant = Applicant()
228        self.importer.addEntry(
229            new_applicant, dict(container_code='dp2011'), self.app)
230        assert len(self.app['applicants']['dp2011'].keys()) == 2
231
232    def test_delEntry(self):
233        assert self.application_number in self.app[
234            'applicants']['dp2011'].keys()
235        self.importer.delEntry(
236            dict(container_code='dp2011',
237                application_number=self.application_number), self.app)
238        assert self.application_number not in self.app[
239            'applicants']['dp2011'].keys()
240
241    def test_import(self):
242        num, num_warns, fin_file, fail_file = self.importer.doImport(
243            self.csv_file, APPLICANT_HEADER_FIELDS)
244        self.assertEqual(num_warns,0)
245        assert len(self.app['applicants']['dp2011'].keys()) == 4
246        shutil.rmtree(os.path.dirname(fin_file))
247
248    def test_import_faulty(self):
249        # we cannot import data with faulty dates. A date is faulty
250        # when in format xx/yy/zzzz as we cannot say whether it is
251        # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd
252        num, num_warns, fin_file, fail_file = self.importer.doImport(
253            self.csv_file_faulty, APPLICANT_HEADER_FIELDS)
254        if fail_file is not None:
255            fail_contents = open(fail_file, 'rb').read()
256            shutil.rmtree(os.path.dirname(fail_file))
257        else:
258            shutil.rmtree(os.path.dirname(fin_file))
259        for applicant in self.app['applicants']['dp2011'].values():
260            if applicant.date_of_birth == datetime.date(1990, 1, 2):
261                self.fail(
262                    'Wrong birthdate of imported applicant '
263                    '(1990-01-02, should be: 1990-02-01)')
264        return
265
266    def test_import_update(self):
267        num, num_warns, fin_file, fail_file = self.importer.doImport(
268            self.csv_file, APPLICANT_HEADER_FIELDS)
269        shutil.rmtree(os.path.dirname(fin_file))
270        num, num_warns, fin_file, fail_file = self.importer.doImport(
271            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
272        self.assertEqual(num_warns,0)
273        shutil.rmtree(os.path.dirname(fin_file))
274
275    def test_import_remove(self):
276        num, num_warns, fin_file, fail_file = self.importer.doImport(
277            self.csv_file, APPLICANT_HEADER_FIELDS)
278        shutil.rmtree(os.path.dirname(fin_file))
279        num, num_warns, fin_file, fail_file = self.importer.doImport(
280            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
281        self.assertEqual(num_warns,0)
282        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.