commit 11cbf61c2dca7cde628e5eec4f3dd4da7867860f Author: Damian Johnson atagar@torproject.org Date: Tue Jan 10 09:49:26 2012 -0800
Mocking utility functions
As the tests grow I've found that I'm repeating a lot of mocking logic. Making a nice, general purpose utility to make mocking easier and less error prone.
A benefit of this (and the original goal) is that we'll be able to remove stem.util.system.CALL_MOCKING. Testing code in the library is icky, and once we swap over the system integ tests we'll be able to remove this override function. --- test/__init__.py | 2 +- test/mocking.py | 75 ++++++++++++++++++++++++++++++++++++++++++++++ test/unit/util/system.py | 40 ++++++++++++++---------- 3 files changed, 100 insertions(+), 17 deletions(-)
diff --git a/test/__init__.py b/test/__init__.py index a663300..93250e2 100644 --- a/test/__init__.py +++ b/test/__init__.py @@ -2,5 +2,5 @@ Unit and integration tests for the stem library. """
-__all__ = ["output", "runner"] +__all__ = ["mocking", "output", "runner"]
diff --git a/test/mocking.py b/test/mocking.py new file mode 100644 index 0000000..740bcb7 --- /dev/null +++ b/test/mocking.py @@ -0,0 +1,75 @@ +""" +Helper functions for creating mock objects and monkey patching to help with +testing. +""" + +import inspect +import itertools + +# Once we've mocked a function we can't rely on its __module__ or __name__ +# attributes, so instead we associate a unique 'mock_id' attribute that maps +# back to the original attributes. + +MOCK_ID = itertools.count(0) + +# mock_id => (module, function_name, original_function) + +MOCK_STATE = {} + +def no_op(): + def _no_op(*args): pass + return _no_op + +def return_value(value): + def _return_value(*args): return value + return _return_value + +def return_true(): return return_value(True) +def return_false(): return return_value(False) +def return_none(): return return_value(None) + +def mock(target, mock_call): + """ + Mocks the given function, saving the initial implementation so it can be + reverted later. + + Arguments: + target (function) - function to be mocked + mock_call (functor) - mocking to replace the function with + """ + + if "mock_id" in target.__dict__: + # we're overriding an already mocked function + mocking_id = target.__dict__["mock_id"] + target_module, target_function, _ = MOCK_STATE[mocking_id] + else: + # this is a new mocking, save the original state + mocking_id = MOCK_ID.next() + target_module = inspect.getmodule(target) + target_function = target.__name__ + MOCK_STATE[mocking_id] = (target_module, target_function, target) + + mock_wrapper = lambda *args: mock_call(*args) + mock_wrapper.__dict__["mock_id"] = mocking_id + + # mocks the function with this wrapper + target_module.__dict__[target_function] = mock_wrapper + +def revert_mocking(): + """ + Reverts any mocking done by this function. + """ + + # Reverting mocks in reverse order. If we properly reuse mock_ids then this + # shouldn't matter, but might as well be safe. + + mock_ids = MOCK_STATE.keys() + mock_ids.sort() + mock_ids.reverse() + + for mock_id in mock_ids: + module, function, impl = MOCK_STATE[mock_id] + module.__dict__[function] = impl + + MOCK_STATE.clear() + diff --git a/test/unit/util/system.py b/test/unit/util/system.py index e3d1225..4821073 100644 --- a/test/unit/util/system.py +++ b/test/unit/util/system.py @@ -7,6 +7,8 @@ system running the tests.
import functools import unittest +import test.mocking +import stem.util.proc import stem.util.system as system
# Base responses for the get_pid_by_name tests. The 'success' and @@ -74,9 +76,13 @@ class TestSystem(unittest.TestCase): Tests the stem.util.system contents. """
+ def setUp(self): + test.mocking.mock(stem.util.proc.is_available, test.mocking.return_false()) + test.mocking.mock(system.is_available, test.mocking.return_true()) + test.mocking.mock(system.call, test.mocking.return_none()) + def tearDown(self): - # resets call mocking back to being disabled - system.CALL_MOCKING = None + test.mocking.revert_mocking()
def test_is_running(self): """ @@ -87,7 +93,7 @@ class TestSystem(unittest.TestCase): running_commands = ["irssi", "moc", "tor", "ps", " firefox "]
for ps_cmd in (system.IS_RUNNING_PS_LINUX, system.IS_RUNNING_PS_BSD): - system.CALL_MOCKING = functools.partial(mock_call, ps_cmd, running_commands) + test.mocking.mock(system.call, functools.partial(mock_call, ps_cmd, running_commands))
self.assertTrue(system.is_running("irssi")) self.assertTrue(system.is_running("moc")) @@ -97,7 +103,7 @@ class TestSystem(unittest.TestCase): self.assertEqual(False, system.is_running("something_else"))
# mock both calls failing - system.CALL_MOCKING = lambda cmd: None + test.mocking.mock(system.call, test.mocking.return_none()) self.assertFalse(system.is_running("irssi")) self.assertEquals(None, system.is_running("irssi"))
@@ -109,7 +115,7 @@ class TestSystem(unittest.TestCase): responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses["success"] = ["1111"] responses["multiple_results"] = ["123", "456", "789"] - system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_NAME_PGREP, responses) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_NAME_PGREP, responses))
for test_input in responses: expected_response = 1111 if test_input == "success" else None @@ -123,7 +129,7 @@ class TestSystem(unittest.TestCase): responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses["success"] = ["1111"] responses["multiple_results"] = ["123 456 789"] - system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_NAME_PIDOF, responses) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_NAME_PIDOF, responses))
for test_input in responses: expected_response = 1111 if test_input == "success" else None @@ -134,10 +140,11 @@ class TestSystem(unittest.TestCase): Tests the get_pid_by_name function with the linux variant of ps. """
+ test.mocking.mock(system.is_bsd, test.mocking.return_false()) responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses["success"] = ["PID", " 1111"] responses["multiple_results"] = ["PID", " 123", " 456", " 789"] - system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_NAME_PS_LINUX, responses) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_NAME_PS_LINUX, responses))
for test_input in responses: expected_response = 1111 if test_input == "success" else None @@ -148,7 +155,8 @@ class TestSystem(unittest.TestCase): Tests the get_pid_by_name function with the bsd variant of ps. """
- system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_NAME_PS_BSD, GET_PID_BY_NAME_PS_BSD) + test.mocking.mock(system.is_bsd, test.mocking.return_true()) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_NAME_PS_BSD, GET_PID_BY_NAME_PS_BSD)) self.assertEquals(1, system.get_pid_by_name("launchd")) self.assertEquals(11, system.get_pid_by_name("DirectoryService")) self.assertEquals(None, system.get_pid_by_name("blarg")) @@ -161,7 +169,7 @@ class TestSystem(unittest.TestCase): responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses["success"] = ["1111"] responses["multiple_results"] = ["123", "456", "789"] - system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_NAME_LSOF, responses) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_NAME_LSOF, responses))
for test_input in responses: expected_response = 1111 if test_input == "success" else None @@ -172,7 +180,7 @@ class TestSystem(unittest.TestCase): Tests the get_pid_by_port function with a netstat response. """
- system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_PORT_NETSTAT, GET_PID_BY_PORT_NETSTAT_RESULTS) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_PORT_NETSTAT, GET_PID_BY_PORT_NETSTAT_RESULTS)) self.assertEquals(1641, system.get_pid_by_port(9051)) self.assertEquals(1641, system.get_pid_by_port("9051")) self.assertEquals(None, system.get_pid_by_port(631)) @@ -183,7 +191,7 @@ class TestSystem(unittest.TestCase): Tests the get_pid_by_port function with a sockstat response. """
- system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_PORT_SOCKSTAT % 9051, GET_PID_BY_PORT_SOCKSTAT_RESULTS) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_PORT_SOCKSTAT % 9051, GET_PID_BY_PORT_SOCKSTAT_RESULTS)) self.assertEquals(4397, system.get_pid_by_port(9051)) self.assertEquals(4397, system.get_pid_by_port("9051")) self.assertEquals(None, system.get_pid_by_port(123)) @@ -193,7 +201,7 @@ class TestSystem(unittest.TestCase): Tests the get_pid_by_port function with a lsof response. """
- system.CALL_MOCKING = functools.partial(mock_call, system.GET_PID_BY_PORT_LSOF, GET_PID_BY_PORT_LSOF_RESULTS) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_PID_BY_PORT_LSOF, GET_PID_BY_PORT_LSOF_RESULTS)) self.assertEquals(1745, system.get_pid_by_port(9051)) self.assertEquals(1745, system.get_pid_by_port("9051")) self.assertEquals(329, system.get_pid_by_port(80)) @@ -205,7 +213,7 @@ class TestSystem(unittest.TestCase): """
lsof_query = system.GET_PID_BY_FILE_LSOF % "/tmp/foo" - system.CALL_MOCKING = functools.partial(mock_call, lsof_query, ["4762"]) + test.mocking.mock(system.call, functools.partial(mock_call, lsof_query, ["4762"])) self.assertEquals(4762, system.get_pid_by_open_file("/tmp/foo")) self.assertEquals(None, system.get_pid_by_open_file("/tmp/somewhere_else"))
@@ -221,7 +229,7 @@ class TestSystem(unittest.TestCase): "7878": None, }
- system.CALL_MOCKING = functools.partial(mock_call, system.GET_CWD_PWDX, responses) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_CWD_PWDX, responses))
for test_input in responses: expected_response = "/home/atagar" if test_input == "3799" else None @@ -238,7 +246,7 @@ class TestSystem(unittest.TestCase): "7878": None, }
- system.CALL_MOCKING = functools.partial(mock_call, system.GET_CWD_LSOF, responses) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_CWD_LSOF, responses))
for test_input in responses: expected_response = "/Users/atagar/tor/src/or" if test_input == "75717" else None @@ -258,7 +266,7 @@ class TestSystem(unittest.TestCase): "6666": None }
- system.CALL_MOCKING = functools.partial(mock_call, system.GET_BSD_JAIL_ID_PS, responses) + test.mocking.mock(system.call, functools.partial(mock_call, system.GET_BSD_JAIL_ID_PS, responses))
for test_input in responses: expected_response = 1 if test_input == "1111" else 0