improve tor/i2p test coverage

src/foolscap/test/test_connection.py

@@ -1,9 +1,11 @@
+import os
 import mock
 from zope.interface import implementer
 from twisted.trial import unittest
 from twisted.internet import endpoints, defer, reactor
 from twisted.internet.defer import inlineCallbacks
 from twisted.application import service
+import txtorcon
 from txsocksx.client import SOCKS5ClientEndpoint
 from foolscap.api import Tub
 from foolscap.connection import get_endpoint
@@ -215,6 +217,9 @@ def test_bad_hint(self):
         self.assertRaises(ipb.InvalidHintError,
                           h.hint_to_endpoint, "tcp:@:1234", reactor)
 
+class Empty:
+    pass
+
 class Tor(unittest.TestCase):
     @inlineCallbacks
     def test_default_socks(self):
@@ -236,6 +241,7 @@ def test_badaddr(self):
         self.assertTrue(isnon("10.0.0.1"))
         self.assertTrue(isnon("127.0.0.1"))
         self.assertTrue(isnon("192.168.78.254"))
+        self.assertTrue(isnon("::1"))
         self.assertFalse(isnon("8.8.8.8"))
         self.assertFalse(isnon("example.org"))
 
@@ -250,7 +256,162 @@ def test_default_socks_badaddr(self):
         f = yield self.assertFailure(d, InvalidHintError)
         self.assertEqual(str(f), "ignoring non-Tor-able ipaddr 127.0.0.1")
 
-    # TODO: exercise launch_tor and with_control_port somehow
+        d = h.hint_to_endpoint("tcp:not@a@hint:123", reactor)
+        f = yield self.assertFailure(d, InvalidHintError)
+        self.assertEqual(str(f), "unrecognized TCP/Tor hint")
+
+    @inlineCallbacks
+    def test_socks_port(self):
+        h = tor.socks_port(100)
+        res = yield h.hint_to_endpoint("tcp:example.com:1234", reactor)
+        ep, host = res
+        self.assertIsInstance(ep, txtorcon.endpoints.TorClientEndpoint)
+        self.assertEqual(host, "example.com")
+        self.assertEqual(ep.socks_hostname, "127.0.0.1")
+        self.assertEqual(ep.socks_port, 100)
+
+    @inlineCallbacks
+    def test_launch(self):
+        tpp = Empty()
+        tpp.tor_protocol = None
+        h = tor.launch()
+        fake_reactor = object()
+        with mock.patch("txtorcon.launch_tor", return_value=tpp) as lt:
+            # we ignore the return value of hint_to_endpoint()
+            yield h.hint_to_endpoint("tor:foo.onion:29212", fake_reactor)
+            self.assertEqual(len(lt.mock_calls), 1)
+            args,kwargs = lt.mock_calls[0][1:]
+            self.assertIdentical(args[0], h.config)
+            self.assertIdentical(args[1], fake_reactor)
+            self.assertEqual(kwargs, {"tor_binary": None})
+        self.assertEqual(h._socks_hostname, "127.0.0.1")
+        # the socks port will be allocated by launch_tor, so it should match
+        self.assertEqual(h._socks_portnum, h.config.SocksPort)
+
+    @inlineCallbacks
+    def test_launch_tor_binary(self):
+        tpp = Empty()
+        tpp.tor_protocol = None
+        h = tor.launch(tor_binary="/bin/tor")
+        fake_reactor = object()
+        with mock.patch("txtorcon.launch_tor", return_value=tpp) as lt:
+            # we ignore the return value of hint_to_endpoint()
+            yield h.hint_to_endpoint("tor:foo.onion:29212", fake_reactor)
+            self.assertEqual(len(lt.mock_calls), 1)
+            args,kwargs = lt.mock_calls[0][1:]
+            self.assertIdentical(args[0], h.config)
+            self.assertIdentical(args[1], fake_reactor)
+            self.assertEqual(kwargs, {"tor_binary": "/bin/tor"})
+        self.assertEqual(h._socks_hostname, "127.0.0.1")
+        # the socks port will be allocated by launch_tor, so it should match
+        self.assertEqual(h._socks_portnum, h.config.SocksPort)
+
+    @inlineCallbacks
+    def test_launch_data_directory(self):
+        datadir = self.mktemp()
+        tpp = Empty()
+        tpp.tor_protocol = None
+        h = tor.launch(data_directory=datadir)
+        fake_reactor = object()
+        with mock.patch("txtorcon.launch_tor", return_value=tpp) as lt:
+            # we ignore the return value of hint_to_endpoint()
+            yield h.hint_to_endpoint("tor:foo.onion:29212", fake_reactor)
+            self.assertEqual(len(lt.mock_calls), 1)
+            args,kwargs = lt.mock_calls[0][1:]
+            self.assertIdentical(args[0], h.config)
+            self.assertIdentical(args[1], fake_reactor)
+            self.assertEqual(kwargs, {"tor_binary": None})
+            self.assertEqual(h.config.DataDirectory, datadir)
+        self.assertEqual(h._socks_hostname, "127.0.0.1")
+        # the socks port will be allocated by launch_tor, so it should match
+        self.assertEqual(h._socks_portnum, h.config.SocksPort)
+
+    @inlineCallbacks
+    def test_launch_data_directory_exists(self):
+        datadir = self.mktemp()
+        os.mkdir(datadir)
+        tpp = Empty()
+        tpp.tor_protocol = None
+        h = tor.launch(data_directory=datadir)
+        fake_reactor = object()
+        with mock.patch("txtorcon.launch_tor", return_value=tpp) as lt:
+            # we ignore the return value of hint_to_endpoint()
+            yield h.hint_to_endpoint("tor:foo.onion:29212", fake_reactor)
+            self.assertEqual(len(lt.mock_calls), 1)
+            args,kwargs = lt.mock_calls[0][1:]
+            self.assertIdentical(args[0], h.config)
+            self.assertIdentical(args[1], fake_reactor)
+            self.assertEqual(kwargs, {"tor_binary": None})
+            self.assertEqual(h.config.DataDirectory, datadir)
+        self.assertEqual(h._socks_hostname, "127.0.0.1")
+        # the socks port will be allocated by launch_tor, so it should match
+        self.assertEqual(h._socks_portnum, h.config.SocksPort)
+
+    @inlineCallbacks
+    def test_control_endpoint(self):
+        control_ep = endpoints.HostnameEndpoint(reactor, "localhost", 9051)
+        h = tor.control_endpoint(control_ep)
+        # We don't actually care about the generated endpoint, just the state
+        # that the handler builds up internally. But we need to provoke a
+        # connection to build that state, and we need to prevent the handler
+        # from actually talking to a Tor daemon (which probably doesn't exist
+        # on this host).
+        config = Empty()
+        config.SocksPort = ["9050"]
+        with mock.patch("txtorcon.build_tor_connection",
+                        return_value=None):
+            with mock.patch("txtorcon.TorConfig.from_protocol",
+                            return_value=config):
+                # we ignore the return value of hint_to_endpoint()
+                yield h.hint_to_endpoint("tor:foo.onion:29212", reactor)
+        self.assertEqual(h._socks_hostname, "127.0.0.1")
+        self.assertEqual(h._socks_portnum, 9050)
+
+    @inlineCallbacks
+    def test_control_endpoint_default(self):
+        control_ep = endpoints.HostnameEndpoint(reactor, "localhost", 9051)
+        h = tor.control_endpoint(control_ep)
+        config = Empty()
+        config.SocksPort = [txtorcon.DEFAULT_VALUE]
+        with mock.patch("txtorcon.build_tor_connection",
+                        return_value=None):
+            with mock.patch("txtorcon.TorConfig.from_protocol",
+                            return_value=config):
+                # we ignore the return value of hint_to_endpoint()
+                yield h.hint_to_endpoint("tor:foo.onion:29212", reactor)
+        self.assertEqual(h._socks_hostname, "127.0.0.1")
+        self.assertEqual(h._socks_portnum, 9050)
+
+    @inlineCallbacks
+    def test_control_endpoint_non_numeric(self):
+        control_ep = endpoints.HostnameEndpoint(reactor, "localhost", 9051)
+        h = tor.control_endpoint(control_ep)
+        config = Empty()
+        config.SocksPort = ["unix:var/run/tor/socks WorldWritable", "9050"]
+        with mock.patch("txtorcon.build_tor_connection",
+                        return_value=None):
+            with mock.patch("txtorcon.TorConfig.from_protocol",
+                            return_value=config):
+                # we ignore the return value of hint_to_endpoint()
+                yield h.hint_to_endpoint("tor:foo.onion:29212", reactor)
+        self.assertEqual(h._socks_hostname, "127.0.0.1")
+        self.assertEqual(h._socks_portnum, 9050)
+
+    @inlineCallbacks
+    def test_control_endpoint_no_port(self):
+        control_ep = endpoints.HostnameEndpoint(reactor, "localhost", 9051)
+        h = tor.control_endpoint(control_ep)
+        config = Empty()
+        config.SocksPort = ["unparseable"]
+        with mock.patch("txtorcon.build_tor_connection",
+                        return_value=None):
+            with mock.patch("txtorcon.TorConfig.from_protocol",
+                            return_value=config):
+                # we ignore the return value of hint_to_endpoint()
+                d = h.hint_to_endpoint("tor:foo.onion:29212", reactor)
+                f = yield self.assertFailure(d, ValueError)
+        self.assertIn("could not use config.SocksPort", str(f))
+
 
 class I2P(unittest.TestCase):
     @inlineCallbacks
@@ -259,18 +420,18 @@ def test_default(self):
             sep.new = n = mock.Mock()
             n.return_value = expected_ep = object()
             h = i2p.default(reactor)
-            res = yield h.hint_to_endpoint("i2p:fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p", reactor)
-            self.assertEqual(len(n.mock_calls), 1)
-            args = n.mock_calls[0][1]
-            got_sep, got_host, got_portnum = args
-            self.assertIsInstance(got_sep, endpoints.TCP4ClientEndpoint)
-            self.failUnlessEqual(got_sep._host, "127.0.0.1") # fragile
-            self.failUnlessEqual(got_sep._port, 7656)
-            self.failUnlessEqual(got_host, "fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p")
-            self.failUnlessEqual(got_portnum, None)
-            ep, host = res
-            self.assertIdentical(ep, expected_ep)
-            self.assertEqual(host, "fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p")
+            res = yield h.hint_to_endpoint("i2p:fppym.b32.i2p", reactor)
+        self.assertEqual(len(n.mock_calls), 1)
+        args = n.mock_calls[0][1]
+        got_sep, got_host, got_portnum = args
+        self.assertIsInstance(got_sep, endpoints.TCP4ClientEndpoint)
+        self.failUnlessEqual(got_sep._host, "127.0.0.1") # fragile
+        self.failUnlessEqual(got_sep._port, 7656)
+        self.failUnlessEqual(got_host, "fppym.b32.i2p")
+        self.failUnlessEqual(got_portnum, None)
+        ep, host = res
+        self.assertIdentical(ep, expected_ep)
+        self.assertEqual(host, "fppym.b32.i2p")
 
     @inlineCallbacks
     def test_default_with_portnum(self):
@@ -280,18 +441,24 @@ def test_default_with_portnum(self):
             sep.new = n = mock.Mock()
             n.return_value = expected_ep = object()
             h = i2p.default(reactor)
-            res = yield h.hint_to_endpoint("i2p:fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p:1234", reactor)
-            self.assertEqual(len(n.mock_calls), 1)
-            args = n.mock_calls[0][1]
-            got_sep, got_host, got_portnum = args
-            self.assertIsInstance(got_sep, endpoints.TCP4ClientEndpoint)
-            self.failUnlessEqual(got_sep._host, "127.0.0.1") # fragile
-            self.failUnlessEqual(got_sep._port, 7656)
-            self.failUnlessEqual(got_host, "fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p")
-            self.failUnlessEqual(got_portnum, 1234)
-            ep, host = res
-            self.assertIdentical(ep, expected_ep)
-            self.assertEqual(host, "fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p")
+            res = yield h.hint_to_endpoint("i2p:fppym.b32.i2p:1234", reactor)
+        self.assertEqual(len(n.mock_calls), 1)
+        args = n.mock_calls[0][1]
+        got_sep, got_host, got_portnum = args
+        self.assertIsInstance(got_sep, endpoints.TCP4ClientEndpoint)
+        self.failUnlessEqual(got_sep._host, "127.0.0.1") # fragile
+        self.failUnlessEqual(got_sep._port, 7656)
+        self.failUnlessEqual(got_host, "fppym.b32.i2p")
+        self.failUnlessEqual(got_portnum, 1234)
+        ep, host = res
+        self.assertIdentical(ep, expected_ep)
+        self.assertEqual(host, "fppym.b32.i2p")
+
+    def test_default_badhint(self):
+        h = i2p.default(reactor)
+        d = defer.maybeDeferred(h.hint_to_endpoint, "i2p:not@a@hint", reactor)
+        f = self.failureResultOf(d, InvalidHintError)
+        self.assertEqual(str(f.value), "unrecognized I2P hint")
 
     @inlineCallbacks
     def test_sam_endpoint(self):
@@ -300,13 +467,13 @@ def test_sam_endpoint(self):
             n.return_value = expected_ep = object()
             my_ep = endpoints.HostnameEndpoint(reactor, "localhost", 1234)
             h = i2p.sam_endpoint(my_ep)
-            res = yield h.hint_to_endpoint("i2p:fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p", reactor)
-            self.assertEqual(len(n.mock_calls), 1)
-            args = n.mock_calls[0][1]
-            got_sep, got_host, got_portnum = args
-            self.assertIdentical(got_sep, my_ep)
-            self.failUnlessEqual(got_host, "fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p")
-            self.failUnlessEqual(got_portnum, None)
-            ep, host = res
-            self.assertIdentical(ep, expected_ep)
-            self.assertEqual(host, "fppymhuqbd3klxfqbxbz67t3fk6puzeludedp3a4avym5s4wqs3a.b32.i2p")
+            res = yield h.hint_to_endpoint("i2p:fppym.b32.i2p", reactor)
+        self.assertEqual(len(n.mock_calls), 1)
+        args = n.mock_calls[0][1]
+        got_sep, got_host, got_portnum = args
+        self.assertIdentical(got_sep, my_ep)
+        self.failUnlessEqual(got_host, "fppym.b32.i2p")
+        self.failUnlessEqual(got_portnum, None)
+        ep, host = res
+        self.assertIdentical(ep, expected_ep)
+        self.assertEqual(host, "fppym.b32.i2p")