graphyte/controller.py - chromiumos/graphyte - Git at Google

 #!/usr/bin/python
 # Copyright 2018 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 """The Google RF Test Framework module.

 This is the main application module that will drive the other modules.
 This module can also be imported and an instance of graphyte created from there.

 Typical Usage (Import to other modules):
   from graphyte.graphyte import Graphyte
   app = Graphyte()
 Please refer to graphyte_main.py.
 """

 import json
 import os

 import graphyte_common  # pylint: disable=unused-import
 from graphyte.default_setting import DEFAULT_CONFIG_FILE
 from graphyte.default_setting import DEFAULT_RESULT_FILE
 from graphyte.default_setting import GRAPHYTE_DIR
 from graphyte.default_setting import logger
 from graphyte.device import DeviceType
 from graphyte.port_config import PortConfig
 from graphyte.result_writer import ResultWriter
 from graphyte.testplan import TestPlan
 from graphyte.utils import graphyte_utils
 from graphyte.utils import sync_utils

 class Controller(object):
   """The controller controls the flow to perform tests.

   Instantiating the class does everything now. Refer to the constructor below.
   """
   def __init__(self, config_file=None, result_file=None):
     """Instantiate a controller to perform all the tests, which includes:

     - Load the graphyte.settings file to find the modules to load.
     - Load the plugins
     - Load test plans
     - Execute all the tests in the test plan.

     The configuration file graphyte_config.sample.json follows the json format.
     The file has the dut and inst module name and config file.
     These entries are initialized to sample_dut.py and sample_inst.py. The dut
     pluings are searched under ./dut/ and the instrument plugins are searched
     under ./inst/. Once the plugins are found, they are loaded.

     The testplan module is responsible for reading the CSV file which defines
     the testplan and generate a list of the test case, that can be passed to
     inst and dut plugins.

     The interaction with the dut and the instrument plugin is dipicted in
     the flow diagram.
     """
     config_file = config_file or DEFAULT_CONFIG_FILE
     result_file = result_file or DEFAULT_RESULT_FILE

     config_path = graphyte_utils.SearchConfig(config_file, GRAPHYTE_DIR)
     config_dir = os.path.dirname(config_path)
     with open(config_path, 'r') as f:
       graphyte_config = json.load(f)

     # Load DUT and Inst class
     self.dut = self._LoadDevice(graphyte_config, 'dut', config_dir)
     self.inst = self._LoadDevice(graphyte_config, 'inst', config_dir)

     # Load test plan
     self.testplan = TestPlan(graphyte_config['test_plan'], config_dir)
     if not self.testplan:
       raise ValueError('Load test plan failed.')

     # Load port config, including port mapping and path loss
     self.port_config = PortConfig(graphyte_config['port_config'], config_dir)

     # Load the test argument
     self.retries = graphyte_config.get('retries', 1)

     # Open a result writer
     self.result_writer = ResultWriter(result_file)

     # Record all tests are pass or not
     self.all_tests_pass = None

   def Close(self):
     self.result_writer.Close()
     self.TerminateDevices()

   def _LoadDevice(self, graphyte_config, device_type, config_dir):
     """Dynamically load the device instance by the global config.

     Args:
       graphyte_config: the global config.
       device_type: 'dut' or 'inst'
       config_dir: the directory of the graphyte config file.
     Returns:
       The device instance.
     """
     class_name_mapping = {'dut': 'DUT',
                           'inst': 'Inst'}
     module_name = graphyte_config[device_type]
     class_name = class_name_mapping[device_type]
     module = __import__('%s.%s' % (device_type, module_name),
                         globals(), level=1, fromlist=[class_name])
     # Load config file.
     default_config_file = module_name.replace('.', os.path.sep) + '.json'
     default_config_path = os.path.join(GRAPHYTE_DIR, device_type,
                                        default_config_file)
     device_config = graphyte_utils.LoadConfig(default_config_path)

     config_key = '%s_config_file' % device_type
     if config_key in graphyte_config:
       config_file = graphyte_utils.SearchConfig(
           graphyte_config[config_key], [config_dir])
       device_config = graphyte_utils.OverrideConfig(
           device_config, graphyte_utils.LoadConfig(config_file))
     config_key = '%s_config' % device_type
     if config_key in graphyte_config:
       device_config = graphyte_utils.OverrideConfig(
           device_config, graphyte_config[config_key])

     logger.info('Loading %s: %s, device config: %s',
                 device_type, module_name, device_config)
     return module.__dict__[class_name](**device_config)


   def Run(self):
     """The main method of the class.

     It returns whether all tests are passed or not.
     """
     # Initialize the test environment.
     self.all_tests_pass = True
     self.InitialzeDevices()

     # Iteratively execute every test case.
     for test_case in self.testplan:
       self.RunTest(test_case)
     logger.info('All tests are pass? %s', self.all_tests_pass)
     self.result_writer.WriteTotalResult(self.all_tests_pass)

     # Terminate the test environment.
     self.Close()
     return self.all_tests_pass

   def InitialzeDevices(self, device=DeviceType.ALL):
     if DeviceType.IsInst(device):
       try:
         self.inst.Initialize()
         self.inst.LockInstrument()
         self.inst.InitPortConfig(self.port_config)
       except Exception:
         logger.error('Failed to initialize instrumenet.')
         raise

     if DeviceType.IsDUT(device):
       try:
         self.dut.Initialize()
       except Exception:
         logger.error('Failed to initialize DUT.')
         raise

   def TerminateDevices(self, device=DeviceType.ALL):
     if DeviceType.IsInst(device):
       try:
         self.inst.UnlockInstrument()
         self.inst.Terminate()
       except Exception:
         logger.exception('Failed to terminate the instrument device.')

     if DeviceType.IsDUT(device):
       try:
         self.dut.Terminate()
       except Exception:
         logger.exception('Failed to terminate DUT.')

   def SetRF(self, rf_type, device=DeviceType.ALL):
     logger.info('Set RF type: %s', rf_type)
     if DeviceType.IsDUT(device):
       self.dut.SetRF(rf_type)
     if DeviceType.IsInst(device):
       self.inst.SetRF(rf_type)

   def OnFailure(self, device=DeviceType.ALL):
     if DeviceType.IsDUT(device):
       self.dut.OnFailure()
     if DeviceType.IsInst(device):
       self.inst.OnFailure()

   def RunTest(self, test_case):
     logger.info('Running test: %s', test_case)
     logger.info('Test arguments: %r', test_case.args)
     logger.info('Test limit: %r', test_case.result_limit)
     self.SetRF(test_case.rf_type)

     self.inst.SetPortConfig(test_case)
     function_map = {
         'TX': self.RunTxTest,
         'RX': self.RunRxTest}

     def _RunTest():
       try:
         ret = function_map[test_case.test_type](test_case)
       except Exception:
         logger.exception('RunTest error. test_case: %s', test_case)
         self.OnFailure()
         return False, {}

       if not ret[0]:
         self.OnFailure()

       return ret

     def _Condition(ret):
       # ret is (is_pass, result)
       return ret[0]

     # pylint: disable=W0633
     is_pass, result = sync_utils.Retry(self.retries, 0, _RunTest, _Condition)
     self.all_tests_pass &= is_pass
     self.result_writer.WriteResult(test_case, result)

   def RunTxTest(self, test_case):
     self.dut.active_controller.TxStart(test_case)
     self.inst.active_controller.TxMeasure(test_case)
     result = self.inst.active_controller.TxGetResult(test_case)
     self.dut.active_controller.TxStop(test_case)
     return result

   def RunRxTest(self, test_case):
     self.dut.active_controller.RxClearResult(test_case)
     self.inst.active_controller.RxGenerate(test_case)
     result = self.dut.active_controller.RxGetResult(test_case)
     self.inst.active_controller.RxStop(test_case)
     return result
	#!/usr/bin/python
	# Copyright 2018 The Chromium OS Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	"""The Google RF Test Framework module.

	This is the main application module that will drive the other modules.
	This module can also be imported and an instance of graphyte created from there.

	Typical Usage (Import to other modules):
	from graphyte.graphyte import Graphyte
	app = Graphyte()
	Please refer to graphyte_main.py.
	"""

	import json
	import os

	import graphyte_common # pylint: disable=unused-import
	from graphyte.default_setting import DEFAULT_CONFIG_FILE
	from graphyte.default_setting import DEFAULT_RESULT_FILE
	from graphyte.default_setting import GRAPHYTE_DIR
	from graphyte.default_setting import logger
	from graphyte.device import DeviceType
	from graphyte.port_config import PortConfig
	from graphyte.result_writer import ResultWriter
	from graphyte.testplan import TestPlan
	from graphyte.utils import graphyte_utils
	from graphyte.utils import sync_utils

	class Controller(object):
	"""The controller controls the flow to perform tests.

	Instantiating the class does everything now. Refer to the constructor below.
	"""
	def __init__(self, config_file=None, result_file=None):
	"""Instantiate a controller to perform all the tests, which includes:

	- Load the graphyte.settings file to find the modules to load.
	- Load the plugins
	- Load test plans
	- Execute all the tests in the test plan.

	The configuration file graphyte_config.sample.json follows the json format.
	The file has the dut and inst module name and config file.
	These entries are initialized to sample_dut.py and sample_inst.py. The dut
	pluings are searched under ./dut/ and the instrument plugins are searched
	under ./inst/. Once the plugins are found, they are loaded.

	The testplan module is responsible for reading the CSV file which defines
	the testplan and generate a list of the test case, that can be passed to
	inst and dut plugins.

	The interaction with the dut and the instrument plugin is dipicted in
	the flow diagram.
	"""
	config_file = config_file or DEFAULT_CONFIG_FILE
	result_file = result_file or DEFAULT_RESULT_FILE

	config_path = graphyte_utils.SearchConfig(config_file, GRAPHYTE_DIR)
	config_dir = os.path.dirname(config_path)
	with open(config_path, 'r') as f:
	graphyte_config = json.load(f)

	# Load DUT and Inst class
	self.dut = self._LoadDevice(graphyte_config, 'dut', config_dir)
	self.inst = self._LoadDevice(graphyte_config, 'inst', config_dir)

	# Load test plan
	self.testplan = TestPlan(graphyte_config['test_plan'], config_dir)
	if not self.testplan:
	raise ValueError('Load test plan failed.')

	# Load port config, including port mapping and path loss
	self.port_config = PortConfig(graphyte_config['port_config'], config_dir)

	# Load the test argument
	self.retries = graphyte_config.get('retries', 1)

	# Open a result writer
	self.result_writer = ResultWriter(result_file)

	# Record all tests are pass or not
	self.all_tests_pass = None

	def Close(self):
	self.result_writer.Close()
	self.TerminateDevices()

	def _LoadDevice(self, graphyte_config, device_type, config_dir):
	"""Dynamically load the device instance by the global config.

	Args:
	graphyte_config: the global config.
	device_type: 'dut' or 'inst'
	config_dir: the directory of the graphyte config file.
	Returns:
	The device instance.
	"""
	class_name_mapping = {'dut': 'DUT',
	'inst': 'Inst'}
	module_name = graphyte_config[device_type]
	class_name = class_name_mapping[device_type]
	module = __import__('%s.%s' % (device_type, module_name),
	globals(), level=1, fromlist=[class_name])
	# Load config file.
	default_config_file = module_name.replace('.', os.path.sep) + '.json'
	default_config_path = os.path.join(GRAPHYTE_DIR, device_type,
	default_config_file)
	device_config = graphyte_utils.LoadConfig(default_config_path)

	config_key = '%s_config_file' % device_type
	if config_key in graphyte_config:
	config_file = graphyte_utils.SearchConfig(
	graphyte_config[config_key], [config_dir])
	device_config = graphyte_utils.OverrideConfig(
	device_config, graphyte_utils.LoadConfig(config_file))
	config_key = '%s_config' % device_type
	if config_key in graphyte_config:
	device_config = graphyte_utils.OverrideConfig(
	device_config, graphyte_config[config_key])

	logger.info('Loading %s: %s, device config: %s',
	device_type, module_name, device_config)
	return module.__dict__[class_name](**device_config)


	def Run(self):
	"""The main method of the class.

	It returns whether all tests are passed or not.
	"""
	# Initialize the test environment.
	self.all_tests_pass = True
	self.InitialzeDevices()

	# Iteratively execute every test case.
	for test_case in self.testplan:
	self.RunTest(test_case)
	logger.info('All tests are pass? %s', self.all_tests_pass)
	self.result_writer.WriteTotalResult(self.all_tests_pass)

	# Terminate the test environment.
	self.Close()
	return self.all_tests_pass

	def InitialzeDevices(self, device=DeviceType.ALL):
	if DeviceType.IsInst(device):
	try:
	self.inst.Initialize()
	self.inst.LockInstrument()
	self.inst.InitPortConfig(self.port_config)
	except Exception:
	logger.error('Failed to initialize instrumenet.')
	raise

	if DeviceType.IsDUT(device):
	try:
	self.dut.Initialize()
	except Exception:
	logger.error('Failed to initialize DUT.')
	raise

	def TerminateDevices(self, device=DeviceType.ALL):
	if DeviceType.IsInst(device):
	try:
	self.inst.UnlockInstrument()
	self.inst.Terminate()
	except Exception:
	logger.exception('Failed to terminate the instrument device.')

	if DeviceType.IsDUT(device):
	try:
	self.dut.Terminate()
	except Exception:
	logger.exception('Failed to terminate DUT.')

	def SetRF(self, rf_type, device=DeviceType.ALL):
	logger.info('Set RF type: %s', rf_type)
	if DeviceType.IsDUT(device):
	self.dut.SetRF(rf_type)
	if DeviceType.IsInst(device):
	self.inst.SetRF(rf_type)

	def OnFailure(self, device=DeviceType.ALL):
	if DeviceType.IsDUT(device):
	self.dut.OnFailure()
	if DeviceType.IsInst(device):
	self.inst.OnFailure()

	def RunTest(self, test_case):
	logger.info('Running test: %s', test_case)
	logger.info('Test arguments: %r', test_case.args)
	logger.info('Test limit: %r', test_case.result_limit)
	self.SetRF(test_case.rf_type)

	self.inst.SetPortConfig(test_case)
	function_map = {
	'TX': self.RunTxTest,
	'RX': self.RunRxTest}

	def _RunTest():
	try:
	ret = function_map[test_case.test_type](test_case)
	except Exception:
	logger.exception('RunTest error. test_case: %s', test_case)
	self.OnFailure()
	return False, {}

	if not ret[0]:
	self.OnFailure()

	return ret

	def _Condition(ret):
	# ret is (is_pass, result)
	return ret[0]

	# pylint: disable=W0633
	is_pass, result = sync_utils.Retry(self.retries, 0, _RunTest, _Condition)
	self.all_tests_pass &= is_pass
	self.result_writer.WriteResult(test_case, result)

	def RunTxTest(self, test_case):
	self.dut.active_controller.TxStart(test_case)
	self.inst.active_controller.TxMeasure(test_case)
	result = self.inst.active_controller.TxGetResult(test_case)
	self.dut.active_controller.TxStop(test_case)
	return result

	def RunRxTest(self, test_case):
	self.dut.active_controller.RxClearResult(test_case)
	self.inst.active_controller.RxGenerate(test_case)
	result = self.dut.active_controller.RxGetResult(test_case)
	self.inst.active_controller.RxStop(test_case)
	return result