## # Copyright (c) 2010-2016 Apple Inc. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. ## """ Tests for L{twext.enterprise.adbapi2}. """ import gc from zope.interface.verify import verifyObject from twisted.python.failure import Failure from twisted.trial.unittest import TestCase from twisted.internet.defer import Deferred, fail, succeed, inlineCallbacks from twisted.test.proto_helpers import StringTransport from twext.enterprise.ienterprise import ConnectionError from twext.enterprise.ienterprise import AlreadyFinishedError from twext.enterprise.adbapi2 import ConnectionPoolClient from twext.enterprise.adbapi2 import ConnectionPoolConnection from twext.enterprise.ienterprise import IAsyncTransaction from twext.enterprise.ienterprise import ICommandBlock from twext.enterprise.adbapi2 import FailsafeException from twext.enterprise.adbapi2 import ConnectionPool from twext.enterprise.fixtures import ConnectionPoolHelper from twext.enterprise.fixtures import resultOf from twext.enterprise.fixtures import ClockWithThreads from twext.enterprise.fixtures import FakeConnectionError from twext.enterprise.fixtures import RollbackFail from twext.enterprise.fixtures import CommitFail from twext.enterprise.adbapi2 import Commit from twext.enterprise.adbapi2 import _HookableOperation class TrashCollector(object): """ Test helper for monitoring gc.garbage. """ def __init__(self, testCase): self.testCase = testCase testCase.addCleanup(self.checkTrash) self.start() def start(self): gc.collect() self.garbageStart = len(gc.garbage) def checkTrash(self): """ Ensure that the test has added no additional garbage. """ gc.collect() newGarbage = gc.garbage[self.garbageStart:] if newGarbage: # Don't clean up twice. self.start() self.testCase.fail("New garbage: " + repr(newGarbage)) class AssertResultHelper(object): """ Mixin for asserting about synchronous Deferred results. """ def assertResultList(self, resultList, expected): """ Assert that a list created with L{resultOf} contais the expected result. @param resultList: The return value of L{resultOf}. @type resultList: L{list} @param expected: The expected value that should be present in the list; a L{Failure} if an exception is expected to be raised. """ if not resultList: self.fail("No result; Deferred didn't fire yet.") else: if isinstance(resultList[0], Failure): if isinstance(expected, Failure): resultList[0].trap(expected.type) else: resultList[0].raiseException() else: self.assertEqual(resultList, [expected]) class ConnectionPoolBootTests(TestCase): """ Tests for the start-up phase of L{ConnectionPool}. """ def test_threadCount(self): """ The reactor associated with a L{ConnectionPool} will have its maximum thread count adjusted when L{ConnectionPool.startService} is called, to accomodate for L{ConnectionPool.maxConnections} additional threads. Stopping the service should restore it to its original value, so that a repeatedly re-started L{ConnectionPool} will not cause the thread ceiling to grow without bound. """ defaultMax = 27 connsMax = 45 combinedMax = defaultMax + connsMax pool = ConnectionPool(None, maxConnections=connsMax) pool.reactor = ClockWithThreads() threadpool = pool.reactor.getThreadPool() pool.reactor.suggestThreadPoolSize(defaultMax) self.assertEquals(threadpool.max, defaultMax) pool.startService() self.assertEquals(threadpool.max, combinedMax) justChecking = [] pool.stopService().addCallback(justChecking.append) # No SQL run, so no threads started, so this deferred should fire # immediately. If not, we're in big trouble, so sanity check. self.assertEquals(justChecking, [None]) self.assertEquals(threadpool.max, defaultMax) def test_isRunning(self): """ L{ConnectionPool.startService} should set its C{running} attribute to true. """ pool = ConnectionPool(None) pool.reactor = ClockWithThreads() self.assertEquals(pool.running, False) pool.startService() self.assertEquals(pool.running, True) class ConnectionPoolNameTests(TestCase): """ Tests for L{ConnectionPool}'s C{name} attribute. """ def test_default(self): """ If no value is given for the C{name} parameter to L{ConnectionPool}'s initializer then L{ConnectionPool.name} is C{None}. """ pool = ConnectionPool(None) self.assertIs(None, pool.name) def test_specified(self): """ If a value is given for the C{name} parameter to L{ConnectionPool}'s initializer then it is used as the value for L{ConnectionPool.name}. """ name = "some test pool" pool = ConnectionPool(None, name=name) self.assertEqual(name, pool.name) class ConnectionPoolTests(ConnectionPoolHelper, TestCase, AssertResultHelper): """ Tests for L{ConnectionPool}. """ def test_tooManyConnections(self): """ When the number of outstanding busy transactions exceeds the number of slots specified by L{ConnectionPool.maxConnections}, L{ConnectionPool.connection} will return a pooled transaction that is not backed by any real database connection; this object will queue its SQL statements until an existing connection becomes available. """ a = self.createTransaction() alphaResult = self.resultOf(a.execSQL("alpha")) [[_ignore_counter, _ignore_echo]] = alphaResult[0] b = self.createTransaction() # "b" should have opened a connection. self.assertEquals(len(self.factory.connections), 2) betaResult = self.resultOf(b.execSQL("beta")) [[bcounter, _ignore_becho]] = betaResult[0] # both "a" and "b" are holding open a connection now; let's try to open # a third one. (The ordering will be deterministic even if this fails, # because those threads are already busy.) c = self.createTransaction() gammaResult = self.resultOf(c.execSQL("gamma")) # Did "c" open a connection? Let's hope not... self.assertEquals(len(self.factory.connections), 2) # SQL shouldn't be executed too soon... self.assertEquals(gammaResult, []) commitResult = self.resultOf(b.commit()) # Now that "b" has committed, "c" should be able to complete. [[ccounter, _ignore_cecho]] = gammaResult[0] # The connection for "a" ought to still be busy, so let's make sure # we're using the one for "c". self.assertEquals(ccounter, bcounter) # Sanity check: the commit should have succeeded! self.assertEquals(commitResult, [None]) def test_stopService(self): """ L{ConnectionPool.stopService} stops all the associated L{ThreadHolder}s and thereby frees up the resources it is holding. """ a = self.createTransaction() alphaResult = self.resultOf(a.execSQL("alpha")) [[[_ignore_counter, _ignore_echo]]] = alphaResult self.assertEquals(len(self.factory.connections), 1) self.assertEquals(len(self.holders), 1) [holder] = self.holders self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, False) self.pool.stopService() self.assertEquals(self.pool.running, False) self.assertEquals(len(self.holders), 1) self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) # Closing fake connections removes them from the list. self.assertEquals(len(self.factory.connections), 1) self.assertEquals(self.factory.connections[0].closed, True) def test_retryAfterConnectError(self): """ When the C{connectionFactory} passed to L{ConnectionPool} raises an exception, the L{ConnectionPool} will log the exception and delay execution of a new connection's SQL methods until an attempt succeeds. """ self.factory.willFail() self.factory.willFail() self.factory.willConnect() c = self.createTransaction() def checkOneFailure(): errors = self.flushLoggedErrors(FakeConnectionError) self.assertEquals(len(errors), 1) checkOneFailure() d = c.execSQL("alpha") happened = [] d.addBoth(happened.append) self.assertEquals(happened, []) self.clock.advance(self.pool.RETRY_TIMEOUT + 0.01) checkOneFailure() self.assertEquals(happened, []) self.clock.advance(self.pool.RETRY_TIMEOUT + 0.01) self.flushHolders() self.assertEquals(happened, [[[1, "alpha"]]]) def test_shutdownDuringRetry(self): """ If a L{ConnectionPool} is attempting to shut down while it's in the process of re-trying a connection attempt that received an error, the connection attempt should be cancelled and the shutdown should complete as normal. """ self.factory.defaultFail() self.createTransaction() errors = self.flushLoggedErrors(FakeConnectionError) self.assertEquals(len(errors), 1) stopd = [] self.pool.stopService().addBoth(stopd.append) self.assertResultList(stopd, None) self.assertEquals(self.clock.calls, []) [holder] = self.holders self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) def test_shutdownDuringAttemptSuccess(self): """ If L{ConnectionPool.stopService} is called while a connection attempt is outstanding, the resulting L{Deferred} won't be fired until the connection attempt has finished; in this case, succeeded. """ self.pauseHolders() self.createTransaction() stopd = [] self.pool.stopService().addBoth(stopd.append) self.assertEquals(stopd, []) self.flushHolders() self.assertResultList(stopd, None) [holder] = self.holders self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) def test_shutdownDuringAttemptFailed(self): """ If L{ConnectionPool.stopService} is called while a connection attempt is outstanding, the resulting L{Deferred} won't be fired until the connection attempt has finished; in this case, failed. """ self.factory.defaultFail() self.pauseHolders() self.createTransaction() stopd = [] self.pool.stopService().addBoth(stopd.append) self.assertEquals(stopd, []) self.flushHolders() errors = self.flushLoggedErrors(FakeConnectionError) self.assertEquals(len(errors), 1) self.assertResultList(stopd, None) [holder] = self.holders self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) def test_stopServicePseudoTxn(self): """ When L{ConnectionPool.stopService} is called with a pending L{_ConnectingPseudoTxn} active, the DB connection being created is closed. """ self.pauseHolders() self.createTransaction() stopResult = self.resultOf(self.pool.stopService()) self.assertEquals(stopResult, []) self.flushHolders() [holder] = self.holders self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) self.assertEquals(len(self.factory.connections), 1) self.assertEquals(self.factory.connections[0].closed, True) def test_stopServiceMidAbort(self): """ When L{ConnectionPool.stopService} is called with deferreds from C{abort} still outstanding, it will wait for the currently-aborting transaction to fully abort before firing the L{Deferred} returned from C{stopService}. """ # TODO: commit() too? self.pauseHolders() c = self.createTransaction() abortResult = self.resultOf(c.abort()) # Should abort instantly, as it hasn't managed to unspool anything yet. # FIXME: kill all Deferreds associated with this thing, make sure that # any outstanding query callback chains get nuked. self.assertEquals(abortResult, [None]) stopResult = self.resultOf(self.pool.stopService()) self.assertEquals(stopResult, []) self.flushHolders() # self.assertEquals(abortResult, [None]) self.assertResultList(stopResult, None) def test_stopServiceWithSpooled(self): """ When L{ConnectionPool.stopService} is called when spooled transactions are outstanding, any pending L{Deferreds} returned by those transactions will be failed with L{ConnectionError}. """ # Use up the free slots so we have to spool. hold = [] hold.append(self.createTransaction()) hold.append(self.createTransaction()) c = self.createTransaction() se = self.resultOf(c.execSQL("alpha")) ce = self.resultOf(c.commit()) self.assertEquals(se, []) self.assertEquals(ce, []) self.resultOf(self.pool.stopService()) self.assertEquals(se[0].type, self.translateError(ConnectionError)) self.assertEquals(ce[0].type, self.translateError(ConnectionError)) def test_repoolSpooled(self): """ Regression test for a somewhat tricky-to-explain bug: when a spooled transaction which has already had commit() called on it before it's received a real connection to start executing on, it will not leave behind any detritus that prevents stopService from working. """ self.pauseHolders() c = self.createTransaction() c2 = self.createTransaction() c3 = self.createTransaction() c.commit() c2.commit() c3.commit() self.flushHolders() self.assertEquals(len(self.factory.connections), 2) stopResult = self.resultOf(self.pool.stopService()) self.assertEquals(stopResult, [None]) self.assertEquals(len(self.factory.connections), 2) self.assertEquals(self.factory.connections[0].closed, True) self.assertEquals(self.factory.connections[1].closed, True) def test_connectAfterStop(self): """ Calls to connection() after stopService() result in transactions which immediately fail all operations. """ stopResults = self.resultOf(self.pool.stopService()) self.assertEquals(stopResults, [None]) self.pauseHolders() postClose = self.createTransaction() queryResult = self.resultOf(postClose.execSQL("hello")) self.assertEquals(len(queryResult), 1) self.assertEquals(queryResult[0].type, self.translateError(ConnectionError)) def test_connectAfterStartedStopping(self): """ Calls to connection() after stopService() has been called but before it has completed will result in transactions which immediately fail all operations. """ self.pauseHolders() preClose = self.createTransaction() preCloseResult = self.resultOf(preClose.execSQL("statement")) stopResult = self.resultOf(self.pool.stopService()) postClose = self.createTransaction() queryResult = self.resultOf(postClose.execSQL("hello")) self.assertEquals(stopResult, []) self.assertEquals(len(queryResult), 1) self.assertEquals( queryResult[0].type, self.translateError(ConnectionError) ) self.assertEquals(len(preCloseResult), 1) self.assertEquals( preCloseResult[0].type, self.translateError(ConnectionError) ) def test_abortFailsDuringStopService(self): """ L{IAsyncTransaction.abort} might fail, most likely because the underlying database connection has already been disconnected. If this happens, shutdown should continue. """ txns = [] txns.append(self.createTransaction()) txns.append(self.createTransaction()) for txn in txns: # Make sure rollback will actually be executed. results = self.resultOf(txn.execSQL("maybe change something!")) [[[_ignore_counter, echo]]] = results self.assertEquals("maybe change something!", echo) # Fail one (and only one) call to rollback(). self.factory.rollbackFail = True stopResult = self.resultOf(self.pool.stopService()) self.assertEquals(stopResult, [None]) self.assertEquals(len(self.flushLoggedErrors(RollbackFail)), 1) self.assertEquals(self.factory.connections[0].closed, True) self.assertEquals(self.factory.connections[1].closed, True) def test_partialTxnFailsDuringStopService(self): """ Using the logic in L{ConnectionPool.stopService}, make sure that an L{_ConnectedTxn} cannot continue to process SQL after L{_ConnectedTxn.abort} is called and before L{_ConnectedTxn.reset} is called. """ txn = self.createTransaction() if hasattr(txn, "_baseTxn"): # Send initial statement txn.execSQL("maybe change something!") # Make it look like the service is stopping txn._baseTxn._connection.close() txn._baseTxn.terminate() # Try to send more SQL - must fail self.failUnlessRaises(RuntimeError, txn.execSQL, "maybe change something else!") def test_abortRecycledTransaction(self): """ L{ConnectionPool.stopService} will shut down if a recycled transaction is still pending. """ recycled = self.createTransaction() self.resultOf(recycled.commit()) remember = [] remember.append(self.createTransaction()) self.assertEquals(self.resultOf(self.pool.stopService()), [None]) def test_abortSpooled(self): """ Aborting a still-spooled transaction (one which has no statements being executed) will result in all of its Deferreds immediately failing and none of the queued statements being executed. """ active = [] # Use up the available connections ... for _ignore in xrange(self.pool.maxConnections): active.append(self.createTransaction()) # ... so that this one has to be spooled. spooled = self.createTransaction() result = self.resultOf(spooled.execSQL("alpha")) # sanity check, it would be bad if this actually executed. self.assertEqual(result, []) self.resultOf(spooled.abort()) self.assertEqual(result[0].type, self.translateError(ConnectionError)) def test_waitForAlreadyAbortedTransaction(self): """ L{ConnectionPool.stopService} will wait for all transactions to shut down before exiting, including those which have already been stopped. """ it = self.createTransaction() self.pauseHolders() abortResult = self.resultOf(it.abort()) # steal it from the queue so we can do it out of order d, _ignore_work = self.holders[0]._q.get() # that should be the only work unit so don't continue if something else # got in there self.assertEquals(list(self.holders[0]._q.queue), []) self.assertEquals(len(self.holders), 1) self.flushHolders() stopResult = self.resultOf(self.pool.stopService()) # Sanity check that we haven't actually stopped it yet self.assertEquals(abortResult, []) # We haven't fired it yet, so the service had better not have # stopped... self.assertEquals(stopResult, []) d.callback(None) self.flushHolders() self.assertEquals(abortResult, [None]) self.assertEquals(stopResult, [None]) def test_garbageCollectedTransactionAborts(self): """ When an L{IAsyncTransaction} is garbage collected, it ought to abort itself. """ t = self.createTransaction() self.resultOf(t.execSQL("echo", [])) conns = self.factory.connections self.assertEquals(len(conns), 1) self.assertEquals(conns[0]._rollbackCount, 0) del t gc.collect() self.flushHolders() self.assertEquals(len(conns), 1) self.assertEquals(conns[0]._rollbackCount, 1) self.assertEquals(conns[0]._commitCount, 0) def circularReferenceTest(self, finish, hook): """ Collecting a completed (committed or aborted) L{IAsyncTransaction} should not leak any circular references. """ tc = TrashCollector(self) commitExecuted = [] def carefullyManagedScope(): t = self.createTransaction() def holdAReference(): """ This is a hook that holds a reference to "t". """ commitExecuted.append(True) return t.execSQL("teardown", []) hook(t, holdAReference) finish(t) self.failIf(commitExecuted, "Commit hook executed.") carefullyManagedScope() tc.checkTrash() def test_noGarbageOnCommit(self): """ Committing a transaction does not cause gc garbage. """ self.circularReferenceTest( lambda txn: txn.commit(), lambda txn, hook: txn.preCommit(hook) ) def test_noGarbageOnCommitWithAbortHook(self): """ Committing a transaction does not cause gc garbage. """ self.circularReferenceTest( lambda txn: txn.commit(), lambda txn, hook: txn.postAbort(hook) ) def test_noGarbageOnAbort(self): """ Aborting a transaction does not cause gc garbage. """ self.circularReferenceTest( lambda txn: txn.abort(), lambda txn, hook: txn.preCommit(hook) ) def test_noGarbageOnAbortWithPostCommitHook(self): """ Aborting a transaction does not cause gc garbage. """ self.circularReferenceTest( lambda txn: txn.abort(), lambda txn, hook: txn.postCommit(hook) ) def test_tooManyConnectionsWhileOthersFinish(self): """ L{ConnectionPool.connection} will not spawn more than the maximum connections if there are finishing transactions outstanding. """ a = self.createTransaction() b = self.createTransaction() self.pauseHolders() a.abort() b.abort() # Remove the holders for the existing connections, so that the "extra" # connection() call wins the race and gets executed first. oldholders = list(self.holders) self.holders[:] = [] self.createTransaction() self.flushHolders() self.assertEquals(len(self.factory.connections), 2) self.holders = oldholders self.flushHolders() def setParamstyle(self, paramstyle): """ Change the paramstyle of the transaction under test. """ self.pool.dbtype = self.pool.dbtype.copyreplace(paramstyle=paramstyle) def test_propagateParamstyle(self): """ Each different type of L{ISQLExecutor} relays the C{paramstyle} attribute from the L{ConnectionPool}. """ TEST_PARAMSTYLE = "justtesting" self.setParamstyle(TEST_PARAMSTYLE) normaltxn = self.createTransaction() self.assertEquals(normaltxn.dbtype.paramstyle, TEST_PARAMSTYLE) self.assertEquals(normaltxn.commandBlock().dbtype.paramstyle, TEST_PARAMSTYLE) self.pauseHolders() extra = [] extra.append(self.createTransaction()) waitingtxn = self.createTransaction() self.assertEquals(waitingtxn.dbtype.paramstyle, TEST_PARAMSTYLE) self.flushHolders() self.pool.stopService() notxn = self.createTransaction() self.assertEquals(notxn.dbtype.paramstyle, TEST_PARAMSTYLE) def setDialect(self, dialect): """ Change the dialect of the transaction under test. """ self.pool.dbtype = self.pool.dbtype.copyreplace(dialect=dialect) def test_propagateDialect(self): """ Each different type of L{ISQLExecutor} relays the C{dialect} attribute from the L{ConnectionPool}. """ TEST_DIALECT = "otherdialect" self.setDialect(TEST_DIALECT) normaltxn = self.createTransaction() self.assertEquals(normaltxn.dbtype.dialect, TEST_DIALECT) self.assertEquals(normaltxn.commandBlock().dbtype.dialect, TEST_DIALECT) self.pauseHolders() extra = [] extra.append(self.createTransaction()) waitingtxn = self.createTransaction() self.assertEquals(waitingtxn.dbtype.dialect, TEST_DIALECT) self.flushHolders() self.pool.stopService() notxn = self.createTransaction() self.assertEquals(notxn.dbtype.dialect, TEST_DIALECT) def test_reConnectWhenFirstExecFails(self): """ Generally speaking, DB-API 2.0 adapters do not provide information about the cause of a failed C{execute} method; they definitely don't provide it in a way which can be identified as related to the syntax of the query, the state of the database itself, the state of the connection, etc. Therefore the best general heuristic for whether the connection to the database has been lost and needs to be re-established is to catch exceptions which are raised by the I{first} statement executed in a transaction. """ # Allow C{connect} to succeed. This should behave basically the same # whether connect() happened to succeed in some previous transaction # and it's recycling the underlying transaction, or connect() just # succeeded. Either way you just have a _SingleTxn wrapping a # _ConnectedTxn. txn = self.createTransaction() self.assertEquals(len(self.factory.connections), 1, "Sanity check failed.") class CustomExecuteFailed(Exception): """ Custom "execute-failed" exception. """ self.factory.connections[0].executeWillFail(CustomExecuteFailed) results = self.resultOf(txn.execSQL("hello, world!")) [[[_ignore_counter, echo]]] = results self.assertEquals("hello, world!", echo) # Two execution attempts should have been made, one on each connection. # The first failed with a RuntimeError, but that is deliberately # obscured, because then we tried again and it succeeded. self.assertEquals( len(self.factory.connections), 2, "No new connection opened." ) self.assertEquals(self.factory.connections[0].executions, 1) self.assertEquals(self.factory.connections[1].executions, 1) self.assertEquals(self.factory.connections[0].closed, True) self.assertEquals(self.factory.connections[1].closed, False) # Nevertheless, since there is currently no classification of "safe" # errors, we should probably log these messages when they occur. self.assertEquals(len(self.flushLoggedErrors(CustomExecuteFailed)), 1) def test_reConnectWhenFirstExecOnExistingConnectionFails( self, moreFailureSetup=lambda factory: None ): """ Another situation that might arise is that a connection will be successfully connected, executed and recycled into the connection pool; then, the database server will shut down and the connections will die, but we will be none the wiser until we try to use them. """ txn = self.createTransaction() moreFailureSetup(self.factory) self.assertEquals( len(self.factory.connections), 1, "Sanity check failed." ) results = self.resultOf(txn.execSQL("hello, world!")) txn.commit() [[[_ignore_counter, echo]]] = results self.assertEquals("hello, world!", echo) txn2 = self.createTransaction() self.assertEquals( len(self.factory.connections), 1, "Sanity check failed." ) class CustomExecFail(Exception): """ Custom C{execute()} failure. """ self.factory.connections[0].executeWillFail(CustomExecFail) results = self.resultOf(txn2.execSQL("second try!")) txn2.commit() [[[_ignore_counter, echo]]] = results self.assertEquals("second try!", echo) self.assertEquals(len(self.flushLoggedErrors(CustomExecFail)), 1) def test_closeExceptionDoesntHinderReconnection(self): """ In some database bindings, if the server closes the connection, C{close()} will fail. If C{close} fails, there's not much that could mean except that the connection is already closed, so similar to the condition described in L{test_reConnectWhenFirstExecOnExistingConnectionFails}, the failure should be logged, but transparent to application code. """ class BindingSpecificException(Exception): """ Exception that's a placeholder for something that a database binding might raise. """ def alsoFailClose(factory): factory.childCloseWillFail(BindingSpecificException()) t = self.test_reConnectWhenFirstExecOnExistingConnectionFails( alsoFailClose ) errors = self.flushLoggedErrors(BindingSpecificException) self.assertEquals(len(errors), 1) return t def test_preCommitSuccess(self): """ Callables passed to L{IAsyncTransaction.preCommit} will be invoked upon commit. """ txn = self.createTransaction() def simple(): simple.done = True simple.done = False txn.preCommit(simple) self.assertEquals(simple.done, False) result = self.resultOf(txn.commit()) self.assertEquals(len(result), 1) self.assertEquals(simple.done, True) def test_deferPreCommit(self): """ If callables passed to L{IAsyncTransaction.preCommit} return L{Deferred}s, they will defer the actual commit operation until it has fired. """ txn = self.createTransaction() d = Deferred() def wait(): wait.started = True def executed(it): wait.sqlResult = it # To make sure the _underlying_ commit operation was Deferred, we # have to execute some SQL to make sure it happens. d.addCallback(lambda ignored: txn.execSQL("some test sql")) d.addCallback(executed) return d wait.started = False wait.sqlResult = None txn.preCommit(wait) result = self.resultOf(txn.commit()) self.flushHolders() self.assertEquals(wait.started, True) self.assertEquals(wait.sqlResult, None) self.assertEquals(result, []) d.callback(None) # allow network I/O for pooled / networked implementation; there should # be the commit message now. self.flushHolders() self.assertEquals(len(result), 1) self.assertEquals(wait.sqlResult, [[1, "some test sql"]]) def test_failPreCommit(self): """ If callables passed to L{IAsyncTransaction.preCommit} raise an exception or return a Failure, subsequent callables will not be run, and the transaction will be aborted. """ def test(flawedCallable, exc): # Set up. test.committed = False test.aborted = False # Create transaction and add monitoring hooks. txn = self.createTransaction() def didCommit(): test.committed = True def didAbort(): test.aborted = True txn.postCommit(didCommit) txn.postAbort(didAbort) txn.preCommit(flawedCallable) result = self.resultOf(txn.commit()) self.flushHolders() self.assertResultList(result, Failure(exc())) self.assertEquals(test.committed, False) self.assertEquals(test.aborted, True) def failer(): return fail(ZeroDivisionError()) def raiser(): raise EOFError() test(failer, ZeroDivisionError) test(raiser, EOFError) def test_noOpCommitDoesntHinderReconnection(self): """ Until you've executed a query or performed a statement on an ADBAPI connection, the connection is semantically idle (between transactions). A .commit() or .rollback() followed immediately by a .commit() is therefore pointless, and can be ignored. Furthermore, actually executing the commit and propagating a possible connection-oriented error causes clients to see errors, when, if those clients had actually executed any statements, the connection would have been recycled and the statement transparently re-executed by the logic tested by L{test_reConnectWhenFirstExecFails}. """ txn = self.createTransaction() self.factory.commitFail = True self.factory.rollbackFail = True [x] = self.resultOf(txn.commit()) # No statements have been executed, so C{commit} will *not* be # executed. self.assertEquals(self.factory.commitFail, True) self.assertIdentical(x, None) self.assertEquals(len(self.pool._free), 1) self.assertEquals(self.pool._finishing, []) self.assertEquals(len(self.factory.connections), 1) self.assertEquals(self.factory.connections[0].closed, False) def test_reConnectWhenSecondExecFailsThenFirstExecFails(self): """ Other connection-oriented errors might raise exceptions if they occur in the middle of a transaction, but that should cause the error to be caught, the transaction to be aborted, and the (closed) connection to be recycled, where the next transaction that attempts to do anything with it will encounter the error immediately and discover it needs to be recycled. It would be better if this behavior were invisible, but that could only be accomplished with more precise database exceptions. We may come up with support in the future for more precisely identifying exceptions, but I{unknown} exceptions should continue to be treated in this manner, relaying the exception back to application code but attempting a re-connection on the next try. """ txn = self.createTransaction() [[[_ignore_counter, _ignore_echo]]] = self.resultOf(txn.execSQL("hello, world!", [])) self.factory.connections[0].executeWillFail(ZeroDivisionError) [f] = self.resultOf(txn.execSQL("divide by zero", [])) f.trap(self.translateError(ZeroDivisionError)) self.assertEquals(self.factory.connections[0].executions, 2) # Reconnection should work exactly as before. self.assertEquals(self.factory.connections[0].closed, False) # Application code has to roll back its transaction at this point, # since it failed (and we don't necessarily know why it failed: not # enough information). self.resultOf(txn.abort()) self.factory.connections[0].executions = 0 # re-set for next test self.assertEquals(len(self.factory.connections), 1) self.test_reConnectWhenFirstExecFails() def test_disconnectOnFailedRollback(self): """ When C{rollback} fails for any reason on a connection object, then we don't know what state it's in. Most likely, it's already been disconnected, so the connection should be closed and the transaction de-pooled instead of recycled. Also, a new connection will immediately be established to keep the pool size the same. """ txn = self.createTransaction() results = self.resultOf(txn.execSQL("maybe change something!")) [[[_ignore_counter, echo]]] = results self.assertEquals("maybe change something!", echo) self.factory.rollbackFail = True [x] = self.resultOf(txn.abort()) # Abort does not propagate the error on, the transaction merely gets # disposed of. self.assertIdentical(x, None) self.assertEquals(len(self.pool._free), 1) self.assertEquals(self.pool._finishing, []) self.assertEquals(len(self.factory.connections), 2) self.assertEquals(self.factory.connections[0].closed, True) self.assertEquals(self.factory.connections[1].closed, False) self.assertEquals(len(self.flushLoggedErrors(RollbackFail)), 1) def test_exceptionPropagatesFailedCommit(self): """ A failed C{rollback} is fine (the premature death of the connection without C{commit} means that the changes are surely gone), but a failed C{commit} has to be relayed to client code, since that actually means some changes didn't hit the database. """ txn = self.createTransaction() self.factory.commitFail = True results = self.resultOf(txn.execSQL("maybe change something!")) [[[_ignore_counter, echo]]] = results self.assertEquals("maybe change something!", echo) [x] = self.resultOf(txn.commit()) x.trap(self.translateError(CommitFail)) self.assertEquals(len(self.pool._free), 1) self.assertEquals(self.pool._finishing, []) self.assertEquals(len(self.factory.connections), 2) self.assertEquals(self.factory.connections[0].closed, True) self.assertEquals(self.factory.connections[1].closed, False) def test_commandBlock(self): """ L{IAsyncTransaction.commandBlock} returns an L{IAsyncTransaction} provider which ensures that a block of commands are executed together. """ txn = self.createTransaction() a = self.resultOf(txn.execSQL("a")) cb = txn.commandBlock() verifyObject(ICommandBlock, cb) b = self.resultOf(cb.execSQL("b")) d = self.resultOf(txn.execSQL("d")) c = self.resultOf(cb.execSQL("c")) cb.end() e = self.resultOf(txn.execSQL("e")) self.assertEquals( self.factory.connections[0].cursors[0].allExecutions, [("a", []), ("b", []), ("c", []), ("d", []), ("e", [])] ) self.assertEquals(len(a), 1) self.assertEquals(len(b), 1) self.assertEquals(len(c), 1) self.assertEquals(len(d), 1) self.assertEquals(len(e), 1) def test_commandBlockWithLatency(self): """ A block returned by L{IAsyncTransaction.commandBlock} won't start executing until all SQL statements scheduled before it have completed. """ self.pauseHolders() txn = self.createTransaction() a = self.resultOf(txn.execSQL("a")) b = self.resultOf(txn.execSQL("b")) cb = txn.commandBlock() c = self.resultOf(cb.execSQL("c")) d = self.resultOf(cb.execSQL("d")) e = self.resultOf(txn.execSQL("e")) cb.end() self.flushHolders() self.assertEquals( self.factory.connections[0].cursors[0].allExecutions, [("a", []), ("b", []), ("c", []), ("d", []), ("e", [])] ) self.assertEquals(len(a), 1) self.assertEquals(len(b), 1) self.assertEquals(len(c), 1) self.assertEquals(len(d), 1) self.assertEquals(len(e), 1) def test_twoCommandBlocks(self, flush=lambda: None): """ When execution of one command block is complete, it will proceed to the next queued block, then to regular SQL executed on the transaction. """ txn = self.createTransaction() cb1 = txn.commandBlock() cb2 = txn.commandBlock() txn.execSQL("e") cb1.execSQL("a") cb2.execSQL("c") cb1.execSQL("b") cb2.execSQL("d") cb2.end() cb1.end() flush() self.flushHolders() self.assertEquals( self.factory.connections[0].cursors[0].allExecutions, [("a", []), ("b", []), ("c", []), ("d", []), ("e", [])] ) def test_twoCommandBlocksLatently(self): """ Same as L{test_twoCommandBlocks}, but with slower callbacks. """ self.pauseHolders() self.test_twoCommandBlocks(self.flushHolders) def test_commandBlockEndTwice(self): """ L{CommandBlock.end} will raise L{AlreadyFinishedError} when called more than once. """ txn = self.createTransaction() block = txn.commandBlock() block.end() self.assertRaises(AlreadyFinishedError, block.end) def test_commandBlockDelaysCommit(self): """ Some command blocks need to run asynchronously, without the overall transaction-managing code knowing how far they've progressed. Therefore when you call {IAsyncTransaction.commit}(), it should not actually take effect if there are any pending command blocks. """ txn = self.createTransaction() block = txn.commandBlock() commitResult = self.resultOf(txn.commit()) self.resultOf(block.execSQL("in block")) self.assertEquals(commitResult, []) self.assertEquals( self.factory.connections[0].cursors[0].allExecutions, [("in block", [])] ) block.end() self.flushHolders() self.assertEquals(commitResult, [None]) def test_commandBlockDoesntDelayAbort(self): """ A L{CommandBlock} can't possibly have anything interesting to say about a transaction that gets rolled back, so C{abort} applies immediately; all outstanding C{execSQL}s will fail immediately, on both command blocks and on the transaction itself. """ txn = self.createTransaction() block = txn.commandBlock() block2 = txn.commandBlock() abortResult = self.resultOf(txn.abort()) self.assertEquals(abortResult, [None]) self.assertRaises(AlreadyFinishedError, block2.execSQL, "bar") self.assertRaises(AlreadyFinishedError, block.execSQL, "foo") self.assertRaises(AlreadyFinishedError, txn.execSQL, "baz") self.assertEquals( self.factory.connections[0].cursors[0].allExecutions, [] ) # end() should _not_ raise an exception, because this is the sort of # thing that might be around a try/finally or try/except; it's just # putting the commandBlock itself into a state consistent with the # transaction. block.end() block2.end() def test_endedBlockDoesntExecuteMoreSQL(self): """ Attempting to execute SQL on a L{CommandBlock} which has had C{end} called on it will result in an L{AlreadyFinishedError}. """ txn = self.createTransaction() block = txn.commandBlock() block.end() self.assertRaises(AlreadyFinishedError, block.execSQL, "hello") self.assertEquals( self.factory.connections[0].cursors[0].allExecutions, [] ) def test_commandBlockAfterCommitRaises(self): """ Once an L{IAsyncTransaction} has been committed, L{commandBlock} raises an exception. """ txn = self.createTransaction() txn.commit() self.assertRaises(AlreadyFinishedError, txn.commandBlock) def test_commandBlockAfterAbortRaises(self): """ Once an L{IAsyncTransaction} has been committed, L{commandBlock} raises an exception. """ txn = self.createTransaction() self.resultOf(txn.abort()) self.assertRaises(AlreadyFinishedError, txn.commandBlock) def test_raiseOnZeroRowCount(self): """ L{IAsyncTransaction.execSQL} will return a L{Deferred} failing with the exception passed as its raiseOnZeroRowCount argument if the underlying query returns no rows. """ self.factory.hasResults = False txn = self.createTransaction() f = self.resultOf( txn.execSQL("hello", raiseOnZeroRowCount=ZeroDivisionError) )[0] self.assertRaises(ZeroDivisionError, f.raiseException) txn.commit() def test_raiseOnZeroRowCountWithUnreliableRowCount(self): """ As it turns out, some databases can't reliably tell you how many rows they're going to fetch via the C{rowcount} attribute before the rows have actually been fetched, so the C{raiseOnZeroRowCount} will I{not} raise an exception if C{rowcount} is zero but C{description} and C{fetchall} indicates the presence of some rows. """ self.factory.hasResults = True self.factory.shouldUpdateRowcount = False txn = self.createTransaction() r = self.resultOf( txn.execSQL("some-rows", raiseOnZeroRowCount=RuntimeError) ) [[[_ignore_counter, echo]]] = r self.assertEquals(echo, "some-rows") class IOPump(object): """ Connect a client and a server. @ivar client: a client protocol @ivar server: a server protocol """ def __init__(self, client, server): self.client = client self.server = server self.clientTransport = StringTransport() self.serverTransport = StringTransport() self.client.makeConnection(self.clientTransport) self.server.makeConnection(self.serverTransport) self.c2s = [self.clientTransport, self.server] self.s2c = [self.serverTransport, self.client] def moveData(self, (outTransport, inProtocol)): """ Move data from a L{StringTransport} to an L{IProtocol}. @return: C{True} if any data was moved, C{False} if no data was moved. """ data = outTransport.io.getvalue() outTransport.io.seek(0) outTransport.io.truncate() if data: inProtocol.dataReceived(data) return True else: return False def pump(self): """ Deliver all input from the client to the server, then from the server to the client. """ a = self.moveData(self.c2s) b = self.moveData(self.s2c) return a or b def flush(self, maxTurns=100): """ Continue pumping until no more data is flowing. """ turns = 0 while self.pump(): turns += 1 if turns > maxTurns: raise RuntimeError("Ran too long!") class NetworkedPoolHelper(ConnectionPoolHelper): """ An extension of L{ConnectionPoolHelper} that can set up a L{ConnectionPoolClient} and L{ConnectionPoolConnection} attached to each other. """ def setUp(self): """ Do the same setup from L{ConnectionPoolBase}, but also establish a loopback connection between a L{ConnectionPoolConnection} and a L{ConnectionPoolClient}. """ super(NetworkedPoolHelper, self).setUp() self.pump = IOPump( ConnectionPoolClient( dbtype=self.dbtype, ), ConnectionPoolConnection(self.pool) ) def flushHolders(self): """ In addition to flushing the L{ThreadHolder} stubs, also flush any pending network I/O. """ self.pump.flush() super(NetworkedPoolHelper, self).flushHolders() self.pump.flush() def createTransaction(self): txn = self.pump.client.newTransaction() self.pump.flush() return txn def translateError(self, err): """ All errors raised locally will unfortunately be translated into UnknownRemoteError, since AMP requires specific enumeration of all of them. Flush the locally logged error of the given type and return L{UnknownRemoteError}. """ if err in Commit.errors: return err self.flushLoggedErrors(err) return FailsafeException def resultOf(self, it): result = resultOf(it) self.pump.flush() return result class NetworkedConnectionPoolTests(NetworkedPoolHelper, ConnectionPoolTests): """ Tests for L{ConnectionPoolConnection} and L{ConnectionPoolClient} interacting with each other. """ def setParamstyle(self, paramstyle): """ Change the paramstyle on both the pool and the client. """ super(NetworkedConnectionPoolTests, self).setParamstyle(paramstyle) self.pump.client.dbtype = self.pump.client.dbtype.copyreplace(paramstyle=paramstyle) def setDialect(self, dialect): """ Change the dialect on both the pool and the client. """ super(NetworkedConnectionPoolTests, self).setDialect(dialect) self.pump.client.dbtype = self.pump.client.dbtype.copyreplace(dialect=dialect) def test_newTransaction(self): """ L{ConnectionPoolClient.newTransaction} returns a provider of L{IAsyncTransaction}, and creates a new transaction on the server side. """ txn = self.pump.client.newTransaction() verifyObject(IAsyncTransaction, txn) self.pump.flush() self.assertEquals(len(self.factory.connections), 1) class HookableOperationTests(TestCase): """ Tests for L{_HookableOperation}. """ @inlineCallbacks def test_clearPreventsSubsequentAddHook(self): """ After clear() or runHooks() are called, subsequent calls to addHook() are NO-OPs. """ def hook(): return succeed(None) hookOp = _HookableOperation() hookOp.addHook(hook) self.assertEquals(len(hookOp._hooks), 1) hookOp.clear() self.assertEquals(hookOp._hooks, None) hookOp = _HookableOperation() hookOp.addHook(hook) yield hookOp.runHooks() self.assertEquals(hookOp._hooks, None) hookOp.addHook(hook) self.assertEquals(hookOp._hooks, None)