commit 8e90c1029e650536cf07ee42ddeb06300f0aa002 Author: Damian Johnson atagar@torproject.org Date: Tue Jun 11 09:09:05 2013 -0700
Using mock for proc unit tests
Now that I have a rough understanding of the open mock it's quite a bit nicer than our prior version (which got especially icky for python 3 compatability). --- test/unit/util/proc.py | 104 ++++++++++++++++++++++++------------------------ 1 file changed, 51 insertions(+), 53 deletions(-)
diff --git a/test/unit/util/proc.py b/test/unit/util/proc.py index 0a6c0a0..4ee598b 100644 --- a/test/unit/util/proc.py +++ b/test/unit/util/proc.py @@ -2,80 +2,84 @@ Unit testing code for the stem.util.proc functions. """
-import os import StringIO import unittest
-import stem.prereq +from mock import Mock, patch
from stem.util import proc from test import mocking
class TestProc(unittest.TestCase): - def tearDown(self): - mocking.revert_mocking() - - def test_get_system_start_time(self): + @patch('stem.util.proc._get_line') + def test_get_system_start_time(self, get_line_mock): """ Tests the get_system_start_time function. """
- mocking.mock(proc._get_line, mocking.return_for_args({ + get_line_mock.side_effect = lambda *params: { ('/proc/stat', 'btime', 'system start time'): 'btime 1001001', - })) + }[params]
self.assertEquals(1001001, proc.get_system_start_time())
- def test_get_physical_memory(self): + @patch('stem.util.proc._get_line') + def test_get_physical_memory(self, get_line_mock): """ Tests the get_physical_memory function. """
- mocking.mock(proc._get_line, mocking.return_for_args({ + get_line_mock.side_effect = lambda *params: { ('/proc/meminfo', 'MemTotal:', 'system physical memory'): 'MemTotal: 12345 kB', - })) + }[params]
self.assertEquals((12345 * 1024), proc.get_physical_memory())
- def test_get_cwd(self): + @patch('os.readlink') + def test_get_cwd(self, readlink_mock): """ Tests the get_cwd function with a given pid. """
- mocking.mock(os.readlink, mocking.return_for_args({ - ('/proc/24019/cwd',): '/home/directory/TEST' - }), os) + readlink_mock.side_effect = lambda param: { + '/proc/24019/cwd': '/home/directory/TEST' + }[param]
self.assertEquals('/home/directory/TEST', proc.get_cwd(24019))
- def test_get_uid(self): + @patch('stem.util.proc._get_line') + def test_get_uid(self, get_line_mock): """ Tests the get_uid function with a given pid. """
for test_value in [(24019, 11111), (0, 22222)]: pid, uid = test_value - mocking.mock(proc._get_line, mocking.return_for_args({ - ("/proc/%s/status" % pid, 'Uid:', 'uid'): 'Uid: %s' % uid - })) + + get_line_mock.side_effect = lambda *params: { + ("/proc/%s/status" % pid, 'Uid:', 'uid'): 'Uid: %s' % uid, + }[params]
self.assertEquals(uid, proc.get_uid(pid))
- def test_get_memory_usage(self): + @patch('stem.util.proc._get_lines') + def test_get_memory_usage(self, get_lines_mock): """ Tests the get_memory_usage function with a given pid. """
- mocking.mock(proc._get_lines, mocking.return_for_args({ + get_lines_mock.side_effect = lambda *params: { ('/proc/1111/status', ('VmRSS:', 'VmSize:'), 'memory usage'): {'VmRSS:': 'VmRSS: 100 kB', 'VmSize:': 'VmSize: 1800 kB'} - })) + }[params]
self.assertEqual((0, 0), proc.get_memory_usage(0)) self.assertEqual((100 * 1024, 1800 * 1024), proc.get_memory_usage(1111))
- def test_get_stats(self): + @patch('stem.util.proc._get_line') + @patch('stem.util.proc.get_system_start_time', Mock(return_value = 10)) + def test_get_stats(self, get_line_mock): """ Tests get_stats() with all combinations of stat_type arguments. """ @@ -91,12 +95,11 @@ class TestProc(unittest.TestCase): stat_path = "/proc/24062/stat" stat = '1 (test_program) 2 3 4 5 6 7 8 9 10 11 12 13.0 14.0 15 16 17 18 19 20 21.0 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43'
- mocking.mock(proc.get_system_start_time, mocking.return_value(10)) - # tests the case where no stat_types are specified - mocking.mock(proc._get_line, mocking.return_for_args({ + + get_line_mock.side_effect = lambda *params: { (stat_path, '24062', 'process '): stat - })) + }[params]
self.assertEquals((), proc.get_stats(24062))
@@ -109,9 +112,9 @@ class TestProc(unittest.TestCase):
args, response = zip(*stats)
- mocking.mock(proc._get_line, mocking.return_for_args({ + get_line_mock.side_effect = lambda *params: { (stat_path, '24062', 'process %s' % ', '.join(args)): stat - })) + }[params]
self.assertEquals(response, proc.get_stats(24062, *args))
@@ -130,45 +133,40 @@ class TestProc(unittest.TestCase): elif arg == 'stime': response += ('0',)
- mocking.mock(proc._get_line, mocking.return_for_args({ + get_line_mock.side_effect = lambda *params: { ('/proc/0/stat', '0', 'process %s' % ', '.join(args)): stat - })) + }[params]
self.assertEquals(response, proc.get_stats(0, *args))
- def test_get_connections(self): + @patch('os.listdir') + @patch('os.readlink') + @patch('stem.util.proc.open', create = True) + def test_get_connections(self, open_mock, readlink_mock, listdir_mock): """ Tests the get_connections function. """
pid = 1111
- mocking.mock(os.listdir, mocking.return_for_args({ - ('/proc/%s/fd' % pid,): ['1', '2', '3', '4'], - }), os) + listdir_mock.side_effect = lambda param: { + '/proc/%s/fd' % pid: ['1', '2', '3', '4'], + }[param]
- mocking.mock(os.readlink, mocking.return_for_args({ - ('/proc/%s/fd/1' % pid,): 'socket:[99999999]', - ('/proc/%s/fd/2' % pid,): 'socket:[IIIIIIII]', - ('/proc/%s/fd/3' % pid,): 'pipe:[30303]', - ('/proc/%s/fd/4' % pid,): 'pipe:[40404]', - }), os) + readlink_mock.side_effect = lambda param: { + '/proc/%s/fd/1' % pid: 'socket:[99999999]', + '/proc/%s/fd/2' % pid: 'socket:[IIIIIIII]', + '/proc/%s/fd/3' % pid: 'pipe:[30303]', + '/proc/%s/fd/4' % pid: 'pipe:[40404]', + }[param]
tcp = '\n 0: 11111111:1111 22222222:2222 01 44444444:44444444 55:55555555 66666666 1111 8 99999999' udp = '\n A: BBBBBBBB:BBBB CCCCCCCC:CCCC DD EEEEEEEE:EEEEEEEE FF:FFFFFFFF GGGGGGGG 1111 H IIIIIIII'
- if stem.prereq.is_python_3(): - import builtins - - mocking.mock(builtins.open, mocking.return_for_args({ - ('/proc/net/tcp',): StringIO.StringIO(tcp), - ('/proc/net/udp',): StringIO.StringIO(udp) - }), builtins) - else: - mocking.mock(open, mocking.return_for_args({ - ('/proc/net/tcp',): StringIO.StringIO(tcp), - ('/proc/net/udp',): StringIO.StringIO(udp) - })) + open_mock.side_effect = lambda param: { + '/proc/net/tcp': StringIO.StringIO(tcp), + '/proc/net/udp': StringIO.StringIO(udp) + }[param]
# tests the edge case of pid = 0 self.assertEquals([], proc.get_connections(0))