---

test_newtls.py

# Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.internet._newtls}. """ from twisted.internet import interfaces from twisted.internet.test.connectionmixins import ( ConnectableProtocol, runProtocolsWithReactor, ) from twisted.internet.test.reactormixins import ReactorBuilder from twisted.internet.test.test_tcp import TCPCreator from twisted.internet.test.test_tls import ( ContextGeneratingMixin, SSLCreator, StartTLSClientCreator, TLSMixin, ) from twisted.trial import unittest try: from twisted.internet import _newtls as __newtls from twisted.protocols import tls except ImportError: _newtls = None else: _newtls = __newtls from zope.interface import implementer class BypassTLSTests(unittest.TestCase): """ Tests for the L{_newtls._BypassTLS} class. """ if not _newtls: skip = "Couldn't import _newtls, perhaps pyOpenSSL is old or missing" def test_loseConnectionPassThrough(self): """ C{_BypassTLS.loseConnection} calls C{loseConnection} on the base class, while preserving any default argument in the base class' C{loseConnection} implementation. """ default = object() result = [] class FakeTransport: def loseConnection(self, _connDone=default): result.append(_connDone) bypass = _newtls._BypassTLS(FakeTransport, FakeTransport()) # The default from FakeTransport is used: bypass.loseConnection() self.assertEqual(result, [default]) # And we can pass our own: notDefault = object() bypass.loseConnection(notDefault) self.assertEqual(result, [default, notDefault]) class FakeProducer: """ A producer that does nothing. """ def pauseProducing(self): pass def resumeProducing(self): pass def stopProducing(self): pass @implementer(interfaces.IHandshakeListener) class ProducerProtocol(ConnectableProtocol): """ Register a producer, unregister it, and verify the producer hooks up to innards of C{TLSMemoryBIOProtocol}. """ def __init__(self, producer, result): self.producer = producer self.result = result def handshakeCompleted(self): if not isinstance(self.transport.protocol, tls.BufferingTLSTransport): # Either the test or the code have a bug... raise RuntimeError("TLSMemoryBIOProtocol not hooked up.") self.transport.registerProducer(self.producer, True) # The producer was registered with the TLSMemoryBIOProtocol: self.result.append(self.transport.protocol._producer._producer) self.transport.unregisterProducer() # The producer was unregistered from the TLSMemoryBIOProtocol: self.result.append(self.transport.protocol._producer) self.transport.loseConnection() class ProducerTestsMixin(ReactorBuilder, TLSMixin, ContextGeneratingMixin): """ Test the new TLS code integrates C{TLSMemoryBIOProtocol} correctly. """ if not _newtls: skip = "Could not import twisted.internet._newtls" def test_producerSSLFromStart(self): """ C{registerProducer} and C{unregisterProducer} on TLS transports created as SSL from the get go are passed to the C{TLSMemoryBIOProtocol}, not the underlying transport directly. """ result = [] producer = FakeProducer() runProtocolsWithReactor( self, ConnectableProtocol(), ProducerProtocol(producer, result), SSLCreator(), ) self.assertEqual(result, [producer, None]) def test_producerAfterStartTLS(self): """ C{registerProducer} and C{unregisterProducer} on TLS transports created by C{startTLS} are passed to the C{TLSMemoryBIOProtocol}, not the underlying transport directly. """ result = [] producer = FakeProducer() runProtocolsWithReactor( self, ConnectableProtocol(), ProducerProtocol(producer, result), StartTLSClientCreator(), ) self.assertEqual(result, [producer, None]) def startTLSAfterRegisterProducer(self, streaming): """ When a producer is registered, and then startTLS is called, the producer is re-registered with the C{TLSMemoryBIOProtocol}. """ clientContext = self.getClientContext() serverContext = self.getServerContext() result = [] producer = FakeProducer() class RegisterTLSProtocol(ConnectableProtocol): def connectionMade(self): self.transport.registerProducer(producer, streaming) self.transport.startTLS(serverContext) # Store TLSMemoryBIOProtocol and underlying transport producer # status: if streaming: # _ProducerMembrane -> producer: result.append(self.transport.protocol._producer._producer) result.append(self.transport.producer._producer) else: # _ProducerMembrane -> _PullToPush -> producer: result.append(self.transport.protocol._producer._producer._producer) result.append(self.transport.producer._producer._producer) self.transport.unregisterProducer() self.transport.loseConnection() class StartTLSProtocol(ConnectableProtocol): def connectionMade(self): self.transport.startTLS(clientContext) runProtocolsWithReactor( self, RegisterTLSProtocol(), StartTLSProtocol(), TCPCreator() ) self.assertEqual(result, [producer, producer]) def test_startTLSAfterRegisterProducerStreaming(self): """ When a streaming producer is registered, and then startTLS is called, the producer is re-registered with the C{TLSMemoryBIOProtocol}. """ self.startTLSAfterRegisterProducer(True) def test_startTLSAfterRegisterProducerNonStreaming(self): """ When a non-streaming producer is registered, and then startTLS is called, the producer is re-registered with the C{TLSMemoryBIOProtocol}. """ self.startTLSAfterRegisterProducer(False) globals().update(ProducerTestsMixin.makeTestCaseClasses())

---

docroot: /var/www/469ed2f6-4d4b-4a8a-9f5e-ee4ef04d3bc8/public_html

---

curdir: /usr/lib/python3/dist-packages/twisted/internet/test

	.		12288
	..		12288
X	__init__.py		112
	__pycache__		4096
X	_posixifaces.py		4389
X	_win32ifaces.py		3975
X	connectionmixins.py		20184
	fake_CAs		4096
X	fakeendpoint.py		1663
X	modulehelpers.py		1672
X	process_cli.py		547
X	process_connectionlost.py		126
X	process_gireactornocompat.py		792
X	process_helper.py		1250
X	reactormixins.py		16431
X	test_abstract.py		2203
X	test_address.py		8266
X	test_asyncioreactor.py		9929
X	test_base.py		14787
X	test_baseprocess.py		2588
X	test_cfreactor.py		3168
X	test_core.py		11125
X	test_default.py		3582
X	test_defer_await.py		6821
X	test_defer_yieldfrom.py		4904
X	test_endpoints.py		150780
X	test_epollreactor.py		7322
X	test_error.py		1106
X	test_fdset.py		13497
X	test_filedescriptor.py		2761
X	test_gireactor.py		7349
X	test_glibbase.py		3090
X	test_inlinecb.py		11482
X	test_inotify.py		18562
X	test_iocp.py		7700
X	test_kqueuereactor.py		1972
X	test_main.py		1355
X	test_newtls.py		6533
X	test_pollingfile.py		1313
X	test_posixbase.py		11837
X	test_posixprocess.py		11203
X	test_process.py		45672
X	test_protocol.py		18518
X	test_reactormixins.py		1972
X	test_resolver.py		19610
X	test_serialport.py		1989
X	test_sigchld.py		3329
X	test_socket.py		9437
X	test_stdio.py		6327
X	test_tcp.py		108237
X	test_testing.py		16883
X	test_threads.py		8130
X	test_time.py		3654
X	test_tls.py		12841
X	test_udp.py		16880
X	test_udp_internals.py		5103
X	test_unix.py		34860
X	test_win32events.py		6449
X	test_win32serialport.py		5306