我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用twisted.internet.protocol.Factory()。
def testWriter(self): f = protocol.Factory() f.protocol = WriterProtocol f.done = 0 f.problem = 0 wrappedF = WiredFactory(f) p = reactor.listenTCP(0, wrappedF, interface="127.0.0.1") n = p.getHost().port self.ports.append(p) clientF = WriterClientFactory() wrappedClientF = WiredFactory(clientF) reactor.connectTCP("127.0.0.1", n, wrappedClientF) def check(ignored): self.failUnless(f.done, "writer didn't finish, it probably died") self.failUnless(f.problem == 0, "writer indicated an error") self.failUnless(clientF.done, "client didn't see connection dropped") expected = "".join(["Hello Cleveland!\n", "Goodbye", " cruel", " world", "\n"]) self.failUnless(clientF.data == expected, "client didn't receive all the data it expected") d = defer.gatherResults([wrappedF.onDisconnect, wrappedClientF.onDisconnect]) return d.addCallback(check)
def testWriter(self): f = protocol.Factory() f.protocol = LargeBufferWriterProtocol f.done = 0 f.problem = 0 f.len = self.datalen wrappedF = FireOnCloseFactory(f) p = reactor.listenTCP(0, wrappedF, interface="127.0.0.1") n = p.getHost().port self.ports.append(p) clientF = LargeBufferReaderClientFactory() wrappedClientF = FireOnCloseFactory(clientF) reactor.connectTCP("127.0.0.1", n, wrappedClientF) d = defer.gatherResults([wrappedF.deferred, wrappedClientF.deferred]) def check(ignored): self.failUnless(f.done, "writer didn't finish, it probably died") self.failUnless(clientF.len == self.datalen, "client didn't receive all the data it expected " "(%d != %d)" % (clientF.len, self.datalen)) self.failUnless(clientF.done, "client didn't see connection dropped") return d.addCallback(check)
def testPeerBind(self): """assert the remote endpoint (getPeer) on the receiving end matches the local endpoint (bind) on the connecting end, for unix sockets""" filename = self.mktemp() peername = self.mktemp() f = Factory(self, filename, peername=peername) l = reactor.listenUNIX(filename, f) self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self._sock.bind(peername) self._sock.connect(filename) d = f.deferred def done(x): self._addPorts(l) self._sock.close() del self._sock return x d.addBoth(done) return d
def testPORTCannotConnect(self): # Login d = self._anonymousLogin() # Listen on a port, and immediately stop listening as a way to find a # port number that is definitely closed. def loggedIn(ignored): port = reactor.listenTCP(0, protocol.Factory(), interface='127.0.0.1') portNum = port.getHost().port d = port.stopListening() d.addCallback(lambda _: portNum) return d d.addCallback(loggedIn) # Tell the server to connect to that port with a PORT command, and # verify that it fails with the right error. def gotPortNum(portNum): return self.assertCommandFailed( 'PORT ' + ftp.encodeHostPort('127.0.0.1', portNum), ["425 Can't open data connection."]) return d.addCallback(gotPortNum) # -- Client Tests -----------------------------------------------------------
def loopbackUNIX(server, client, noisy=True): """Run session between server and client protocol instances over UNIX socket.""" path = tempfile.mktemp() from twisted.internet import reactor f = policies.WrappingFactory(protocol.Factory()) serverWrapper = _FireOnClose(f, server) f.noisy = noisy f.buildProtocol = lambda addr: serverWrapper serverPort = reactor.listenUNIX(path, f) clientF = LoopbackClientFactory(client) clientF.noisy = noisy reactor.connectUNIX(path, clientF) d = clientF.deferred d.addCallback(lambda x: serverWrapper.deferred) d.addCallback(lambda x: serverPort.stopListening()) return d
def test_simpleSuccess(self): """ If C{getaddrinfo} gives one L{GAIEndpoint.connect}. """ gaiendpoint = self.makeEndpoint() protos = [] f = Factory() f.protocol = Protocol gaiendpoint.connect(f).addCallback(protos.append) WHO_CARES = 0 WHAT_EVER = "" self.gaiResult(AF_INET, SOCK_STREAM, WHO_CARES, WHAT_EVER, ("1.2.3.4", 4321)) self.clock.advance(1.0) attempt = self.fakeRealEndpoints[0]._attempt attempt.callback(self.fakeRealEndpoints[0]._factory.buildProtocol(None)) self.assertEqual(len(protos), 1)
def buildProtocol(self, addr): """ Create an instance of the server side of the SSH protocol. @type addr: L{twisted.internet.interfaces.IAddress} provider @param addr: The address at which the server will listen. @rtype: L{twisted.conch.ssh.transport.SSHServerTransport} @return: The built transport. """ t = protocol.Factory.buildProtocol(self, addr) t.supportedPublicKeys = self.privateKeys.keys() if not self.primes: log.msg('disabling non-fixed-group key exchange algorithms ' 'because we cannot find moduli file') t.supportedKeyExchanges = [ kexAlgorithm for kexAlgorithm in t.supportedKeyExchanges if _kex.isFixedGroup(kexAlgorithm) or _kex.isEllipticCurve(kexAlgorithm)] return t
def connectToAgent(self, endpoint): """ Set up a connection to the authentication agent and trigger its initialization. @param endpoint: An endpoint which can be used to connect to the authentication agent. @type endpoint: L{IStreamClientEndpoint} provider @return: A L{Deferred} which fires when the agent connection is ready for use. """ factory = Factory() factory.protocol = SSHAgentClient d = endpoint.connect(factory) def connected(agent): self.agent = agent return agent.getPublicKeys() d.addCallback(connected) return d
def connect(self, protocolFactory): """ Set up an SSH connection, use a channel from that connection to launch a command, and hook the stdin and stdout of that command up as a transport for a protocol created by the given factory. @param protocolFactory: A L{Factory} to use to create the protocol which will be connected to the stdin and stdout of the command on the SSH server. @return: A L{Deferred} which will fire with an error if the connection cannot be set up for any reason or with the protocol instance created by C{protocolFactory} once it has been connected to the command. """ d = self._creator.secureConnection() d.addCallback(self._executeCommand, protocolFactory) return d
def test_processAddress(self): """ The address passed to the factory's buildProtocol in the endpoint is a _ProcessAddress instance. """ class TestAddrFactory(protocol.Factory): protocol = StubApplicationProtocol address = None def buildProtocol(self, addr): self.address = addr p = self.protocol() p.factory = self return p myFactory = TestAddrFactory() d = self.ep.connect(myFactory) self.successResultOf(d) self.assertIsInstance(myFactory.address, _ProcessAddress)
def test_deferBadEncodingToConnect(self): """ Since any client of L{IStreamClientEndpoint} needs to handle Deferred failures from C{connect}, L{HostnameEndpoint}'s constructor will not raise exceptions when given bad host names, instead deferring to returning a failing L{Deferred} from C{connect}. """ endpoint = endpoints.HostnameEndpoint( deterministicResolvingReactor(MemoryReactor(), ['127.0.0.1']), b'\xff-garbage-\xff', 80 ) deferred = endpoint.connect(Factory.forProtocol(Protocol)) err = self.failureResultOf(deferred, ValueError) self.assertIn("\\xff-garbage-\\xff", str(err)) endpoint = endpoints.HostnameEndpoint( deterministicResolvingReactor(MemoryReactor(), ['127.0.0.1']), u'\u2ff0-garbage-\u2ff0', 80 ) deferred = endpoint.connect(Factory()) err = self.failureResultOf(deferred, ValueError) self.assertIn("\\u2ff0-garbage-\\u2ff0", str(err))
def test_connectProtocolCreatesFactory(self): """ C{endpoints.connectProtocol} calls the given endpoint's C{connect()} method with a factory that will build the given protocol. """ reactor = MemoryReactor() endpoint = endpoints.TCP4ClientEndpoint(reactor, "127.0.0.1", 0) theProtocol = object() endpoints.connectProtocol(endpoint, theProtocol) # A TCP connection was made via the given endpoint: self.assertEqual(len(reactor.tcpClients), 1) # TCP4ClientEndpoint uses a _WrapperFactory around the underlying # factory, so we need to unwrap it: factory = reactor.tcpClients[0][2]._wrappedFactory self.assertIsInstance(factory, protocol.Factory) self.assertIs(factory.buildProtocol(None), theProtocol)
def testWriter(self): f = protocol.Factory() f.protocol = LargeBufferWriterProtocol f.done = 0 f.problem = 0 f.len = self.datalen wrappedF = FireOnCloseFactory(f) p = reactor.listenTCP(0, wrappedF, interface="127.0.0.1") self.addCleanup(p.stopListening) n = p.getHost().port clientF = LargeBufferReaderClientFactory() wrappedClientF = FireOnCloseFactory(clientF) reactor.connectTCP("127.0.0.1", n, wrappedClientF) d = defer.gatherResults([wrappedF.deferred, wrappedClientF.deferred]) def check(ignored): self.assertTrue(f.done, "writer didn't finish, it probably died") self.assertTrue(clientF.len == self.datalen, "client didn't receive all the data it expected " "(%d != %d)" % (clientF.len, self.datalen)) self.assertTrue(clientF.done, "client didn't see connection dropped") return d.addCallback(check)
def test_service(self): """ L{strports.service} returns a L{StreamServerEndpointService} constructed with an endpoint produced from L{endpoint.serverFromString}, using the same syntax. """ reactor = object() # the cake is a lie aFactory = Factory() aGoodPort = 1337 svc = strports.service( 'tcp:' + str(aGoodPort), aFactory, reactor=reactor) self.assertIsInstance(svc, internet.StreamServerEndpointService) # See twisted.application.test.test_internet.EndpointServiceTests. # test_synchronousRaiseRaisesSynchronously self.assertTrue(svc._raiseSynchronously) self.assertIsInstance(svc.endpoint, TCP4ServerEndpoint) # Maybe we should implement equality for endpoints. self.assertEqual(svc.endpoint._port, aGoodPort) self.assertIs(svc.factory, aFactory) self.assertIs(svc.endpoint._reactor, reactor)
def test_listenDefaultHost(self): """ L{MemoryReactor.listenTCP}, L{MemoryReactor.listenSSL} and L{MemoryReactor.listenUNIX} will return an L{IListeningPort} whose C{getHost} method returns an L{IAddress}; C{listenTCP} and C{listenSSL} will have a default host of C{'0.0.0.0'}, and a port that reflects the value passed, and C{listenUNIX} will have a name that reflects the path passed. """ memoryReactor = MemoryReactor() for port in [memoryReactor.listenTCP(8242, Factory()), memoryReactor.listenSSL(8242, Factory(), None)]: verifyObject(IListeningPort, port) address = port.getHost() verifyObject(IAddress, address) self.assertEqual(address.host, '0.0.0.0') self.assertEqual(address.port, 8242) port = memoryReactor.listenUNIX(b"/path/to/socket", Factory()) verifyObject(IListeningPort, port) address = port.getHost() verifyObject(IAddress, address) self.assertEqual(address.name, b"/path/to/socket")
def test_portRange(self): """ L{FTP.passivePortRange} should determine the ports which L{FTP.getDTPPort} attempts to bind. If no port from that iterator can be bound, L{error.CannotListenError} should be raised, otherwise the first successful result from L{FTP.listenFactory} should be returned. """ def listenFactory(portNumber, factory): if portNumber in (22032, 22033, 22034): raise error.CannotListenError('localhost', portNumber, 'error') return portNumber self.serverProtocol.listenFactory = listenFactory port = self.serverProtocol.getDTPPort(protocol.Factory()) self.assertEqual(port, 0) self.serverProtocol.passivePortRange = xrange(22032, 65536) port = self.serverProtocol.getDTPPort(protocol.Factory()) self.assertEqual(port, 22035) self.serverProtocol.passivePortRange = xrange(22032, 22035) self.assertRaises(error.CannotListenError, self.serverProtocol.getDTPPort, protocol.Factory())
def test_PORTCannotConnect(self): """ Listen on a port, and immediately stop listening as a way to find a port number that is definitely closed. """ # Login d = self._anonymousLogin() def loggedIn(ignored): port = reactor.listenTCP(0, protocol.Factory(), interface='127.0.0.1') portNum = port.getHost().port d = port.stopListening() d.addCallback(lambda _: portNum) return d d.addCallback(loggedIn) # Tell the server to connect to that port with a PORT command, and # verify that it fails with the right error. def gotPortNum(portNum): return self.assertCommandFailed( 'PORT ' + ftp.encodeHostPort('127.0.0.1', portNum), ["425 Can't open data connection."]) return d.addCallback(gotPortNum)
def test_listen_starts_service(self): """ ``AutoTLSEndpoint.listen`` starts an ``AcmeIssuingService``. Stopping the port stops the service. """ factory = Factory() d = self.endpoint.listen(factory) self.assertThat( d, succeeded( MatchesPredicate( IListeningPort.providedBy, '%r does not provide IListeningPort'))) port = d.result self.assertThat( self.endpoint.service, MatchesStructure(running=Equals(True))) self.assertThat(port.stopListening(), succeeded(Always())) self.assertThat( self.endpoint.service, MatchesStructure(running=Equals(False)))
def connect(sdata, command, username, host, port=22, key_file=None, password=None): """ Connect to an SSH host (as it happens, persistently). """ sdata.set_conn_state('connecting') try: keys = [Key.fromFile(key_file)] if key_file else None except exceptions.IOError as e: print('### key load error:', str(e)) push_failure_message(str(e), sdata) return endpoint = SSHCommandClientEndpoint.newConnection( reactor, command, username, host, port=int(port), keys=keys, password=password, ui=None, knownHosts=PermissiveKnownHosts()) factory = Factory() factory.protocol = LineProtocol factory.sdata = sdata d = endpoint.connect(factory) # Very small race condition between here and the replacement # in connectionMade() above, but I've never managed to hit it. def disconnect(): sdata.log('Disconnecting while still attempting to connect, by request') d.cancel() sdata.transport_drop_cb = disconnect d.addErrback(lambda reason: push_failure_message(reason, sdata)) return d
def testDumber(self): filename = self.mktemp() f = Factory(self, filename) l = reactor.listenUNIX(filename, f) tcf = TestClientFactory(self, filename) c = reactor.connectUNIX(filename, tcf) d = defer.gatherResults([f.deferred, tcf.deferred]) d.addCallback(lambda x : self._addPorts(l, c.transport, tcf.protocol.transport, f.protocol.transport)) return d
def testMode(self): filename = self.mktemp() f = Factory(self, filename) l = reactor.listenUNIX(filename, f, mode = 0600) self.assertEquals(stat.S_IMODE(os.stat(filename)[0]), 0600) tcf = TestClientFactory(self, filename) c = reactor.connectUNIX(filename, tcf) self._addPorts(l, c.transport)
def testSocketLocking(self): filename = self.mktemp() f = Factory(self, filename) l = reactor.listenUNIX(filename, f, wantPID=True) self.assertRaises( error.CannotListenError, reactor.listenUNIX, filename, f, wantPID=True) def stoppedListening(ign): l = reactor.listenUNIX(filename, f, wantPID=True) return l.stopListening() return l.stopListening().addCallback(stoppedListening)
def testUncleanServerSocketLocking(self): def ranStupidChild(ign): # If this next call succeeds, our lock handling is correct. p = reactor.listenUNIX(self.filename, Factory(self, self.filename), wantPID=True) return p.stopListening() return self._uncleanSocketTest(ranStupidChild)
def testRepr(self): filename = self.mktemp() f = Factory(self, filename) p = reactor.listenUNIX(filename, f) self.failIf(str(p).find(filename) == -1) def stoppedListening(ign): self.failIf(str(p).find(filename) != -1) return defer.maybeDeferred(p.stopListening).addCallback(stoppedListening)
def test_socketsLeftOpen(self): f = protocol.Factory() f.protocol = protocol.Protocol reactor.listenTCP(0, f)
def loopbackTCP(server, client, port=0, noisy=True): """Run session between server and client protocol instances over TCP.""" from twisted.internet import reactor f = policies.WrappingFactory(protocol.Factory()) serverWrapper = _FireOnClose(f, server) f.noisy = noisy f.buildProtocol = lambda addr: serverWrapper serverPort = reactor.listenTCP(port, f, interface='127.0.0.1') clientF = LoopbackClientFactory(client) clientF.noisy = noisy reactor.connectTCP('127.0.0.1', serverPort.getHost().port, clientF) d = clientF.deferred d.addCallback(lambda x: serverWrapper.deferred) d.addCallback(lambda x: serverPort.stopListening()) return d
def main(): if len(sys.argv) < 2: printMessage("POP3 with no messages") else: args = sys.argv[1:] for arg in args: processArg(arg) f = Factory() f.protocol = POP3TestServer reactor.listenTCP(PORT, f) reactor.run()
def autostart(reason, **kwargs): if reason == 0: from twisted.internet import reactor try: os.remove("/tmp/hotplug.socket") except OSError: pass factory = Factory() factory.protocol = Hotplug reactor.listenUNIX("/tmp/hotplug.socket", factory)
def setUp(self): factory = protocol.Factory() factory.protocol = iscp.ISCP self.proto = factory.buildProtocol('/dev/ttyUSB0') self.tr = proto_helpers.StringTransport() self.proto.makeConnection(self.tr) self.tr.clear()
def __init__(self, eiscp_port=60128): """ Args: eiscp_port (int): UDP port to listen for discovery requests on. """ protocol.Factory.__init__(self) self.eiscp_port = eiscp_port
def main(): logging.basicConfig(filename='rsa.log', level=logging.DEBUG) # Create the server f = Factory() f.protocol = Notary reactor.listenTCP(1977, f) reactor.run()
def autostart(reason, **kwargs): if reason == 0: print "[Hotplug] starting hotplug handler" from twisted.internet import reactor import os try: os.remove("/tmp/hotplug.socket") except OSError: pass factory = Factory() factory.protocol = Hotplug reactor.listenUNIX("/tmp/hotplug.socket", factory)
def startedConnecting(self, connector): # test logger.info('Factory - Connecting')
