<s> from twisted . internet import defer , reactor <EOL> from twisted . python import usage <EOL> from twisted . web . client import Agent , readBody <EOL> from ooni . templates . process import ProcessTest <EOL> from ooni . utils import log <EOL> from ooni . errors import handleAllFailures , failureToString <EOL> import distutils . spawn <EOL> import re <EOL> class UsageOptions ( usage . Options ) : <EOL> optParameters = [ <EOL> [ '<STR_LIT:url>' , '<STR_LIT:u>' , None , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:c>' , None , '<STR_LIT>' ] , <EOL> ] <EOL> class OpenVPNTest ( ProcessTest ) : <EOL> """<STR_LIT>""" <EOL> name = "<STR_LIT>" <EOL> description = "<STR_LIT>" <EOL> author = "<STR_LIT>" <EOL> version = "<STR_LIT>" <EOL> timeout = <NUM_LIT:20> <EOL> usageOptions = UsageOptions <EOL> requiredOptions = [ '<STR_LIT:url>' , '<STR_LIT>' ] <EOL> requiresRoot = True <EOL> def setUp ( self ) : <EOL> self . bootstrapped = defer . Deferred ( ) <EOL> self . command = [ distutils . spawn . find_executable ( "<STR_LIT>" ) ] <EOL> self . exited = False <EOL> self . url = self . localOptions . get ( '<STR_LIT:url>' ) <EOL> if self . localOptions . get ( '<STR_LIT>' ) : <EOL> openvpn_config = self . localOptions . get ( '<STR_LIT>' ) <EOL> self . command . extend ( [ '<STR_LIT>' , openvpn_config ] ) <EOL> def stop ( self , reason = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . exited : <EOL> self . processDirector . close ( ) <EOL> self . processDirector . transport . signalProcess ( '<STR_LIT>' ) <EOL> self . exited = True <EOL> def inConnectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> if not self . bootstrapped . called : <EOL> self . bootstrapped . errback ( Exception ( "<STR_LIT>" ) ) <EOL> def processExited ( self , reason ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" % handleAllFailures ( reason ) ) <EOL> if not self . bootstrapped . called : <EOL> self . bootstrapped . errback ( Exception ( "<STR_LIT>" ) ) <EOL> def handleRead ( self , stdout = None , stderr = None ) : <EOL> """<STR_LIT>""" <EOL> if not self . bootstrapped . called : <EOL> if re . search ( r'<STR_LIT>' , self . processDirector . stdout ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . bootstrapped . errback ( Exception ( "<STR_LIT>" ) ) <EOL> elif "<STR_LIT>" in self . processDirector . stdout : <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . processDirector . cancelTimer ( ) <EOL> self . bootstrapped . callback ( "<STR_LIT>" ) <EOL> def test_openvpn_circumvent ( self ) : <EOL> def addResultToReport ( result ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . report [ '<STR_LIT:body>' ] = result <EOL> self . report [ '<STR_LIT:success>' ] = True <EOL> def addFailureToReport ( failure ) : <EOL> log . debug ( "<STR_LIT>" % failureToString ( failure ) ) <EOL> self . report [ '<STR_LIT>' ] = failureToString ( failure ) <EOL> self . report [ '<STR_LIT:success>' ] = False <EOL> def doRequest ( noreason ) : <EOL> """<STR_LIT>""" <EOL> agent = Agent ( reactor ) <EOL> log . debug ( "<STR_LIT>" % self . url ) <EOL> request = agent . request ( "<STR_LIT:GET>" , self . url ) <EOL> request . addCallback ( readBody ) <EOL> request . addCallback ( addResultToReport ) <EOL> request . addErrback ( addFailureToReport ) <EOL> return request <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . d = self . run ( self . command ) <EOL> self . processDirector . inConnectionLost = self . inConnectionLost <EOL> self . processDirector . processExited = self . processExited <EOL> self . bootstrapped . addCallback ( doRequest ) <EOL> self . bootstrapped . addErrback ( addFailureToReport ) <EOL> self . bootstrapped . addBoth ( self . stop ) <EOL> return self . d </s>
