granicus.if.org Git - esp-idf/blob - tools/tiny-test-fw/Env.py

   1 # Copyright 2015-2017 Espressif Systems (Shanghai) PTE LTD
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http:#www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 """ Test Env, manages DUT, App and EnvConfig, interface for test cases to access these components """
  16 import os
  17 import threading
  18 import functools
  19
  20 import netifaces
  21
  22 import EnvConfig
  23
  24
  25 def _synced(func):
  26     @functools.wraps(func)
  27     def decorator(self, *args, **kwargs):
  28         with self.lock:
  29             ret = func(self, *args, **kwargs)
  30         return ret
  31
  32     decorator.__doc__ = func.__doc__
  33     return decorator
  34
  35
  36 class Env(object):
  37     """
  38     test env, manages DUTs and env configs.
  39
  40     :keyword app: class for default application
  41     :keyword dut: class for default DUT
  42     :keyword env_tag: test env tag, used to select configs from env config file
  43     :keyword env_config_file: test env config file path
  44     :keyword test_name: test suite name, used when generate log folder name
  45     """
  46
  47     def __init__(self,
  48                  app=None,
  49                  dut=None,
  50                  env_tag=None,
  51                  env_config_file=None,
  52                  test_suite_name=None,
  53                  **kwargs):
  54         self.app_cls = app
  55         self.default_dut_cls = dut
  56         self.config = EnvConfig.Config(env_config_file, env_tag)
  57         self.log_path = self.app_cls.get_log_folder(test_suite_name)
  58         if not os.path.exists(self.log_path):
  59             os.makedirs(self.log_path)
  60
  61         self.allocated_duts = dict()
  62         self.lock = threading.RLock()
  63
  64     @_synced
  65     def get_dut(self, dut_name, app_path, dut_class=None, app_class=None):
  66         """
  67         get_dut(dut_name, app_path, dut_class=None, app_class=None)
  68
  69         :param dut_name: user defined name for DUT
  70         :param app_path: application path, app instance will use this path to process application info
  71         :param dut_class: dut class, if not specified will use default dut class of env
  72         :param app_class: app class, if not specified will use default app of env
  73         :return: dut instance
  74         """
  75         if dut_name in self.allocated_duts:
  76             dut = self.allocated_duts[dut_name]["dut"]
  77         else:
  78             if dut_class is None:
  79                 dut_class = self.default_dut_cls
  80             if app_class is None:
  81                 app_class = self.app_cls
  82             app_inst = app_class(app_path)
  83             try:
  84                 port = self.config.get_variable(dut_name)
  85             except ValueError:
  86                 # try to auto detect ports
  87                 allocated_ports = [self.allocated_duts[x]["port"] for x in self.allocated_duts]
  88                 available_ports = dut_class.list_available_ports()
  89                 for port in available_ports:
  90                     if port not in allocated_ports:
  91                         if dut_class.confirm_dut(port, app_inst):
  92                             break
  93                 else:
  94                     port = None
  95             if port:
  96                 try:
  97                     dut_config = self.get_variable(dut_name + "_port_config")
  98                 except ValueError:
  99                     dut_config = dict()
 100                 dut = self.default_dut_cls(dut_name, port,
 101                                            os.path.join(self.log_path, dut_name + ".log"),
 102                                            app_inst,
 103                                            **dut_config)
 104                 self.allocated_duts[dut_name] = {"port": port, "dut": dut}
 105             else:
 106                 raise ValueError("Failed to get DUT")
 107         return dut
 108
 109     @_synced
 110     def close_dut(self, dut_name):
 111         """
 112         close_dut(dut_name)
 113         close one DUT by name if DUT name is valid (the name used by ``get_dut``). otherwise will do nothing.
 114
 115         :param dut_name: user defined name for DUT
 116         :return: None
 117         """
 118         try:
 119             dut = self.allocated_duts.pop(dut_name)["dut"]
 120             dut.close()
 121         except KeyError:
 122             pass
 123
 124     @_synced
 125     def get_variable(self, variable_name):
 126         """
 127         get_variable(variable_name)
 128         get variable from config file. If failed then try to auto-detected it.
 129
 130         :param variable_name: name of the variable
 131         :return: value of variable if successfully found. otherwise None.
 132         """
 133         return self.config.get_variable(variable_name)
 134
 135     PROTO_MAP = {
 136         "ipv4": netifaces.AF_INET,
 137         "ipv6": netifaces.AF_INET6,
 138         "mac": netifaces.AF_LINK,
 139     }
 140
 141     @_synced
 142     def get_pc_nic_info(self, nic_name="pc_nic", proto="ipv4"):
 143         """
 144         get_pc_nic_info(nic_name="pc_nic")
 145         try to get info of a specified NIC and protocol.
 146
 147         :param nic_name: pc nic name. allows passing variable name, nic name value.
 148         :param proto: "ipv4", "ipv6" or "mac"
 149         :return: a dict of nic info if successfully found. otherwise None.
 150                  nic info keys could be different for different protocols.
 151                  key "addr" is available for both mac, ipv4 and ipv6 pic info.
 152         """
 153         interfaces = netifaces.interfaces()
 154         if nic_name in interfaces:
 155             # the name is in the interface list, we regard it as NIC name
 156             if_addr = netifaces.ifaddresses(nic_name)
 157         else:
 158             # it's not in interface name list, we assume it's variable name
 159             _nic_name = self.get_variable(nic_name)
 160             if_addr = netifaces.ifaddresses(_nic_name)
 161
 162         return if_addr[self.PROTO_MAP[proto]][0]
 163
 164     @_synced
 165     def close(self, dut_debug=False):
 166         """
 167         close()
 168         close all DUTs of the Env.
 169
 170         :param dut_debug: if dut_debug is True, then print all dut expect failures before close it
 171         :return: None
 172         """
 173         for dut_name in self.allocated_duts:
 174             dut = self.allocated_duts[dut_name]["dut"]
 175             if dut_debug:
 176                 dut.print_debug_info()
 177             dut.close()
 178         self.allocated_duts = dict()