Connection status

.travis.yml

@@ -6,6 +6,8 @@ python:
 # command to install dependencies
 before_install:
  - sudo apt-get install wireless-tools
+ - sudo pip install -U pip wheel setuptools
+ - sudo pip install -U mock
 install:
  - sudo python setup.py -q install
 # command to run tests

tests/test_parsing.py

@@ -68,6 +68,7 @@ def test_absolute_quality(self):
 class ScanningTest(TestCase):
  def test_scanning(self):
  self.assertRaises(InterfaceError, Cell.all, 'fake-interface')
+ self.assertRaises(InterfaceError, Cell.all, 'fake-interface', sudo=True)
 
 
 IWLIST_SCAN_NO_ENCRYPTION = """Cell 02 - Address: 38:83:45:CC:58:74
@@ -278,3 +279,4 @@ def test_scanning(self):
  Encryption key:off
  Bit Rates:144 Mb/s
 """
+

tests/test_scan.py

@@ -0,0 +1,22 @@
+from unittest import TestCase
+
+from wifi.scan import Cell
+from wifi import subprocess_compat as subprocess
+from mock import patch
+from test_parsing import IWLIST_SCAN_WEP, IWLIST_SCAN_WPA2
+
+
+expected_output = '\n'.join([IWLIST_SCAN_WEP, IWLIST_SCAN_WPA2])
+
+class ScanTest(TestCase):
+ def test_all_calls_check_output_with_good_args(self):
+ args = ['/sbin/iwlist', 'interface', 'scan']
+ kwargs = {'stderr':subprocess.STDOUT}
+ with patch.object(subprocess, 'check_output',
+ return_value=expected_output):
+ Cell.all('interface')
+ subprocess.check_output.assert_called_with(args, **kwargs)
+ args.insert(0, 'sudo')
+ Cell.all('interface', sudo=True)
+ subprocess.check_output.assert_called_with(args, **kwargs)
+

tests/test_schemes.py

@@ -5,6 +5,14 @@
 from wifi import Cell
 from wifi.scheme import extract_schemes, Scheme
 from wifi.exceptions import ConnectionError
+from wifi import subprocess_compat as subprocess
+from mock import patch
+from wifi.utils import MyStringIO
+
+try:
+ from io import StringIO
+except ImportError: # Python < 3
+ from StringIO import StringIO
 
 
 NETWORK_INTERFACES_FILE = """
@@ -87,18 +95,35 @@ def test_save(self):
 
  assert self.Scheme.find('wlan0', 'test')
 
+def do_nothing(interface_current=None, scheme_current=None, config=None):
+ # mock of utils.set_properties because to avoid errors in tests
+ pass
 
 class TestActivation(TestCase):
+
  def test_successful_connection(self):
- scheme = Scheme('wlan0', 'test')
- connection = scheme.parse_ifup_output(SUCCESSFUL_IFUP_OUTPUT)
- self.assertEqual(connection.scheme, scheme)
- self.assertEqual(connection.ip_address, '192.168.1.113')
+ with patch('wifi.utils.set_properties', side_effect=do_nothing):
+ scheme = Scheme('wlan0', 'test')
+ connection = scheme.parse_ifup_output(SUCCESSFUL_IFUP_OUTPUT)
+ self.assertEqual(connection.scheme, scheme)
+ self.assertEqual(connection.ip_address, '192.168.1.113')
 
  def test_failed_connection(self):
  scheme = Scheme('wlan0', 'test')
  self.assertRaises(ConnectionError, scheme.parse_ifup_output, FAILED_IFUP_OUTPUT)
 
+ def test_activate_is_called_with_good_args(self):
+ args = ['sudo', '/sbin/ifdown', 'wlan0']
+ kwargs = {'stderr':subprocess.STDOUT}
+ scheme = Scheme('wlan0', 'test')
+ with patch.object(subprocess, 'check_output',
+ return_value=SUCCESSFUL_IFUP_OUTPUT):
+ scheme.activate(sudo=True)
+ subprocess.check_output.assert_any_call(args, **kwargs)
+ args = ['/sbin/ifdown', 'wlan0']
+ scheme.activate()
+ subprocess.check_output.assert_any_call(args, **kwargs)
+
 
 class TestForCell(TestCase):
  def test_unencrypted(self):
@@ -218,3 +243,4 @@ def test_wpa(self):
 No DHCPOFFERS received.
 No working leases in persistent database - sleeping.
 """
+

tests/test_utils.py

@@ -8,7 +8,11 @@
 except ImportError: # Python < 3
  from StringIO import StringIO
 
-from wifi.utils import print_table, match, db2dbm
+from wifi.utils import (print_table, match, db2dbm, set_properties,
+get_properties, get_property, ensure_file_exists, MyStringIO)
+from mock import patch, MagicMock
+#from wifi import subprocess_compat as subprocess
+from wifi.scheme import Scheme
 
 
 print_table_in = [
@@ -52,3 +56,122 @@ def test_db2dbm(self):
  self.assertEqual(db2dbm(100), -50)
  self.assertEqual(db2dbm(101), -50)
  self.assertEqual(db2dbm(200), -50)
+
+
+properties_file_content = """scheme_current=test-scheme
+interface_current=test-interface
+scheme_active=True
+"""
+properties_file = MyStringIO(properties_file_content)
+
+connected_output = """test-interface00 IEEE 802.11bgn ESSID:"test-essid" 
+ Mode:Managed Frequency:2.412 GHz Access Point: 70:62:B8:52:7A:00 
+ Bit Rate=72.2 Mb/s Tx-Power=20 dBm 
+ Retry short limit:7 RTS thr=2347 B Fragment thr:off
+ Power Management:off
+ Link Quality=50/70 Signal level=-60 dBm 
+ Rx invalid nwid:0 Rx invalid crypt:0 Rx invalid frag:0
+ Tx excessive retries:0 Invalid misc:8 Missed beacon:0
+"""
+
+disconnected_output = """test-interface00 IEEE 802.11bgn ESSID:off/any 
+ Mode:Managed Access Point: Not-Associated Tx-Power=20 dBm 
+ Retry short limit:7 RTS thr=2347 B Fragment thr:off
+ Power Management:off
+"""
+
+config = {'wireless-essid' : 'test-essid',}
+test_scheme00 = Scheme('test-interface00', 'test-scheme00')
+test_scheme = Scheme('test-interface', 'test-scheme')
+
+#def return_scheme(interface_current, scheme_current):
+# return Scheme(interface_current, scheme_current, options=config)
+
+class propertiesTest(TestCase):
+ def test_get_properties(self):
+ with patch('__builtin__.open', return_value=properties_file):
+ properties = get_properties()
+ self.assertEqual(properties['scheme_current'], 'test-scheme')
+ self.assertEqual(properties['interface_current'], 'test-interface')
+ self.assertEqual(properties['scheme_active'], 'True')
+
+ def test_set_properties(self):
+ with patch('wifi.scheme.Scheme.find', return_value=test_scheme00):
+ # when disconnected
+ with patch('__builtin__.open', return_value=properties_file):
+ with patch('wifi.subprocess_compat.check_output',
+ return_value=disconnected_output):
+ properties_to_set = {
+ 'scheme_current' : 'test-scheme00',
+ 'interface_current' : 'test-interface00'}
+ set_properties(config=config, **properties_to_set)
+ properties = get_properties()
+ self.assertEqual(properties['scheme_current'],
+ 'test-scheme00')
+ self.assertEqual(properties['interface_current'],
+ 'test-interface00')
+ self.assertEqual(properties['scheme_active'], 'False')
+ properties_to_set = {
+ 'interface_current' : 'test-interface00'}
+ set_properties(config=config, **properties_to_set)
+ properties = get_properties()
+ self.assertEqual(properties['scheme_current'],
+ 'test-scheme00')
+ self.assertEqual(properties['interface_current'],
+ 'test-interface00')
+ self.assertEqual(properties['scheme_active'], 'False')
+ properties_to_set = {
+ 'scheme_current' : 'test-scheme00'}
+ set_properties(config=config, **properties_to_set)
+ properties = get_properties()
+ self.assertEqual(properties['scheme_current'],
+ 'test-scheme00')
+ self.assertEqual(properties['interface_current'],
+ 'test-interface00')
+ self.assertEqual(properties['scheme_active'], 'False')
+ # when actually connected
+ with patch('__builtin__.open', return_value=properties_file):
+ with patch('wifi.subprocess_compat.check_output',
+ return_value=connected_output):
+ properties_to_set = {
+ 'scheme_current' : 'test-scheme00',
+ 'interface_current' : 'test-interface00'}
+ set_properties(config=config, **properties_to_set)
+ properties = get_properties()
+ self.assertEqual(properties['scheme_current'],
+ 'test-scheme00')
+ self.assertEqual(properties['interface_current'],
+ 'test-interface00')
+ self.assertEqual(properties['scheme_active'], 'True')
+ properties_to_set = {
+ 'interface_current' : 'test-interface00'}
+ set_properties(config=config, **properties_to_set)
+ properties = get_properties()
+ self.assertEqual(properties['scheme_current'],
+ 'test-scheme00')
+ self.assertEqual(properties['interface_current'],
+ 'test-interface00')
+ self.assertEqual(properties['scheme_active'], 'True')
+ properties_to_set = {
+ 'scheme_current' : 'test-scheme00'}
+ set_properties(config=config, **properties_to_set)
+ properties = get_properties()
+ self.assertEqual(properties['scheme_current'],
+ 'test-scheme00')
+ self.assertEqual(properties['interface_current'],
+ 'test-interface00')
+ self.assertEqual(properties['scheme_active'], 'True')
+ # trying to activate current scheme
+ with patch('wifi.utils.get_properties',
+ return_value=properties):
+ with patch('wifi.subprocess_compat.check_output',
+ return_value=connected_output):
+ set_properties(config=config)
+ self.assertEqual(properties['scheme_active'],
+ 'True')
+ with patch('wifi.subprocess_compat.check_output',
+ return_value=disconnected_output):
+ set_properties(config=config)
+ self.assertEqual(properties['scheme_active'],
+ 'False')
+

wifi/scan.py

@@ -20,12 +20,15 @@ def __repr__(self):
  return 'Cell(ssid={ssid})'.format(**vars(self))
 
  @classmethod
- def all(cls, interface):
+ def all(cls, interface, sudo=False):
  """
  Returns a list of all cells extracted from the output of iwlist.
  """
+ args = ['/sbin/iwlist', interface, 'scan']
+ if sudo:
+ args.insert(0, 'sudo')
  try:
- iwlist_scan = subprocess.check_output(['/sbin/iwlist', interface, 'scan'],
+ iwlist_scan = subprocess.check_output(args,
  stderr=subprocess.STDOUT)
  except subprocess.CalledProcessError as e:
  raise InterfaceError(e.output.strip())
@@ -154,3 +157,4 @@ def normalize(cell_block):
  cell.encryption_type = 'wep'
 
  return cell
+

wifi/scheme.py

@@ -164,20 +164,37 @@ def as_args(self):
 
  return [self.interface + '=' + self.iface] + args
 
- def activate(self):
+ def activate(self, sudo=False):
  """
  Connects to the network as configured in this scheme.
  """
 
- subprocess.check_output(['/sbin/ifdown', self.interface], stderr=subprocess.STDOUT)
- ifup_output = subprocess.check_output(['/sbin/ifup'] + self.as_args(), stderr=subprocess.STDOUT)
+ args_ifdown = ['/sbin/ifdown', self.interface]
+ args_ifup = ['/sbin/ifup']
+ if sudo:
+ args_ifdown.insert(0, 'sudo')
+ args_ifup.insert(0, 'sudo')
+ subprocess.check_output(args_ifdown, stderr=subprocess.STDOUT)
+ ifup_output = subprocess.check_output(args_ifup +
+ self.as_args(),
+ stderr=subprocess.STDOUT)
  ifup_output = ifup_output.decode('utf-8')
+ # set the running config file with current values
+ # we import here to avoid failures in tests
+ from wifi.utils import set_properties
+ properties = {'interface_current' : self.interface,
+ 'scheme_current' : self.name}
+ set_properties(**properties)
 
  return self.parse_ifup_output(ifup_output)
 
  def parse_ifup_output(self, output):
  matches = bound_ip_re.search(output)
  if matches:
+ # set the running config file with current values
+ # we import here to avoid failures in tests
+ from wifi.utils import set_properties
+ set_properties(config=self.options)
  return Connection(scheme=self, ip_address=matches.group('ip_address'))
  else:
  raise ConnectionError("Failed to connect to %r" % self)