""" Unit tests for the stem.util.system functions. This works by mocking the stem.util.system.call function to selectively exercise other functions. None of these tests actually make system calls, use proc, or otherwise deal with the system running the tests. """ import functools import ntpath import posixpath import unittest from stem.util import system try: # added in python 3.3 from unittest.mock import Mock, patch except ImportError: from mock import Mock, patch # Base responses for the get_pid_by_name tests. The 'success' and # 'multiple_results' entries are filled in by tests. GET_PID_BY_NAME_BASE_RESULTS = { 'success': [], 'multiple_results': [], 'malformed_data': ['bad data'], 'no_results': [], 'command_fails': None, } # testing output for system calls GET_PID_BY_NAME_PS_BSD = [ ' PID TT STAT TIME COMMAND', ' 1 ?? Ss 9:00.22 launchd', ' 10 ?? Ss 0:09.97 kextd', ' 11 ?? Ss 5:47.36 DirectoryService', ' 12 ?? Ss 3:01.44 notifyd'] GET_PID_BY_NAME_PS_BSD_MULTIPLE = [ ' PID TT STAT TIME COMMAND', ' 1 ?? Ss 9:00.22 launchd', ' 10 ?? Ss 0:09.97 kextd', ' 41 ?? Ss 9:00.22 launchd'] GET_PID_BY_PORT_NETSTAT_RESULTS = [ 'Active Internet connections (only servers)', 'Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name', 'tcp 0 0 127.0.0.1:631 0.0.0.0:* LISTEN - ', 'tcp 0 0 127.0.0.1:9051 0.0.0.0:* LISTEN 1641/tor ', 'tcp6 0 0 ::1:631 :::* LISTEN - ', 'udp 0 0 0.0.0.0:5353 0.0.0.0:* - ', 'udp6 0 0 fe80::7ae4:ff:fe2f::123 :::* - '] GET_PID_BY_PORT_SOCKSTAT_RESULTS = [ '_tor tor 4397 7 tcp4 51.64.7.84:9051 *:*', '_tor tor 4397 12 tcp4 51.64.7.84:54011 80.3.121.7:9051', '_tor tor 4397 15 tcp4 51.64.7.84:59374 7.42.1.102:9051'] GET_PID_BY_PORT_LSOF_RESULTS = [ 'COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME', 'tor 1745 atagar 6u IPv4 14229 0t0 TCP 127.0.0.1:9051 (LISTEN)', 'apache 329 atagar 6u IPv4 14229 0t0 TCP 127.0.0.1:80 (LISTEN)'] GET_BSD_JAIL_PATH_RESULTS = [ ' JID IP Address Hostname Path', ' 1 10.0.0.2 tor-jail /usr/jails/tor-jail', ] def mock_call(base_cmd, responses): """ Provides mocking for the system module's call function. There are a couple ways of using this... - Simple usage is for base_cmd is the system call we want to respond to and responses is a list containing the respnose. For instance... mock_call('ls my_dir', ['file1', 'file2', 'file3']) - The base_cmd can be a formatted string and responses are a dictionary of completions for tat string to the responses. For instance... mock_call('ls %s', {'dir1': ['file1', 'file2'], 'dir2': ['file3', 'file4']}) :param str base_cmd: command to match against :param list,dict responses: either list with the response, or mapping of base_cmd formatted string completions to responses :returns: **functor** to override stem.util.system.call with """ def _mock_call(base_cmd, responses, command, default = None): if isinstance(responses, list): if base_cmd == command: return responses else: return default else: for cmd_completion in responses: if command == base_cmd % cmd_completion: return responses[cmd_completion] return default return functools.partial(_mock_call, base_cmd, responses) class TestSystem(unittest.TestCase): @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_is_running(self, call_mock): """ Exercises multiple use cases for the is_running function. """ # mock response with a linux and bsd resolver running_commands = [u'irssi', u'moc', u'tor', u'ps', u' firefox '] for ps_cmd in (system.IS_RUNNING_PS_LINUX, system.IS_RUNNING_PS_BSD): call_mock.side_effect = mock_call(ps_cmd, running_commands) self.assertTrue(system.is_running('irssi')) self.assertTrue(system.is_running('moc')) self.assertTrue(system.is_running('tor')) self.assertTrue(system.is_running('ps')) self.assertTrue(system.is_running('firefox')) self.assertEqual(False, system.is_running('something_else')) # mock both calls failing call_mock.return_value = None call_mock.side_effect = None self.assertFalse(system.is_running('irssi')) self.assertEquals(None, system.is_running('irssi')) @patch('stem.util.system.call') @patch('stem.util.proc.is_available', Mock(return_value = False)) @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_name_by_pid_ps(self, call_mock): """ Tests the get_name_by_pid function with ps responses. """ responses = { 'success': ['COMMAND', 'vim'], 'malformed_command_1': ['COMMAND'], 'malformed_command_2': ['foobar'], 'malformed_command_3': ['NOT_COMMAND', 'vim'], 'no_results': [], 'command_fails': None, } call_mock.side_effect = mock_call(system.GET_NAME_BY_PID_PS, responses) for test_input in responses: expected_response = 'vim' if test_input == 'success' else None self.assertEquals(expected_response, system.get_name_by_pid(test_input)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_name_pgrep(self, call_mock): """ Tests the get_pid_by_name function with pgrep responses. """ responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses['success'] = ['1111'] responses['multiple_results'] = ['123', '456', '789'] call_mock.side_effect = mock_call(system.GET_PID_BY_NAME_PGREP, responses) for test_input in responses: expected_response = 1111 if test_input == 'success' else None self.assertEquals(expected_response, system.get_pid_by_name(test_input)) self.assertEquals([123, 456, 789], system.get_pid_by_name('multiple_results', multiple = True)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_name_pidof(self, call_mock): """ Tests the get_pid_by_name function with pidof responses. """ responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses['success'] = ['1111'] responses['multiple_results'] = ['123 456 789'] call_mock.side_effect = mock_call(system.GET_PID_BY_NAME_PIDOF, responses) for test_input in responses: expected_response = 1111 if test_input == 'success' else None self.assertEquals(expected_response, system.get_pid_by_name(test_input)) self.assertEquals([123, 456, 789], system.get_pid_by_name('multiple_results', multiple = True)) @patch('stem.util.system.call') @patch('stem.util.system.is_bsd', Mock(return_value = False)) @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_name_ps_linux(self, call_mock): """ Tests the get_pid_by_name function with the linux variant of ps. """ responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses['success'] = ['PID', ' 1111'] responses['multiple_results'] = ['PID', ' 123', ' 456', ' 789'] call_mock.side_effect = mock_call(system.GET_PID_BY_NAME_PS_LINUX, responses) for test_input in responses: expected_response = 1111 if test_input == 'success' else None self.assertEquals(expected_response, system.get_pid_by_name(test_input)) self.assertEquals([123, 456, 789], system.get_pid_by_name('multiple_results', multiple = True)) @patch('stem.util.system.call') @patch('stem.util.system.is_bsd', Mock(return_value = True)) @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_name_ps_bsd(self, call_mock): """ Tests the get_pid_by_name function with the bsd variant of ps. """ call_mock.side_effect = 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')) call_mock.side_effect = mock_call(system.GET_PID_BY_NAME_PS_BSD, GET_PID_BY_NAME_PS_BSD_MULTIPLE) self.assertEquals([1, 41], system.get_pid_by_name('launchd', multiple = True)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_name_lsof(self, call_mock): """ Tests the get_pid_by_name function with lsof responses. """ responses = dict(GET_PID_BY_NAME_BASE_RESULTS) responses['success'] = ['1111'] responses['multiple_results'] = ['123', '456', '789'] call_mock.side_effect = mock_call(system.GET_PID_BY_NAME_LSOF, responses) for test_input in responses: expected_response = 1111 if test_input == 'success' else None self.assertEquals(expected_response, system.get_pid_by_name(test_input)) self.assertEquals([123, 456, 789], system.get_pid_by_name('multiple_results', multiple = True)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_port_netstat(self, call_mock): """ Tests the get_pid_by_port function with a netstat response. """ call_mock.side_effect = 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)) self.assertEquals(None, system.get_pid_by_port(123)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_port_sockstat(self, call_mock): """ Tests the get_pid_by_port function with a sockstat response. """ call_mock.side_effect = 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)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_port_lsof(self, call_mock): """ Tests the get_pid_by_port function with a lsof response. """ call_mock.side_effect = 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)) self.assertEquals(None, system.get_pid_by_port(123)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_pid_by_open_file_lsof(self, call_mock): """ Tests the get_pid_by_open_file function with a lsof response. """ lsof_query = system.GET_PID_BY_FILE_LSOF % '/tmp/foo' call_mock.side_effect = mock_call(lsof_query, ['4762']) self.assertEquals(4762, system.get_pid_by_open_file('/tmp/foo')) call_mock.return_value = [] call_mock.side_effect = None self.assertEquals(None, system.get_pid_by_open_file('/tmp/somewhere_else')) @patch('stem.util.system.call') @patch('stem.util.proc.is_available', Mock(return_value = False)) @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_cwd_pwdx(self, call_mock): """ Tests the get_cwd function with a pwdx response. """ responses = { '3799': ['3799: /home/atagar'], '5839': ['5839: No such process'], '1234': ['malformed output'], '7878': None, } call_mock.side_effect = mock_call(system.GET_CWD_PWDX, responses) for test_input in responses: expected_response = '/home/atagar' if test_input == '3799' else None self.assertEquals(expected_response, system.get_cwd(test_input)) @patch('stem.util.system.call') @patch('stem.util.proc.is_available', Mock(return_value = False)) @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_cwd_lsof(self, call_mock): """ Tests the get_cwd function with a lsof response. """ responses = { '75717': ['p75717', 'n/Users/atagar/tor/src/or'], '1234': ['malformed output'], '7878': [], } call_mock.side_effect = 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 self.assertEquals(expected_response, system.get_cwd(test_input)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_bsd_jail_id(self, call_mock): """ Tests the get_bsd_jail_id function. """ responses = { '1111': ['JID', ' 1'], '2222': ['JID', ' 0'], '3333': ['JID', 'bad data'], '4444': ['bad data'], '5555': [], '6666': [] } call_mock.side_effect = mock_call(system.GET_BSD_JAIL_ID_PS, responses) for test_input in responses: expected_response = 1 if test_input == '1111' else 0 self.assertEquals(expected_response, system.get_bsd_jail_id(test_input)) @patch('stem.util.system.call') @patch('stem.util.system.is_available', Mock(return_value = True)) def test_get_bsd_jail_path(self, call_mock): """ Tests the get_bsd_jail_path function. """ # check when we don't have a jail call_mock.return_value = [] self.assertEquals(None, system.get_bsd_jail_path(1)) call_mock.side_effect = mock_call(system.GET_BSD_JAIL_PATH % '1', GET_BSD_JAIL_PATH_RESULTS) self.assertEquals('/usr/jails/tor-jail', system.get_bsd_jail_path(1)) @patch('platform.system', Mock(return_value = 'Linux')) @patch('os.path.join', Mock(side_effect = posixpath.join)) def test_expand_path_unix(self): """ Tests the expand_path function. This does not exercise home directory expansions since that deals with our environment (that's left to integ tests). """ self.assertEquals('', system.expand_path('')) self.assertEquals('/tmp', system.expand_path('/tmp')) self.assertEquals('/tmp', system.expand_path('/tmp/')) self.assertEquals('/tmp', system.expand_path('.', '/tmp')) self.assertEquals('/tmp', system.expand_path('./', '/tmp')) self.assertEquals('/tmp/foo', system.expand_path('foo', '/tmp')) self.assertEquals('/tmp/foo', system.expand_path('./foo', '/tmp')) @patch('platform.system', Mock(return_value = 'Windows')) @patch('os.path.join', Mock(side_effect = ntpath.join)) def test_expand_path_windows(self): """ Tests the expand_path function on windows. This does not exercise home directory expansions since that deals with our environment (that's left to integ tests). """ self.assertEquals('', system.expand_path('')) self.assertEquals('C:\\tmp', system.expand_path('C:\\tmp')) self.assertEquals('C:\\tmp', system.expand_path('C:\\tmp\\')) self.assertEquals('C:\\tmp', system.expand_path('.', 'C:\\tmp')) self.assertEquals('C:\\tmp', system.expand_path('.\\', 'C:\\tmp')) self.assertEquals('C:\\tmp\\foo', system.expand_path('foo', 'C:\\tmp')) self.assertEquals('C:\\tmp\\foo', system.expand_path('.\\foo', 'C:\\tmp'))