## ## test_batching.py ## Login : ## Started on Fri Jun 3 00:28:43 2011 Uli Fouquet ## $Id$ ## ## Copyright (C) 2011 Uli Fouquet ## 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 ## import doctest import logging import os import shutil import tempfile import unittest from zope import schema from zope.app.testing.functional import FunctionalTestCase from zope.component import provideUtility from zope.component.factory import Factory from zope.component.hooks import setSite, clearSite from zope.component.interfaces import IFactory from zope.interface import Interface, implements from waeup.sirp.app import University from waeup.sirp.testing import FunctionalLayer, doctestsuite_for_module from waeup.sirp.utils.helpers import attrs_to_fields optionflags = ( doctest.REPORT_NDIFF + doctest.ELLIPSIS + doctest.NORMALIZE_WHITESPACE) class ICave(Interface): """A cave.""" id_num = schema.TextLine( title = u'internal id', default = u'default', required = True, readonly = True, ) name = schema.TextLine( title = u'Cave name', default = u'Unnamed', required = True) dinoports = schema.Int( title = u'Number of DinoPorts (tm)', required = False, default = 1) owner = schema.TextLine( title = u'Owner name', required = True, missing_value = 'Fred Estates Inc.') taxpayer = schema.Bool( title = u'Payes taxes', required = True, default = False) class Cave(object): implements(ICave) def __init__(self, name=u'Unnamed', dinoports=2, owner='Fred Estates Inc.', taxpayer=False): self.name = name self.dinoports = 2 self.owner = owner self.taxpayer = taxpayer #Cave = attrs_to_fields(Cave) stoneville = dict from waeup.sirp.utils.batching import BatchProcessor class CaveProcessor(BatchProcessor): util_name = 'caveprocessor' #grok.name(util_name) name = 'Cave Processor' iface = ICave location_fields = ['name'] factory_name = 'Lovely Cave' def parentsExist(self, row, site): return True def getParent(self, row, site): return stoneville def entryExists(self, row, site): return row['name'] in stoneville.keys() def getEntry(self, row, site): if not self.entryExists(row, site): return None return stoneville[row['name']] def delEntry(self, row, site): del stoneville[row['name']] def addEntry(self, obj, row, site): stoneville[row['name']] = obj def updateEntry(self, obj, row, site): # This is not strictly necessary, as the default # updateEntry method does exactly the same for key, value in row.items(): setattr(obj, key, value) class BatchProcessorTests(FunctionalTestCase): layer = FunctionalLayer def setupLogger(self): self.logger = logging.getLogger('stoneville') self.logger.setLevel(logging.DEBUG) self.logger.propagate = False self.logfile = os.path.join(self.workdir, 'stoneville.log') self.handler = logging.FileHandler(self.logfile, 'w') self.logger.addHandler(self.handler) def setUp(self): global stoneville super(BatchProcessorTests, 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.workdir = tempfile.mkdtemp() factory = Factory(Cave) provideUtility(factory, IFactory, 'Lovely Cave') # Provide sample data self.newcomers_csv = os.path.join(self.workdir, 'newcomers.csv') open(self.newcomers_csv, 'wb').write( """name,dinoports,owner,taxpayer Barneys Home,2,Barney,1 Wilmas Asylum,1,Wilma,1 Freds Dinoburgers,10,Fred,0 Joeys Drive-in,110,Joey,0 """) self.setupLogger() self.stoneville = stoneville stoneville = dict() self.resultpath = None return def tearDown(self): super(BatchProcessorTests, self).tearDown() shutil.rmtree(self.workdir) shutil.rmtree(self.dc_root) self.logger.removeHandler(self.handler) clearSite() if not isinstance(self.resultpath, list): self.resultpath = [self.resultpath] for path in self.resultpath: if not isinstance(path, basestring): continue if not os.path.isdir(path): path = os.path.dirname(path) if os.path.exists(path): shutil.rmtree(path) return def test_import(self): processor = CaveProcessor() result = processor.doImport( self.newcomers_csv, ['name', 'dinoports', 'owner', 'taxpayer'], mode='create', user='Bob', logger=self.logger) num_succ, num_fail, finished_path, failed_path = result self.resultpath = [finished_path, failed_path] assert num_succ == 4 assert num_fail == 0 assert finished_path.endswith('/newcomers.finished.csv') assert failed_path is None def test_import_stoneville(self): processor = CaveProcessor() result = processor.doImport( self.newcomers_csv, ['name', 'dinoports', 'owner', 'taxpayer'], mode='create', user='Bob', logger=self.logger) num_succ, num_fail, finished_path, failed_path = result self.resultpath = [finished_path, failed_path] assert len(self.stoneville) == 4 self.assertEqual( sorted(self.stoneville.keys()), [u'Barneys Home', u'Freds Dinoburgers', u'Joeys Drive-in', u'Wilmas Asylum']) def test_import_correct_type(self): processor = CaveProcessor() result = processor.doImport( self.newcomers_csv, ['name', 'dinoports', 'owner', 'taxpayer'], mode='create', user='Bob', logger=self.logger) num_succ, num_fail, finished_path, failed_path = result self.resultpath = [finished_path, failed_path] assert isinstance(self.stoneville['Barneys Home'].dinoports, int) def test_log(self): """ >>> print log_contents -------------------- Bob: Batch processing finished: OK Bob: Source: /.../newcomers.csv Bob: Mode: create Bob: User: Bob Bob: Processing time: ... s (... s/item) Bob: Processed: 4 lines (4 successful/ 0 failed) -------------------- """ processor = CaveProcessor() result = processor.doImport( self.newcomers_csv, ['name', 'dinoports', 'owner', 'taxpayer'], mode='create', user='Bob', logger=self.logger) num_succ, num_fail, finished_path, failed_path = result self.resultpath = [finished_path, failed_path] log_contents = open(self.logfile, 'rb').read() doctest.run_docstring_examples( self.test_log, locals(), False, 'test_log', None, optionflags) return #def test_suite(): # return unittest.TestSuite()