<s> """<STR_LIT>""" <EOL> __metaclass__ = type <EOL> from twisted . internet . test . reactormixins import ReactorBuilder <EOL> from twisted . internet . protocol import ServerFactory , ClientFactory , Protocol <EOL> from twisted . internet . interfaces import ITLSTransport <EOL> from twisted . internet . defer import Deferred , DeferredList <EOL> from twisted . internet . error import ConnectionClosed <EOL> from twisted . trial . unittest import SkipTest <EOL> from twisted . python . runtime import platform <EOL> try : <EOL> from OpenSSL . crypto import FILETYPE_PEM <EOL> except ImportError : <EOL> FILETYPE_PEM = None <EOL> else : <EOL> from twisted . internet . ssl import PrivateCertificate , KeyPair <EOL> from twisted . internet . ssl import ClientContextFactory <EOL> class SSLClientTestsMixin ( ReactorBuilder ) : <EOL> """<STR_LIT>""" <EOL> if FILETYPE_PEM is None : <EOL> skip = "<STR_LIT>" <EOL> if platform . isWindows ( ) : <EOL> msg = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> skippedReactors = { <EOL> "<STR_LIT>" : msg , <EOL> "<STR_LIT>" : msg } <EOL> _certificateText = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> _privateKeyText = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def getServerContext ( self ) : <EOL> """<STR_LIT>""" <EOL> cert = PrivateCertificate . load ( <EOL> self . _certificateText , <EOL> KeyPair . load ( self . _privateKeyText , FILETYPE_PEM ) , <EOL> FILETYPE_PEM ) <EOL> return cert . options ( ) <EOL> def test_disconnectAfterWriteAfterStartTLS ( self ) : <EOL> """<STR_LIT>""" <EOL> class ShortProtocol ( Protocol ) : <EOL> def connectionMade ( self ) : <EOL> if not ITLSTransport . providedBy ( self . transport ) : <EOL> finished = self . factory . finished <EOL> self . factory . finished = None <EOL> finished . errback ( SkipTest ( "<STR_LIT>" ) ) <EOL> return <EOL> self . transport . startTLS ( self . factory . context ) <EOL> self . transport . write ( "<STR_LIT:x>" ) <EOL> def dataReceived ( self , data ) : <EOL> self . transport . write ( "<STR_LIT:y>" ) <EOL> self . transport . loseConnection ( ) <EOL> def connectionLost ( self , reason ) : <EOL> finished = self . factory . finished <EOL> if finished is not None : <EOL> self . factory . finished = None <EOL> finished . callback ( reason ) <EOL> serverFactory = ServerFactory ( ) <EOL> serverFactory . finished = Deferred ( ) <EOL> serverFactory . protocol = ShortProtocol <EOL> serverFactory . context = self . getServerContext ( ) <EOL> clientFactory = ClientFactory ( ) <EOL> clientFactory . finished = Deferred ( ) <EOL> clientFactory . protocol = ShortProtocol <EOL> clientFactory . context = ClientContextFactory ( ) <EOL> clientFactory . context . method = serverFactory . context . method <EOL> lostConnectionResults = [ ] <EOL> finished = DeferredList ( <EOL> [ serverFactory . finished , clientFactory . finished ] , <EOL> consumeErrors = True ) <EOL> def cbFinished ( results ) : <EOL> lostConnectionResults . extend ( [ results [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , results [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] ] ) <EOL> finished . addCallback ( cbFinished ) <EOL> reactor = self . buildReactor ( ) <EOL> port = reactor . listenTCP ( <NUM_LIT:0> , serverFactory , interface = '<STR_LIT:127.0.0.1>' ) <EOL> self . addCleanup ( port . stopListening ) <EOL> connector = reactor . connectTCP ( <EOL> port . getHost ( ) . host , port . getHost ( ) . port , clientFactory ) <EOL> self . addCleanup ( connector . disconnect ) <EOL> finished . addCallback ( lambda ign : reactor . stop ( ) ) <EOL> self . runReactor ( reactor ) <EOL> lostConnectionResults [ <NUM_LIT:0> ] . trap ( ConnectionClosed ) <EOL> lostConnectionResults [ <NUM_LIT:1> ] . trap ( ConnectionClosed ) <EOL> globals ( ) . update ( SSLClientTestsMixin . makeTestCaseClasses ( ) ) </s>
