[tor-commits] [stem/master] Refactoring protocolinfo integ tests

commit 51e21cf88643e63953e897d0cd05875d00c26991 Author: Damian Johnson atagar@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)