[Repoze-checkins] r1077 - in repoze.zope2/trunk: . repoze/zope2 repoze/zope2/tests
Chris McDonough
chrism at agendaless.com
Wed Jun 11 23:37:09 EDT 2008
Author: Chris McDonough <chrism at agendaless.com>
Date: Wed Jun 11 23:37:09 2008
New Revision: 1077
Log:
- Don't attempt retries at startup.
- Real unit tests for getDB.
Modified:
repoze.zope2/trunk/CHANGES.txt
repoze.zope2/trunk/repoze/zope2/db.py
repoze.zope2/trunk/repoze/zope2/tests/test_db.py
Modified: repoze.zope2/trunk/CHANGES.txt
==============================================================================
--- repoze.zope2/trunk/CHANGES.txt (original)
+++ repoze.zope2/trunk/CHANGES.txt Wed Jun 11 23:37:09 2008
@@ -1,3 +1,8 @@
+Next release
+
+ - Dont attempt any retries on conflict errors at startup.
+ This was never tested, and probably didn't work.
+
0.4.2 (2008-06-11)
- Deal with Unauthorized exceptions properly: allow
Modified: repoze.zope2/trunk/repoze/zope2/db.py
==============================================================================
--- repoze.zope2/trunk/repoze/zope2/db.py (original)
+++ repoze.zope2/trunk/repoze/zope2/db.py Wed Jun 11 23:37:09 2008
@@ -14,38 +14,35 @@
""" Utilies for constructing a ZODB database object """
-def retry(f, numtries):
- from ZODB.POSException import ConflictError
- for x in range(0, numtries):
- try:
- return f()
- break
- except ConflictError, why:
- if x == (numtries-1):
- raise
-
def getDB(config):
config_file = config['zope.conf']
- from Zope2.Startup import options, handlers, get_starter
- import App.config
+ # these are imported without "from" statements in them so we can
+ # unit test them more easily
import Zope2
- starter = get_starter()
- opts = options.ZopeOptions()
+ import Zope2.Startup
+ import Zope2.Startup.options
+ import Zope2.Startup.handlers
+ import Zope2.App.startup
+ import App.config
+ import OFS.Application
+ import transaction
+
+ starter = Zope2.Startup.get_starter()
+ opts = Zope2.Startup.options.ZopeOptions()
opts.configfile = config_file
opts.realize(args=[], doc="", raise_getopt_errs=0)
- handlers.handleConfig(opts.configroot, opts.confighandlers)
+ Zope2.Startup.handlers.handleConfig(opts.configroot, opts.confighandlers)
App.config.setConfiguration(opts.configroot)
starter.setConfiguration(opts.configroot)
starter.setupInitialLogging()
starter.setupSecurityOptions()
starter.setupPublisher()
- from Zope2.App.startup import startup as _startup
- # _startup is what causes Products to be installed; Product
+ # startup is what causes Products to be installed; Product
# initialization causes database writes, which cause conflict
- # errors when used with multiple clients vs. ZEO. We need to retry
- # these conflict errors. I wish Zope didn't write to the database
- # at startup time.
- retry(_startup, 5)
+ # errors when used with multiple clients vs. ZEO. We need some
+ # way to retry these conflict errors. I wish Zope didn't write to
+ # the database at startup time.
+ Zope2.App.startup.startup()
starter.setupFinalLogging()
Zope2.zpublisher_transactions_manager = None # middleware does this
# Zope2.bobo_application is used by: App.ZApplication
@@ -53,16 +50,13 @@
# Zope2.__init__, Zope2.App.startup
Zope2.bobo_application = None
db = Zope2.DB # set the global
- from OFS.Application import Application
conn = db.open()
root = conn.root()
- import transaction
appname = config.get('appname', 'Application')
- def addroot():
- if not root.has_key(appname):
- root[appname] = Application()
- transaction.commit()
- retry(addroot, 5)
+ # this can also conflict
+ if not root.has_key(appname):
+ root[appname] = OFS.Application.Application()
+ transaction.commit()
conn.close()
return db
Modified: repoze.zope2/trunk/repoze/zope2/tests/test_db.py
==============================================================================
--- repoze.zope2/trunk/repoze/zope2/tests/test_db.py (original)
+++ repoze.zope2/trunk/repoze/zope2/tests/test_db.py Wed Jun 11 23:37:09 2008
@@ -1,32 +1,189 @@
import unittest
-class TestRetry(unittest.TestCase):
+_root = {}
+
+class TestGetDB(unittest.TestCase):
+ def setUp(self):
+ self.root = _root
+
+ class FakeZopeOptions:
+ configroot = None
+ confighandlers = None
+ def realize(self, args, doc, raise_getopt_errs):
+ self.realized = True
+
+ options = FakeZopeOptions()
+ self.options = options
+
+ class FakeStartupOptions:
+ def ZopeOptions(self):
+ return options
+
+ class FakeStarter:
+ def setConfiguration(self, configroot):
+ self.configroot = configroot
+ def setupInitialLogging(self):
+ self.setup_initial_logging = True
+ def setupSecurityOptions(self):
+ self.security_setup = True
+ def setupPublisher(self):
+ self.publisher_setup = True
+ def setupFinalLogging(self):
+ self.setup_final_logging = True
+
+ starter = FakeStarter()
+ self.starter = starter
+
+ class FakeStartupHandlers:
+ def handleConfig(self, configroot, confighandlers):
+ self.configroot = configroot
+ self.confighandlers = confighandlers
+
+ class FakeStartup:
+ options = FakeStartupOptions()
+ handlers = FakeStartupHandlers()
+ def get_starter(self):
+ return starter
+
+ startup = FakeStartup()
+ self.startup = startup
+
+ class FakeZope2AppStartup:
+ def startup(self):
+ self.started = True
+
+ appstartup = FakeZope2AppStartup()
+ self.appstartup = appstartup
+
+ class FakeZope2App:
+ startup = appstartup
+
+ zope2app = FakeZope2App()
+
+ class FakeConn:
+ def root(self):
+ return _root
+
+ def close(self):
+ self.closed = True
+
+ conn = FakeConn()
+ self.conn = conn
+
+ class FakeDB:
+ def open(self):
+ return conn
+
+ db = FakeDB()
+ self.db = db
+
+ class FakeZope2:
+ Startup = startup
+ App = FakeZope2App()
+ DB = db
+ bobo_application = 1
+ zpublisher_transactions_manager = 1
+
+ zope2 = FakeZope2()
+ self.zope2 = zope2
+
+ class FakeZope2Startup:
+ pass
+
+ class FakeAppConfig:
+ def setConfiguration(self, configroot):
+ self.configroot = configroot
+
+ class FakeApp:
+ config = FakeAppConfig()
+
+ app = FakeApp()
+
+ class FakeOFSApplicationApplication:
+ pass
+
+ dbapp = FakeOFSApplicationApplication()
+ self.dbapp = dbapp
+
+ class FakeOFSApplication:
+ def Application(self):
+ return dbapp
+
+ class FakeOFS:
+ Application = FakeOFSApplication()
+
+ class FakeTransaction:
+ committed = False
+ def commit(self):
+ self.committed = True
+
+ txn = FakeTransaction()
+ self.txn = txn
+
+ import Zope2
+ import Zope2.App
+ import Zope2.App.startup
+ import App
+ import App.config
+ import OFS
+ import OFS.Application
+ import transaction
+ self.old = (Zope2, Zope2.App, Zope2.App.startup,
+ App, App.config, OFS, OFS.Application, transaction)
+ import sys
+ sys.modules['Zope2'] = zope2
+ sys.modules['Zope2.App'] = zope2.App
+ sys.modules['Zope2.App.startup'] = appstartup
+ sys.modules['App'] = app
+ sys.modules['App.config'] = app.config
+ sys.modules['OFS'] = FakeOFS
+ sys.modules['OFS.Application'] = FakeOFSApplication()
+ sys.modules['transaction'] = txn
+
+ def tearDown(self):
+ import sys
+ sys.modules['Zope2'] = self.old[0]
+ sys.modules['Zope2.App'] = self.old[1]
+ sys.modules['Zope2.App.startup'] = self.old[2]
+ sys.modules['App'] = self.old[3]
+ sys.modules['App.config'] = self.old[4]
+ sys.modules['OFS'] = self.old[5]
+ sys.modules['OFS.Application'] = self.old[6]
+ sys.modules['transaction'] = self.old[7]
+ del self.old
+ _root.clear()
+
def _getFUT(self):
- from repoze.zope2.db import retry
- return retry
+ from repoze.zope2.db import getDB
+ return getDB
+
+ def test_getDB_appmissing(self):
+ config = {'zope.conf':None, 'appname':'MyApp'}
+ f = self._getFUT()
+ db = f(config)
+ self.assertEqual(len(self.root), 1)
+ self.assertEqual(self.root['MyApp'], self.dbapp)
+ self.assertEqual(db, self.db)
+ self.assertEqual(self.zope2.bobo_application, None)
+ self.assertEqual(self.zope2.zpublisher_transactions_manager, None)
+ self.assertEqual(self.txn.committed, True)
+ self.assertEqual(self.options.realized, True)
+ self.assertEqual(self.starter.configroot, None)
+ self.assertEqual(self.starter.setup_initial_logging, True)
+ self.assertEqual(self.starter.security_setup, True)
+ self.assertEqual(self.starter.publisher_setup, True)
+ self.assertEqual(self.starter.setup_final_logging, True)
+ self.assertEqual(self.conn.closed, True)
+ self.assertEqual(self.appstartup.started, True)
+
+ def test_getDB_appexists(self):
+ config = {'zope.conf':None, 'appname':'MyApp'}
+ _root['MyApp'] = True
+ f = self._getFUT()
+ db = f(config)
+ self.assertEqual(_root['MyApp'], True)
+ self.assertEqual(self.txn.committed, False)
+
+
+
- def test_retry_success(self):
- retry = self._getFUT()
- raiser = DummyConflictErrorRaiser(5)
- val = retry(raiser, 10)
- self.assertEqual(val, 5)
-
- def test_retry_fail(self):
- retry = self._getFUT()
- raiser = DummyConflictErrorRaiser(11)
- from ZODB.POSException import ConflictError
- self.assertRaises(ConflictError, retry, raiser, 10)
-
-class DummyConflictErrorRaiser:
- def __init__(self, retries):
- self.retries = retries
- self.tried = 0
-
- def __call__(self):
- if self.tried >= self.retries:
- return self.tried
- self.tried += 1
- from ZODB.POSException import ConflictError
- raise ConflictError
-
-
More information about the Repoze-checkins
mailing list