[Pkg-privacy-commits] [txtorcon] 21/96: More test coverage
Jérémy Bobbio
lunar at moszumanska.debian.org
Sun Sep 6 18:33:34 UTC 2015
This is an automated email from the git hooks/post-receive script.
lunar pushed a commit to branch master
in repository txtorcon.
commit ea33bd080b0bb76f035a63de3fdf0a7c67e48173
Author: meejah <meejah at meejah.ca>
Date: Sat Jan 17 00:39:17 2015 -0700
More test coverage
---
test/test_endpoints.py | 32 ++++++++++++++++--
test/test_torconfig.py | 73 ++++++++++++++++++++++++++++++-----------
test/test_torcontrolprotocol.py | 29 ++++++++++++++++
test/test_torinfo.py | 31 +++++++++++++++--
test/test_torstate.py | 38 ++++++++++++++++++++-
test/test_util.py | 23 ++++++++++++-
test/test_util_imports.py | 12 +++++--
txtorcon/endpoints.py | 14 ++++----
txtorcon/stream.py | 1 +
9 files changed, 217 insertions(+), 36 deletions(-)
diff --git a/test/test_endpoints.py b/test/test_endpoints.py
index c4ceed9..41f82d8 100644
--- a/test/test_endpoints.py
+++ b/test/test_endpoints.py
@@ -166,6 +166,34 @@ class EndpointTests(unittest.TestCase):
ep._tor_progress_update(*args)
ding.assert_called_with(*args)
+ @patch('txtorcon.endpoints.launch_tor')
+ def test_progress_updates_private_tor(self, tor):
+ ep = TCPHiddenServiceEndpoint.private_tor(self.reactor, 1234)
+ tor.call_args[1]['progress_updates'](40, 'FOO', 'foo to the bar')
+ return ep
+
+ def __test_progress_updates_system_tor(self):
+ ep = TCPHiddenServiceEndpoint.system_tor(self.reactor, 1234)
+ ep._tor_progress_update(40, "FOO", "foo to bar")
+ return ep
+
+ @patch('txtorcon.endpoints.get_global_tor')
+ def test_progress_updates_global_tor(self, tor):
+ ep = TCPHiddenServiceEndpoint.global_tor(self.reactor, 1234)
+ tor.call_args[1]['progress_updates'](40, 'FOO', 'foo to the bar')
+ return ep
+
+ def test_hiddenservice_key_unfound(self):
+ ep = TCPHiddenServiceEndpoint.private_tor(self.reactor, 1234, hidden_service_dir='/dev/null')
+ # FIXME Mock() should work somehow for this, but I couldn't make it "go"
+ class Blam(object):
+ @property
+ def private_key(self):
+ raise IOError("blam")
+ ep.hiddenservice = Blam()
+ self.assertEqual(ep.onion_private_key, None)
+ return ep
+
def test_multiple_listen(self):
ep = TCPHiddenServiceEndpoint(self.reactor, self.config, 123)
d0 = ep.listen(NoOpProtocolFactory())
@@ -397,8 +425,8 @@ def port_generator():
yield x
-from test_torconfig import FakeReactor # FIXME
-from test_torconfig import FakeProcessTransport # FIXME
+from test_torconfig import FakeReactor # FIXME put in util or something?
+from test_torconfig import FakeProcessTransport # FIXME importing from other test sucks
from test_torconfig import FakeControlProtocol # FIXME
diff --git a/test/test_torconfig.py b/test/test_torconfig.py
index 94a3d37..d58544b 100644
--- a/test/test_torconfig.py
+++ b/test/test_torconfig.py
@@ -2,9 +2,10 @@ import os
import shutil
import tempfile
import functools
+from getpass import getuser
from StringIO import StringIO
-from mock import Mock
+from mock import Mock, patch
from zope.interface import implements
from twisted.trial import unittest
@@ -356,6 +357,38 @@ class ConfigTests(unittest.TestCase):
self.assertEqual(self.protocol.sets[0], ('Log', 'foo'))
self.assertEqual(self.protocol.sets[1], ('Log', 'bar'))
+ @defer.inlineCallbacks
+ def test_attach_protocol(self):
+ self.protocol.answers.append('config/names=\nLog LineList')
+ self.protocol.answers.append({'Log': 'foo'})
+
+ conf = TorConfig()
+ d = conf.attach_protocol(self.protocol)
+ yield d
+
+ conf.log.append('bar')
+ yield conf.save()
+
+ self.assertEqual(len(self.protocol.sets), 2)
+ self.assertEqual(self.protocol.sets[0], ('Log', 'foo'))
+ self.assertEqual(self.protocol.sets[1], ('Log', 'bar'))
+
+ def test_attach_protocol_but_already_have_one(self):
+ conf = TorConfig(self.protocol)
+ self.assertRaises(RuntimeError, conf.attach_protocol, self.protocol)
+
+ def test_no_confchanged_event(self):
+ conf = TorConfig(self.protocol)
+ self.protocol.add_event_listener = Mock(side_effect=RuntimeError)
+ d = defer.Deferred()
+ self.protocol.get_info_raw = Mock(return_value=d)
+ conf.bootstrap()
+ # this should log a message, do we really care what?
+
+ def test_attribute_access(self):
+ conf = TorConfig(self.protocol)
+ self.assertNotIn('_slutty_', conf.__dict__)
+ self.assertNotIn('foo', conf)
class LogTests(unittest.TestCase):
@@ -400,6 +433,7 @@ class LogTests(unittest.TestCase):
self.assertTrue(conf.needs_save())
conf.save()
+ self.assertEqual(1, len(self.protocol.sets))
self.assertEqual(self.protocol.sets[0], ('Log', 'info file /tmp/foo.log'))
def test_log_set_pop(self):
@@ -581,7 +615,7 @@ HiddenServiceAuthorizeClient Dependant''')
def test_add_hidden_service_to_empty_config(self):
conf = TorConfig()
h = HiddenService(conf, '/fake/path', ['80 127.0.0.1:1234'], '', 3)
- conf.hiddenservices.append(h)
+ conf.HiddenServices.append(h)
self.assertEqual(len(conf.hiddenservices), 1)
self.assertEqual(h, conf.hiddenservices[0])
self.assertTrue(conf.needs_save())
@@ -765,10 +799,14 @@ class LaunchTorTests(unittest.TestCase):
self.assertTrue(not os.path.exists(f))
return None
- def test_basic_launch(self):
+ @patch('txtorcon.torconfig.os.geteuid')
+ def test_basic_launch(self, geteuid):
+ # pretend we're root to exercise the "maybe chown data dir" codepath
+ geteuid.return_value = 0
config = TorConfig()
config.ORPort = 1234
config.SOCKSPort = 9999
+ config.User = getuser()
def connector(proto, trans):
proto._set_valid_events('STATUS_CLIENT')
@@ -1135,11 +1173,11 @@ class LaunchTorTests(unittest.TestCase):
return pp
+from mock import patch
class ErrorTests(unittest.TestCase):
- def test_no_tor_binary(self):
+ @patch('txtorcon.torconfig.find_tor_binary')
+ def test_no_tor_binary(self, ftb):
"""FIXME: do I really need all this crap in here?"""
- from txtorcon import torconfig
- oldone = torconfig.find_tor_binary
self.transport = proto_helpers.StringTransport()
config = TorConfig()
d = None
@@ -1151,19 +1189,16 @@ class ErrorTests(unittest.TestCase):
proto.post_bootstrap.callback(proto)
return proto.post_bootstrap
+ self.protocol = FakeControlProtocol([])
+ torconfig.find_tor_binary = lambda: None
+ trans = FakeProcessTransport()
+ trans.protocol = self.protocol
+ creator = functools.partial(Connector(), self.protocol, self.transport)
try:
- self.protocol = FakeControlProtocol([])
- torconfig.find_tor_binary = lambda: None
- trans = FakeProcessTransport()
- trans.protocol = self.protocol
- creator = functools.partial(Connector(), self.protocol, self.transport)
- try:
- d = launch_tor(config, FakeReactor(self, trans, lambda x: None), connection_creator=creator)
- self.fail()
-
- except TorNotFound:
- pass # success!
- finally:
- torconfig.find_tor_binary = oldone
+ d = launch_tor(config, FakeReactor(self, trans, lambda x: None), connection_creator=creator)
+ self.fail()
+
+ except TorNotFound:
+ pass # success!
return d
diff --git a/test/test_torcontrolprotocol.py b/test/test_torcontrolprotocol.py
index 8674a5d..38b4957 100644
--- a/test/test_torcontrolprotocol.py
+++ b/test/test_torcontrolprotocol.py
@@ -1,5 +1,7 @@
from __future__ import with_statement
+from os.path import exists
+
from twisted.python import log, failure
from twisted.trial import unittest
from twisted.test import proto_helpers
@@ -117,6 +119,19 @@ class AuthenticationTests(unittest.TestCase):
## now make sure we DID try to authenticate
self.assertEqual(self.transport.value(), 'AUTHENTICATE %s\r\n' % "foo".encode("hex"))
+ def test_authenticate_password_deferred_but_no_password(self):
+ d = defer.Deferred()
+ self.protocol.password_function = lambda: d
+ self.protocol.makeConnection(self.transport)
+ self.assertEqual(self.transport.value(), 'PROTOCOLINFO 1\r\n')
+ self.transport.clear()
+ self.send('250-PROTOCOLINFO 1')
+ self.send('250-AUTH METHODS=HASHEDPASSWORD')
+ self.send('250-VERSION Tor="0.2.2.34"')
+ self.send('250 OK')
+ d.callback(None)
+ return self.assertFailure(self.protocol.post_bootstrap, RuntimeError)
+
def confirmAuthFailed(self, *args):
self.auth_failed = True
@@ -457,6 +472,20 @@ OK''' % cookietmp.name)
self._wait(d)
self.assertEqual(self.transport.value(), "SETCONF foo=bar baz=1\r\n")
+ def test_quit(self):
+ d = self.protocol.quit()
+ self.send("250 OK")
+ self._wait(d)
+ self.assertEqual(self.transport.value(), "QUIT\r\n")
+
+ def test_dot(self):
+ # just checking we don't expode
+ self.protocol.graphviz_data()
+
+ def test_debug(self):
+ self.protocol.start_debug()
+ self.assertTrue(exists('txtorcon-debug.log'))
+
def error(self, failure):
print "ERROR", failure
self.assertTrue(False)
diff --git a/test/test_torinfo.py b/test/test_torinfo.py
index 8610b7c..7a7d2d1 100644
--- a/test/test_torinfo.py
+++ b/test/test_torinfo.py
@@ -151,6 +151,7 @@ something/two a second documentation string
d.addCallback(CheckAnswer(self, 'bar'))
return d
+ @defer.inlineCallbacks
def test_attribute_access(self):
'''
test that our post-setup TorInfo pretends to only have
@@ -162,11 +163,23 @@ something/one a documentation string
something/two a second documentation string
''')
info = TorInfo(self.protocol)
-
- self.assertTrue(dir(info) == ['something'])
+ yield self.protocol.post_bootstrap
+ self.assertTrue('something' in dir(info))
self.assertTrue(dir(info.something) == ['one', 'two'] or
dir(info.something) == ['two', 'one'])
+ def test_member_access(self):
+ self.protocol.answers.append('info/names blam a thinkg\r\n')
+ info = TorInfo(self.protocol)
+ from txtorcon import torinfo
+ c = torinfo.MagicContainer(None)
+ c._setup = True
+ self.assertEqual([], c.__members__)
+ self.assertEqual(['info'], info.__members__)
+ # make sure __magic__ attr access doesn't throw
+ c.__class__
+ self.assertRaises(AttributeError, lambda: c.foo_mc_bar_bar)
+
def test_iterator_access(self):
'''
confirm we can use the iterator protocol
@@ -190,6 +203,18 @@ something/two a second documentation string
all.append(x)
self.assertTrue(len(all) == 2)
+
+ def test_accessors_not_setup(self):
+ info = TorInfo(self.protocol)
+ self.assertTrue(info.__dict__['_setup'] == False)
+ self.assertRaises(TypeError, len, info)
+ dir(info)
+ try:
+ info[0]
+ self.fail("Should have raised TypeError")
+ except TypeError:
+ pass
+
def handle_error(self, f):
if 'Already had something' in f.getErrorMessage():
self.error_happened = True
@@ -263,7 +288,7 @@ multi/path/arg/* a documentation string
config/* a documentation string
''')
info = TorInfo(self.protocol)
- self.assertTrue(dir(info) == [])
+ self.assertEqual(dir(info), [])
def test_other_bootstrap(self):
self.protocol.answers.append('''info/names=
diff --git a/test/test_torstate.py b/test/test_torstate.py
index ad86a61..a420ba0 100644
--- a/test/test_torstate.py
+++ b/test/test_torstate.py
@@ -8,7 +8,7 @@ from twisted.internet.interfaces import IStreamClientEndpoint, IReactorCore
import os
import tempfile
-from txtorcon import TorControlProtocol, TorProtocolError, TorState, Stream, Circuit, build_tor_connection
+from txtorcon import TorControlProtocol, TorProtocolError, TorState, Stream, Circuit, build_tor_connection, build_local_tor_connection
from txtorcon.interface import ITorControlProtocol, IStreamAttacher, ICircuitListener, IStreamListener, StreamListenerMixin, CircuitListenerMixin
@@ -256,6 +256,32 @@ class BootstrapTests(unittest.TestCase):
p.proto.post_bootstrap.callback(p.proto)
return d
+ def test_build_with_answers_guards_unfound_entry(self):
+ p = FakeEndpointAnswers(['', # ns/all
+ '', # circuit-status
+ '', # stream-status
+ '', # address-mappings/all
+ '\n\nkerblam up\nOK\n' # entry-guards
+ ])
+
+ d = build_tor_connection(p, build_state=True)
+ d.addCallback(self.confirm_state)
+ d.addCallback(self.confirm_no_pid)
+ p.proto.post_bootstrap.callback(p.proto)
+ return d
+
+ def test_build_local_unix(self):
+ reactor = FakeReactor(self)
+ d = build_local_tor_connection(reactor)
+ d.addErrback(lambda _: None)
+ return d
+
+ def test_build_local_tcp(self):
+ reactor = FakeReactor(self)
+ d = build_local_tor_connection(reactor, socket=None)
+ d.addErrback(lambda _: None)
+ return d
+
class StateTests(unittest.TestCase):
@@ -292,6 +318,11 @@ class StateTests(unittest.TestCase):
self.state._stream_status('stream-status=123 SUCCEEDED 496 www.example.com:6667')
self.assertEqual(len(self.state.streams), 1)
+ def test_multiple_streams(self):
+ self.state.circuits[496] = FakeCircuit(496)
+ self.state._stream_status('stream-status=\r\n123 SUCCEEDED 496 www.example.com:6667\r\n124 SUCCEEDED 496 www.example.com:6667')
+ self.assertEqual(len(self.state.streams), 2)
+
def send(self, line):
self.protocol.dataReceived(line.strip() + "\r\n")
@@ -304,6 +335,7 @@ class StateTests(unittest.TestCase):
d = self.state.post_bootstrap
self.protocol._set_valid_events(' '.join(self.state.event_map.keys()))
+ self.protocol.is_owned = 999
self.state._bootstrap()
self.send("250+ns/all=")
@@ -1028,6 +1060,10 @@ s Fast Guard Running Stable Valid
self.state.build_circuit()
self.assertEqual(self.transport.value(), 'EXTENDCIRCUIT 0\r\n')
+ def test_build_circuit_unfound_router(self):
+ self.state.build_circuit(routers=['AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'])
+ self.assertEqual(self.transport.value(), 'EXTENDCIRCUIT 0 AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\r\n')
+
def circuit_callback(self, circ):
self.assertTrue(isinstance(circ, Circuit))
self.assertEqual(circ.id, 1234)
diff --git a/test/test_util.py b/test/test_util.py
index a82a9a7..e53a814 100644
--- a/test/test_util.py
+++ b/test/test_util.py
@@ -1,10 +1,11 @@
+from mock import patch
from twisted.trial import unittest
from twisted.internet import defer
from twisted.internet.endpoints import TCP4ServerEndpoint
from twisted.internet.interfaces import IProtocolFactory
from zope.interface import implements
-from txtorcon.util import process_from_address, delete_file_or_tree, find_keywords, ip_from_int, find_tor_binary
+from txtorcon.util import process_from_address, delete_file_or_tree, find_keywords, ip_from_int, find_tor_binary, maybe_ip_addr
import os
import tempfile
@@ -230,3 +231,23 @@ class TestFindTor(unittest.TestCase):
def test_find_tor_unfound(self):
"test searching by globs"
self.assertEqual(None, find_tor_binary(system_tor=False, globs=()))
+
+ @patch('txtorcon.util.subprocess.Popen')
+ def test_find_ioerror(self, popen):
+ "test searching with which, but it fails"
+ popen.side_effect = OSError
+ self.assertEqual(None, find_tor_binary(system_tor=True, globs=()))
+
+
+class TestIpAddr(unittest.TestCase):
+
+ @patch('txtorcon.util.ipaddr')
+ def test_create_ipaddr(self, ipaddr):
+ ip = maybe_ip_addr('1.2.3.4')
+
+ @patch('txtorcon.util.ipaddr')
+ def test_create_ipaddr(self, ipaddr):
+ def foo(blam):
+ raise ValueError('testing')
+ ipaddr.IPAddress.side_effect = foo
+ ip = maybe_ip_addr('1.2.3.4')
diff --git a/test/test_util_imports.py b/test/test_util_imports.py
index 6845b7d..ed1f59f 100644
--- a/test/test_util_imports.py
+++ b/test/test_util_imports.py
@@ -3,6 +3,7 @@ from twisted.trial import unittest
import sys
import types
import functools
+from unittest import skipIf
def fake_import(orig, name, *args, **kw):
@@ -14,16 +15,20 @@ def fake_import(orig, name, *args, **kw):
class TestImports(unittest.TestCase):
+ @skipIf('pypy' in sys.version.lower(), "Doesn't work in PYPY")
def test_no_GeoIP(self):
"""
- Make sure we don't explode if there's no ipaddr module
+ Make sure we don't explode if there's no GeoIP module
"""
global __import__
orig = __import__
try:
# attempt to ensure we've unimportted txtorcon.util
- del sys.modules['txtorcon.util']
+ try:
+ del sys.modules['txtorcon.util']
+ except KeyError:
+ pass
import gc
gc.collect()
@@ -41,9 +46,10 @@ class TestImports(unittest.TestCase):
finally:
__import__ = orig
+ @skipIf('pypy' in sys.version.lower(), "Doesn't work in PYPY")
def test_no_ipaddr(self):
"""
- make sure the code we run if there's no GeoIP installed
+ make sure the code we run if there's no ipaddr installed
doesn't do anything horrific
"""
diff --git a/txtorcon/endpoints.py b/txtorcon/endpoints.py
index 6d78394..bcae171 100644
--- a/txtorcon/endpoints.py
+++ b/txtorcon/endpoints.py
@@ -103,8 +103,7 @@ class IProgressProvider(Interface):
@implementer(IStreamServerEndpoint, IProgressProvider)
class TCPHiddenServiceEndpoint(object):
- """
- This represents something listening on an arbitrary local port
+ """This represents something listening on an arbitrary local port
that has a Tor configured with a Hidden Service pointing at
it. :api:`twisted.internet.endpoints.TCP4ServerEndpoint
<TCP4ServerEndpoint>` is used under the hood to do the local
@@ -137,11 +136,11 @@ class TCPHiddenServiceEndpoint(object):
No matter how you came by your instance, calling `listen()` on it
causes Tor to be launched or connected-to, your hidden service to
- be added, (XXX and check the descriptor is uploaded! FIXME) and
- you get a ``Deferred`` with an ``IListeningPort`` whose
- ``getHost()`` will return a :class:`txtorcon.TorOnionAddress`. The port
- object will also implement :class:`txtorcon.IHiddenService` so you can get
- the locally-listening address and hidden serivce directory::
+ be added, checks that the descriptor is uploaded and you get a
+ ``Deferred`` with an ``IListeningPort`` whose ``getHost()`` will
+ return a :class:`txtorcon.TorOnionAddress`. The port object will
+ also implement :class:`txtorcon.IHiddenService` so you can get the
+ locally-listening address and hidden serivce directory::
endpoint = ...
port = yield endpoint.listen(...)
@@ -161,6 +160,7 @@ class TCPHiddenServiceEndpoint(object):
:ivar hiddenServiceDir: the data directory, either passed in or created
with ``tempfile.mkstemp``
+
"""
@classmethod
diff --git a/txtorcon/stream.py b/txtorcon/stream.py
index 5db6263..ea3ab85 100644
--- a/txtorcon/stream.py
+++ b/txtorcon/stream.py
@@ -174,6 +174,7 @@ class Stream(object):
self.source_port = int(kw['SOURCE_ADDR'][last_colon + 1:])
self.state = args[1]
+ # XXX why not using the state-machine stuff? ;)
if self.state in ['NEW', 'NEWRESOLVE', 'SUCCEEDED']:
if self.target_host is None:
last_colon = args[3].rfind(':')
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/pkg-privacy/packages/txtorcon.git
More information about the Pkg-privacy-commits
mailing list