[PATCH 1/8] tests: tests start using general cmd_execute() func
Jonathan Afek
jonathan at wizery.com
Thu Jun 23 10:16:29 PDT 2016
The hwsim tests used to execute shell commands in the tests
using the subprocess python module.
Start using the cmd_execute() general function for executing
shell commands so that this would also work on remote setups.
Signed-off-by: Jonathan Afek <jonathanx.afek at intel.com>
---
tests/hwsim/hwsim_utils.py | 1 -
tests/hwsim/test_ap_acs.py | 7 ++--
tests/hwsim/test_ap_ft.py | 1 -
tests/hwsim/test_ap_hs20.py | 1 -
tests/hwsim/test_ap_open.py | 12 +++---
tests/hwsim/test_ap_params.py | 19 +++++-----
tests/hwsim/test_ap_pmf.py | 11 ++----
tests/hwsim/test_ap_psk.py | 81 +++++++++++++++++++----------------------
tests/hwsim/test_ap_qosmap.py | 1 -
tests/hwsim/test_ap_roam.py | 1 -
tests/hwsim/test_cfg80211.py | 1 -
tests/hwsim/test_nfc_p2p.py | 1 -
tests/hwsim/test_p2p_autogo.py | 23 +++++-------
tests/hwsim/test_pmksa_cache.py | 20 +++++-----
tests/hwsim/test_radio_work.py | 1 -
tests/hwsim/test_sae.py | 1 -
tests/hwsim/test_scan.py | 23 +++++-------
tests/hwsim/test_wpas_ctrl.py | 8 ++--
18 files changed, 94 insertions(+), 119 deletions(-)
diff --git a/tests/hwsim/hwsim_utils.py b/tests/hwsim/hwsim_utils.py
index fd5f70c..cba9271 100644
--- a/tests/hwsim/hwsim_utils.py
+++ b/tests/hwsim/hwsim_utils.py
@@ -5,7 +5,6 @@
# See README for more details.
import os
-import subprocess
import time
import logging
logger = logging.getLogger()
diff --git a/tests/hwsim/test_ap_acs.py b/tests/hwsim/test_ap_acs.py
index 0f9bc8c..fbfecb7 100644
--- a/tests/hwsim/test_ap_acs.py
+++ b/tests/hwsim/test_ap_acs.py
@@ -6,7 +6,6 @@
import logging
logger = logging.getLogger()
-import subprocess
import time
import hostapd
@@ -144,7 +143,7 @@ def test_ap_acs_5ghz(dev, apdev):
dev[0].request("DISCONNECT")
if hapd:
hapd.request("DISABLE")
- subprocess.call(['iw', 'reg', 'set', '00'])
+ hostapd.cmd_execute(apdev[0], ['iw', 'reg', 'set', '00'])
dev[0].flush_scan_cache()
def test_ap_acs_5ghz_40mhz(dev, apdev):
@@ -173,7 +172,7 @@ def test_ap_acs_5ghz_40mhz(dev, apdev):
dev[0].request("DISCONNECT")
if hapd:
hapd.request("DISABLE")
- subprocess.call(['iw', 'reg', 'set', '00'])
+ hostapd.cmd_execute(apdev[0], ['iw', 'reg', 'set', '00'])
dev[0].flush_scan_cache()
def test_ap_acs_vht(dev, apdev):
@@ -204,7 +203,7 @@ def test_ap_acs_vht(dev, apdev):
dev[0].request("DISCONNECT")
if hapd:
hapd.request("DISABLE")
- subprocess.call(['iw', 'reg', 'set', '00'])
+ hostapd.cmd_execute(apdev[0], ['iw', 'reg', 'set', '00'])
dev[0].flush_scan_cache()
def test_ap_acs_bias(dev, apdev):
diff --git a/tests/hwsim/test_ap_ft.py b/tests/hwsim/test_ap_ft.py
index 7b2c7f4..ccdefb2 100644
--- a/tests/hwsim/test_ap_ft.py
+++ b/tests/hwsim/test_ap_ft.py
@@ -7,7 +7,6 @@
import binascii
import os
import time
-import subprocess
import logging
logger = logging.getLogger()
diff --git a/tests/hwsim/test_ap_hs20.py b/tests/hwsim/test_ap_hs20.py
index 3a93544..1d2ca1e 100644
--- a/tests/hwsim/test_ap_hs20.py
+++ b/tests/hwsim/test_ap_hs20.py
@@ -8,7 +8,6 @@ import base64
import binascii
import struct
import time
-import subprocess
import logging
logger = logging.getLogger()
import os
diff --git a/tests/hwsim/test_ap_open.py b/tests/hwsim/test_ap_open.py
index c9cb265..69d2564 100644
--- a/tests/hwsim/test_ap_open.py
+++ b/tests/hwsim/test_ap_open.py
@@ -160,8 +160,8 @@ def test_ap_open_unexpected_assoc_event(dev, apdev):
dev[0].wait_disconnected(timeout=15)
dev[0].dump_monitor()
# This will be accepted due to matching network
- subprocess.call(['iw', 'dev', dev[0].ifname, 'connect', 'open', "2412",
- apdev[0]['bssid']])
+ dev[0].cmd_execute(['iw', 'dev', dev[0].ifname, 'connect', 'open', "2412",
+ apdev[0]['bssid']])
dev[0].wait_connected(timeout=15)
dev[0].dump_monitor()
@@ -169,8 +169,8 @@ def test_ap_open_unexpected_assoc_event(dev, apdev):
dev[0].wait_disconnected(timeout=5)
dev[0].dump_monitor()
# This will result in disconnection due to no matching network
- subprocess.call(['iw', 'dev', dev[0].ifname, 'connect', 'open', "2412",
- apdev[0]['bssid']])
+ dev[0].cmd_execute(['iw', 'dev', dev[0].ifname, 'connect', 'open', "2412",
+ apdev[0]['bssid']])
dev[0].wait_disconnected(timeout=15)
def test_ap_bss_load(dev, apdev):
@@ -345,7 +345,7 @@ def test_ap_open_ifdown(dev, apdev):
dev[0].connect("open", key_mgmt="NONE", scan_freq="2412")
dev[1].connect("open", key_mgmt="NONE", scan_freq="2412")
- subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'down'])
+ hapd.cmd_execute(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'down'])
ev = hapd.wait_event(["AP-STA-DISCONNECTED"], timeout=10)
if ev is None:
raise Exception("Timeout on AP-STA-DISCONNECTED (1)")
@@ -358,7 +358,7 @@ def test_ap_open_ifdown(dev, apdev):
# The following wait tests beacon loss detection in mac80211 on dev0.
# dev1 is used to test stopping of AP side functionality on client polling.
dev[1].request("REMOVE_NETWORK all")
- subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'up'])
+ hapd.cmd_execute(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'up'])
dev[0].wait_disconnected()
dev[1].wait_disconnected()
ev = hapd.wait_event(["INTERFACE-ENABLED"], timeout=10)
diff --git a/tests/hwsim/test_ap_params.py b/tests/hwsim/test_ap_params.py
index bfba5f8..4dff354 100644
--- a/tests/hwsim/test_ap_params.py
+++ b/tests/hwsim/test_ap_params.py
@@ -7,7 +7,6 @@
import logging
logger = logging.getLogger()
import os
-import subprocess
import hwsim_utils
import hostapd
@@ -112,7 +111,7 @@ def test_ap_country(dev, apdev):
dev[0].request("DISCONNECT")
if hapd:
hapd.request("DISABLE")
- subprocess.call(['iw', 'reg', 'set', '00'])
+ hostapd.cmd_execute(apdev[0], ['iw', 'reg', 'set', '00'])
dev[0].flush_scan_cache()
def test_ap_acl_accept(dev, apdev):
@@ -161,17 +160,17 @@ def test_ap_wds_sta(dev, apdev):
hapd = hostapd.add_ap(apdev[0], params)
try:
- subprocess.call(['brctl', 'addbr', 'wds-br0'])
- subprocess.call(['brctl', 'setfd', 'wds-br0', '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', 'wds-br0', 'up'])
- subprocess.call(['iw', dev[0].ifname, 'set', '4addr', 'on'])
+ dev[0].cmd_execute(['brctl', 'addbr', 'wds-br0'])
+ dev[0].cmd_execute(['brctl', 'setfd', 'wds-br0', '0'])
+ dev[0].cmd_execute(['ip', 'link', 'set', 'dev', 'wds-br0', 'up'])
+ dev[0].cmd_execute(['iw', dev[0].ifname, 'set', '4addr', 'on'])
dev[0].connect(ssid, psk=passphrase, scan_freq="2412")
hwsim_utils.test_connectivity_iface(dev[0], hapd, "wds-br0",
max_tries=15)
finally:
- subprocess.call(['iw', dev[0].ifname, 'set', '4addr', 'off'])
- subprocess.call(['ip', 'link', 'set', 'dev', 'wds-br0', 'down'])
- subprocess.call(['brctl', 'delbr', 'wds-br0'])
+ dev[0].cmd_execute(['iw', dev[0].ifname, 'set', '4addr', 'off'])
+ dev[0].cmd_execute(['ip', 'link', 'set', 'dev', 'wds-br0', 'down'])
+ dev[0].cmd_execute(['brctl', 'delbr', 'wds-br0'])
def test_ap_inactivity_poll(dev, apdev):
"""AP using inactivity poll"""
@@ -247,7 +246,7 @@ def test_ap_spectrum_management_required(dev, apdev):
dev[0].request("DISCONNECT")
if hapd:
hapd.request("DISABLE")
- subprocess.call(['iw', 'reg', 'set', '00'])
+ hostapd.cmd_execute(apdev[0], ['iw', 'reg', 'set', '00'])
dev[0].flush_scan_cache()
def test_ap_max_listen_interval(dev, apdev):
diff --git a/tests/hwsim/test_ap_pmf.py b/tests/hwsim/test_ap_pmf.py
index c9d7403..28d5489 100644
--- a/tests/hwsim/test_ap_pmf.py
+++ b/tests/hwsim/test_ap_pmf.py
@@ -5,7 +5,6 @@
# See README for more details.
import time
-import subprocess
import logging
logger = logging.getLogger()
@@ -413,9 +412,8 @@ def _test_ap_pmf_toggle(dev, apdev):
sta = hapd.get_sta(addr)
if '[MFP]' in sta['flags']:
raise Exception("MFP flag unexpectedly present for STA")
- cmd = subprocess.Popen(['iw', 'dev', apdev[0]['ifname'], 'station', 'get',
- addr], stdout=subprocess.PIPE)
- (data,err) = cmd.communicate()
+ err, data = hapd.cmd_execute(['iw', 'dev', apdev[0]['ifname'], 'station',
+ 'get', addr])
if "yes" in [l for l in data.splitlines() if "MFP" in l][0]:
raise Exception("Kernel STA entry had MFP enabled")
@@ -426,9 +424,8 @@ def _test_ap_pmf_toggle(dev, apdev):
sta = hapd.get_sta(addr)
if '[MFP]' not in sta['flags']:
raise Exception("MFP flag not present for STA")
- cmd = subprocess.Popen(['iw', 'dev', apdev[0]['ifname'], 'station', 'get',
- addr], stdout=subprocess.PIPE)
- (data,err) = cmd.communicate()
+ err, data = hapd.cmd_execute(['iw', 'dev', apdev[0]['ifname'], 'station',
+ 'get', addr])
if "yes" not in [l for l in data.splitlines() if "MFP" in l][0]:
raise Exception("Kernel STA entry did not have MFP enabled")
diff --git a/tests/hwsim/test_ap_psk.py b/tests/hwsim/test_ap_psk.py
index ac63837..3e43610 100644
--- a/tests/hwsim/test_ap_psk.py
+++ b/tests/hwsim/test_ap_psk.py
@@ -340,30 +340,27 @@ def test_ap_wpa2_strict_rekey(dev, apdev):
def test_ap_wpa2_bridge_fdb(dev, apdev):
"""Bridge FDB entry removal"""
+ hapd = None
try:
ssid = "test-wpa2-psk"
passphrase = "12345678"
params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase)
params['bridge'] = 'ap-br0'
- hostapd.add_ap(apdev[0], params)
- subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
+ hapd = hostapd.add_ap(apdev[0], params)
+ hapd.cmd_execute(['brctl', 'setfd', 'ap-br0', '0'])
+ hapd.cmd_execute(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
dev[0].connect(ssid, psk=passphrase, scan_freq="2412",
bssid=apdev[0]['bssid'])
dev[1].connect(ssid, psk=passphrase, scan_freq="2412",
bssid=apdev[0]['bssid'])
addr0 = dev[0].p2p_interface_addr()
hwsim_utils.test_connectivity_sta(dev[0], dev[1])
- cmd = subprocess.Popen(['brctl', 'showmacs', 'ap-br0'],
- stdout=subprocess.PIPE)
- macs1 = cmd.stdout.read()
- cmd = subprocess.call(['brctl', 'setageing', 'ap-br0', '1'])
+ err, macs1 = hapd.cmd_execute(['brctl', 'showmacs', 'ap-br0'])[1]
+ hapd.cmd_execute(['brctl', 'setageing', 'ap-br0', '1'])
dev[0].request("DISCONNECT")
dev[1].request("DISCONNECT")
time.sleep(1)
- cmd = subprocess.Popen(['brctl', 'showmacs', 'ap-br0'],
- stdout=subprocess.PIPE)
- macs2 = cmd.stdout.read()
+ err, macs2 = hapd.cmd_execute(['brctl', 'showmacs', 'ap-br0'])
addr1 = dev[1].p2p_interface_addr()
if addr0 not in macs1 or addr1 not in macs1:
@@ -371,8 +368,8 @@ def test_ap_wpa2_bridge_fdb(dev, apdev):
if addr0 in macs2 or addr1 in macs2:
raise Exception("Bridge FDB entry was not removed")
finally:
- subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'down'])
- subprocess.call(['brctl', 'delbr', 'ap-br0'])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', 'ap-br0', 'down'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', 'ap-br0'])
def test_ap_wpa2_already_in_bridge(dev, apdev):
"""hostapd behavior with interface already in bridge"""
@@ -381,21 +378,21 @@ def test_ap_wpa2_already_in_bridge(dev, apdev):
try:
ssid = "test-wpa2-psk"
passphrase = "12345678"
- subprocess.call(['brctl', 'addbr', br_ifname])
- subprocess.call(['brctl', 'setfd', br_ifname, '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', br_ifname, 'up'])
- subprocess.call(['iw', ifname, 'set', 'type', '__ap'])
- subprocess.call(['brctl', 'addif', br_ifname, ifname])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'addbr', br_ifname])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'setfd', br_ifname, '0'])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', br_ifname, 'up'])
+ hostapd.cmd_execute(apdev[0], ['iw', ifname, 'set', 'type', '__ap'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'addif', br_ifname, ifname])
params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase)
hapd = hostapd.add_ap(apdev[0], params)
if hapd.get_driver_status_field('brname') != br_ifname:
raise Exception("Bridge name not identified correctly")
dev[0].connect(ssid, psk=passphrase, scan_freq="2412")
finally:
- subprocess.call(['ip', 'link', 'set', 'dev', br_ifname, 'down'])
- subprocess.call(['brctl', 'delif', br_ifname, ifname])
- subprocess.call(['iw', ifname, 'set', 'type', 'station'])
- subprocess.call(['brctl', 'delbr', br_ifname])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', br_ifname, 'down'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delif', br_ifname, ifname])
+ hostapd.cmd_execute(apdev[0], ['iw', ifname, 'set', 'type', 'station'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', br_ifname])
def test_ap_wpa2_in_different_bridge(dev, apdev):
"""hostapd behavior with interface in different bridge"""
@@ -404,17 +401,17 @@ def test_ap_wpa2_in_different_bridge(dev, apdev):
try:
ssid = "test-wpa2-psk"
passphrase = "12345678"
- subprocess.call(['brctl', 'addbr', br_ifname])
- subprocess.call(['brctl', 'setfd', br_ifname, '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', br_ifname, 'up'])
- subprocess.call(['iw', ifname, 'set', 'type', '__ap'])
- subprocess.call(['brctl', 'addif', br_ifname, ifname])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'addbr', br_ifname])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'setfd', br_ifname, '0'])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', br_ifname, 'up'])
+ hostapd.cmd_execute(apdev[0], ['iw', ifname, 'set', 'type', '__ap'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'addif', br_ifname, ifname])
time.sleep(0.5)
params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase)
params['bridge'] = 'ap-br0'
hapd = hostapd.add_ap(apdev[0], params)
- subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'setfd', 'ap-br0', '0'])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
brname = hapd.get_driver_status_field('brname')
if brname != 'ap-br0':
raise Exception("Incorrect bridge: " + brname)
@@ -427,10 +424,9 @@ def test_ap_wpa2_in_different_bridge(dev, apdev):
dev[0].request("DISCONNECT")
hapd.disable()
finally:
- subprocess.call(['ip', 'link', 'set', 'dev', br_ifname, 'down'])
- subprocess.call(['brctl', 'delif', br_ifname, ifname],
- stderr=open('/dev/null', 'w'))
- subprocess.call(['brctl', 'delbr', br_ifname])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', br_ifname, 'down'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delif', br_ifname, ifname, "2>", "/dev/null"])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', br_ifname])
def test_ap_wpa2_ext_add_to_bridge(dev, apdev):
"""hostapd behavior with interface added to bridge externally"""
@@ -442,17 +438,17 @@ def test_ap_wpa2_ext_add_to_bridge(dev, apdev):
params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase)
hapd = hostapd.add_ap(apdev[0], params)
- subprocess.call(['brctl', 'addbr', br_ifname])
- subprocess.call(['brctl', 'setfd', br_ifname, '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', br_ifname, 'up'])
- subprocess.call(['brctl', 'addif', br_ifname, ifname])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'addbr', br_ifname])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'setfd', br_ifname, '0'])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', br_ifname, 'up'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'addif', br_ifname, ifname])
dev[0].connect(ssid, psk=passphrase, scan_freq="2412")
if hapd.get_driver_status_field('brname') != br_ifname:
raise Exception("Bridge name not identified correctly")
finally:
- subprocess.call(['ip', 'link', 'set', 'dev', br_ifname, 'down'])
- subprocess.call(['brctl', 'delif', br_ifname, ifname])
- subprocess.call(['brctl', 'delbr', br_ifname])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', br_ifname, 'down'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delif', br_ifname, ifname])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', br_ifname])
def test_ap_wpa2_psk_ext(dev, apdev):
"""WPA2-PSK AP using external EAPOL I/O"""
@@ -1666,8 +1662,7 @@ def test_ap_wpa2_psk_supp_proto_gtk_not_encrypted(dev, apdev):
def find_wpas_process(dev):
ifname = dev.ifname
- cmd = subprocess.Popen(['ps', 'ax'], stdout=subprocess.PIPE)
- (data,err) = cmd.communicate()
+ err, data = dev.cmd_execute(['ps', 'ax'])
for l in data.splitlines():
if "wpa_supplicant" not in l:
continue
@@ -1889,13 +1884,13 @@ def test_ap_wpa2_psk_ifdown(dev, apdev):
bssid = apdev[0]['bssid']
dev[0].connect(ssid, psk=passphrase, scan_freq="2412")
- subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'down'])
+ hapd.cmd_execute(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'down'])
ev = hapd.wait_event(["INTERFACE-DISABLED"], timeout=10)
if ev is None:
raise Exception("No INTERFACE-DISABLED event")
# this wait tests beacon loss detection in mac80211
dev[0].wait_disconnected()
- subprocess.call(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'up'])
+ hapd.cmd_execute(['ip', 'link', 'set', 'dev', apdev[0]['ifname'], 'up'])
ev = hapd.wait_event(["INTERFACE-ENABLED"], timeout=10)
if ev is None:
raise Exception("No INTERFACE-ENABLED event")
diff --git a/tests/hwsim/test_ap_qosmap.py b/tests/hwsim/test_ap_qosmap.py
index 6084521..0030d94 100644
--- a/tests/hwsim/test_ap_qosmap.py
+++ b/tests/hwsim/test_ap_qosmap.py
@@ -5,7 +5,6 @@
# See README for more details.
import time
-import subprocess
import logging
logger = logging.getLogger()
diff --git a/tests/hwsim/test_ap_roam.py b/tests/hwsim/test_ap_roam.py
index cfa5a9d..89d0d9c 100644
--- a/tests/hwsim/test_ap_roam.py
+++ b/tests/hwsim/test_ap_roam.py
@@ -5,7 +5,6 @@
# See README for more details.
import time
-import subprocess
import logging
logger = logging.getLogger()
diff --git a/tests/hwsim/test_cfg80211.py b/tests/hwsim/test_cfg80211.py
index 2162721..56d197b 100644
--- a/tests/hwsim/test_cfg80211.py
+++ b/tests/hwsim/test_cfg80211.py
@@ -8,7 +8,6 @@ import logging
logger = logging.getLogger()
import binascii
import os
-import subprocess
import time
import hostapd
diff --git a/tests/hwsim/test_nfc_p2p.py b/tests/hwsim/test_nfc_p2p.py
index 4425a06..24ae075 100644
--- a/tests/hwsim/test_nfc_p2p.py
+++ b/tests/hwsim/test_nfc_p2p.py
@@ -5,7 +5,6 @@
# See README for more details.
import time
-import subprocess
import logging
logger = logging.getLogger(__name__)
diff --git a/tests/hwsim/test_p2p_autogo.py b/tests/hwsim/test_p2p_autogo.py
index 55c4c0f..2f05d0e 100644
--- a/tests/hwsim/test_p2p_autogo.py
+++ b/tests/hwsim/test_p2p_autogo.py
@@ -426,16 +426,16 @@ def test_autogo_bridge(dev):
raise Exception("Failed to set autoscan")
autogo(dev[0])
ifname = dev[0].get_group_ifname()
- subprocess.call(['brctl', 'addbr', 'p2p-br0'])
- subprocess.call(['brctl', 'setfd', 'p2p-br0', '0'])
- subprocess.call(['brctl', 'addif', 'p2p-br0', ifname])
- subprocess.call(['ip', 'link', 'set', 'dev', 'p2p-br0', 'up'])
+ dev[0].cmd_execute(['brctl', 'addbr', 'p2p-br0'])
+ dev[0].cmd_execute(['brctl', 'setfd', 'p2p-br0', '0'])
+ dev[0].cmd_execute(['brctl', 'addif', 'p2p-br0', ifname])
+ dev[0].cmd_execute(['ip', 'link', 'set', 'dev', 'p2p-br0', 'up'])
time.sleep(0.1)
- subprocess.call(['brctl', 'delif', 'p2p-br0', ifname])
+ dev[0].cmd_execute(['brctl', 'delif', 'p2p-br0', ifname])
time.sleep(0.1)
- subprocess.call(['ip', 'link', 'set', 'dev', 'p2p-br0', 'down'])
+ dev[0].cmd_execute(['ip', 'link', 'set', 'dev', 'p2p-br0', 'down'])
time.sleep(0.1)
- subprocess.call(['brctl', 'delbr', 'p2p-br0'])
+ dev[0].cmd_execute(['brctl', 'delbr', 'p2p-br0'])
ev = dev[0].wait_global_event(["P2P-GROUP-REMOVED"], timeout=1)
if ev is not None:
raise Exception("P2P group removed unexpectedly")
@@ -444,12 +444,9 @@ def test_autogo_bridge(dev):
dev[0].remove_group()
finally:
dev[0].request("AUTOSCAN ")
- subprocess.Popen(['brctl', 'delif', 'p2p-br0', ifname],
- stderr=open('/dev/null', 'w'))
- subprocess.Popen(['ip', 'link', 'set', 'dev', 'p2p-br0', 'down'],
- stderr=open('/dev/null', 'w'))
- subprocess.Popen(['brctl', 'delbr', 'p2p-br0'],
- stderr=open('/dev/null', 'w'))
+ dev[0].cmd_execute(['brctl', 'delif', 'p2p-br0', ifname, '2>', '/dev/null'])
+ dev[0].cmd_execute(['ip', 'link', 'set', 'dev', 'p2p-br0', 'down', '2>', '/dev/null'])
+ dev[0].cmd_execute(['brctl', 'delbr', 'p2p-br0', '2>', '/dev/null'])
def test_presence_req_on_group_interface(dev):
"""P2P_PRESENCE_REQ on group interface"""
diff --git a/tests/hwsim/test_pmksa_cache.py b/tests/hwsim/test_pmksa_cache.py
index 77ec1cf..f8f62cd 100644
--- a/tests/hwsim/test_pmksa_cache.py
+++ b/tests/hwsim/test_pmksa_cache.py
@@ -393,8 +393,8 @@ def generic_pmksa_cache_preauth(dev, apdev, extraparams, identity, databridge,
params[key] = value
hapd = hostapd.add_ap(apdev[0], params)
- subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
+ hapd.cmd_execute(['brctl', 'setfd', 'ap-br0', '0'])
+ hapd.cmd_execute(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
eap_connect(dev[0], hapd, "PAX", identity,
password_hex="0123456789abcdef0123456789abcdef")
@@ -456,10 +456,10 @@ def generic_pmksa_cache_preauth(dev, apdev, extraparams, identity, databridge,
hapd.request("DISABLE")
finally:
- subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'down'],
- stderr=open('/dev/null', 'w'))
- subprocess.call(['brctl', 'delbr', 'ap-br0'],
- stderr=open('/dev/null', 'w'))
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev',
+ 'ap-br0', 'down', '2>', '/dev/null'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', 'ap-br0',
+ '2>' , '/dev/null'])
def test_pmksa_cache_preauth(dev, apdev):
"""RSN pre-authentication to generate PMKSA cache entry"""
@@ -701,15 +701,15 @@ def test_pmksa_cache_preauth_oom(dev, apdev):
try:
_test_pmksa_cache_preauth_oom(dev, apdev)
finally:
- subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'down'])
- subprocess.call(['brctl', 'delbr', 'ap-br0'])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', 'ap-br0', 'down'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'delbr', 'ap-br0'])
def _test_pmksa_cache_preauth_oom(dev, apdev):
params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
params['bridge'] = 'ap-br0'
hapd = hostapd.add_ap(apdev[0], params)
- subprocess.call(['brctl', 'setfd', 'ap-br0', '0'])
- subprocess.call(['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
+ hostapd.cmd_execute(apdev[0], ['brctl', 'setfd', 'ap-br0', '0'])
+ hostapd.cmd_execute(apdev[0], ['ip', 'link', 'set', 'dev', 'ap-br0', 'up'])
eap_connect(dev[0], hapd, "PAX", "pax.user at example.com",
password_hex="0123456789abcdef0123456789abcdef",
bssid=apdev[0]['bssid'])
diff --git a/tests/hwsim/test_radio_work.py b/tests/hwsim/test_radio_work.py
index 482ffda..805d087 100644
--- a/tests/hwsim/test_radio_work.py
+++ b/tests/hwsim/test_radio_work.py
@@ -7,7 +7,6 @@
import time
import logging
logger = logging.getLogger()
-import subprocess
import hostapd
from wpasupplicant import WpaSupplicant
diff --git a/tests/hwsim/test_sae.py b/tests/hwsim/test_sae.py
index 724d3f4..0dbdd5d 100644
--- a/tests/hwsim/test_sae.py
+++ b/tests/hwsim/test_sae.py
@@ -7,7 +7,6 @@
import binascii
import os
import time
-import subprocess
import logging
logger = logging.getLogger()
diff --git a/tests/hwsim/test_scan.py b/tests/hwsim/test_scan.py
index 2e3b839..87e24cf 100644
--- a/tests/hwsim/test_scan.py
+++ b/tests/hwsim/test_scan.py
@@ -138,7 +138,7 @@ def test_scan_external_trigger(dev, apdev):
"""Avoid operations during externally triggered scan"""
hostapd.add_ap(apdev[0], { "ssid": "test-scan" })
bssid = apdev[0]['bssid']
- subprocess.call(['iw', dev[0].ifname, 'scan', 'trigger'])
+ dev[0].cmd_execute(['iw', dev[0].ifname, 'scan', 'trigger'])
check_scan(dev[0], "use_id=1", other_started=True)
def test_scan_bss_expiration_count(dev, apdev):
@@ -362,10 +362,9 @@ def test_scan_for_auth(dev, apdev):
wait_connect=False)
dev[0].dump_monitor()
# Clear cfg80211 BSS table.
- try:
- subprocess.check_call(['iw', dev[0].ifname, 'scan', 'trigger',
- 'freq', '2457', 'flush'])
- except subprocess.CalledProcessError, e:
+ res, data = dev[0].cmd_execute(['iw', dev[0].ifname, 'scan', 'trigger',
+ 'freq', '2457', 'flush'])
+ if 0 != res:
raise HwsimSkip("iw scan trigger flush not supported")
ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], 5)
if ev is None:
@@ -393,10 +392,9 @@ def test_scan_for_auth_fail(dev, apdev):
dev[0].dump_monitor()
hapd.disable()
# Clear cfg80211 BSS table.
- try:
- subprocess.check_call(['iw', dev[0].ifname, 'scan', 'trigger',
- 'freq', '2457', 'flush'])
- except subprocess.CalledProcessError, e:
+ res, data = dev[0].cmd_execute(['iw', dev[0].ifname, 'scan', 'trigger',
+ 'freq', '2457', 'flush'])
+ if 0 != res:
raise HwsimSkip("iw scan trigger flush not supported")
ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], 5)
if ev is None:
@@ -432,10 +430,9 @@ def test_scan_for_auth_wep(dev, apdev):
auth_alg="SHARED", scan_freq="2412", wait_connect=False)
dev[0].dump_monitor()
# Clear cfg80211 BSS table.
- try:
- subprocess.check_call(['iw', dev[0].ifname, 'scan', 'trigger',
- 'freq', '2457', 'flush'])
- except subprocess.CalledProcessError, e:
+ res, data = dev[0].cmd_execute(['iw', dev[0].ifname, 'scan', 'trigger',
+ 'freq', '2457', 'flush'])
+ if 0 != res:
raise HwsimSkip("iw scan trigger flush not supported")
ev = dev[0].wait_event(["CTRL-EVENT-SCAN-RESULTS"], 5)
if ev is None:
diff --git a/tests/hwsim/test_wpas_ctrl.py b/tests/hwsim/test_wpas_ctrl.py
index eb5d113..8a22106 100644
--- a/tests/hwsim/test_wpas_ctrl.py
+++ b/tests/hwsim/test_wpas_ctrl.py
@@ -1259,14 +1259,14 @@ def test_wpas_ctrl_roam(dev, apdev):
def test_wpas_ctrl_ipaddr(dev, apdev):
"""wpa_supplicant IP address in STATUS"""
try:
- subprocess.call(['ip', 'addr', 'add', '10.174.65.207/32', 'dev',
- dev[0].ifname])
+ dev[0].cmd_execute(['ip', 'addr', 'add', '10.174.65.207/32', 'dev',
+ dev[0].ifname])
ipaddr = dev[0].get_status_field('ip_address')
if ipaddr != '10.174.65.207':
raise Exception("IP address not in STATUS output")
finally:
- subprocess.call(['ip', 'addr', 'del', '10.174.65.207/32', 'dev',
- dev[0].ifname])
+ dev[0].cmd_execute(['ip', 'addr', 'del', '10.174.65.207/32', 'dev',
+ dev[0].ifname])
def test_wpas_ctrl_rsp(dev, apdev):
"""wpa_supplicant ctrl_iface CTRL-RSP-"""
--
1.9.1
More information about the Hostap
mailing list