我们从Python开源项目中，提取了以下33个代码示例，用于说明如何使用twisted.python.failure.trap()。
def ftp_PORT(self, address): addr = map(int, address.split(',')) ip = '%d.%d.%d.%d' % tuple(addr[:4]) port = addr[4] << 8 | addr[5] # if we have a DTP port set up, lose it. if self.dtpFactory is not None: self.cleanupDTP() self.dtpFactory = DTPFactory(pi=self, peerHost=self.transport.getPeer().host) self.dtpFactory.setTimeout(self.dtpTimeout) self.dtpPort = reactor.connectTCP(ip, port, self.dtpFactory) def connected(ignored): return ENTERING_PORT_MODE def connFailed(err): err.trap(PortConnectionError) return CANT_OPEN_DATA_CNX return self.dtpFactory.deferred.addCallbacks(connected, connFailed)
def _ebRoundRobinBackoff(self, failure, fakeProto): failure.trap(defer.TimeoutError) # Assert that each server is tried with a particular timeout # before the timeout is increased and the attempts are repeated. for t in (1, 3, 11, 45): tries = fakeProto.queries[:len(self.testServers)] del fakeProto.queries[:len(self.testServers)] tries.sort() expected = list(self.testServers) expected.sort() for ((addr, query, timeout, id), expectedAddr) in zip(tries, expected): self.assertEquals(addr, (expectedAddr, 53)) self.assertEquals(timeout, t) self.failIf(fakeProto.queries)
def testServerTimeout(self): self.server.timeoutTest = True self.client.timeout = 5 #seconds self.selectedArgs = None def login(): d = self.client.login('testuser', 'password-test') d.addErrback(timedOut) return d def timedOut(failure): self._cbStopClient(None) failure.trap(error.TimeoutError) d = self.connected.addCallback(strip(login)) d.addErrback(self._ebGeneral) return defer.gatherResults([d, self.loopback()])
def test_waitUntilLockedWithTimeoutUnlocked(self): """ Test that a lock can be acquired while a lock is held but the lock is unlocked before our timeout. """ def onTimeout(f): f.trap(defer.TimeoutError) self.fail("Should not have timed out") self.assertTrue(self.lock.lock()) self.clock.callLater(1, self.lock.unlock) d = self.lock.deferUntilLocked(timeout=10) d.addErrback(onTimeout) self.clock.pump([1] * 10) return d
def test_errbackAddedBeforeTimeoutSuppressesCancellation(self): """ An errback added before a timeout is added errbacks with a L{defer.CancelledError} when the timeout fires. If the errback suppresses the L{defer.CancelledError}, the deferred successfully completes. """ clock = Clock() d = defer.Deferred() dErrbacked = [None] def errback(f): dErrbacked[0] = f f.trap(defer.CancelledError) d.addErrback(errback) d.addTimeout(10, clock) clock.advance(15) self.assertIsInstance(dErrbacked[0], failure.Failure) self.assertIsInstance(dErrbacked[0].value, defer.CancelledError) self.successResultOf(d)
def _ebRoundRobinBackoff(self, failure, fakeProto): failure.trap(defer.TimeoutError) # Assert that each server is tried with a particular timeout # before the timeout is increased and the attempts are repeated. for t in (1, 3, 11, 45): tries = fakeProto.queries[:len(self.testServers)] del fakeProto.queries[:len(self.testServers)] tries.sort() expected = list(self.testServers) expected.sort() for ((addr, query, timeout, id), expectedAddr) in zip(tries, expected): self.assertEqual(addr, (expectedAddr, 53)) self.assertEqual(timeout, t) self.assertFalse(fakeProto.queries)
def testFailedStartTLS(self): failures = [] def breakServerTLS(ign): self.server.canStartTLS = False self.connected.addCallback(breakServerTLS) self.connected.addCallback(lambda ign: self.client.startTLS()) self.connected.addErrback( lambda err: failures.append(err.trap(imap4.IMAP4Exception))) self.connected.addCallback(self._cbStopClient) self.connected.addErrback(self._ebGeneral) def check(ignored): self.assertTrue(failures) self.assertIdentical(failures[0], imap4.IMAP4Exception) return self.loopback().addCallback(check)
def _ebLogin(failure): failure.trap(cred_error.UnauthorizedLogin, cred_error.UnhandledCredentials) self.state = self.UNAUTH raise AuthorizationError
def _unwrapFirstError(failure): failure.trap(defer.FirstError) return failure.value.subFailure
def _rcodeTest(self, rcode, exc): m = Message(rCode=rcode) err = self.resolver.filterAnswers(m) err.trap(exc)
def defaultErrback(self, failure): failure.trap(ConnectionAborted, ConnectionClosed, ConnectionDone, ConnectionLost, UnknownRemoteError) reactor.stop()
def defaultErrback(self, failure): """TODO better network error handling. """ failure.trap(ConnectionAborted, ConnectionClosed, ConnectionDone, ConnectionLost)
def test_anonymousSTOR(self): """ Try to make an STOR as anonymous, and check that we got a permission denied error. """ def eb(res): res.trap(ftp.CommandFailed) self.assertEqual(res.value.args[0][0], '550 foo: Permission denied.') d1, d2 = self.client.storeFile('foo') d2.addErrback(eb) return defer.gatherResults([d1, d2])
def test_STORtransferErrorIsReturned(self): """ Any FTP error raised by STOR while transferring the file is returned to the client. """ # Make a failing file writer. class FailingFileWriter(ftp._FileWriter): def receive(self): return defer.fail(ftp.IsADirectoryError("failing_file")) def failingSTOR(a, b): return defer.succeed(FailingFileWriter(None)) # Monkey patch the shell so it returns a file writer that will # fail during transfer. self.patch(ftp.FTPAnonymousShell, 'openForWriting', failingSTOR) def eb(res): res.trap(ftp.CommandFailed) logs = self.flushLoggedErrors() self.assertEqual(1, len(logs)) self.assertIsInstance(logs[0].value, ftp.IsADirectoryError) self.assertEqual( res.value.args[0][0], "550 failing_file: is a directory") d1, d2 = self.client.storeFile('failing_file') d2.addErrback(eb) return defer.gatherResults([d1, d2])
def test_STORunknownTransferErrorBecomesAbort(self): """ Any non FTP error raised by STOR while transferring the file is converted into a critical error and transfer is closed. The unknown error is logged. """ class FailingFileWriter(ftp._FileWriter): def receive(self): return defer.fail(AssertionError()) def failingSTOR(a, b): return defer.succeed(FailingFileWriter(None)) # Monkey patch the shell so it returns a file writer that will # fail during transfer. self.patch(ftp.FTPAnonymousShell, 'openForWriting', failingSTOR) def eb(res): res.trap(ftp.CommandFailed) logs = self.flushLoggedErrors() self.assertEqual(1, len(logs)) self.assertIsInstance(logs[0].value, AssertionError) self.assertEqual( res.value.args[0][0], "426 Transfer aborted. Data connection closed.") d1, d2 = self.client.storeFile('failing_file') d2.addErrback(eb) return defer.gatherResults([d1, d2])
def test_RETRreadError(self): """ Any errors during reading a file inside a RETR should be returned to the client. """ # Make a failing file reading. class FailingFileReader(ftp._FileReader): def send(self, consumer): return defer.fail(ftp.IsADirectoryError("blah")) def failingRETR(a, b): return defer.succeed(FailingFileReader(None)) # Monkey patch the shell so it returns a file reader that will # fail. self.patch(ftp.FTPAnonymousShell, 'openForReading', failingRETR) def check_response(failure): self.flushLoggedErrors() failure.trap(ftp.CommandFailed) self.assertEqual( failure.value.args[0][0], "125 Data connection already open, starting transfer") self.assertEqual( failure.value.args[0][1], "550 blah: is a directory") proto = _BufferingProtocol() d = self.client.retrieveFile('failing_file', proto) d.addErrback(check_response) return d
def test_downloadFolder(self): """ When RETR is called for a folder, it will fail complaining that the path is a folder. """ # Make a directory in the current working directory self.dirPath.child('foo').createDirectory() # Login d = self._anonymousLogin() d.addCallback(self._makeDataConnection) def retrFolder(downloader): downloader.transport.loseConnection() deferred = self.client.queueStringCommand('RETR foo') return deferred d.addCallback(retrFolder) def failOnSuccess(result): raise AssertionError('Downloading a folder should not succeed.') d.addCallback(failOnSuccess) def checkError(failure): failure.trap(ftp.CommandFailed) self.assertEqual( ['550 foo: is a directory'], failure.value.args[0]) current_errors = self.flushLoggedErrors() self.assertEqual( 0, len(current_errors), 'No errors should be logged while downloading a folder.') d.addErrback(checkError) return d
def _check(self): """ Check the output of the log observer to see if the error is present. """ c2 = self._loggedErrors() self.assertEqual(len(c2), 2) c2[1]["failure"].trap(ZeroDivisionError) self.flushLoggedErrors(ZeroDivisionError)
def test_chainedErrorCleanup(self): """ If one Deferred with an error result is returned from a callback on another Deferred, when the first Deferred is garbage collected it does not log its error. """ d = defer.Deferred() d.addCallback(lambda ign: defer.fail(RuntimeError("zoop"))) d.callback(None) # Sanity check - this isn't too interesting, but we do want the original # Deferred to have gotten the failure. results = [] errors = [] d.addCallbacks(results.append, errors.append) self.assertEqual(results, []) self.assertEqual(len(errors), 1) errors[0].trap(Exception) # Get rid of any references we might have to the inner Deferred (none of # these should really refer to it, but we're just being safe). del results, errors, d # Force a collection cycle so that there's a chance for an error to be # logged, if it's going to be logged. gc.collect() # And make sure it is not. self.assertEqual(self._loggedErrors(), [])
def ftp_PASS(self, password): """ Second part of login. Get the password the peer wants to authenticate with. """ if self.factory.allowAnonymous and self._user == self.factory.userAnonymous: # anonymous login creds = credentials.Anonymous() reply = GUEST_LOGGED_IN_PROCEED else: # user login creds = credentials.UsernamePassword(self._user, password) reply = USR_LOGGED_IN_PROCEED del self._user def _cbLogin(result): (interface, avatar, logout) = result assert interface is IFTPShell, "The realm is busted, jerk." self.shell = avatar self.logout = logout self.workingDirectory = [] self.state = self.AUTHED return reply def _ebLogin(failure): failure.trap(cred_error.UnauthorizedLogin, cred_error.UnhandledCredentials) self.state = self.UNAUTH raise AuthorizationError d = self.portal.login(creds, None, IFTPShell) d.addCallbacks(_cbLogin, _ebLogin) return d
def _rcodeTest(self, rcode, exc): m = dns.Message(rCode=rcode) err = self.resolver.filterAnswers(m) err.trap(exc)
def _trapCancellation(failure): failure.trap(defer.CancelledError)
def _sessionClosed(self, maybeFailure, sessionID): del self.sessions[sessionID] if isinstance(maybeFailure, failure.Failure): failure.trap(error.ConnectionDone, error.ConnectionLost, SessionTimeout) else: return maybeFailure
def test_reentrantTCPQueryErrbackOnConnectionFailure(self): """ An errback on the deferred returned by L{client.Resolver.queryTCP} may trigger another TCP query. """ reactor = proto_helpers.MemoryReactor() resolver = client.Resolver( servers=[('127.0.0.1', 10053)], reactor=reactor) q = dns.Query('example.com') # First query sent d = resolver.queryTCP(q) # Repeat the query when the first query fails def reissue(e): e.trap(ConnectionRefusedError) return resolver.queryTCP(q) d.addErrback(reissue) self.assertEqual(len(reactor.tcpClients), 1) self.assertEqual(len(reactor.connectors), 1) host, port, factory, timeout, bindAddress = reactor.tcpClients[0] # First query fails f1 = failure.Failure(ConnectionRefusedError()) factory.clientConnectionFailed( reactor.connectors[0], f1) # A second TCP connection is immediately attempted self.assertEqual(len(reactor.tcpClients), 2) self.assertEqual(len(reactor.connectors), 2) # No result expected until the second chained query returns self.assertNoResult(d) # Second query fails f2 = failure.Failure(ConnectionRefusedError()) factory.clientConnectionFailed( reactor.connectors[1], f2) # Original deferred now fires with the second failure f = self.failureResultOf(d, ConnectionRefusedError) self.assertIs(f, f2)
