test/unit/connections.py

   1 #   test.connections - tests for connection manager
   2 #
   3 #   Part of WiFi Radar: A utility for managing WiFi profiles on GNU/Linux.
   4 #
   5 #   Copyright (C) 2014 Sean Robinson <robinson@tuxfamily.org>
   6 #
   7 #   This program is free software; you can redistribute it and/or modify
   8 #   it under the terms of the GNU General Public License as published by
   9 #   the Free Software Foundation; version 2 of the License.
  10 #
  11 #   This program is distributed in the hope that it will be useful,
  12 #   but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 #   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 #   GNU General Public License in LICENSE.GPL for more details.
  15 #
  16 #   You should have received a copy of the GNU General Public License
  17 #   along with this program; if not, write to:
  18 #      Free Software Foundation, Inc.
  19 #      51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
  20 #
  21
  22
  23 from multiprocessing import Pipe
  24 from signal import SIGTERM
  25 from subprocess import CalledProcessError
  26 from threading import Thread
  27 import unittest
  28
  29 import mock
  30
  31 import wifiradar.connections
  32 from wifiradar.config import ConfigManager
  33 from wifiradar.misc import DeviceError, get_new_profile
  34 from wifiradar.pubsub import Message
  35
  36
  37 class TestScanner(unittest.TestCase):
  38     @mock.patch('wifiradar.connections.Popen')
  39     def test_scanner(self, mock_popen):
  40         """ Test scanner function. """
  41         # Prepare the profile information to test against mock scan data.
  42         profiles = list()
  43         profile = get_new_profile()
  44         profile.update({'protocol': 'bg', 'bssid': '10:15:54:9E:2E:AA',
  45             'encrypted': True, 'signal': '-58', 'essid': 'startingenlivening',
  46             'channel': '2', 'mode': 'Master'})
  47         profiles.append(profile)
  48         profile = get_new_profile()
  49         profile.update({'protocol': 'bg', 'bssid': 'B2:B0:03:1A:8F:6B',
  50             'encrypted': False, 'signal': '-55', 'essid': 'filtersgenuinely',
  51             'channel': '3', 'mode': 'Master'})
  52         profiles.append(profile)
  53         profile = get_new_profile()
  54         profile.update({'protocol': 'bg', 'bssid': '10:A8:08:4C:18:26',
  55             'encrypted': False, 'signal': '-54', 'essid': 'tractorsenslave',
  56             'channel': '11', 'mode': 'Managed'})
  57         profiles.append(profile)
  58
  59         # Read mock scan data from file.
  60         mock_popen.return_value.stdout = open('./test/data/mock-scan-001.txt', 'r')
  61
  62         # Create control Pipe
  63         a, b = Pipe()
  64         test_thread = Thread(None, wifiradar.connections.scanner, args=(mock.Mock(), b))
  65         test_thread.start()
  66         for profile in profiles:
  67             # assert on poll() ensures a timeout if any expected msgs are not sent.
  68             self.assertTrue(a.poll(5))
  69             msg = a.recv()
  70             self.assertEqual('ACCESSPOINT', msg.topic)
  71             self.assertEqual(profile, msg.details)
  72         a.send(Message('EXIT', ''))
  73         test_thread.join()
  74
  75
  76 class TestConnectionManager(unittest.TestCase):
  77     def setUp(self):
  78         self.conn_manager = wifiradar.connections.ConnectionManager(mock.Mock())
  79         self.mock_config = mock.Mock()
  80         self.conn_manager.config = self.mock_config
  81
  82     def test_set_config(self):
  83         """ Test set_config method. """
  84         self.assertRaises(TypeError, self.conn_manager.set_config, None)
  85         self.assertEqual(None, self.conn_manager.set_config(ConfigManager({})))
  86
  87     @mock.patch('wifiradar.connections.shellcmd')
  88     def test__run_script(self, mock_shellcmd):
  89         """ Test _run_script method. """
  90         script_name = 'con_script'
  91         device = 'wlan0'
  92         ip = '192.168.1.1'
  93         essid = 'WinterPalace'
  94         bssid = '01:02:03:04:05:06'
  95         profile = {
  96             'con_script': 'mysterious_bash_command',
  97             'essid': essid,
  98             'bssid': bssid,
  99             'use_wpa': False,
 100             'key': '',
 101             'security': ''}
 102         self.conn_manager.get_current_ip = mock.Mock(return_value=ip)
 103         self.conn_manager.get_current_essid = mock.Mock(return_value=essid)
 104         self.conn_manager.get_current_bssid = mock.Mock(return_value=bssid)
 105
 106         self.assertEqual(None,
 107             self.conn_manager._run_script(script_name, profile, device))
 108
 109         custom_env = {
 110             "WIFIRADAR_IP": ip,
 111             "WIFIRADAR_ESSID": essid,
 112             "WIFIRADAR_BSSID": bssid,
 113             "WIFIRADAR_PROFILE": '{}:{}'.format(essid, bssid),
 114             "WIFIRADAR_ENCMODE": 'none',
 115             "WIFIRADAR_SECMODE": '',
 116             "WIFIRADAR_IF": device}
 117         mock_shellcmd.assert_called_with(['mysterious_bash_command'],
 118             custom_env)
 119
 120     @mock.patch('wifiradar.connections.shellcmd')
 121     def test__prepare_nic(self, mock_shellcmd):
 122         """ Test _prepare_nic method. """
 123         self.mock_config.get_opt.return_value = 'mock_iwconfig'
 124         self.mock_config.get_opt_as_bool.return_value = True
 125         essid = 'WinterPalace'
 126         bssid = '01:02:03:04:05:06'
 127         profile = {
 128             'essid': essid,
 129             'bssid': bssid,
 130             'key': '',
 131             'mode': 'managed',
 132             'channel': 11}
 133         device = 'wlan0'
 134         self.assertEqual(None,
 135             self.conn_manager._prepare_nic(profile, device))
 136         mock_shellcmd.assert_called_with(['mock_iwconfig', device,
 137             'essid', "'{}'".format(essid), 'key', 'off',
 138             'mode', 'managed', 'channel', 11, 'ap', bssid, 'commit'])
 139
 140         # Test failure.
 141         self.mock_config.get_opt.return_value = 'mock_iwconfig'
 142         self.mock_config.get_opt_as_bool.return_value = True
 143         mock_shellcmd.side_effect = CalledProcessError(250, 'mock_iwconfig')
 144         self.assertRaises(DeviceError,
 145             self.conn_manager._prepare_nic, profile, device)
 146
 147     @mock.patch('__builtin__.open', )
 148     @mock.patch('os.kill')
 149     @mock.patch('os.access')
 150     @mock.patch('wifiradar.connections.shellcmd')
 151     def test__stop_dhcp(self, mock_shellcmd, mock_access, mock_kill, mock_open):
 152         """ Test _stop_dhcp method. """
 153         dhcp_cmd = 'dhcpcd'
 154         kill_args = '-k -d'
 155         kill_args_list = ['-k', '-d']
 156         self.mock_config.get_opt.side_effect = ['mock_pidfile', kill_args,
 157             dhcp_cmd, kill_args]
 158         mock_access.return_value = True
 159         device = 'wlan0'
 160
 161         self.assertEqual(None, self.conn_manager._stop_dhcp(device))
 162         call_list = [dhcp_cmd]
 163         call_list.extend(kill_args_list)
 164         call_list.append(device)
 165         mock_shellcmd.assert_called_with(call_list)
 166
 167         self.mock_config.get_opt.side_effect = ['mock_pidfile', '',
 168             dhcp_cmd, '', 'mock_pidfile']
 169         PID = 42
 170         mock_open.return_value.readline.return_value = PID
 171         self.assertEqual(None, self.conn_manager._stop_dhcp(device))
 172         mock_kill.assert_called_with(PID, SIGTERM)
 173
 174     @mock.patch('wifiradar.connections.Popen')
 175     def test__start_dhcp(self, mock_popen):
 176         """ Test _start_dhcp method. """
 177         dhcp_cmd = 'dhcpcd'
 178         dhcp_args = '-t %(timeout)s -w'
 179         dhcp_args_list = ['-t', '%(timeout)s', '-w']
 180         dhcp_timeout = 1
 181         device = 'wlan0'
 182         self.mock_config.get_opt.side_effect = [dhcp_cmd, dhcp_args]
 183         self.mock_config.get_opt_as_int.return_value = dhcp_timeout
 184         self.conn_manager.get_current_ip = mock.Mock(return_value='0.0.0.0')
 185         self.assertEqual(None, self.conn_manager._start_dhcp(device))
 186         call_list = [dhcp_cmd]
 187         call_list.extend(dhcp_args_list)
 188         call_list.append(device)
 189         mock_popen.assert_called_with(call_list, stderr=None, stdout=None)
 190         # Test missing DHCP command.
 191         self.mock_config.get_opt.side_effect = [dhcp_cmd, dhcp_args]
 192         mock_popen.side_effect = OSError(2, 'No such file or directory')
 193         self.assertEqual(None, self.conn_manager._start_dhcp(device))
 194
 195     @mock.patch('__builtin__.open', )
 196     @mock.patch('os.kill')
 197     @mock.patch('os.access')
 198     @mock.patch('wifiradar.connections.shellcmd')
 199     def test__stop_wpa(self, mock_shellcmd, mock_access, mock_kill, mock_open):
 200         """ Test _stop_wpa method. """
 201         kill_cmd = ''
 202         self.mock_config.get_opt.side_effect = ['mock_pidfile', kill_cmd,
 203             'mock_pidfile']
 204         mock_access.return_value = True
 205
 206         PID = 42
 207         mock_open.return_value.readline.return_value = PID
 208         self.assertEqual(None, self.conn_manager._stop_wpa())
 209         mock_kill.assert_called_with(PID, SIGTERM)
 210
 211     @mock.patch('wifiradar.connections.shellcmd')
 212     def test__start_wpa(self, mock_shellcmd):
 213         """ Test _start_wpa method. """
 214         wpa_cmd = 'mock_wpa'
 215         wpa_args = '-B -i %(interface)s -c %(configuration)s'
 216         wpa_args_list = ['-B', '-i', '%(interface)s', '-c', '%(configuration)s']
 217         self.mock_config.get_opt.side_effect = [wpa_cmd, wpa_args]
 218         self.assertEqual(None,
 219             self.conn_manager._start_wpa())
 220         call_list = [wpa_cmd]
 221         call_list.extend(wpa_args_list)
 222         mock_shellcmd.assert_called_with(call_list)
 223
 224     @mock.patch('__builtin__.open', )
 225     @mock.patch('wifiradar.connections.shellcmd')
 226     def test__start_manual_network(self, mock_shellcmd, mock_open):
 227         """ Test _start_manual_network method. """
 228         ifconfig = 'mock_ifconfig'
 229         route = 'mock_route'
 230         ip = '192.168.1.100'
 231         netmask = '192.168.1.255'
 232         gateway = '192.168.1.1'
 233         domain = 'winterpalace.org'
 234         dns1 = '192.168.1.200'
 235         dns2 = '192.168.1.201'
 236         profile = {
 237             'ip': ip,
 238             'netmask': netmask,
 239             'gateway': gateway,
 240             'domain': domain,
 241             'dns1': dns1,
 242             'dns2': dns2}
 243         device = 'wlan0'
 244         self.mock_config.get_opt.side_effect = [ifconfig, ifconfig, route]
 245
 246         self.assertEqual(None,
 247             self.conn_manager._start_manual_network(profile, device))
 248         mock_shellcmd.assert_any_call([ifconfig, device, 'down'])
 249         mock_shellcmd.assert_any_call([ifconfig, device, ip,
 250             'netmask', netmask])
 251         mock_shellcmd.assert_any_call([route, 'add', 'default',
 252             'gw', gateway])
 253         # Test mock-writing the resolv file.
 254         mock_open.assert_called_with('/etc/resolv.conf', 'w')
 255         # Check that resolv_file.write was called with the proper args.
 256         mock_open.return_value.__enter__.return_value.write.assert_called_with(
 257             'domain {}\nnameserver {}\nnameserver {}\n'.format(domain, dns1, dns2))
 258
 259     def test_connect(self):
 260         """ Test connect method. """
 261         profile = {
 262             'essid': 'WinterPalace',
 263             'bssid': '01:02:03:04:05:06',
 264             'use_wpa': True,
 265             'use_dhcp': True}
 266         device = 'wlan0'
 267         self.mock_config.get_network_device.return_value = device
 268
 269         self.conn_manager._run_script = mock.Mock(name='_run_script')
 270         self.conn_manager._prepare_nic = mock.Mock(name='_prepare_nic')
 271         self.conn_manager._stop_dhcp = mock.Mock(name='_stop_dhcp')
 272         self.conn_manager._start_dhcp = mock.Mock(name='_start_dhcp')
 273         self.conn_manager._stop_wpa = mock.Mock(name='_stop_wpa')
 274         self.conn_manager._start_wpa = mock.Mock(name='_start_wpa')
 275         self.conn_manager._start_manual_network = mock.Mock(
 276             name='_start_manual_network')
 277
 278         # Test DHCP and WPA network configuration.
 279         self.assertEqual(None, self.conn_manager.connect(profile))
 280         self.conn_manager._run_script.assert_any_call('con_prescript',
 281             profile, device)
 282         self.conn_manager._run_script.assert_any_call('con_postscript',
 283             profile, device)
 284         self.conn_manager._prepare_nic.assert_called_with(profile, device)
 285         self.conn_manager._stop_dhcp.assert_called_with(device)
 286         self.conn_manager._start_dhcp.assert_called_with(device)
 287         self.conn_manager._stop_wpa.assert_called_with()
 288         self.conn_manager._start_wpa.assert_called_with()
 289
 290         # Test manual network configuration.
 291         profile['use_dhcp'] = False
 292         self.assertEqual(None, self.conn_manager.connect(profile))
 293         self.conn_manager._start_manual_network.assert_called_with(profile, device)
 294
 295     @mock.patch('wifiradar.connections.shellcmd')
 296     def test_disconnect(self, mock_shellcmd):
 297         """ Test disconnect method. """
 298         iwconfig = 'mock_iwconfig'
 299         ifconfig = 'mock_ifconfig'
 300         profile = {}
 301         device = 'wlan0'
 302         self.mock_config.get_network_device.return_value = device
 303         self.mock_config.get_opt.side_effect = [iwconfig, ifconfig]
 304         self.conn_manager._run_script = mock.Mock(name='_run_script')
 305         self.conn_manager._stop_dhcp = mock.Mock(name='_stop_dhcp')
 306         self.conn_manager._stop_wpa = mock.Mock(name='_stop_wpa')
 307         self.conn_manager.if_change = mock.Mock(name='if_change')
 308
 309         self.assertEqual(None, self.conn_manager.disconnect(profile))
 310         self.conn_manager._run_script.assert_any_call('dis_prescript',
 311             profile, device)
 312         self.conn_manager._run_script.assert_any_call('dis_postscript',
 313             profile, device)
 314         self.conn_manager._stop_dhcp.assert_called_with(device)
 315         self.conn_manager._stop_wpa.assert_called_with()
 316
 317         mock_shellcmd.assert_any_call([iwconfig, device,
 318             'essid', 'any', 'key', 'off', 'mode', 'managed',
 319             'channel', 'auto', 'ap', 'off'])
 320         mock_shellcmd.assert_any_call([ifconfig, device, '0.0.0.0'])
 321
 322     @mock.patch('wifiradar.connections.Popen')
 323     def test_if_change(self, mock_popen):
 324         """ Test if_change method. """
 325         self.mock_config.get_opt.return_value = 'mock_ifconfig'
 326         self.mock_config.get_network_device.return_value = 'wlan0'
 327
 328         # Test normal operation.
 329         self.assertEqual(None, self.conn_manager.if_change('up'))
 330         self.assertEqual(None, self.conn_manager.if_change('down'))
 331         self.assertEqual(None, self.conn_manager.if_change('UP'))
 332         self.assertEqual(None, self.conn_manager.if_change('DOWN'))
 333
 334         # Test unknown state.
 335         self.assertRaises(ValueError, self.conn_manager.if_change, 'sideways')
 336
 337         # Test failed state change with found ifconfig.
 338         mock_popen_instance = mock.MagicMock()
 339         mock_popen_instance.stdout.__iter__.return_value = [
 340             'wlan0: ERROR while getting interface flags: No such device']
 341         mock_popen.return_value = mock_popen_instance
 342         self.assertRaises(DeviceError, self.conn_manager.if_change, 'up')
 343
 344         # Test OSError.
 345         mock_popen.side_effect = OSError(2, 'No such file or directory')
 346         self.assertRaises(OSError, self.conn_manager.if_change, 'up')
 347
 348     @mock.patch('wifiradar.connections.Popen')
 349     def test_get_current_ip(self, mock_popen):
 350         """ Test get_current_ip method. """
 351         self.mock_config.get_opt.return_value = 'mock_ifconfig'
 352         self.mock_config.get_network_device.return_value = 'wlan0'
 353
 354         # Test connected and with an IP address.
 355         mock_popen_instance = mock.Mock()
 356         mock_popen_instance.stdout.read.return_value = \
 357             'wlan0: flags=4098<BROADCAST,MULTICAST>  mtu 1500\n' + \
 358             'inet addr:192.168.1.1  netmask 255.255.255.0  broadcast 192.168.1.255\n' + \
 359             'ether 00:00:00:00:00:00  txqueuelen 1000  (Ethernet)\n' + \
 360             'RX packets 246  bytes 259631 (253.5 KiB)\n' + \
 361             'RX errors 0  dropped 1  overruns 0  frame 0\n' + \
 362             'TX packets 123  bytes 140216 (136.9 KiB)\n' + \
 363             'TX errors 0  dropped 0 overruns 0  carrier 0  collisions 0\n'
 364         mock_popen.return_value = mock_popen_instance
 365         self.assertEqual('192.168.1.1', self.conn_manager.get_current_ip())
 366
 367         # Test not connected and without an IP address.
 368         mock_popen_instance.stdout.read.return_value = \
 369             'wlan0: flags=4098<BROADCAST,MULTICAST>  mtu 1500\n' + \
 370             'ether 00:00:00:00:00:00  txqueuelen 1000  (Ethernet)\n' + \
 371             'RX packets 246  bytes 259631 (253.5 KiB)\n' + \
 372             'RX errors 0  dropped 1  overruns 0  frame 0\n' + \
 373             'TX packets 123  bytes 140216 (136.9 KiB)\n' + \
 374             'TX errors 0  dropped 0 overruns 0  carrier 0  collisions 0\n'
 375         mock_popen.return_value = mock_popen_instance
 376         self.assertEqual(None, self.conn_manager.get_current_ip())
 377
 378     @mock.patch('wifiradar.connections.Popen')
 379     def test_get_current_essid(self, mock_popen):
 380         """ Test get_current_essid method. """
 381         self.mock_config.get_opt.return_value = 'mock_iwconfig'
 382         self.mock_config.get_network_device.return_value = 'wlan0'
 383
 384         # Test associated interface.
 385         mock_popen_instance = mock.Mock()
 386         mock_popen_instance.stdout.read.return_value = \
 387             'wlan0     IEEE 802.11abgn  ESSID:"WinterPalace"\n' + \
 388             '          Mode:Managed  Frequency:5.26 GHz  Access Point: 00:00:00:00:00:00\n' + \
 389             '          Tx-Power=15 dBm\n' + \
 390             '          Retry  long limit:7   RTS thr:off   Fragment thr:off\n' + \
 391             '          Power Management:off\n'
 392         mock_popen.return_value = mock_popen_instance
 393         self.assertEqual('WinterPalace', self.conn_manager.get_current_essid())
 394
 395         # Test not associated interface.
 396         mock_popen_instance = mock.Mock()
 397         mock_popen_instance.stdout.read.return_value = \
 398             'wlan0     IEEE 802.11abgn  ESSID:off/any\n' + \
 399             '          Mode:Managed  Frequency:5.26 GHz  Access Point: Not-Associated\n' + \
 400             '          Tx-Power=15 dBm\n' + \
 401             '          Retry  long limit:7   RTS thr:off   Fragment thr:off\n' + \
 402             '          Power Management:off\n'
 403         mock_popen.return_value = mock_popen_instance
 404         self.assertEqual(None, self.conn_manager.get_current_essid())
 405
 406     @mock.patch('wifiradar.connections.Popen')
 407     def test_get_current_bssid(self, mock_popen):
 408         """ Test get_current_bssid method. """
 409         self.mock_config.get_opt.return_value = 'mock_iwconfig'
 410         self.mock_config.get_network_device.return_value = 'wlan0'
 411
 412         # Test associated interface.
 413         mock_popen_instance = mock.Mock()
 414         mock_popen_instance.stdout.read.return_value = \
 415             'wlan0     IEEE 802.11abgn  ESSID:"WinterPalace"\n' + \
 416             '          Mode:Managed  Frequency:5.26 GHz  Access Point: 00:00:00:00:00:00\n' + \
 417             '          Tx-Power=15 dBm\n' + \
 418             '          Retry  long limit:7   RTS thr:off   Fragment thr:off\n' + \
 419             '          Power Management:off\n'
 420
 421         mock_popen.return_value = mock_popen_instance
 422         self.assertEqual('00:00:00:00:00:00', self.conn_manager.get_current_bssid())
 423
 424         # Test not associated interface.
 425         mock_popen_instance = mock.Mock()
 426         mock_popen_instance.stdout.read.return_value = \
 427             'wlan0     IEEE 802.11abgn  ESSID:off/any\n' + \
 428             '          Mode:Managed  Frequency:5.26 GHz  Access Point: Not-Associated\n' + \
 429             '          Tx-Power=15 dBm\n' + \
 430             '          Retry  long limit:7   RTS thr:off   Fragment thr:off\n' + \
 431             '          Power Management:off\n'
 432         mock_popen.return_value = mock_popen_instance
 433         self.assertEqual(None, self.conn_manager.get_current_bssid())