jeltz/hostap
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

tests: Use wpa_supplicant global control interface for P2P

Browse source 
This allows more consistent interface to be used regardless of which
P2P driver design is used (especially for P2P management operations
over netdev vs. dedicated P2P_DEVICE).

Signed-hostap: Jouni Malinen <j@w1.fi>
This commit is contained in:
Jouni Malinen 2013-07-01 00:13:11 +03:00
parent 78f79fe5fc
commit 0fa28afe35
5 changed files with 66 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
tests/hwsim/run-tests.py
View file

@ -49,9 +49,9 @@ def main():
else: else:
test_filter = None test_filter = None
dev0 = WpaSupplicant('wlan0') dev0 = WpaSupplicant('wlan0', '/tmp/wpas-wlan0')
dev1 = WpaSupplicant('wlan1') dev1 = WpaSupplicant('wlan1', '/tmp/wpas-wlan1')
dev2 = WpaSupplicant('wlan2') dev2 = WpaSupplicant('wlan2', '/tmp/wpas-wlan2')
dev = [ dev0, dev1, dev2 ] dev = [ dev0, dev1, dev2 ]
apdev = [ ] apdev = [ ]
apdev.append({"ifname": 'wlan3', "bssid": "02:00:00:00:03:00"}) apdev.append({"ifname": 'wlan3', "bssid": "02:00:00:00:03:00"})

6
tests/hwsim/start.sh
View file

@ -12,9 +12,9 @@ DATE=`date +%s`
sudo ifconfig hwsim0 up sudo ifconfig hwsim0 up
sudo $WLANTEST -i hwsim0 -c -d > $DIR/logs/$DATE-hwsim0 & sudo $WLANTEST -i hwsim0 -c -d > $DIR/logs/$DATE-hwsim0 &
sudo tcpdump -ni hwsim0 -s 2500 -w $DIR/logs/$DATE-hwsim0.dump & sudo tcpdump -ni hwsim0 -s 2500 -w $DIR/logs/$DATE-hwsim0.dump &
sudo $WPAS -g @abstract:wpas-wlan0 -Dnl80211 -iwlan0 -c $DIR/p2p0.conf -ddKt > $DIR/logs/$DATE-log0 & sudo $WPAS -g /tmp/wpas-wlan0 -Gadmin -Dnl80211 -iwlan0 -c $DIR/p2p0.conf -ddKt > $DIR/logs/$DATE-log0 &
sudo $WPAS -g @abstract:wpas-wlan1 -Dnl80211 -iwlan1 -c $DIR/p2p1.conf -ddKt > $DIR/logs/$DATE-log1 & sudo $WPAS -g /tmp/wpas-wlan1 -Gadmin -Dnl80211 -iwlan1 -c $DIR/p2p1.conf -ddKt > $DIR/logs/$DATE-log1 &
sudo $WPAS -g @abstract:wpas-wlan2 -Dnl80211 -iwlan2 -c $DIR/p2p2.conf -ddKt > $DIR/logs/$DATE-log2 & sudo $WPAS -g /tmp/wpas-wlan2 -Gadmin -Dnl80211 -iwlan2 -c $DIR/p2p2.conf -ddKt > $DIR/logs/$DATE-log2 &
sudo $HAPD -ddKt -g /var/run/hostapd-global -G admin -ddKt > $DIR/logs/$DATE-hostapd & sudo $HAPD -ddKt -g /var/run/hostapd-global -G admin -ddKt > $DIR/logs/$DATE-hostapd &
sleep 1 sleep 1
sudo chown $USER $DIR/logs/$DATE-hwsim0.dump sudo chown $USER $DIR/logs/$DATE-hwsim0.dump

26
tests/hwsim/test_p2p_discovery.py
View file

@ -23,14 +23,14 @@ def test_discovery(dev):
raise Exception("Device discovery timed out") raise Exception("Device discovery timed out")
logger.info("Test provision discovery for display") logger.info("Test provision discovery for display")
dev[0].request("P2P_PROV_DISC " + addr1 + " display") dev[0].global_request("P2P_PROV_DISC " + addr1 + " display")
ev1 = dev[1].wait_event(["P2P-PROV-DISC-SHOW-PIN"], timeout=15) ev1 = dev[1].wait_global_event(["P2P-PROV-DISC-SHOW-PIN"], timeout=15)
if ev1 is None: if ev1 is None:
raise Exception("Provision discovery timed out (display/dev1)") raise Exception("Provision discovery timed out (display/dev1)")
if addr0 not in ev1: if addr0 not in ev1:
raise Exception("Dev0 not in provision discovery event") raise Exception("Dev0 not in provision discovery event")
ev0 = dev[0].wait_event(["P2P-PROV-DISC-ENTER-PIN", ev0 = dev[0].wait_global_event(["P2P-PROV-DISC-ENTER-PIN",
"P2P-PROV-DISC-FAILURE"], timeout=15) "P2P-PROV-DISC-FAILURE"], timeout=15)
if ev0 is None: if ev0 is None:
raise Exception("Provision discovery timed out (display/dev0)") raise Exception("Provision discovery timed out (display/dev0)")
if "P2P-PROV-DISC-FAILURE" in ev0: if "P2P-PROV-DISC-FAILURE" in ev0:
@ -39,14 +39,15 @@ def test_discovery(dev):
raise Exception("Dev1 not in provision discovery event") raise Exception("Dev1 not in provision discovery event")
logger.info("Test provision discovery for keypad") logger.info("Test provision discovery for keypad")
dev[0].request("P2P_PROV_DISC " + addr1 + " keypad") dev[0].global_request("P2P_PROV_DISC " + addr1 + " keypad")
ev1 = dev[1].wait_event(["P2P-PROV-DISC-ENTER-PIN"], timeout=15) ev1 = dev[1].wait_global_event(["P2P-PROV-DISC-ENTER-PIN"], timeout=15)
if ev1 is None: if ev1 is None:
raise Exception("Provision discovery timed out (keypad/dev1)") raise Exception("Provision discovery timed out (keypad/dev1)")
if addr0 not in ev1: if addr0 not in ev1:
raise Exception("Dev0 not in provision discovery event") raise Exception("Dev0 not in provision discovery event")
ev0 = dev[0].wait_event(["P2P-PROV-DISC-SHOW-PIN", "P2P-PROV-DISC-FAILURE"], ev0 = dev[0].wait_global_event(["P2P-PROV-DISC-SHOW-PIN",
timeout=15) "P2P-PROV-DISC-FAILURE"],
timeout=15)
if ev0 is None: if ev0 is None:
raise Exception("Provision discovery timed out (keypad/dev0)") raise Exception("Provision discovery timed out (keypad/dev0)")
if "P2P-PROV-DISC-FAILURE" in ev0: if "P2P-PROV-DISC-FAILURE" in ev0:
@ -55,14 +56,15 @@ def test_discovery(dev):
raise Exception("Dev1 not in provision discovery event") raise Exception("Dev1 not in provision discovery event")
logger.info("Test provision discovery for push button") logger.info("Test provision discovery for push button")
dev[0].request("P2P_PROV_DISC " + addr1 + " pbc") dev[0].global_request("P2P_PROV_DISC " + addr1 + " pbc")
ev1 = dev[1].wait_event(["P2P-PROV-DISC-PBC-REQ"], timeout=15) ev1 = dev[1].wait_global_event(["P2P-PROV-DISC-PBC-REQ"], timeout=15)
if ev1 is None: if ev1 is None:
raise Exception("Provision discovery timed out (pbc/dev1)") raise Exception("Provision discovery timed out (pbc/dev1)")
if addr0 not in ev1: if addr0 not in ev1:
raise Exception("Dev0 not in provision discovery event") raise Exception("Dev0 not in provision discovery event")
ev0 = dev[0].wait_event(["P2P-PROV-DISC-PBC-RESP", "P2P-PROV-DISC-FAILURE"], ev0 = dev[0].wait_global_event(["P2P-PROV-DISC-PBC-RESP",
timeout=15) "P2P-PROV-DISC-FAILURE"],
timeout=15)
if ev0 is None: if ev0 is None:
raise Exception("Provision discovery timed out (pbc/dev0)") raise Exception("Provision discovery timed out (pbc/dev0)")
if "P2P-PROV-DISC-FAILURE" in ev0: if "P2P-PROV-DISC-FAILURE" in ev0:

4
tests/hwsim/test_p2p_grpform.py
View file

@ -40,7 +40,7 @@ def go_neg_pin(i_dev, r_dev, i_intent=None, r_intent=None, i_method='enter', r_m
t = threading.Thread(target=go_neg_init, args=(i_dev, r_dev, pin, i_method, i_intent, res)) t = threading.Thread(target=go_neg_init, args=(i_dev, r_dev, pin, i_method, i_intent, res))
t.start() t.start()
logger.debug("Wait for GO Negotiation Request on r_dev") logger.debug("Wait for GO Negotiation Request on r_dev")
ev = r_dev.wait_event(["P2P-GO-NEG-REQUEST"], timeout=15) ev = r_dev.wait_global_event(["P2P-GO-NEG-REQUEST"], timeout=15)
if ev is None: if ev is None:
raise Exception("GO Negotiation timed out") raise Exception("GO Negotiation timed out")
r_dev.dump_monitor() r_dev.dump_monitor()
@ -92,7 +92,7 @@ def go_neg_pbc(i_dev, r_dev, i_intent=None, r_intent=None):
t = threading.Thread(target=go_neg_init_pbc, args=(i_dev, r_dev, i_intent, res)) t = threading.Thread(target=go_neg_init_pbc, args=(i_dev, r_dev, i_intent, res))
t.start() t.start()
logger.debug("Wait for GO Negotiation Request on r_dev") logger.debug("Wait for GO Negotiation Request on r_dev")
ev = r_dev.wait_event(["P2P-GO-NEG-REQUEST"], timeout=15) ev = r_dev.wait_global_event(["P2P-GO-NEG-REQUEST"], timeout=15)
if ev is None: if ev is None:
raise Exception("GO Negotiation timed out") raise Exception("GO Negotiation timed out")
r_dev.dump_monitor() r_dev.dump_monitor()

59
tests/hwsim/wpasupplicant.py
View file

@ -16,17 +16,30 @@ logger = logging.getLogger(__name__)
wpas_ctrl = '/var/run/wpa_supplicant' wpas_ctrl = '/var/run/wpa_supplicant'
class WpaSupplicant: class WpaSupplicant:
def __init__(self, ifname): def __init__(self, ifname, global_iface=None):
self.ifname = ifname self.ifname = ifname
self.group_ifname = None self.group_ifname = None
self.ctrl = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname)) self.ctrl = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname))
self.mon = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname)) self.mon = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname))
self.mon.attach() self.mon.attach()
self.global_iface = global_iface
if global_iface:
self.global_ctrl = wpaspy.Ctrl(global_iface)
self.global_mon = wpaspy.Ctrl(global_iface)
self.global_mon.attach()
def request(self, cmd): def request(self, cmd):
logger.debug(self.ifname + ": CTRL: " + cmd) logger.debug(self.ifname + ": CTRL: " + cmd)
return self.ctrl.request(cmd) return self.ctrl.request(cmd)
def global_request(self, cmd):
if self.global_iface is None:
self.request(cmd)
else:
logger.debug(self.ifname + ": CTRL: " + cmd)
return self.global_ctrl.request(cmd)
def group_request(self, cmd): def group_request(self, cmd):
if self.group_ifname and self.group_ifname != self.ifname: if self.group_ifname and self.group_ifname != self.ifname:
logger.debug(self.group_ifname + ": CTRL: " + cmd) logger.debug(self.group_ifname + ": CTRL: " + cmd)
@ -142,15 +155,15 @@ class WpaSupplicant:
return self.get_group_status_field("address") return self.get_group_status_field("address")
def p2p_listen(self): def p2p_listen(self):
return self.request("P2P_LISTEN") return self.global_request("P2P_LISTEN")
def p2p_find(self, social=False): def p2p_find(self, social=False):
if social: if social:
return self.request("P2P_FIND type=social") return self.global_request("P2P_FIND type=social")
return self.request("P2P_FIND") return self.global_request("P2P_FIND")
def p2p_stop_find(self): def p2p_stop_find(self):
return self.request("P2P_STOP_FIND") return self.global_request("P2P_STOP_FIND")
def wps_read_pin(self): def wps_read_pin(self):
#TODO: make this random #TODO: make this random
@ -158,7 +171,7 @@ class WpaSupplicant:
return self.pin return self.pin
def peer_known(self, peer, full=True): def peer_known(self, peer, full=True):
res = self.request("P2P_PEER " + peer) res = self.global_request("P2P_PEER " + peer)
if peer.lower() not in res.lower(): if peer.lower() not in res.lower():
return False return False
if not full: if not full:
@ -221,12 +234,12 @@ class WpaSupplicant:
cmd = "P2P_CONNECT " + peer + " " + pin + " " + method + " auth" cmd = "P2P_CONNECT " + peer + " " + pin + " " + method + " auth"
if go_intent: if go_intent:
cmd = cmd + ' go_intent=' + str(go_intent) cmd = cmd + ' go_intent=' + str(go_intent)
if "OK" in self.request(cmd): if "OK" in self.global_request(cmd):
return None return None
raise Exception("P2P_CONNECT (auth) failed") raise Exception("P2P_CONNECT (auth) failed")
def p2p_go_neg_auth_result(self, timeout=1, expect_failure=False): def p2p_go_neg_auth_result(self, timeout=1, expect_failure=False):
ev = self.wait_event(["P2P-GROUP-STARTED","P2P-GO-NEG-FAILURE"], timeout); ev = self.wait_global_event(["P2P-GROUP-STARTED","P2P-GO-NEG-FAILURE"], timeout);
if ev is None: if ev is None:
if expect_failure: if expect_failure:
return None return None
@ -244,11 +257,11 @@ class WpaSupplicant:
cmd = "P2P_CONNECT " + peer + " " + method cmd = "P2P_CONNECT " + peer + " " + method
if go_intent: if go_intent:
cmd = cmd + ' go_intent=' + str(go_intent) cmd = cmd + ' go_intent=' + str(go_intent)
if "OK" in self.request(cmd): if "OK" in self.global_request(cmd):
if timeout == 0: if timeout == 0:
self.dump_monitor() self.dump_monitor()
return None return None
ev = self.wait_event(["P2P-GROUP-STARTED","P2P-GO-NEG-FAILURE"], timeout) ev = self.wait_global_event(["P2P-GROUP-STARTED","P2P-GO-NEG-FAILURE"], timeout)
if ev is None: if ev is None:
if expect_failure: if expect_failure:
return None return None
@ -270,6 +283,22 @@ class WpaSupplicant:
return ev return ev
return None return None
def wait_global_event(self, events, timeout):
if self.global_iface is None:
self.wait_event(events, timeout)
else:
count = 0
while count < timeout * 10:
count = count + 1
time.sleep(0.1)
while self.global_mon.pending():
ev = self.global_mon.recv()
logger.debug(self.ifname + ": " + ev)
for event in events:
if event in ev:
return ev
return None
def dump_monitor(self): def dump_monitor(self):
while self.mon.pending(): while self.mon.pending():
ev = self.mon.recv() ev = self.mon.recv()
@ -278,7 +307,7 @@ class WpaSupplicant:
def remove_group(self, ifname=None): def remove_group(self, ifname=None):
if ifname is None: if ifname is None:
ifname = self.group_ifname if self.group_ifname else self.iname ifname = self.group_ifname if self.group_ifname else self.iname
if "OK" not in self.request("P2P_GROUP_REMOVE " + ifname): if "OK" not in self.global_request("P2P_GROUP_REMOVE " + ifname):
raise Exception("Group could not be removed") raise Exception("Group could not be removed")
self.group_ifname = None self.group_ifname = None
@ -293,8 +322,8 @@ class WpaSupplicant:
cmd = cmd + " persistent=" + str(persistent) cmd = cmd + " persistent=" + str(persistent)
if freq: if freq:
cmd = cmd + " freq=" + freq cmd = cmd + " freq=" + freq
if "OK" in self.request(cmd): if "OK" in self.global_request(cmd):
ev = self.wait_event(["P2P-GROUP-STARTED"], timeout=5) ev = self.wait_global_event(["P2P-GROUP-STARTED"], timeout=5)
if ev is None: if ev is None:
raise Exception("GO start up timed out") raise Exception("GO start up timed out")
self.dump_monitor() self.dump_monitor()
@ -313,11 +342,11 @@ class WpaSupplicant:
raise Exception("GO " + go_addr + " not found") raise Exception("GO " + go_addr + " not found")
self.dump_monitor() self.dump_monitor()
cmd = "P2P_CONNECT " + go_addr + " " + pin + " join" cmd = "P2P_CONNECT " + go_addr + " " + pin + " join"
if "OK" in self.request(cmd): if "OK" in self.global_request(cmd):
if timeout == 0: if timeout == 0:
self.dump_monitor() self.dump_monitor()
return None return None
ev = self.wait_event(["P2P-GROUP-STARTED"], timeout) ev = self.wait_global_event(["P2P-GROUP-STARTED"], timeout)
if ev is None: if ev is None:
raise Exception("Joining the group timed out") raise Exception("Joining the group timed out")
self.dump_monitor() self.dump_monitor()