libeap/tests/hwsim/utils.py

   1 # Testing utilities
   2 # Copyright (c) 2013-2015, Jouni Malinen <j@w1.fi>
   3 #
   4 # This software may be distributed under the terms of the BSD license.
   5 # See README for more details.
   6
   7 import os
   8 import time
   9 import remotehost
  10
  11 def get_ifnames():
  12     ifnames = []
  13     with open("/proc/net/dev", "r") as f:
  14         lines = f.readlines()
  15         for l in lines:
  16             val = l.split(':', 1)
  17             if len(val) == 2:
  18                 ifnames.append(val[0].strip(' '))
  19     return ifnames
  20
  21 class HwsimSkip(Exception):
  22     def __init__(self, reason):
  23         self.reason = reason
  24     def __str__(self):
  25         return self.reason
  26
  27 class alloc_fail(object):
  28     def __init__(self, dev, count, funcs):
  29         self._dev = dev
  30         self._count = count
  31         self._funcs = funcs
  32     def __enter__(self):
  33         cmd = "TEST_ALLOC_FAIL %d:%s" % (self._count, self._funcs)
  34         if "OK" not in self._dev.request(cmd):
  35             raise HwsimSkip("TEST_ALLOC_FAIL not supported")
  36     def __exit__(self, type, value, traceback):
  37         if type is None:
  38             if self._dev.request("GET_ALLOC_FAIL") != "0:%s" % self._funcs:
  39                 raise Exception("Allocation failure did not trigger")
  40
  41 class fail_test(object):
  42     def __init__(self, dev, count, funcs):
  43         self._dev = dev
  44         self._count = count
  45         self._funcs = funcs
  46     def __enter__(self):
  47         cmd = "TEST_FAIL %d:%s" % (self._count, self._funcs)
  48         if "OK" not in self._dev.request(cmd):
  49             raise HwsimSkip("TEST_FAIL not supported")
  50     def __exit__(self, type, value, traceback):
  51         if type is None:
  52             if self._dev.request("GET_FAIL") != "0:%s" % self._funcs:
  53                 raise Exception("Test failure did not trigger")
  54
  55 def wait_fail_trigger(dev, cmd, note="Failure not triggered"):
  56     for i in range(0, 40):
  57         if dev.request(cmd).startswith("0:"):
  58             break
  59         if i == 39:
  60             raise Exception(note)
  61         time.sleep(0.05)
  62
  63 def require_under_vm():
  64     with open('/proc/1/cmdline', 'r') as f:
  65         cmd = f.read()
  66         if "inside.sh" not in cmd:
  67             raise HwsimSkip("Not running under VM")
  68
  69 def iface_is_in_bridge(bridge, ifname):
  70     fname = "/sys/class/net/"+ifname+"/brport/bridge"
  71     if not os.path.exists(fname):
  72         return False
  73     if not os.path.islink(fname):
  74         return False
  75     truebridge = os.path.basename(os.readlink(fname))
  76     if bridge == truebridge:
  77         return True
  78     return False
  79
  80 def skip_with_fips(dev, reason="Not supported in FIPS mode"):
  81     res = dev.get_capability("fips")
  82     if res and 'FIPS' in res:
  83         raise HwsimSkip(reason)
  84
  85 def get_phy(ap, ifname=None):
  86     phy = "phy3"
  87     try:
  88         hostname = ap['hostname']
  89     except:
  90         hostname = None
  91     host = remotehost.Host(hostname)
  92
  93     if ifname == None:
  94         ifname = ap['ifname']
  95     status, buf = host.execute(["iw", "dev", ifname, "info"])
  96     if status != 0:
  97         raise Exception("iw " + ifname + " info failed")
  98     lines = buf.split("\n")
  99     for line in lines:
 100         if "wiphy" in line:
 101             words = line.split()
 102             phy = "phy" + words[1]
 103             break
 104     return phy