/usr/lib/python3/dist-packages/wormhole/test/test_tor_manager.py is in magic-wormhole 0.10.3-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | from __future__ import print_function, unicode_literals
import mock, io
from twisted.trial import unittest
from twisted.internet import defer
from twisted.internet.error import ConnectError
from ..tor_manager import get_tor, SocksOnlyTor
from ..errors import NoTorError
from .._interfaces import ITorManager
class X():
pass
class Tor(unittest.TestCase):
def test_no_txtorcon(self):
with mock.patch("wormhole.tor_manager.txtorcon", None):
self.failureResultOf(get_tor(None), NoTorError)
def test_bad_args(self):
f = self.failureResultOf(get_tor(None, launch_tor="not boolean"),
TypeError)
self.assertEqual(str(f.value), "launch_tor= must be boolean")
f = self.failureResultOf(get_tor(None, tor_control_port=1234),
TypeError)
self.assertEqual(str(f.value), "tor_control_port= must be str or None")
f = self.failureResultOf(get_tor(None, launch_tor=True,
tor_control_port="tcp:127.0.0.1:1234"),
ValueError)
self.assertEqual(str(f.value),
"cannot combine --launch-tor and --tor-control-port=")
def test_launch(self):
reactor = object()
my_tor = X() # object() didn't like providedBy()
launch_d = defer.Deferred()
stderr = io.StringIO()
with mock.patch("wormhole.tor_manager.txtorcon.launch",
side_effect=launch_d) as launch:
d = get_tor(reactor, launch_tor=True, stderr=stderr)
self.assertNoResult(d)
self.assertEqual(launch.mock_calls, [mock.call(reactor)])
launch_d.callback(my_tor)
tor = self.successResultOf(d)
self.assertIs(tor, my_tor)
self.assert_(ITorManager.providedBy(tor))
self.assertEqual(stderr.getvalue(),
" launching a new Tor process, this may take a while..\n")
def test_connect(self):
reactor = object()
my_tor = X() # object() didn't like providedBy()
tcp = "port"
connect_d = defer.Deferred()
stderr = io.StringIO()
with mock.patch("wormhole.tor_manager.txtorcon.connect",
side_effect=connect_d) as connect:
d = get_tor(reactor, tor_control_port=tcp, stderr=stderr)
self.assertNoResult(d)
self.assertEqual(connect.mock_calls, [mock.call(reactor, tcp)])
connect_d.callback(my_tor)
tor = self.successResultOf(d)
self.assertIs(tor, my_tor)
self.assert_(ITorManager.providedBy(tor))
self.assertEqual(stderr.getvalue(), " using Tor via control port\n")
def test_connect_fails(self):
reactor = object()
tcp = "port"
connect_d = defer.Deferred()
stderr = io.StringIO()
with mock.patch("wormhole.tor_manager.txtorcon.connect",
side_effect=connect_d) as connect:
d = get_tor(reactor, tor_control_port=tcp, stderr=stderr)
self.assertNoResult(d)
self.assertEqual(connect.mock_calls, [mock.call(reactor, tcp)])
connect_d.errback(ConnectError())
tor = self.successResultOf(d)
self.assertIsInstance(tor, SocksOnlyTor)
self.assert_(ITorManager.providedBy(tor))
self.assertEqual(tor._reactor, reactor)
self.assertEqual(stderr.getvalue(),
" unable to find Tor control port, using SOCKS\n")
class SocksOnly(unittest.TestCase):
def test_tor(self):
reactor = object()
sot = SocksOnlyTor(reactor)
fake_ep = object()
with mock.patch("wormhole.tor_manager.txtorcon.TorClientEndpoint",
return_value=fake_ep) as tce:
ep = sot.stream_via("host", "port")
self.assertIs(ep, fake_ep)
self.assertEqual(tce.mock_calls, [mock.call("host", "port",
socks_endpoint=None,
tls=False,
reactor=reactor)])
|