def test_full_tor_connection(self):
config = txtorcon.TorConfig()
config.ControlPort = net.randomFreePort()
config.SocksPort = net.randomFreePort()
config.DataDirectory = self.tor_datadir
log.msg(
"Connecting to tor %s" %
(onion.tor_details['version']))
config.log = ['notice stdout', 'notice file %s' % self.tor_logfile]
config.save()
def updates(prog, tag, summary):
log.msg("Progress is at: %s%%" % (prog))
self.report['tor_progress'] = int(prog)
self.report['tor_progress_tag'] = tag
self.report['tor_progress_summary'] = summary
d = txtorcon.launch_tor(config, reactor, timeout=self.timeout,
progress_updates=updates)
@d.addCallback
def setup_complete(proto):
try:
proto.transport.signalProcess('TERM')
except error.ProcessExitedAlready:
proto.transport.loseConnection()
log.msg("Successfully connected to Tor")
self.report['success'] = True
@d.addErrback
def setup_failed(failure):
log.msg("Failed to connect to Tor")
self.report['success'] = False
self.report['error'] = 'timeout-reached'
return
@d.addCallback
def write_log(_):
with open(self.tor_logfile) as f:
self.report['tor_log'] = f.read()
os.remove(self.tor_logfile)
try:
shutil.rmtree(self.tor_datadir)
except:
pass
return d
评论列表
文章目录