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

Last change on this file since 7287 was 7266, checked in by uli, 13 years ago

Fix indent.
  • Property svn:keywords set to Id
File size: 9.8 KB
Line 
1## $Id: test_batching.py 7266 2011-12-04 13:38:20Z 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 os
21import shutil
22import tempfile
23import unittest
24from zope.component.hooks import setSite, clearSite
25from zope.component import createObject
26from zope.interface.verify import verifyClass, verifyObject
27
28from waeup.sirp.app import University
29from waeup.sirp.applicants.batching import (
30    ApplicantsContainerImporter, ApplicantImporter)
31from waeup.sirp.applicants.container import ApplicantsContainer
32from waeup.sirp.applicants.applicant import Applicant
33from waeup.sirp.university.faculty import Faculty
34from waeup.sirp.university.department import Department
35from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase
36from waeup.sirp.interfaces import IBatchProcessor
37
38
39# Sample data we can use in tests...
40APPS_CONTAINER_SAMPLE_DATA = open(
41    os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'),
42    'rb').read()
43
44# The header fields of the above CSV snippet
45APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split(
46    '\n')[0].split(',')
47
48# The same for students
49APPLICANT_SAMPLE_DATA = open(
50    os.path.join(os.path.dirname(__file__), 'sample_applicant_data.csv'),
51    'rb').read()
52
53APPLICANT_HEADER_FIELDS = APPLICANT_SAMPLE_DATA.split(
54    '\n')[0].split(',')
55
56APPLICANT_SAMPLE_DATA_UPDATE = open(
57    os.path.join(os.path.dirname(__file__), 'sample_applicant_data_update.csv'),
58    'rb').read()
59
60APPLICANT_HEADER_FIELDS_UPDATE = APPLICANT_SAMPLE_DATA_UPDATE.split(
61    '\n')[0].split(',')
62
63class ApplicantsContainerImporterTest(FunctionalTestCase):
64
65    layer = FunctionalLayer
66
67    def setUp(self):
68        super(ApplicantsContainerImporterTest, self).setUp()
69
70        # Setup a sample site for each test
71        app = University()
72        self.dc_root = tempfile.mkdtemp()
73        app['datacenter'].setStoragePath(self.dc_root)
74
75        # Prepopulate the ZODB...
76        self.getRootFolder()['app'] = app
77        self.app = self.getRootFolder()['app']
78        self.container = ApplicantsContainer()
79        self.container.code = u'dp2011'
80        self.app['applicants']['dp2011'] = self.container
81
82        self.importer = ApplicantsContainerImporter()
83        self.workdir = tempfile.mkdtemp()
84        self.csv_file = os.path.join(self.workdir, 'sampledata.csv')
85        open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA)
86        setSite(self.app)
87        return
88
89    def tearDown(self):
90        super(ApplicantsContainerImporterTest, self).tearDown()
91        shutil.rmtree(self.workdir)
92        shutil.rmtree(self.dc_root)
93        clearSite()
94        return
95
96    def test_interface(self):
97        # Make sure we fulfill the interface contracts.
98        assert verifyObject(IBatchProcessor, self.importer) is True
99        assert verifyClass(
100            IBatchProcessor, ApplicantsContainerImporter) is True
101
102    def test_parentsExist(self):
103        assert self.importer.parentsExist(None, dict()) is False
104        assert self.importer.parentsExist(None, self.app) is True
105
106    def test_entryExists(self):
107        assert self.importer.entryExists(
108            dict(code='REG_NONE'), self.app) is False
109        assert self.importer.entryExists(
110            dict(code='dp2011'), self.app) is True
111
112    def test_getParent(self):
113        parent = self.importer.getParent(None, self.app)
114        assert parent is self.app['applicants']
115
116    def test_getEntry(self):
117        assert self.importer.getEntry(
118            dict(code='REG_NONE'), self.app) is None
119        assert self.importer.getEntry(
120            dict(code='dp2011'), self.app) is self.container
121
122    def test_addEntry(self):
123        self.importer.addEntry(
124            'New application', dict(code='dp2012'), self.app)
125        assert self.app['applicants']['dp2012'] == 'New application'
126
127    def test_delEntry(self):
128        self.importer.delEntry(dict(code='dp2011'), self.app)
129        assert 'dp2011' not in self.app['applicants'].keys()
130
131    def test_import(self):
132        # Do a real import
133        # see local sample_container.csv file for input
134        num, num_warns, fin_file, fail_file = self.importer.doImport(
135            self.csv_file, APPS_CONTAINER_HEADER_FIELDS)
136        avail_containers = [x for x in self.app['applicants'].keys()]
137        self.assertTrue(u'CODE1' in avail_containers)
138        self.assertTrue(u'CODE2' in avail_containers)
139        shutil.rmtree(os.path.dirname(fin_file))
140
141class ApplicantImporterTest(FunctionalTestCase):
142
143    layer = FunctionalLayer
144
145    def setUp(self):
146        super(ApplicantImporterTest, self).setUp()
147        # Setup a sample site for each test
148        app = University()
149        self.dc_root = tempfile.mkdtemp()
150        app['datacenter'].setStoragePath(self.dc_root)
151
152        # Prepopulate the ZODB...
153        self.getRootFolder()['app'] = app
154        # we add the site immediately after creation to the
155        # ZODB. Catalogs and other local utilities are not setup
156        # before that step.
157        self.app = self.getRootFolder()['app']
158        # Set site here. Some of the following setup code might need
159        # to access grok.getSite() and should get our new app then
160        setSite(app)
161
162        # Add an applicants container
163        self.container = ApplicantsContainer()
164        self.container.code = u'dp2011'
165        self.app['applicants']['dp2011'] = self.container
166
167        # Populate university
168        self.certificate = createObject('waeup.Certificate')
169        self.certificate.code = 'CERT1'
170        self.certificate.application_category = 'basic'
171        self.certificate.start_level = 100
172        self.certificate.end_level = 500
173        self.app['faculties']['fac1'] = Faculty()
174        self.app['faculties']['fac1']['dep1'] = Department()
175        self.app['faculties']['fac1']['dep1'].certificates.addCertificate(
176            self.certificate)
177
178        # Add applicant with subobjects
179        applicant = Applicant()
180        applicant.firstname = u'Anna'
181        applicant.firstname = u'Tester'
182        self.app['applicants']['dp2011'].addApplicant(applicant)
183        self.application_number = applicant.application_number
184        self.applicant = self.app['applicants']['dp2011'][
185            self.application_number]
186        self.importer = ApplicantImporter()
187        self.workdir = tempfile.mkdtemp()
188        self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv')
189        self.csv_file_update = os.path.join(
190            self.workdir, 'sample_applicant_data_update.csv')
191        open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA)
192        open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE)
193
194    def tearDown(self):
195        super(ApplicantImporterTest, self).tearDown()
196        shutil.rmtree(self.workdir)
197        shutil.rmtree(self.dc_root)
198        clearSite()
199        return
200
201    def test_interface(self):
202        # Make sure we fulfill the interface contracts.
203        assert verifyObject(IBatchProcessor, self.importer) is True
204        assert verifyClass(
205            IBatchProcessor, ApplicantImporter) is True
206
207    def test_entryExists(self):
208        assert self.importer.entryExists(
209            dict(container_code='dp2011', application_number='999'),
210            self.app) is False
211
212    def test_getEntry(self):
213        applicant = self.importer.getEntry(
214            dict(container_code='dp2011',
215                 application_number=self.application_number), self.app)
216        self.assertEqual(applicant.applicant_id, self.applicant.applicant_id)
217
218    def test_addEntry(self):
219        new_applicant = Applicant()
220        self.importer.addEntry(
221            new_applicant, dict(container_code='dp2011'), self.app)
222        assert len(self.app['applicants']['dp2011'].keys()) == 2
223
224    def test_delEntry(self):
225        assert self.application_number in self.app[
226            'applicants']['dp2011'].keys()
227        self.importer.delEntry(
228            dict(container_code='dp2011',
229                application_number=self.application_number), self.app)
230        assert self.application_number not in self.app[
231            'applicants']['dp2011'].keys()
232
233    def test_import(self):
234        num, num_warns, fin_file, fail_file = self.importer.doImport(
235            self.csv_file, APPLICANT_HEADER_FIELDS)
236        self.assertEqual(num_warns,0)
237        assert len(self.app['applicants']['dp2011'].keys()) == 4
238        shutil.rmtree(os.path.dirname(fin_file))
239
240    def test_import_update(self):
241        num, num_warns, fin_file, fail_file = self.importer.doImport(
242            self.csv_file, APPLICANT_HEADER_FIELDS)
243        shutil.rmtree(os.path.dirname(fin_file))
244        num, num_warns, fin_file, fail_file = self.importer.doImport(
245            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update')
246        self.assertEqual(num_warns,0)
247        shutil.rmtree(os.path.dirname(fin_file))
248
249    def test_import_remove(self):
250        num, num_warns, fin_file, fail_file = self.importer.doImport(
251            self.csv_file, APPLICANT_HEADER_FIELDS)
252        shutil.rmtree(os.path.dirname(fin_file))
253        num, num_warns, fin_file, fail_file = self.importer.doImport(
254            self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove')
255        self.assertEqual(num_warns,0)
256        shutil.rmtree(os.path.dirname(fin_file))
Note: See TracBrowser for help on using the repository browser.