lava/lava_interface.py - vendor/fairphone/tools - Gitiles

 # Copyright 2019 Fairphone B.V.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #      http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #

 """Utilities for accessing a LAVA instance from Python."""

 from enum import Enum
 from pathlib import Path
 import subprocess
 from typing import Any, Optional, Set

 import yaml


 class LavaError(Exception):
     """Exception raised for failed calls to the LAVA XMLRPC API."""

     def __init__(self, called_process_error: subprocess.CalledProcessError):
         """Initialize a LavaError instance."""
         super().__init__()
         self.called_process_error = called_process_error

     def __str__(self) -> str:
         """Create a string representation of this object."""
         return (
             "Command '{}' failed with exit status {}.\n"
             "stdout: {}\n"
             "stderr: {}".format(
                 self.called_process_error.cmd,
                 self.called_process_error.returncode,
                 self.called_process_error.stdout,
                 self.called_process_error.stderr,
             )
         )


 class Lavacli:
     """Wrapper class for lavacli.

     This class allows to call lavacli from python. It assumes that connection
     and authorization to a LAVA instance is set up in lavacli before using this
     class.
     """

     def __init__(self, identity: str):
         """Initialize a Lavacli instance.

         Arguments:
             identity:
                 lavacli identity for connecting and authorizing at a LAVA
                 instance through lavacli. This identity needs to be configured
                 in lavacli before making calls through this class.
         """
         self._identity = identity

     def cmd(self, *argv: str) -> str:
         """Execute a command through lavacli.

         Arguments:
             argv:
                 List if command line arguments for lavacli.

         Returns:
             Standard output string of the executed command.

         Throws:
             LavaError:
                 If lavacli is not available or quits with non-zero exit code.

         """
         try:
             # subprocess.check_output is not typed, so ignore type checks here.
             return subprocess.check_output(  # type: ignore
                 ["lavacli", "-i", self._identity, *argv], encoding="utf-8"
             )
         except subprocess.CalledProcessError as e:
             # Re-raise the exception in the own type to make semantics clearer.
             raise LavaError(e)

     def cmd_yaml(self, *argv: str) -> Any:
         """Call `cmd` and parse output as YAML document.

         This function assumes that the arguments given in argv instruct lavacli
         to print a YAML document to the standard output (usually achieved with
         `--yaml` parameter in supporting subcommands). The standard output
         string must contain exactly one YAML document with only standard YAML
         tags.

         Refer to `cmd` for arguments and exception specification.

         Returns:
             Python object constructed from the standard output of the executed
             command.

         """
         return yaml.safe_load(self.cmd(*argv))

     def tags_cmd(self, *argv: str) -> str:
         """Call the 'tag' subcommand in lavacli."""
         return self.cmd("tags", *argv)

     def tags_cmd_yaml(self, *argv: str) -> Any:
         """Call `cmd_yaml` for the 'tag' subcommand in lavacli."""
         return self.cmd_yaml("tags", *argv)

     def device_types_cmd(self, *argv: str) -> str:
         """Call the 'device-types' subcommand in lavacli."""
         return self.cmd("device-types", *argv)

     def device_types_cmd_yaml(self, *argv: str) -> Any:
         """Call `cmd_yaml` for the 'device-types' subcommand in lavacli."""
         return self.cmd_yaml("device-types", *argv)

     def devices_cmd(self, *argv: str) -> str:
         """Call the 'devices' subcommand in lavacli."""
         return self.cmd("devices", *argv)

     def devices_cmd_yaml(self, *argv: str) -> Any:
         """Call `cmd_yaml` for the 'devices' subcommand in lavacli."""
         return self.cmd_yaml("devices", *argv)


 class LavaDeviceHealth(Enum):
     """Enumeration of possible LAVA device health states.

     This enumeration also provides a translation between different string
     representations in lavacli: When setting the device health state, the
     uppercase notation is required. When querying the state, it is notated with
     first letter in uppercase only.

     """

     GOOD = "Good"
     UNKNOWN = "Unknown"
     LOOPING = "Looping"
     BAD = "Bad"
     MAINTENANCE = "Maintenance"
     RETIRED = "Retired"


 class LavaDevice:
     """Wrapper of a device in LAVA."""

     def __init__(
         self,
         hostname: str,
         lavacli: Lavacli,
         health: Optional[LavaDeviceHealth] = None,
     ):
         """Initialize a LavaDevice instance.

         Arguments:
             hostname: Device hostname in LAVA.
             lavacli: lavacli wrapper instance.
             health:
                 Health state of the device. This must be None if the device is
                 not yet present in the LAVA instance and a valid
                 LavaDeviceHealth enum value otherwise.

         """
         self._hostname = hostname
         self._lavacli = lavacli
         self._health = health

     @property
     def hostname(self) -> str:
         """Return the hostname that identifies the device in a LAVA instance."""
         return self._hostname

     @property
     def health(self) -> Optional[LavaDeviceHealth]:
         """Return the cached value of the device health state."""
         return self._health

     @health.setter
     def health(self, health: LavaDeviceHealth) -> None:
         """Set device state if requested and cache value differ."""
         if self.health == health:
             return

         self._lavacli.devices_cmd(
             "update", "--health", health.name, self.hostname
         )
         self._health = health

     def set_tags(self, new_tags: Set[str]) -> bool:
         """Set the list of tags assigned to the device.

         Returns: True if the set of tags changed, False otherwise.

         """
         current_tags = set(
             self._lavacli.devices_cmd_yaml(
                 "tags", "list", "--yaml", self.hostname
             )
         )
         if current_tags == new_tags:
             return False

         for tag in new_tags - current_tags:
             self._lavacli.devices_cmd("tags", "add", self.hostname, tag)
         for tag in current_tags - new_tags:
             self._lavacli.devices_cmd("tags", "delete", self.hostname, tag)
         return True

     def set_device_dict(self, device_dict_file: Path) -> bool:
         """Set the device dictionary.

         Returns: True if the dictionary changed, False otherwise.
         """
         current_dict: Optional[str]
         try:
             # Fetch the dict if it is set. Silently ignore if this fails due to
             # a not yet existing configuration.
             current_dict = self._lavacli.devices_cmd(
                 "dict", "get", self.hostname
             )
         except LavaError as e:
             expected_error = (
                 '''Fault 404: "Device '{}' does not have a configuration"'''
                 "".format(self.hostname)
             )
             if expected_error not in e.called_process_error.stdout:
                 raise e
             current_dict = None

         new_dict = device_dict_file.read_text()
         if new_dict == current_dict:
             return False

         self._lavacli.devices_cmd(
             "dict", "set", self.hostname, str(device_dict_file)
         )
         return True

     def add_update_config(
         self,
         lava_worker: str,
         device_type_name: str,
         tags: Set[str],
         device_dict_file: Path,
     ) -> None:
         """Ensure that the device is present and configured.

         This adds the device to the LAVA instance if it is not present yet and
         ensures that all supplied parameters are up-to-date. If the device was
         previously retired (removed from the LAVA instance), it will be
         reactivated.

         This will trigger a health check on the device if anything needed to be
         changed on the configuration.
         """
         request_health_check = False
         maintenance_params = [
             "--health",
             LavaDeviceHealth.MAINTENANCE.name,
             "--worker",
             lava_worker,
             self.hostname,
         ]
         if self.health is None:  # New to the LAVA instance.
             # Start in maintenance mode so that the health check doesn't run
             # before the device is fully configured.
             self._lavacli.devices_cmd(
                 "add", "--type", device_type_name, *maintenance_params
             )
             request_health_check = True
         else:
             if self.health == LavaDeviceHealth.RETIRED:
                 # Reactivate the device if it was previously retired.
                 request_health_check = True
                 current_worker = None
             else:
                 # A health check will be required if the worker changes.
                 current_worker = self._lavacli.devices_cmd_yaml(
                     "show", "--yaml", self.hostname
                 )["worker"]

             if current_worker != lava_worker:
                 self._lavacli.devices_cmd("update", *maintenance_params)
                 request_health_check = True

         if self.set_tags(new_tags=set(tags)):
             request_health_check = True
         if self.set_device_dict(device_dict_file=device_dict_file):
             request_health_check = True

         # Run a health check if something changed.
         if request_health_check:
             self.health = LavaDeviceHealth.UNKNOWN
	# Copyright 2019 Fairphone B.V.
	#
	# Licensed under the Apache License, Version 2.0 (the "License");
	# you may not use this file except in compliance with the License.
	# You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.
	#

	"""Utilities for accessing a LAVA instance from Python."""

	from enum import Enum
	from pathlib import Path
	import subprocess
	from typing import Any, Optional, Set

	import yaml


	class LavaError(Exception):
	"""Exception raised for failed calls to the LAVA XMLRPC API."""

	def __init__(self, called_process_error: subprocess.CalledProcessError):
	"""Initialize a LavaError instance."""
	super().__init__()
	self.called_process_error = called_process_error

	def __str__(self) -> str:
	"""Create a string representation of this object."""
	return (
	"Command '{}' failed with exit status {}.\n"
	"stdout: {}\n"
	"stderr: {}".format(
	self.called_process_error.cmd,
	self.called_process_error.returncode,
	self.called_process_error.stdout,
	self.called_process_error.stderr,
	)
	)


	class Lavacli:
	"""Wrapper class for lavacli.

	This class allows to call lavacli from python. It assumes that connection
	and authorization to a LAVA instance is set up in lavacli before using this
	class.
	"""

	def __init__(self, identity: str):
	"""Initialize a Lavacli instance.

	Arguments:
	identity:
	lavacli identity for connecting and authorizing at a LAVA
	instance through lavacli. This identity needs to be configured
	in lavacli before making calls through this class.
	"""
	self._identity = identity

	def cmd(self, *argv: str) -> str:
	"""Execute a command through lavacli.

	Arguments:
	argv:
	List if command line arguments for lavacli.

	Returns:
	Standard output string of the executed command.

	Throws:
	LavaError:
	If lavacli is not available or quits with non-zero exit code.

	"""
	try:
	# subprocess.check_output is not typed, so ignore type checks here.
	return subprocess.check_output( # type: ignore
	["lavacli", "-i", self._identity, *argv], encoding="utf-8"
	)
	except subprocess.CalledProcessError as e:
	# Re-raise the exception in the own type to make semantics clearer.
	raise LavaError(e)

	def cmd_yaml(self, *argv: str) -> Any:
	"""Call `cmd` and parse output as YAML document.

	This function assumes that the arguments given in argv instruct lavacli
	to print a YAML document to the standard output (usually achieved with
	`--yaml` parameter in supporting subcommands). The standard output
	string must contain exactly one YAML document with only standard YAML
	tags.

	Refer to `cmd` for arguments and exception specification.

	Returns:
	Python object constructed from the standard output of the executed
	command.

	"""
	return yaml.safe_load(self.cmd(*argv))

	def tags_cmd(self, *argv: str) -> str:
	"""Call the 'tag' subcommand in lavacli."""
	return self.cmd("tags", *argv)

	def tags_cmd_yaml(self, *argv: str) -> Any:
	"""Call `cmd_yaml` for the 'tag' subcommand in lavacli."""
	return self.cmd_yaml("tags", *argv)

	def device_types_cmd(self, *argv: str) -> str:
	"""Call the 'device-types' subcommand in lavacli."""
	return self.cmd("device-types", *argv)

	def device_types_cmd_yaml(self, *argv: str) -> Any:
	"""Call `cmd_yaml` for the 'device-types' subcommand in lavacli."""
	return self.cmd_yaml("device-types", *argv)

	def devices_cmd(self, *argv: str) -> str:
	"""Call the 'devices' subcommand in lavacli."""
	return self.cmd("devices", *argv)

	def devices_cmd_yaml(self, *argv: str) -> Any:
	"""Call `cmd_yaml` for the 'devices' subcommand in lavacli."""
	return self.cmd_yaml("devices", *argv)


	class LavaDeviceHealth(Enum):
	"""Enumeration of possible LAVA device health states.

	This enumeration also provides a translation between different string
	representations in lavacli: When setting the device health state, the
	uppercase notation is required. When querying the state, it is notated with
	first letter in uppercase only.

	"""

	GOOD = "Good"
	UNKNOWN = "Unknown"
	LOOPING = "Looping"
	BAD = "Bad"
	MAINTENANCE = "Maintenance"
	RETIRED = "Retired"


	class LavaDevice:
	"""Wrapper of a device in LAVA."""

	def __init__(
	self,
	hostname: str,
	lavacli: Lavacli,
	health: Optional[LavaDeviceHealth] = None,
	):
	"""Initialize a LavaDevice instance.

	Arguments:
	hostname: Device hostname in LAVA.
	lavacli: lavacli wrapper instance.
	health:
	Health state of the device. This must be None if the device is
	not yet present in the LAVA instance and a valid
	LavaDeviceHealth enum value otherwise.

	"""
	self._hostname = hostname
	self._lavacli = lavacli
	self._health = health

	@property
	def hostname(self) -> str:
	"""Return the hostname that identifies the device in a LAVA instance."""
	return self._hostname

	@property
	def health(self) -> Optional[LavaDeviceHealth]:
	"""Return the cached value of the device health state."""
	return self._health

	@health.setter
	def health(self, health: LavaDeviceHealth) -> None:
	"""Set device state if requested and cache value differ."""
	if self.health == health:
	return

	self._lavacli.devices_cmd(
	"update", "--health", health.name, self.hostname
	)
	self._health = health

	def set_tags(self, new_tags: Set[str]) -> bool:
	"""Set the list of tags assigned to the device.

	Returns: True if the set of tags changed, False otherwise.

	"""
	current_tags = set(
	self._lavacli.devices_cmd_yaml(
	"tags", "list", "--yaml", self.hostname
	)
	)
	if current_tags == new_tags:
	return False

	for tag in new_tags - current_tags:
	self._lavacli.devices_cmd("tags", "add", self.hostname, tag)
	for tag in current_tags - new_tags:
	self._lavacli.devices_cmd("tags", "delete", self.hostname, tag)
	return True

	def set_device_dict(self, device_dict_file: Path) -> bool:
	"""Set the device dictionary.

	Returns: True if the dictionary changed, False otherwise.
	"""
	current_dict: Optional[str]
	try:
	# Fetch the dict if it is set. Silently ignore if this fails due to
	# a not yet existing configuration.
	current_dict = self._lavacli.devices_cmd(
	"dict", "get", self.hostname
	)
	except LavaError as e:
	expected_error = (
	'''Fault 404: "Device '{}' does not have a configuration"'''
	"".format(self.hostname)
	)
	if expected_error not in e.called_process_error.stdout:
	raise e
	current_dict = None

	new_dict = device_dict_file.read_text()
	if new_dict == current_dict:
	return False

	self._lavacli.devices_cmd(
	"dict", "set", self.hostname, str(device_dict_file)
	)
	return True

	def add_update_config(
	self,
	lava_worker: str,
	device_type_name: str,
	tags: Set[str],
	device_dict_file: Path,
	) -> None:
	"""Ensure that the device is present and configured.

	This adds the device to the LAVA instance if it is not present yet and
	ensures that all supplied parameters are up-to-date. If the device was
	previously retired (removed from the LAVA instance), it will be
	reactivated.

	This will trigger a health check on the device if anything needed to be
	changed on the configuration.
	"""
	request_health_check = False
	maintenance_params = [
	"--health",
	LavaDeviceHealth.MAINTENANCE.name,
	"--worker",
	lava_worker,
	self.hostname,
	]
	if self.health is None: # New to the LAVA instance.
	# Start in maintenance mode so that the health check doesn't run
	# before the device is fully configured.
	self._lavacli.devices_cmd(
	"add", "--type", device_type_name, *maintenance_params
	)
	request_health_check = True
	else:
	if self.health == LavaDeviceHealth.RETIRED:
	# Reactivate the device if it was previously retired.
	request_health_check = True
	current_worker = None
	else:
	# A health check will be required if the worker changes.
	current_worker = self._lavacli.devices_cmd_yaml(
	"show", "--yaml", self.hostname
	)["worker"]

	if current_worker != lava_worker:
	self._lavacli.devices_cmd("update", *maintenance_params)
	request_health_check = True

	if self.set_tags(new_tags=set(tags)):
	request_health_check = True
	if self.set_device_dict(device_dict_file=device_dict_file):
	request_health_check = True

	# Run a health check if something changed.
	if request_health_check:
	self.health = LavaDeviceHealth.UNKNOWN