Skip to content
This repository has been archived by the owner on Aug 11, 2020. It is now read-only.

Commit

Permalink
Check connection status.
Browse files Browse the repository at this point in the history
	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.
  • Loading branch information
Erf- committed Aug 12, 2015
1 parent 837f975 commit c534deb
Show file tree
Hide file tree
Showing 4 changed files with 157 additions and 48 deletions.
33 changes: 15 additions & 18 deletions tests/test_schemes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand All @@ -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):
Expand Down
126 changes: 103 additions & 23 deletions tests/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down Expand Up @@ -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):
Expand All @@ -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')

9 changes: 8 additions & 1 deletion wifi/scheme.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down
37 changes: 31 additions & 6 deletions wifi/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import os
import sys


if sys.version < '3':
str = unicode

Expand All @@ -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.
Expand Down Expand Up @@ -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'
Expand All @@ -88,7 +110,10 @@ def get_properties():

def get_property(prop):
properties = get_properties()
return properties[prop]
try:
return properties[prop]
except KeyError:
return None

class MyStringIO(StringIO):
# extends StringIO buit-in class to override write and close methods
Expand Down

0 comments on commit c534deb

Please sign in to comment.