use_stem cleanups

 - put all Stem-related tests in their own suite
 - re-jig where imports happen, and test them separately
 - improve documentation and comments

test/test_torcontrolprotocol.py

@@ -204,37 +204,39 @@ def it_was_called(*args):
         self.assertTrue(it_was_called.yes)
 
 
-class ProtocolTests(unittest.TestCase):
+def _answer_bootstrap_requests(protocol):
+    """
+    Helper used by ProtocolTests and ProtocolTestsUsingStem
+    """
+    def send(line):
+        protocol.dataReceived(line.strip() + "\r\n")
 
-    def setUp(self):
-        self.protocol = TorControlProtocol()
-        self.protocol.connectionMade = lambda: None
-        self.transport = proto_helpers.StringTransport()
-        self.protocol.makeConnection(self.transport)
+    send("250-signal/names=")
+    send("250 OK")
 
-    def tearDown(self):
-        self.protocol = None
+    send("250-version=0.2.5.10 (git-42b42605f8d8eac2)")
+    send("250 OK")
 
-    def setup_use_stem(self):
-        self.protocol = TorControlProtocol(use_stem = True)
-        self.protocol.connectionMade = lambda: None
-        self.protocol.makeConnection(self.transport)
+    send("250-events/names=")
+    send("250 OK")
 
-    def answer_bootstrap_requests(self):
-        self.send("250-signal/names=")
-        self.send("250 OK")
+    send("250 OK")
 
-        self.send("250-version=0.2.5.10 (git-42b42605f8d8eac2)")
-        self.send("250 OK")
 
-        self.send("250-events/names=")
-        self.send("250 OK")
+class ProtocolTests(unittest.TestCase):
 
-        self.send("250 OK")
+    def setUp(self):
+        self.protocol = TorControlProtocol()
+        self.protocol.connectionMade = lambda: None
+        self.transport = proto_helpers.StringTransport()
+        self.protocol.makeConnection(self.transport)
 
     def send(self, line):
         self.protocol.dataReceived(line.strip() + "\r\n")
 
+    def tearDown(self):
+        self.protocol = None
+
     def test_statemachine_broadcast_no_code(self):
         try:
             self.protocol._broadcast_response("foo")
@@ -402,71 +404,9 @@ def test_bootstrap_callback(self):
 
         return d
 
-    def test_bootstrap_tor_does_not_support_signal_names(self):
-        self.protocol._bootstrap()
-        self.send('552 Unrecognized key "signal/names"')
-        valid_signals = ["RELOAD", "DUMP", "DEBUG", "NEWNYM", "CLEARDNSCACHE"]
-        self.assertEqual(self.protocol.valid_signals, valid_signals)
-
-    def confirm_version_use_stem(self, arg):
-        self.assertEqual(self.protocol.version.major, 0)
-        self.assertEqual(self.protocol.version.minor, 2)
-        self.assertEqual(self.protocol.version.micro, 5)
-        self.assertEqual(self.protocol.version.patch, 10)
-        self.assertEqual(self.protocol.version.git_commit, '42b42605f8d8eac2')
-        self.assertEqual(self.protocol.version.extra, 'git-42b42605f8d8eac2')
-        self.assertEqual(self.protocol.version.version_str, '0.2.5.10 (git-42b42605f8d8eac2)')
-
-    def test_bootstrap_callback_version_use_stem(self):
-        """
-        Test that the Tor version makes use of
-        stem.version.Version when the user specifies use_stem=True
-        """
-        self.setup_use_stem()
-        d = self.protocol.post_bootstrap
-        d.addCallback(CallbackChecker(self.protocol))
-        d.addCallback(self.confirm_version_use_stem)
-
-        self.protocol._bootstrap()
-
-        # Answer all the requests generated by bootstrapping
-        self.answer_bootstrap_requests()
-
-        return d
-
-    def confirm_version_no_stem(self, arg, orig):
-        __builtins__['__import__'] = orig
-        self.assertEqual(type(self.protocol.version).__name__, 'str')
-        self.assertEqual(self.protocol.version, '0.2.5.10 (git-42b42605f8d8eac2)')
-
-    def fake_import(self, orig, name, *args, **kw):
-        if name == 'stem.version':
-            raise ImportError('testing!')
-        return orig(*((name,) + args), **kw)
-
-    @skipIf('pypy' in sys.version.lower(), "Doesn't work in PYPY")
-    def test_bootstrap_callback_version_no_stem(self):
-        """
-        Test that the Tor version is a string
-        when there is no stem module found
-        """
-        # XXX FIXME this should use mock.patch instead, probably
-        orig = __builtins__['__import__']
-        __builtins__['__import__'] = functools.partial(self.fake_import, orig)
-        self.setup_use_stem()
-        d = self.protocol.post_bootstrap
-        d.addCallback(CallbackChecker(self.protocol))
-        d.addCallback(self.confirm_version_no_stem, orig)
-
-        self.protocol._bootstrap()
-
-        # Answer all the requests generated by bootstrapping
-        self.answer_bootstrap_requests()
-
-        return d
-
     def confirm_version_dont_use_stem(self, arg):
-        self.assertEqual(type(self.protocol.version).__name__, 'str')
+        # XXX six.string_type
+        self.assertTrue(isinstance(self.protocol.version, str))
         self.assertEqual(self.protocol.version, '0.2.5.10 (git-42b42605f8d8eac2)')
 
     def test_bootstrap_callback_version_dont_use_stem(self):
@@ -480,10 +420,16 @@ def test_bootstrap_callback_version_dont_use_stem(self):
         self.protocol._bootstrap()
 
         # Answer all the requests generated by bootstrapping
-        self.answer_bootstrap_requests()
+        _answer_bootstrap_requests(self.protocol)
 
         return d
 
+    def test_bootstrap_tor_does_not_support_signal_names(self):
+        self.protocol._bootstrap()
+        self.send('552 Unrecognized key "signal/names"')
+        valid_signals = ["RELOAD", "DUMP", "DEBUG", "NEWNYM", "CLEARDNSCACHE"]
+        self.assertEqual(self.protocol.valid_signals, valid_signals)
+
     def test_async(self):
         """
         test the example from control-spec.txt to see that we
@@ -943,8 +889,8 @@ def test_invalid_stem_event_information(self, stdout):
         self.protocol.add_event_listener('STREAM', CallbackChecker(""))
         self.send("650 STREAM UNKNOWN_RESPONSE")
 
-        expected = "ProtocolError STREAM event didn't have a target: STREAM UNKNOWN_RESPONSE\nUnable to convert into stem.response.ControlMessage instance: UNKNOWN_RESPONSE\n"
-        self.assertEqual(stdout.getvalue(), expected)
+        expected = "STREAM event didn't have a target"
+        self.assertTrue(expected in stdout.getvalue())
 
 
 class ParseTests(unittest.TestCase):
@@ -1076,3 +1022,48 @@ def test_circuit_status(self):
         self.controller.routers_by_name.clear()
         self.controller.routers_by_hash.clear()
         self.controller.circuits.clear()
+
+
+class ProtocolTestsUsingStem(unittest.TestCase):
+
+    def setUp(self):
+        self.protocol = TorControlProtocol(use_stem=True)
+        self.protocol.connectionMade = lambda: None
+        self.transport = proto_helpers.StringTransport()
+        self.protocol.makeConnection(self.transport)
+
+    def confirm_version(self, arg):
+        import stem.version
+        self.assertTrue(isinstance(self.protocol.version, stem.version.Version))
+        self.assertEqual(self.protocol.version.major, 0)
+        self.assertEqual(self.protocol.version.minor, 2)
+        self.assertEqual(self.protocol.version.micro, 5)
+        self.assertEqual(self.protocol.version.patch, 10)
+        self.assertEqual(self.protocol.version.git_commit, '42b42605f8d8eac2')
+        self.assertEqual(self.protocol.version.extra, 'git-42b42605f8d8eac2')
+        self.assertEqual(self.protocol.version.version_str, '0.2.5.10 (git-42b42605f8d8eac2)')
+
+    def test_bootstrap_callback_version(self):
+        """
+        Test that the Tor version makes use of
+        stem.version.Version when the user specifies use_stem=True
+        """
+        d = self.protocol.post_bootstrap
+        d.addCallback(CallbackChecker(self.protocol))
+        d.addCallback(self.confirm_version)
+
+        self.protocol._bootstrap()
+
+        # Answer all the requests generated by bootstrapping
+        _answer_bootstrap_requests(self.protocol)
+
+        return d
+
+    def test_bootstrap_callback_version_no_stem(self):
+        """
+        Test that the Tor version is a string when there is no stem module
+        found
+        """
+        with patch('txtorcon.torcontrolprotocol._HAVE_STEM', False):
+            self.assertRaises(RuntimeError, TorControlProtocol, use_stem=True)
+

test/test_util_imports.py → test/test_zzz_imports.py

@@ -1,27 +1,31 @@
 from twisted.trial import unittest
 
+import gc
 import sys
 import types
 import functools
 from unittest import skipIf
 
 
 def fake_import(orig, name, *args, **kw):
-    if name in ['GeoIP', 'ipaddr']:
+    if name in ['GeoIP', 'ipaddr', 'stem']:
         raise ImportError('testing!')
     return orig(*((name,) + args), **kw)
 
 
 class TestImports(unittest.TestCase):
+    # XXX FIXME this messes up "os" imports, of all things, for some
+    # reason, so it gets the "zzz" in its name to be "last". But
+    # that's not a very good solution.
 
     @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 GeoIP module
         """
 
-        global __import__
-        orig = __import__
+        global __builtins__
+        orig = __builtins__['__import__']
         try:
             # attempt to ensure we've unimportted txtorcon.util
             try:
@@ -33,7 +37,6 @@ def test_no_GeoIP(self):
 
             # replace global import with our test import, which will
             # throw on GeoIP import no matter what
-            global __builtins__
             __builtins__['__import__'] = functools.partial(fake_import, orig)
 
             # now ensure we set up all the databases as "None" when we
@@ -43,7 +46,7 @@ def test_no_GeoIP(self):
             self.assertTrue(isinstance(ipa, types.StringType))
 
         finally:
-            __import__ = orig
+            __builtins__['__import__'] = orig
 
     @skipIf('pypy' in sys.version.lower(), "Doesn't work in PYPY")
     def test_no_ipaddr(self):
@@ -52,17 +55,15 @@ def test_no_ipaddr(self):
         doesn't do anything horrific
         """
 
-        global __import__
-        orig = __import__
+        global __builtins__
+        orig = __builtins__['__import__']
         try:
             # attempt to ensure we've unimportted txtorcon.util
-            del sys.modules['txtorcon.util']
-            import gc
-            gc.collect()
-
-            # replace global import with our test import, which will
-            # throw on GeoIP import no matter what
-            global __builtins__
+            try:
+                del sys.modules['txtorcon.util']
+                gc.collect()
+            except KeyError:
+                pass
             __builtins__['__import__'] = functools.partial(fake_import, orig)
 
             # now ensure we set up all the databases as "None" when we
@@ -73,4 +74,30 @@ def test_no_ipaddr(self):
             self.assertEqual(None, txtorcon.util.country)
 
         finally:
-            __import__ = orig
+            __builtins__['__import__'] = orig
+
+    @skipIf('pypy' in sys.version.lower(), "Doesn't work in PYPY")
+    def test_no_Stem(self):
+        """
+        Ensure we work without Stem installed
+        """
+
+        global __builtins__
+        orig = __builtins__['__import__']
+        try:
+            # attempt to ensure we've unimportted txtorcon.util
+            try:
+                del sys.modules['txtorcon.torcontrolprotocol']
+            except KeyError:
+                pass
+            import gc
+            gc.collect()
+
+            __builtins__['__import__'] = functools.partial(fake_import, orig)
+
+            # make sure we marked that we don't have Stem
+            import txtorcon.torcontrolprotocol
+            self.assertFalse(txtorcon.torcontrolprotocol._HAVE_STEM)
+
+        finally:
+            __builtins__['__import__'] = orig