From 65b48b25c87c67a91134951136af6b19d4102eea Mon Sep 17 00:00:00 2001 From: Erf- Date: Mon, 10 Aug 2015 15:57:51 +0200 Subject: [PATCH] Check connection status. modified tests/test_utils.py modified: wifi/scheme.py modified: wifi/utils.py Check connection status by verifying the output of iwconfig command. Part 2. --- tests/test_schemes.py | 33 +++++------ tests/test_utils.py | 126 ++++++++++++++++++++++++++++++++++-------- wifi/scheme.py | 9 ++- wifi/utils.py | 34 ++++++++++-- 4 files changed, 154 insertions(+), 48 deletions(-) diff --git a/tests/test_schemes.py b/tests/test_schemes.py index 20f1a15..4e83be4 100644 --- a/tests/test_schemes.py +++ b/tests/test_schemes.py @@ -95,20 +95,18 @@ def test_save(self): assert self.Scheme.find('wlan0', 'test') - -properties_file_content = """scheme_current=test-scheme -interface_current=test-interface -scheme_active=True -""" -properties_file = MyStringIO(unicode(properties_file_content)) - +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') @@ -119,13 +117,12 @@ def test_activate_is_called_with_good_args(self): kwargs = {'stderr':subprocess.STDOUT} scheme = Scheme('wlan0', 'test') with patch.object(subprocess, 'check_output', - return_value=SUCCESSFUL_IFUP_OUTPUT): - with patch('__builtin__.open', return_value=properties_file): - 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) + 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): diff --git a/tests/test_utils.py b/tests/test_utils.py index c239ac8..88e4b6a 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -11,6 +11,8 @@ 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 = [ @@ -62,6 +64,29 @@ def test_db2dbm(self): """ 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): @@ -71,27 +96,82 @@ def test_get_properties(self): self.assertEqual(properties['scheme_active'], 'True') def test_set_properties(self): - with patch('__builtin__.open', return_value=properties_file): - properties_to_set = {'scheme_current' : 'test-scheme00', - 'scheme_active' : 'False', - 'interface_current' : 'test-interface00'} - set_properties(**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', - 'scheme_active' : 'True'} - set_properties(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', - 'scheme_active' : 'True'} - set_properties(**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') + 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') diff --git a/wifi/scheme.py b/wifi/scheme.py index f2b9bf7..e8f60e7 100644 --- a/wifi/scheme.py +++ b/wifi/scheme.py @@ -3,7 +3,7 @@ import wifi.subprocess_compat as subprocess from pbkdf2 import PBKDF2 -from wifi.utils import ensure_file_exists, set_properties +from wifi.utils import ensure_file_exists from wifi.exceptions import ConnectionError @@ -179,6 +179,9 @@ def activate(self, sudo=False): 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) @@ -188,6 +191,10 @@ def activate(self, sudo=False): 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) diff --git a/wifi/utils.py b/wifi/utils.py index f6bf9e1..7fe49ec 100644 --- a/wifi/utils.py +++ b/wifi/utils.py @@ -3,7 +3,6 @@ import os import sys - if sys.version < '3': str = unicode @@ -12,6 +11,9 @@ except ImportError: # Python < 3 from StringIO import StringIO +import wifi.subprocess_compat as subprocess +from wifi.exceptions import InterfaceError + def match(needle, haystack): """ Command-T-style string matching. @@ -61,16 +63,36 @@ def ensure_file_exists(filename): open(filename, 'a').close() rconf_file = '.runningconfig' -#runnig config file +# runnig config file ensure_file_exists(rconf_file) -def set_properties(interface_current=None, scheme_current=None, scheme_active=False): +def set_properties(interface_current=None, scheme_current=None, config=None): properties = get_properties() if not (bool(interface_current) ^ bool(scheme_current)): - properties['scheme_active'] = scheme_active - if interface_current and scheme_active: + if interface_current and scheme_current: properties['interface_current'] = interface_current properties['scheme_current'] = scheme_current + # set scheme_active value from the output of iwconfig + if config: + if 'wireless-essid' in config: + ssid = config['wireless-essid'] + else: + ssid = config['wpa-ssid'] + else: + ssid = '' + if interface_current: + interface = interface_current + else: + interface = properties['interface_current'] + args = ['/sbin/iwconfig', interface] + try: + iwconfig_output = subprocess.check_output(args, + stderr=subprocess.STDOUT) + except subprocess.CalledProcessError as e: + raise InterfaceError(e.output.strip()) + iwconfig_output = iwconfig_output.decode('utf-8') + scheme_active = str(iwconfig_output.find(ssid)!=-1) + properties['scheme_active'] = scheme_active f = open(rconf_file, 'w') for prop in properties: prop_line = str(prop) + '=' + str(properties[prop]) + '\n' @@ -88,7 +110,7 @@ def get_properties(): def get_property(prop): properties = get_properties() - return properties[prop] + return properties[prop] or None class MyStringIO(StringIO): # extends StringIO buit-in class to override write and close methods