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