server/hosts/cros_label.py - chromium.googlesource.com/chromiumos/platform/tauto - Gitiles

 # Lint as: python2, python3
 # Copyright 2016 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.

 """This class defines the CrosHost Label class."""

 from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function

 import collections
 import logging
 import re

 import common

 from autotest_lib.client.bin import utils
 from autotest_lib.server.cros.dynamic_suite import constants as ds_constants
 from autotest_lib.server.hosts import base_label
 from autotest_lib.server.hosts import common_label
 from autotest_lib.server.hosts import servo_constants
 from six.moves import zip

 # pylint: disable=missing-docstring
 LsbOutput = collections.namedtuple('LsbOutput', ['unibuild', 'board'])

 # Repair and Deploy taskName
 REPAIR_TASK_NAME = 'repair'
 DEPLOY_TASK_NAME = 'deploy'


 def _parse_lsb_output(host):
     """Parses the LSB output and returns key data points for labeling.

     @param host: Host that the command will be executed against
     @returns: LsbOutput with the result of parsing the /etc/lsb-release output
     """
     release_info = utils.parse_cmd_output('cat /etc/lsb-release',
                                           run_method=host.run)

     unibuild = release_info.get('CHROMEOS_RELEASE_UNIBUILD') == '1'
     return LsbOutput(unibuild, release_info['CHROMEOS_RELEASE_BOARD'])


 class DeviceSkuLabel(base_label.StringPrefixLabel):
     """Determine the correct device_sku label for the device."""

     _NAME =  ds_constants.DEVICE_SKU_LABEL

     def generate_labels(self, host):
         device_sku = host.host_info_store.get().device_sku
         if device_sku:
             return [device_sku]

         mosys_cmd = 'mosys platform sku'
         result = host.run(command=mosys_cmd, ignore_status=True)
         if result.exit_status == 0:
             return [result.stdout.strip()]

         return []

     def update_for_task(self, task_name):
         # This label is stored in the lab config.
         return task_name in (DEPLOY_TASK_NAME, REPAIR_TASK_NAME, '')


 class BrandCodeLabel(base_label.StringPrefixLabel):
     """Determine the correct brand_code (aka RLZ-code) for the device."""

     _NAME =  ds_constants.BRAND_CODE_LABEL

     def generate_labels(self, host):
         brand_code = host.host_info_store.get().brand_code
         if brand_code:
             return [brand_code]

         cros_config_cmd = 'cros_config / brand-code'
         result = host.run(command=cros_config_cmd, ignore_status=True)
         if result.exit_status == 0:
             return [result.stdout.strip()]

         return []


 class BluetoothPeerLabel(base_label.StringPrefixLabel):
     """Return the Bluetooth peer labels.

     working_bluetooth_btpeer label is applied if a Raspberry Pi Bluetooth peer
     is detected.There can be up to 4 Bluetooth peers. Labels
     working_bluetooth_btpeer:[1-4] will be assigned depending on the number of
     peers present.

     """

     _NAME = 'working_bluetooth_btpeer'

     def exists(self, host):
         return  len(host._btpeer_host_list) > 0

     def generate_labels(self, host):
         labels_list = []
         count = 1

         for (btpeer, btpeer_host) in \
                         zip(host.btpeer_list, host._btpeer_host_list):
             try:
                 # Initialize one device type to make sure the peer is working
                 bt_hid_device = btpeer.get_bluetooth_hid_mouse()
                 if bt_hid_device.CheckSerialConnection():
                     labels_list.append(str(count))
                     count += 1
             except Exception as e:
                 logging.error('Error with initializing bt_hid_mouse on '
                               'btpeer %s %s', btpeer_host.hostname, e)

         logging.info('Bluetooth Peer labels are %s', labels_list)
         return labels_list

     def update_for_task(self, task_name):
         # This label is stored in the state config, so only repair tasks update
         # it or when no task name is mentioned.
         return task_name in (REPAIR_TASK_NAME, '')


 class Cr50Label(base_label.StringPrefixLabel):
     """Label indicating the cr50 image type."""

     _NAME = 'cr50'

     def __init__(self):
         self.ver = None

     def exists(self, host):
         # Make sure the gsctool version command runs ok
         self.ver = host.run('gsctool -a -f', ignore_status=True)
         return self.ver.exit_status == 0

     def _get_version(self, region):
         """Get the version number of the given region"""
         return re.search(region + ' (\d+\.\d+\.\d+)', self.ver.stdout).group(1)

     def generate_labels(self, host):
         # Check the major version to determine prePVT vs PVT
         version = self._get_version('RW')
         major_version = int(version.split('.')[1])
         # PVT images have a odd major version prePVT have even
         return ['pvt' if (major_version % 2) else 'prepvt']

     def update_for_task(self, task_name):
         # This label is stored in the state config, so only repair tasks update
         # it or when no task name is mentioned.
         return task_name in (REPAIR_TASK_NAME, '')


 class Cr50RWKeyidLabel(Cr50Label):
     """Label indicating the cr50 RW version."""
     _REGION = 'RW'
     _NAME = 'cr50-rw-keyid'

     def _get_keyid_info(self, region):
         """Get the keyid of the given region."""
         match = re.search('keyids:.*%s (\S+)' % region, self.ver.stdout)
         keyid = match.group(1).rstrip(',')
         is_prod = int(keyid, 16) & (1 << 2)
         return [keyid, 'prod' if is_prod else 'dev']

     def generate_labels(self, host):
         """Get the key type."""
         return self._get_keyid_info(self._REGION)


 class Cr50ROKeyidLabel(Cr50RWKeyidLabel):
     """Label indicating the RO key type."""
     _REGION = 'RO'
     _NAME = 'cr50-ro-keyid'


 class ChameleonLabel(base_label.BaseLabel):
     """Determine if a Chameleon is connected to this host."""

     _NAME = 'chameleon'

     def exists(self, host):
         # See crbug.com/1004500#2 for details.
         has_chameleon = host._chameleon_host is not None
         # TODO(crbug.com/995900) -- debug why chameleon label is flipping
         try:
             logging.info("has_chameleon %s", has_chameleon)
             logging.info("_chameleon_host %s",
                          getattr(host, "_chameleon_host", "NO_ATTRIBUTE"))
             logging.info("chameleon %s",
                          getattr(host, "chameleon", "NO_ATTRIBUTE"))
         except:
             pass
         return has_chameleon

     def update_for_task(self, task_name):
         # This label is stored in the state config, so only repair tasks update
         # it or when no task name is mentioned.
         return task_name in (REPAIR_TASK_NAME, '')


 class ChameleonConnectionLabel(base_label.StringPrefixLabel):
     """Return the Chameleon connection label."""

     _NAME = 'chameleon'

     def exists(self, host):
         return host._chameleon_host is not None

     def generate_labels(self, host):
         return [host.chameleon.get_label()]

     def update_for_task(self, task_name):
         # This label is stored in the lab config, so only deploy tasks update it
         # or when no task name is mentioned.
         return task_name in (DEPLOY_TASK_NAME, '')


 class AudioLoopbackDongleLabel(base_label.BaseLabel):
     """Return the label if an audio loopback dongle is plugged in."""

     _NAME = 'audio_loopback_dongle'

     def exists(self, host):
         # Based on crbug.com/991285, AudioLoopbackDongle sometimes flips.
         # Ensure that AudioLoopbackDongle.exists returns True
         # forever, after it returns True *once*.
         if self._cached_exists(host):
             # If the current state is True, return it, don't run the command on
             # the DUT and potentially flip the state.
             return True
         # If the current state is not True, run the command on
         # the DUT. The new state will be set to whatever the command
         # produces.
         return self._host_run_exists(host)

     def _node_type_is_plugged(self, node_type, nodes_info):
         """Determine if there is any node of node_type plugged.

         This method is used in the AudioLoopbackDongleLabel class, where the
         call is executed on autotest server. Use get_cras_nodes instead if
         the call can be executed on Cros device.

         Since Cras only reports the plugged node in GetNodes, we can
         parse the return value to see if there is any node with the given type.
         For example, if INTERNAL_MIC is of intereset, the pattern we are
         looking for is:

         dict entry(
            string "Type"
            variant             string "INTERNAL_MIC"
         )

         @param node_type: A str representing node type defined in CRAS_NODE_TYPES.
         @param nodes_info: A str containing output of command get_nodes_cmd.

         @returns: True if there is any node of node_type plugged. False otherwise.

         """
         match = re.search(r'string "Type"\s+variant\s+string "%s"' % node_type,
                           nodes_info)
         return True if match else False

     def _cached_exists(self, host):
         """Get the state of AudioLoopbackDongle in the data store"""
         info = host.host_info_store.get()
         for label in info.labels:
             if label.startswith(self._NAME):
                 return True
         return False

     def _host_run_exists(self, host):
         """Detect presence of audio_loopback_dongle by physically
         running a command on the DUT."""
         cras_cmd = ('dbus-send --system --type=method_call --print-reply '
                     '--dest=org.chromium.cras /org/chromium/cras '
                     'org.chromium.cras.Control.GetNodes')
         nodes_info = host.run(command=cras_cmd, ignore_status=True).stdout
         if (self._node_type_is_plugged('HEADPHONE', nodes_info)
                     and self._node_type_is_plugged('MIC', nodes_info)):
             return True
         return False

     def update_for_task(self, task_name):
         # This label is stored in the state config, so only repair tasks update
         # it or when no task name is mentioned.
         return task_name in (REPAIR_TASK_NAME, '')


 class ServoTypeLabel(base_label.StringPrefixLabel):
     _NAME = servo_constants.SERVO_TYPE_LABEL_PREFIX

     def generate_labels(self, host):
         info = host.host_info_store.get()

         servo_type = self._get_from_labels(info)
         if servo_type != '':
             logging.info("Using servo_type: %s from cache!", servo_type)
             return [servo_type]

         if host.servo is not None:
             try:
                 servo_type = host.servo.get_servo_version()
                 if servo_type != '':
                     return [servo_type]
                 logging.warning('Cannot collect servo_type from servo'
                 ' by `dut-control servo_type`! Please file a bug'
                 ' and inform infra team as we are not expected '
                 ' to reach this point.')
             except Exception as e:
                 # We don't want fail the label and break DUTs here just
                 # because of servo issue.
                 logging.error("Failed to update servo_type, %s", str(e))
         return []

     def _get_from_labels(self, info):
         prefix = self._NAME + ':'
         for label in info.labels:
             if  label.startswith(prefix):
                 suffix_length = len(prefix)
                 return label[suffix_length:]
         return ''

     def update_for_task(self, task_name):
         # This label is stored in the lab config,
         # only deploy and repair tasks update it
         # or when no task name is mentioned.
         return task_name in (DEPLOY_TASK_NAME, '')


 def _parse_hwid_labels(hwid_info_list):
     if len(hwid_info_list) == 0:
         return hwid_info_list

     res = []
     # See crbug.com/997816#c7 for details of two potential formats of returns
     # from HWID server.
     if isinstance(hwid_info_list[0], dict):
         # Format of hwid_info:
         # [{u'name': u'sku', u'value': u'xxx'}, ..., ]
         for hwid_info in hwid_info_list:
             value = hwid_info.get('value', '')
             name = hwid_info.get('name', '')
             # There should always be a name but just in case there is not.
             if name:
                 new_label = name if not value else '%s:%s' % (name, value)
                 res.append(new_label)
     else:
         # Format of hwid_info:
         # [<DUTLabel name: 'sku' value: u'xxx'>, ..., ]
         for hwid_info in hwid_info_list:
             new_label = str(hwid_info)
             logging.info('processing hwid label: %s', new_label)
             res.append(new_label)

     return res


 CROS_LABELS = [
     AudioLoopbackDongleLabel(), #STATECONFIG
     BluetoothPeerLabel(), #STATECONFIG
     ChameleonConnectionLabel(), #LABCONFIG
     ChameleonLabel(), #STATECONFIG
     common_label.OSLabel(),
     DeviceSkuLabel(), #LABCONFIG
     ServoTypeLabel(), #LABCONFIG
     # Temporarily add back as there's no way to reference cr50 configs.
     # See crbug.com/1057145 for the root cause.
     # See crbug.com/1057719 for future tracking.
     Cr50Label(),
     Cr50ROKeyidLabel(),
 ]

 LABSTATION_LABELS = [
     common_label.OSLabel(),
 ]
	# Lint as: python2, python3
	# Copyright 2016 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.

	"""This class defines the CrosHost Label class."""

	from __future__ import absolute_import
	from __future__ import division
	from __future__ import print_function

	import collections
	import logging
	import re

	import common

	from autotest_lib.client.bin import utils
	from autotest_lib.server.cros.dynamic_suite import constants as ds_constants
	from autotest_lib.server.hosts import base_label
	from autotest_lib.server.hosts import common_label
	from autotest_lib.server.hosts import servo_constants
	from six.moves import zip

	# pylint: disable=missing-docstring
	LsbOutput = collections.namedtuple('LsbOutput', ['unibuild', 'board'])

	# Repair and Deploy taskName
	REPAIR_TASK_NAME = 'repair'
	DEPLOY_TASK_NAME = 'deploy'


	def _parse_lsb_output(host):
	"""Parses the LSB output and returns key data points for labeling.

	@param host: Host that the command will be executed against
	@returns: LsbOutput with the result of parsing the /etc/lsb-release output
	"""
	release_info = utils.parse_cmd_output('cat /etc/lsb-release',
	run_method=host.run)

	unibuild = release_info.get('CHROMEOS_RELEASE_UNIBUILD') == '1'
	return LsbOutput(unibuild, release_info['CHROMEOS_RELEASE_BOARD'])


	class DeviceSkuLabel(base_label.StringPrefixLabel):
	"""Determine the correct device_sku label for the device."""

	_NAME = ds_constants.DEVICE_SKU_LABEL

	def generate_labels(self, host):
	device_sku = host.host_info_store.get().device_sku
	if device_sku:
	return [device_sku]

	mosys_cmd = 'mosys platform sku'
	result = host.run(command=mosys_cmd, ignore_status=True)
	if result.exit_status == 0:
	return [result.stdout.strip()]

	return []

	def update_for_task(self, task_name):
	# This label is stored in the lab config.
	return task_name in (DEPLOY_TASK_NAME, REPAIR_TASK_NAME, '')


	class BrandCodeLabel(base_label.StringPrefixLabel):
	"""Determine the correct brand_code (aka RLZ-code) for the device."""

	_NAME = ds_constants.BRAND_CODE_LABEL

	def generate_labels(self, host):
	brand_code = host.host_info_store.get().brand_code
	if brand_code:
	return [brand_code]

	cros_config_cmd = 'cros_config / brand-code'
	result = host.run(command=cros_config_cmd, ignore_status=True)
	if result.exit_status == 0:
	return [result.stdout.strip()]

	return []


	class BluetoothPeerLabel(base_label.StringPrefixLabel):
	"""Return the Bluetooth peer labels.

	working_bluetooth_btpeer label is applied if a Raspberry Pi Bluetooth peer
	is detected.There can be up to 4 Bluetooth peers. Labels
	working_bluetooth_btpeer:[1-4] will be assigned depending on the number of
	peers present.

	"""

	_NAME = 'working_bluetooth_btpeer'

	def exists(self, host):
	return len(host._btpeer_host_list) > 0

	def generate_labels(self, host):
	labels_list = []
	count = 1

	for (btpeer, btpeer_host) in \
	zip(host.btpeer_list, host._btpeer_host_list):
	try:
	# Initialize one device type to make sure the peer is working
	bt_hid_device = btpeer.get_bluetooth_hid_mouse()
	if bt_hid_device.CheckSerialConnection():
	labels_list.append(str(count))
	count += 1
	except Exception as e:
	logging.error('Error with initializing bt_hid_mouse on '
	'btpeer %s %s', btpeer_host.hostname, e)

	logging.info('Bluetooth Peer labels are %s', labels_list)
	return labels_list

	def update_for_task(self, task_name):
	# This label is stored in the state config, so only repair tasks update
	# it or when no task name is mentioned.
	return task_name in (REPAIR_TASK_NAME, '')


	class Cr50Label(base_label.StringPrefixLabel):
	"""Label indicating the cr50 image type."""

	_NAME = 'cr50'

	def __init__(self):
	self.ver = None

	def exists(self, host):
	# Make sure the gsctool version command runs ok
	self.ver = host.run('gsctool -a -f', ignore_status=True)
	return self.ver.exit_status == 0

	def _get_version(self, region):
	"""Get the version number of the given region"""
	return re.search(region + ' (\d+\.\d+\.\d+)', self.ver.stdout).group(1)

	def generate_labels(self, host):
	# Check the major version to determine prePVT vs PVT
	version = self._get_version('RW')
	major_version = int(version.split('.')[1])
	# PVT images have a odd major version prePVT have even
	return ['pvt' if (major_version % 2) else 'prepvt']

	def update_for_task(self, task_name):
	# This label is stored in the state config, so only repair tasks update
	# it or when no task name is mentioned.
	return task_name in (REPAIR_TASK_NAME, '')


	class Cr50RWKeyidLabel(Cr50Label):
	"""Label indicating the cr50 RW version."""
	_REGION = 'RW'
	_NAME = 'cr50-rw-keyid'

	def _get_keyid_info(self, region):
	"""Get the keyid of the given region."""
	match = re.search('keyids:.*%s (\S+)' % region, self.ver.stdout)
	keyid = match.group(1).rstrip(',')
	is_prod = int(keyid, 16) & (1 << 2)
	return [keyid, 'prod' if is_prod else 'dev']

	def generate_labels(self, host):
	"""Get the key type."""
	return self._get_keyid_info(self._REGION)


	class Cr50ROKeyidLabel(Cr50RWKeyidLabel):
	"""Label indicating the RO key type."""
	_REGION = 'RO'
	_NAME = 'cr50-ro-keyid'


	class ChameleonLabel(base_label.BaseLabel):
	"""Determine if a Chameleon is connected to this host."""

	_NAME = 'chameleon'

	def exists(self, host):
	# See crbug.com/1004500#2 for details.
	has_chameleon = host._chameleon_host is not None
	# TODO(crbug.com/995900) -- debug why chameleon label is flipping
	try:
	logging.info("has_chameleon %s", has_chameleon)
	logging.info("_chameleon_host %s",
	getattr(host, "_chameleon_host", "NO_ATTRIBUTE"))
	logging.info("chameleon %s",
	getattr(host, "chameleon", "NO_ATTRIBUTE"))
	except:
	pass
	return has_chameleon

	def update_for_task(self, task_name):
	# This label is stored in the state config, so only repair tasks update
	# it or when no task name is mentioned.
	return task_name in (REPAIR_TASK_NAME, '')


	class ChameleonConnectionLabel(base_label.StringPrefixLabel):
	"""Return the Chameleon connection label."""

	_NAME = 'chameleon'

	def exists(self, host):
	return host._chameleon_host is not None

	def generate_labels(self, host):
	return [host.chameleon.get_label()]

	def update_for_task(self, task_name):
	# This label is stored in the lab config, so only deploy tasks update it
	# or when no task name is mentioned.
	return task_name in (DEPLOY_TASK_NAME, '')


	class AudioLoopbackDongleLabel(base_label.BaseLabel):
	"""Return the label if an audio loopback dongle is plugged in."""

	_NAME = 'audio_loopback_dongle'

	def exists(self, host):
	# Based on crbug.com/991285, AudioLoopbackDongle sometimes flips.
	# Ensure that AudioLoopbackDongle.exists returns True
	# forever, after it returns True once.
	if self._cached_exists(host):
	# If the current state is True, return it, don't run the command on
	# the DUT and potentially flip the state.
	return True
	# If the current state is not True, run the command on
	# the DUT. The new state will be set to whatever the command
	# produces.
	return self._host_run_exists(host)

	def _node_type_is_plugged(self, node_type, nodes_info):
	"""Determine if there is any node of node_type plugged.

	This method is used in the AudioLoopbackDongleLabel class, where the
	call is executed on autotest server. Use get_cras_nodes instead if
	the call can be executed on Cros device.

	Since Cras only reports the plugged node in GetNodes, we can
	parse the return value to see if there is any node with the given type.
	For example, if INTERNAL_MIC is of intereset, the pattern we are
	looking for is:

	dict entry(
	string "Type"
	variant string "INTERNAL_MIC"
	)

	@param node_type: A str representing node type defined in CRAS_NODE_TYPES.
	@param nodes_info: A str containing output of command get_nodes_cmd.

	@returns: True if there is any node of node_type plugged. False otherwise.

	"""
	match = re.search(r'string "Type"\s+variant\s+string "%s"' % node_type,
	nodes_info)
	return True if match else False

	def _cached_exists(self, host):
	"""Get the state of AudioLoopbackDongle in the data store"""
	info = host.host_info_store.get()
	for label in info.labels:
	if label.startswith(self._NAME):
	return True
	return False

	def _host_run_exists(self, host):
	"""Detect presence of audio_loopback_dongle by physically
	running a command on the DUT."""
	cras_cmd = ('dbus-send --system --type=method_call --print-reply '
	'--dest=org.chromium.cras /org/chromium/cras '
	'org.chromium.cras.Control.GetNodes')
	nodes_info = host.run(command=cras_cmd, ignore_status=True).stdout
	if (self._node_type_is_plugged('HEADPHONE', nodes_info)
	and self._node_type_is_plugged('MIC', nodes_info)):
	return True
	return False

	def update_for_task(self, task_name):
	# This label is stored in the state config, so only repair tasks update
	# it or when no task name is mentioned.
	return task_name in (REPAIR_TASK_NAME, '')


	class ServoTypeLabel(base_label.StringPrefixLabel):
	_NAME = servo_constants.SERVO_TYPE_LABEL_PREFIX

	def generate_labels(self, host):
	info = host.host_info_store.get()

	servo_type = self._get_from_labels(info)
	if servo_type != '':
	logging.info("Using servo_type: %s from cache!", servo_type)
	return [servo_type]

	if host.servo is not None:
	try:
	servo_type = host.servo.get_servo_version()
	if servo_type != '':
	return [servo_type]
	logging.warning('Cannot collect servo_type from servo'
	' by `dut-control servo_type`! Please file a bug'
	' and inform infra team as we are not expected '
	' to reach this point.')
	except Exception as e:
	# We don't want fail the label and break DUTs here just
	# because of servo issue.
	logging.error("Failed to update servo_type, %s", str(e))
	return []

	def _get_from_labels(self, info):
	prefix = self._NAME + ':'
	for label in info.labels:
	if label.startswith(prefix):
	suffix_length = len(prefix)
	return label[suffix_length:]
	return ''

	def update_for_task(self, task_name):
	# This label is stored in the lab config,
	# only deploy and repair tasks update it
	# or when no task name is mentioned.
	return task_name in (DEPLOY_TASK_NAME, '')


	def _parse_hwid_labels(hwid_info_list):
	if len(hwid_info_list) == 0:
	return hwid_info_list

	res = []
	# See crbug.com/997816#c7 for details of two potential formats of returns
	# from HWID server.
	if isinstance(hwid_info_list[0], dict):
	# Format of hwid_info:
	# [{u'name': u'sku', u'value': u'xxx'}, ..., ]
	for hwid_info in hwid_info_list:
	value = hwid_info.get('value', '')
	name = hwid_info.get('name', '')
	# There should always be a name but just in case there is not.
	if name:
	new_label = name if not value else '%s:%s' % (name, value)
	res.append(new_label)
	else:
	# Format of hwid_info:
	# [<DUTLabel name: 'sku' value: u'xxx'>, ..., ]
	for hwid_info in hwid_info_list:
	new_label = str(hwid_info)
	logging.info('processing hwid label: %s', new_label)
	res.append(new_label)

	return res


	CROS_LABELS = [
	AudioLoopbackDongleLabel(), #STATECONFIG
	BluetoothPeerLabel(), #STATECONFIG
	ChameleonConnectionLabel(), #LABCONFIG
	ChameleonLabel(), #STATECONFIG
	common_label.OSLabel(),
	DeviceSkuLabel(), #LABCONFIG
	ServoTypeLabel(), #LABCONFIG
	# Temporarily add back as there's no way to reference cr50 configs.
	# See crbug.com/1057145 for the root cause.
	# See crbug.com/1057719 for future tracking.
	Cr50Label(),
	Cr50ROKeyidLabel(),
	]

	LABSTATION_LABELS = [
	common_label.OSLabel(),
	]