[tor-commits] [stem/master] Mocking utility functions
atagar at torproject.org
atagar at torproject.org
Wed Jan 11 18:21:13 UTC 2012
commit 11cbf61c2dca7cde628e5eec4f3dd4da7867860f
Author: Damian Johnson <atagar at 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
More information about the tor-commits
mailing list