[tor-commits] [stem/master] Refactoring protocolinfo integ tests
atagar at torproject.org
atagar at torproject.org
Sun Jan 29 08:54:42 UTC 2012
commit 51e21cf88643e63953e897d0cd05875d00c26991
Author: Damian Johnson <atagar at torproject.org>
Date: Sun Jan 29 00:49:29 2012 -0800
Refactoring protocolinfo integ tests
Cleaning up the protocolinfo integ tests, joining a couple that had a similar
prupose and other miscellaneous refactoring.
---
test/integ/connection/protocolinfo.py | 152 +++++++++++++--------------------
1 files changed, 59 insertions(+), 93 deletions(-)
diff --git a/test/integ/connection/protocolinfo.py b/test/integ/connection/protocolinfo.py
index 6c9519e..a0e1699 100644
--- a/test/integ/connection/protocolinfo.py
+++ b/test/integ/connection/protocolinfo.py
@@ -9,11 +9,31 @@ import test.runner
import stem.socket
import stem.connection
import stem.util.system
+import test.mocking as mocking
+
+def filter_system_call(prefixes):
+ """
+ Provides a functor that passes calls on to the stem.util.system.call()
+ function if it matches one of the prefixes, and acts as a no-op otherwise.
+ """
+
+ def _filter_system_call(command):
+ for prefix in prefixes:
+ if command.startswith(prefix):
+ return stem.util.system.call(command)
+
+ return _filter_system_call
class TestProtocolInfo(unittest.TestCase):
+ def setUp(self):
+ if not test.runner.get_runner().is_accessible():
+ self.skipTest("(no connection)")
+
+ mocking.mock(stem.util.proc.is_available, mocking.return_false())
+ mocking.mock(stem.util.system.is_available, mocking.return_true())
+
def tearDown(self):
- # resets call mocking back to being disabled
- stem.util.system.CALL_MOCKING = None
+ mocking.revert_mocking()
def test_parsing(self):
"""
@@ -21,13 +41,7 @@ class TestProtocolInfo(unittest.TestCase):
connection.
"""
- runner = test.runner.get_runner()
-
- if not runner.is_accessible():
- self.skipTest("(no connection)")
-
- control_socket = runner.get_tor_socket(False)
-
+ control_socket = test.runner.get_runner().get_tor_socket(False)
control_socket.send("PROTOCOLINFO 1")
protocolinfo_response = control_socket.recv()
stem.connection.ProtocolInfoResponse.convert(protocolinfo_response)
@@ -40,76 +54,44 @@ class TestProtocolInfo(unittest.TestCase):
self.assertNotEqual(None, protocolinfo_response.tor_version)
self.assertNotEqual(None, protocolinfo_response.auth_methods)
- self.assert_protocolinfo_attr(protocolinfo_response)
+ self.assert_matches_test_config(protocolinfo_response)
- def test_get_protocolinfo_by_port(self):
- """
- Exercises the stem.connection.get_protocolinfo function with a control
- port.
+ def test_get_protocolinfo_path_expansion(self):
"""
+ If we're running with the 'RELATIVE' target then test_parsing() will
+ exercise cookie path expansion when we're able to query the pid by our
+ prcess name. This test selectively disables system.call() so we exercise
+ the expansion via our control port or socket file.
- # If we have both the 'RELATIVE' target and a cookie then test_parsing
- # should exercise cookie expansion using a pid lookup by process name.
- # Disabling those lookups so we exercise the lookup by port/socket file
- # too. Gotta remember the get_cwd functions too.
-
- cwd_by_port_lookup_prefixes = (
- stem.util.system.GET_PID_BY_PORT_NETSTAT,
- stem.util.system.GET_PID_BY_PORT_SOCKSTAT % "",
- stem.util.system.GET_PID_BY_PORT_LSOF,
- stem.util.system.GET_CWD_PWDX % "",
- "lsof -a -p ")
-
- def port_lookup_filter(command):
- for prefix in cwd_by_port_lookup_prefixes:
- if command.startswith(prefix): return True
-
- return False
-
- stem.util.system.CALL_MOCKING = port_lookup_filter
+ This test is largely redundant with test_parsing() if we aren't running
+ with the 'RELATIVE' target.
+ """
if test.runner.Torrc.PORT in test.runner.get_runner().get_options():
- control_socket = stem.socket.ControlPort(control_port = test.runner.CONTROL_PORT)
- protocolinfo_response = stem.connection.get_protocolinfo(control_socket)
- self.assert_protocolinfo_attr(protocolinfo_response)
+ cwd_by_port_lookup_prefixes = (
+ stem.util.system.GET_PID_BY_PORT_NETSTAT,
+ stem.util.system.GET_PID_BY_PORT_SOCKSTAT % "",
+ stem.util.system.GET_PID_BY_PORT_LSOF,
+ stem.util.system.GET_CWD_PWDX % "",
+ "lsof -a -p ")
- # we should have a usable socket at this point
- self.assertTrue(control_socket.is_alive())
- control_socket.close()
+ mocking.mock(stem.util.system.call, filter_system_call(cwd_by_port_lookup_prefixes))
+ control_socket = stem.socket.ControlPort(control_port = test.runner.CONTROL_PORT)
else:
- # we don't have a control port
- self.assertRaises(stem.socket.SocketError, stem.socket.ControlPort, "127.0.0.1", test.runner.CONTROL_PORT)
-
- def test_get_protocolinfo_by_socket(self):
- """
- Exercises the stem.connection.get_protocolinfo function with a control
- socket.
- """
-
- cwd_by_socket_lookup_prefixes = (
- stem.util.system.GET_PID_BY_FILE_LSOF % "",
- stem.util.system.GET_CWD_PWDX % "",
- "lsof -a -p ")
-
- def socket_lookup_filter(command):
- for prefix in cwd_by_socket_lookup_prefixes:
- if command.startswith(prefix): return True
+ cwd_by_socket_lookup_prefixes = (
+ stem.util.system.GET_PID_BY_FILE_LSOF % "",
+ stem.util.system.GET_CWD_PWDX % "",
+ "lsof -a -p ")
- return False
+ mocking.mock(stem.util.system.call, filter_system_call(cwd_by_socket_lookup_prefixes))
+ control_socket = stem.socket.ControlSocketFile(test.runner.CONTROL_SOCKET_PATH)
- stem.util.system.CALL_MOCKING = socket_lookup_filter
+ protocolinfo_response = stem.connection.get_protocolinfo(control_socket)
+ self.assert_matches_test_config(protocolinfo_response)
- if test.runner.Torrc.SOCKET in test.runner.get_runner().get_options():
- control_socket = stem.socket.ControlSocketFile(test.runner.CONTROL_SOCKET_PATH)
- protocolinfo_response = stem.connection.get_protocolinfo(control_socket)
- self.assert_protocolinfo_attr(protocolinfo_response)
-
- # we should have a usable socket at this point
- self.assertTrue(control_socket.is_alive())
- control_socket.close()
- else:
- # we don't have a control socket
- self.assertRaises(stem.socket.SocketError, stem.socket.ControlSocketFile, test.runner.CONTROL_SOCKET_PATH)
+ # we should have a usable socket at this point
+ self.assertTrue(control_socket.is_alive())
+ control_socket.close()
def test_multiple_protocolinfo_calls(self):
"""
@@ -118,34 +100,23 @@ class TestProtocolInfo(unittest.TestCase):
re-establish it.
"""
- runner = test.runner.get_runner()
-
- if not runner.is_accessible():
- self.skipTest("(no connection)")
-
- control_socket = runner.get_tor_socket(False)
-
- for i in range(5):
- protocolinfo_response = stem.connection.get_protocolinfo(control_socket)
- self.assert_protocolinfo_attr(protocolinfo_response)
-
- control_socket.close()
+ with test.runner.get_runner().get_tor_socket(False) as control_socket:
+ for i in range(5):
+ protocolinfo_response = stem.connection.get_protocolinfo(control_socket)
+ self.assert_matches_test_config(protocolinfo_response)
- def assert_protocolinfo_attr(self, protocolinfo_response):
+ def assert_matches_test_config(self, protocolinfo_response):
"""
Makes assertions that the protocolinfo response's attributes match those of
- a given connection type.
+ the test configuration.
"""
- # This should never have test.runner.TorConnection.NONE. If we somehow got
- # a protocolinfo_response from that config then we have an issue. :)
-
tor_options = test.runner.get_runner().get_options()
-
- auth_methods = []
+ auth_methods, auth_cookie_path = [], None
if test.runner.Torrc.COOKIE in tor_options:
auth_methods.append(stem.connection.AuthMethod.COOKIE)
+ auth_cookie_path = test.runner.get_runner().get_auth_cookie_path()
if test.runner.Torrc.PASSWORD in tor_options:
auth_methods.append(stem.connection.AuthMethod.PASSWORD)
@@ -155,10 +126,5 @@ class TestProtocolInfo(unittest.TestCase):
self.assertEqual((), protocolinfo_response.unknown_auth_methods)
self.assertEqual(tuple(auth_methods), protocolinfo_response.auth_methods)
-
- auth_cookie_path = None
- if test.runner.Torrc.COOKIE in tor_options:
- auth_cookie_path = test.runner.get_runner().get_auth_cookie_path()
-
self.assertEqual(auth_cookie_path, protocolinfo_response.cookie_path)
More information about the tor-commits
mailing list