tests/hwsim/test_p2p_grpform.py

   1 #!/usr/bin/python
   2 #
   3 # P2P group formation test cases
   4 # Copyright (c) 2013, Jouni Malinen <j@w1.fi>
   5 #
   6 # This software may be distributed under the terms of the BSD license.
   7 # See README for more details.
   8
   9 import logging
  10 logger = logging.getLogger(__name__)
  11 import time
  12 import threading
  13 import Queue
  14
  15 import hwsim_utils
  16
  17 def check_grpform_results(i_res, r_res):
  18     if i_res['result'] != 'success' or r_res['result'] != 'success':
  19         raise Exception("Failed group formation")
  20     if i_res['ssid'] != r_res['ssid']:
  21         raise Exception("SSID mismatch")
  22     if i_res['freq'] != r_res['freq']:
  23         raise Exception("SSID mismatch")
  24     if i_res['go_dev_addr'] != r_res['go_dev_addr']:
  25         raise Exception("GO Device Address mismatch")
  26
  27 def go_neg_init(i_dev, r_dev, pin, i_method, i_intent, res):
  28     logger.debug("Initiate GO Negotiation from i_dev")
  29     i_res = i_dev.p2p_go_neg_init(r_dev.p2p_dev_addr(), pin, i_method, timeout=15, go_intent=i_intent)
  30     logger.debug("i_res: " + str(i_res))
  31     res.put(i_res)
  32
  33 def go_neg_pin(i_dev, r_dev, i_intent=None, r_intent=None, i_method='enter', r_method='display'):
  34     r_dev.p2p_listen()
  35     i_dev.p2p_listen()
  36     pin = r_dev.wps_read_pin()
  37     logger.info("Start GO negotiation " + i_dev.ifname + " -> " + r_dev.ifname)
  38     r_dev.dump_monitor()
  39     res = Queue.Queue()
  40     t = threading.Thread(target=go_neg_init, args=(i_dev, r_dev, pin, i_method, i_intent, res))
  41     t.start()
  42     logger.debug("Wait for GO Negotiation Request on r_dev")
  43     ev = r_dev.wait_event(["P2P-GO-NEG-REQUEST"], timeout=15)
  44     if ev is None:
  45         raise Exception("GO Negotiation timed out")
  46     r_dev.dump_monitor()
  47     logger.debug("Re-initiate GO Negotiation from r_dev")
  48     r_res = r_dev.p2p_go_neg_init(i_dev.p2p_dev_addr(), pin, r_method, go_intent=r_intent, timeout=15)
  49     logger.debug("r_res: " + str(r_res))
  50     r_dev.dump_monitor()
  51     t.join()
  52     i_res = res.get()
  53     logger.debug("i_res: " + str(i_res))
  54     logger.info("Group formed")
  55     hwsim_utils.test_connectivity_p2p(r_dev, i_dev)
  56     i_dev.dump_monitor()
  57
  58 def go_neg_pin_authorized(i_dev, r_dev, i_intent=None, r_intent=None, expect_failure=False, i_go_neg_status=None, i_method='enter', r_method='display'):
  59     r_dev.p2p_listen()
  60     i_dev.p2p_listen()
  61     pin = r_dev.wps_read_pin()
  62     logger.info("Start GO negotiation " + i_dev.ifname + " -> " + r_dev.ifname)
  63     r_dev.p2p_go_neg_auth(i_dev.p2p_dev_addr(), pin, r_method, go_intent=r_intent)
  64     i_res = i_dev.p2p_go_neg_init(r_dev.p2p_dev_addr(), pin, i_method, timeout=15, go_intent=i_intent, expect_failure=expect_failure)
  65     r_res = r_dev.p2p_go_neg_auth_result(expect_failure=expect_failure)
  66     logger.debug("i_res: " + str(i_res))
  67     logger.debug("r_res: " + str(r_res))
  68     r_dev.dump_monitor()
  69     i_dev.dump_monitor()
  70     if i_go_neg_status:
  71         if i_res['result'] != 'go-neg-failed':
  72             raise Exception("Expected GO Negotiation failure not reported")
  73         if i_res['status'] != i_go_neg_status:
  74             raise Exception("Expected GO Negotiation status not seen")
  75     if expect_failure:
  76         return
  77     logger.info("Group formed")
  78     hwsim_utils.test_connectivity_p2p(r_dev, i_dev)
  79
  80 def go_neg_init_pbc(i_dev, r_dev, i_intent, res):
  81     logger.debug("Initiate GO Negotiation from i_dev")
  82     i_res = i_dev.p2p_go_neg_init(r_dev.p2p_dev_addr(), None, "pbc", timeout=15, go_intent=i_intent)
  83     logger.debug("i_res: " + str(i_res))
  84     res.put(i_res)
  85
  86 def go_neg_pbc(i_dev, r_dev, i_intent=None, r_intent=None):
  87     r_dev.p2p_find()
  88     i_dev.p2p_find()
  89     logger.info("Start GO negotiation " + i_dev.ifname + " -> " + r_dev.ifname)
  90     r_dev.dump_monitor()
  91     res = Queue.Queue()
  92     t = threading.Thread(target=go_neg_init_pbc, args=(i_dev, r_dev, i_intent, res))
  93     t.start()
  94     logger.debug("Wait for GO Negotiation Request on r_dev")
  95     ev = r_dev.wait_event(["P2P-GO-NEG-REQUEST"], timeout=15)
  96     if ev is None:
  97         raise Exception("GO Negotiation timed out")
  98     r_dev.dump_monitor()
  99     logger.debug("Re-initiate GO Negotiation from r_dev")
 100     r_res = r_dev.p2p_go_neg_init(i_dev.p2p_dev_addr(), None, "pbc", go_intent=r_intent, timeout=15)
 101     logger.debug("r_res: " + str(r_res))
 102     r_dev.dump_monitor()
 103     t.join()
 104     i_res = res.get()
 105     logger.debug("i_res: " + str(i_res))
 106     logger.info("Group formed")
 107     hwsim_utils.test_connectivity_p2p(r_dev, i_dev)
 108     i_dev.dump_monitor()
 109     return [i_res, r_res]
 110
 111 def test_grpform(dev):
 112     """P2P group formation using PIN and authorized connection (init -> GO)"""
 113     go_neg_pin_authorized(i_dev=dev[0], i_intent=15, r_dev=dev[1], r_intent=0)
 114     dev[0].remove_group()
 115     try:
 116         dev[1].remove_group()
 117     except:
 118         pass
 119
 120 def test_grpform2(dev):
 121     """P2P group formation using PIN and authorized connection (resp -> GO)"""
 122     go_neg_pin_authorized(i_dev=dev[0], i_intent=0, r_dev=dev[1], r_intent=15)
 123     dev[0].remove_group()
 124     try:
 125         dev[1].remove_group()
 126     except:
 127         pass
 128
 129 def test_grpform3(dev):
 130     """P2P group formation using PIN and re-init GO Negotiation"""
 131     go_neg_pin(i_dev=dev[0], i_intent=15, r_dev=dev[1], r_intent=0)
 132     dev[0].remove_group()
 133     try:
 134         dev[1].remove_group()
 135     except:
 136         pass
 137
 138 def test_grpform_pbc(dev):
 139     """P2P group formation using PBC and re-init GO Negotiation"""
 140     [i_res, r_res] = go_neg_pbc(i_dev=dev[0], i_intent=15, r_dev=dev[1], r_intent=0)
 141     check_grpform_results(i_res, r_res)
 142     if i_res['role'] != 'GO' or r_res['role'] != 'client':
 143         raise Exception("Unexpected device roles")
 144     dev[0].remove_group()
 145     try:
 146         dev[1].remove_group()
 147     except:
 148         pass
 149
 150 def test_both_go_intent_15(dev):
 151     """P2P GO Negotiation with both devices using GO intent 15"""
 152     go_neg_pin_authorized(i_dev=dev[0], i_intent=15, r_dev=dev[1], r_intent=15, expect_failure=True, i_go_neg_status=9)
 153
 154 def test_both_go_neg_display(dev):
 155     """P2P GO Negotiation with both devices trying to display PIN"""
 156     go_neg_pin_authorized(i_dev=dev[0], r_dev=dev[1], expect_failure=True, i_go_neg_status=10, i_method='display', r_method='display')
 157
 158 def test_both_go_neg_enter(dev):
 159     """P2P GO Negotiation with both devices trying to enter PIN"""
 160     go_neg_pin_authorized(i_dev=dev[0], r_dev=dev[1], expect_failure=True, i_go_neg_status=10, i_method='enter', r_method='enter')
 161
 162 def add_tests(tests):
 163     tests.append(test_grpform)
 164     tests.append(test_grpform2)
 165     tests.append(test_grpform3)
 166     tests.append(test_grpform_pbc)
 167     tests.append(test_both_go_intent_15)
 168     tests.append(test_both_go_neg_display)
 169     tests.append(test_both_go_neg_enter)