## $Id: test_batching.py 7509 2012-01-25 12:52:43Z uli $ ## ## Copyright (C) 2011 Uli Fouquet & Henrik Bettermann ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2 of the License, or ## (at your option) any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License ## along with this program; if not, write to the Free Software ## Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA ## """Unit tests for applicants-related data importers. """ import datetime import os import shutil import tempfile import unittest from zope.component.hooks import setSite, clearSite from zope.component import createObject from zope.interface.verify import verifyClass, verifyObject from waeup.sirp.app import University from waeup.sirp.applicants.batching import ( ApplicantsContainerImporter, ApplicantImporter) from waeup.sirp.applicants.container import ApplicantsContainer from waeup.sirp.applicants.applicant import Applicant from waeup.sirp.university.faculty import Faculty from waeup.sirp.university.department import Department from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase from waeup.sirp.interfaces import IBatchProcessor # Sample data we can use in tests... APPS_CONTAINER_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_container_data.csv'), 'rb').read() # The header fields of the above CSV snippet APPS_CONTAINER_HEADER_FIELDS = APPS_CONTAINER_SAMPLE_DATA.split( '\n')[0].split(',') # The same for students APPLICANT_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_applicant_data.csv'), 'rb').read() FAULTY_APPLICANT_SAMPLE_DATA = open( os.path.join(os.path.dirname(__file__), 'sample_faulty_applicant_data.csv'), 'rb').read() APPLICANT_HEADER_FIELDS = APPLICANT_SAMPLE_DATA.split( '\n')[0].split(',') APPLICANT_SAMPLE_DATA_UPDATE = open( os.path.join(os.path.dirname(__file__), 'sample_applicant_data_update.csv'), 'rb').read() APPLICANT_HEADER_FIELDS_UPDATE = APPLICANT_SAMPLE_DATA_UPDATE.split( '\n')[0].split(',') class ApplicantsContainerImporterTest(FunctionalTestCase): layer = FunctionalLayer def setUp(self): super(ApplicantsContainerImporterTest, self).setUp() # Setup a sample site for each test app = University() self.dc_root = tempfile.mkdtemp() app['datacenter'].setStoragePath(self.dc_root) # Prepopulate the ZODB... self.getRootFolder()['app'] = app self.app = self.getRootFolder()['app'] self.container = ApplicantsContainer() self.container.code = u'dp2011' self.app['applicants']['dp2011'] = self.container self.importer = ApplicantsContainerImporter() self.workdir = tempfile.mkdtemp() self.csv_file = os.path.join(self.workdir, 'sampledata.csv') open(self.csv_file, 'wb').write(APPS_CONTAINER_SAMPLE_DATA) setSite(self.app) return def tearDown(self): super(ApplicantsContainerImporterTest, self).tearDown() shutil.rmtree(self.workdir) shutil.rmtree(self.dc_root) clearSite() return def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.importer) is True assert verifyClass( IBatchProcessor, ApplicantsContainerImporter) is True def test_parentsExist(self): assert self.importer.parentsExist(None, dict()) is False assert self.importer.parentsExist(None, self.app) is True def test_entryExists(self): assert self.importer.entryExists( dict(code='REG_NONE'), self.app) is False assert self.importer.entryExists( dict(code='dp2011'), self.app) is True def test_getParent(self): parent = self.importer.getParent(None, self.app) assert parent is self.app['applicants'] def test_getEntry(self): assert self.importer.getEntry( dict(code='REG_NONE'), self.app) is None assert self.importer.getEntry( dict(code='dp2011'), self.app) is self.container def test_addEntry(self): self.importer.addEntry( 'New application', dict(code='dp2012'), self.app) assert self.app['applicants']['dp2012'] == 'New application' def test_delEntry(self): self.importer.delEntry(dict(code='dp2011'), self.app) assert 'dp2011' not in self.app['applicants'].keys() def test_import(self): # Do a real import # see local sample_container.csv file for input num, num_warns, fin_file, fail_file = self.importer.doImport( self.csv_file, APPS_CONTAINER_HEADER_FIELDS) avail_containers = [x for x in self.app['applicants'].keys()] self.assertTrue(u'CODE1' in avail_containers) self.assertTrue(u'CODE2' in avail_containers) shutil.rmtree(os.path.dirname(fin_file)) class ApplicantImporterTest(FunctionalTestCase): layer = FunctionalLayer def setUp(self): super(ApplicantImporterTest, self).setUp() # Setup a sample site for each test app = University() self.dc_root = tempfile.mkdtemp() app['datacenter'].setStoragePath(self.dc_root) # Prepopulate the ZODB... self.getRootFolder()['app'] = app # we add the site immediately after creation to the # ZODB. Catalogs and other local utilities are not setup # before that step. self.app = self.getRootFolder()['app'] # Set site here. Some of the following setup code might need # to access grok.getSite() and should get our new app then setSite(app) # Add an applicants container self.container = ApplicantsContainer() self.container.code = u'dp2011' self.app['applicants']['dp2011'] = self.container # Populate university self.certificate = createObject('waeup.Certificate') self.certificate.code = 'CERT1' self.certificate.application_category = 'basic' self.certificate.start_level = 100 self.certificate.end_level = 500 self.app['faculties']['fac1'] = Faculty() self.app['faculties']['fac1']['dep1'] = Department() self.app['faculties']['fac1']['dep1'].certificates.addCertificate( self.certificate) # Add applicant with subobjects applicant = Applicant() applicant.firstname = u'Anna' applicant.firstname = u'Tester' self.app['applicants']['dp2011'].addApplicant(applicant) self.application_number = applicant.application_number self.applicant = self.app['applicants']['dp2011'][ self.application_number] self.importer = ApplicantImporter() self.workdir = tempfile.mkdtemp() self.csv_file = os.path.join(self.workdir, 'sample_applicant_data.csv') self.csv_file_faulty = os.path.join(self.workdir, 'faulty_applicant_data.csv') self.csv_file_update = os.path.join( self.workdir, 'sample_applicant_data_update.csv') open(self.csv_file, 'wb').write(APPLICANT_SAMPLE_DATA) open(self.csv_file_faulty, 'wb').write(FAULTY_APPLICANT_SAMPLE_DATA) open(self.csv_file_update, 'wb').write(APPLICANT_SAMPLE_DATA_UPDATE) def tearDown(self): super(ApplicantImporterTest, self).tearDown() shutil.rmtree(self.workdir) shutil.rmtree(self.dc_root) clearSite() return def test_interface(self): # Make sure we fulfill the interface contracts. assert verifyObject(IBatchProcessor, self.importer) is True assert verifyClass( IBatchProcessor, ApplicantImporter) is True def test_entryExists(self): assert self.importer.entryExists( dict(container_code='dp2011', application_number='999'), self.app) is False def test_getEntry(self): applicant = self.importer.getEntry( dict(container_code='dp2011', application_number=self.application_number), self.app) self.assertEqual(applicant.applicant_id, self.applicant.applicant_id) def test_addEntry(self): new_applicant = Applicant() self.importer.addEntry( new_applicant, dict(container_code='dp2011'), self.app) assert len(self.app['applicants']['dp2011'].keys()) == 2 def test_delEntry(self): assert self.application_number in self.app[ 'applicants']['dp2011'].keys() self.importer.delEntry( dict(container_code='dp2011', application_number=self.application_number), self.app) assert self.application_number not in self.app[ 'applicants']['dp2011'].keys() def test_import(self): num, num_warns, fin_file, fail_file = self.importer.doImport( self.csv_file, APPLICANT_HEADER_FIELDS) self.assertEqual(num_warns,0) assert len(self.app['applicants']['dp2011'].keys()) == 4 shutil.rmtree(os.path.dirname(fin_file)) def test_import_faulty(self): # we cannot import data with faulty dates. A date is faulty # when in format xx/yy/zzzz as we cannot say whether it is # meant as dd/mm/yyyy or mm/dd/yyyy. We therefore require yyyy-mm-dd num, num_warns, fin_file, fail_file = self.importer.doImport( self.csv_file_faulty, APPLICANT_HEADER_FIELDS) if fail_file is not None: fail_contents = open(fail_file, 'rb').read() shutil.rmtree(os.path.dirname(fail_file)) else: shutil.rmtree(os.path.dirname(fin_file)) for applicant in self.app['applicants']['dp2011'].values(): if applicant.date_of_birth == datetime.date(1990, 1, 2): self.fail( 'Wrong birthdate of imported applicant ' '(1990-01-02, should be: 1990-02-01)') return def test_import_update(self): num, num_warns, fin_file, fail_file = self.importer.doImport( self.csv_file, APPLICANT_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.importer.doImport( self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'update') self.assertEqual(num_warns,0) shutil.rmtree(os.path.dirname(fin_file)) def test_import_remove(self): num, num_warns, fin_file, fail_file = self.importer.doImport( self.csv_file, APPLICANT_HEADER_FIELDS) shutil.rmtree(os.path.dirname(fin_file)) num, num_warns, fin_file, fail_file = self.importer.doImport( self.csv_file_update, APPLICANT_HEADER_FIELDS_UPDATE, 'remove') self.assertEqual(num_warns,0) shutil.rmtree(os.path.dirname(fin_file))