blob: 4be58ac97895fa51d06af2f238d5273825987a6a [file] [log] [blame]
#!/usr/bin/env python3
# Copyright 2018-2020 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.
#
"""Generate LAVA job definitions to test a software build.
generate-lava-jobs enables to generate job definitions for Tradefed-based test
suites (CTS, GTS, STS) and Fairphone's Sanity Check (based on the SmartViser
ViSer app). Additionally, it allows to generate upgrade path test jobs. The
latter validate that a list of OTA images successfully applies one by one and
leads from an initial build to the final build under test. Optionally, Sanity
Check can be run on each single build on the path. Tradefed-based test suites
will run only on the final build, not on each step on the upgrade path.
Environment variables:
ARTIFACTORIAL_TOKEN: Secret token for authenticating at Artifactorial. Test
artifacts are exported to this service, because the test environments
are non-persistent. If this is not specified, Artifactorial upload will
be skipped and artifacts will get lost once test runs finish.
"""
import argparse
import itertools
import os
from pathlib import Path, PurePosixPath
import sys
from typing import Any, ClassVar, Dict, List, Optional, Set, Tuple, Type
from urllib.parse import urljoin
from jinja2 import Environment, FileSystemLoader
import yaml
RESOURCE_HOST = "http://172.16.17.12:8080"
HOST_IMAGES_ROOT_PATH = PurePosixPath("/opt/fairphone")
class HardwareConfig:
# pylint: disable=too-few-public-methods
"""Hardware configuration of a device type.
A hardware configuration is a specific combination of hardware modules or
revisions of those modules. It is only meaningful in context of a specific
device type.
Attributes:
name: Name of the configuration.
tags: LAVA tags required to select devices of the configuration. May be
empty, so that all devices of the current type are selected.
"""
name: str
tags: List[str]
def __init__(self, name: str, tags: List[str]):
"""Create a HardwareConfig instance.
Arguments:
name: Name of the configuration.
tags: LAVA tags required to select devices of the configuration.
"""
self.name = name
self.tags = tags
@classmethod
def parse_from_yaml(
cls: Type["HardwareConfig"], yaml_hw_configs: Dict[str, Dict[str, Any]]
) -> Dict[str, "HardwareConfig"]:
"""Parse hardware configurations from device type YAML.
Arguments:
yaml_hw_configs: Nested dictionary from a device type YAML
describing available hardware configurations.
Returns:
List of HardwareConfigs representing the values from the original
YAML dictionary.
"""
return {
name: cls(name=name, tags=values["tags"])
for name, values in yaml_hw_configs.items()
}
class DeviceTypeConfig:
# pylint: disable=too-few-public-methods
"""Configurations and information related to a specific device type.
Attributes:
device_type: Device type that is configured.
hardware_configs:
Combinations of hardware components or revisions to run tests on.
Each configuration is expressed by a list of LAVA device tags. When
requesting a specific configuration, tests will only run on devices
that have all tags of the configuration assigned to them.
upgrade_paths:
List of upgrade paths. Upgrade paths are represented here as a list
of build IDs, starting from the initial build that is cleanly
installed to a device up to the last build on the path.
builds:
Dictionary pointing from build IDs to named attributes of those
builds. This must at least contain all builds that are part of
`upgrade_paths`.
otas:
Nested dictionary mapping from source builds to target builds and
finally to attributes of the OTAs between two builds.
"""
device_type: str
hardware_configs: Dict[str, HardwareConfig]
upgrade_paths: List[List[str]]
builds: Dict[str, Dict[str, str]]
otas: Dict[str, Dict[str, Dict[str, str]]]
def __init__(self, config_dir: Path, device_type: str):
"""Initialize an instance from a YAML file for the device type.
Arguments:
config_dir: Root path to read configurations from.
device_type: Device type to represent by this instance.
"""
yaml_path = config_dir / "{}.yaml".format(device_type)
if not yaml_path.exists():
print(
"ERROR: No configuration found for the specified device type: "
"{}.".format(device_type)
)
sys.exit(1)
yaml_data = yaml.safe_load(yaml_path.read_text())
self.device_type = device_type
try:
self.hardware_configs = HardwareConfig.parse_from_yaml(
yaml_data["hardware_configs"]
)
self.builds = yaml_data["builds"]
self.otas = yaml_data["otas"]
self.upgrade_paths = yaml_data["upgrade_paths"]
except KeyError as e:
print(
"ERROR: YAML configuration for device type {} is incomplete: {}"
"".format(device_type, e)
)
sys.exit(1)
def get_upgrade_paths_for_build(
self, build_id: str
) -> Set[Tuple[str, ...]]:
"""List all upgrade paths for a build.
This selects out of all upgrade paths defined for a device type the ones
that include the selected build and trims them down to upgrade only
up to that build. Returned paths consist of at least two builds (thus at
least one upgrade step).
Arguments:
build_id: String identifier of the build under test.
Returns:
Set of upgrade paths.
"""
return {
tuple(path[0 : path.index(build_id) + 1])
for path in self.upgrade_paths
if build_id in path and path.index(build_id) > 0
}
def select_hardware_configs(
self, config_names: Optional[List[str]]
) -> List[HardwareConfig]:
"""Build list of selected hardware configuration names.
Arguments:
config_names: A list of configuration names. If none are given, all
but the "generic" one are returned.
Returns:
List of selected and valid hardware configurations.
"""
if not config_names:
return [
config
for config in self.hardware_configs.values()
if config.name != "generic"
]
try:
return [self.hardware_configs[name] for name in config_names]
except KeyError as e:
print(
"ERROR: Non-existent hardware configuration selected for "
"device type {}: {}".format(self.device_type, e)
)
sys.exit(1)
class TestSuite:
"""A test suite to be run in LAVA.
Different types of test suites are represented by leaf subclasses of
TestSuite.
Class variables:
name: The test suite name.
templates: the test suite templates.
Attributes:
device_type: LAVA device type to run tests on.
build: ID of the build under test.
git_branch:
Branch to checkout in the test-definitions repository. Leave empty
to use the repository's default branch.
artifactorial_token:
Secret token for authenticating at Artifactorial. Leave empty to
omit artifact upload.
"""
name: ClassVar[str]
templates: ClassVar[List[str]]
device_type: str
build: str
android_release: str
git_branch: Optional[str]
artifactorial_token: Optional[str]
@classmethod
def suites(cls: Type["TestSuite"]) -> List[Type["TestSuite"]]:
"""List all concrete TestSuite classes.
Returns:
List of all leaf classes of cls.
"""
suites = []
for subclass in cls.__subclasses__():
if not subclass.__subclasses__():
suites.append(subclass)
else:
suites += subclass.suites()
return suites
@classmethod
def check_args(cls, args: argparse.Namespace) -> bool:
# pylint: disable=unused-argument
# unused-argument: args parameter available for overriding methods
"""Check arguments that are additionally required for specific suites.
Arguments that are always required do not need to be checked; they
should be enforced by the argument parser.
Arguments:
args: Arguments for configuring the test suite.
Returns:
Boolean indicating if all required arguments are set.
"""
return True
@classmethod
def select_suites(
cls: Type["TestSuite"], suite_names: Optional[List[str]] = None
) -> List[Type["TestSuite"]]:
"""Select the test suites matching the given names.
Arguments:
suite_names:
A list of suite names. If none are given, all suites are
selected. Name matching is case-insensitive.
Returns:
The list of TestSuite which names match the given names.
"""
if not suite_names:
return cls.suites()
all_suites = {suite.name.lower(): suite for suite in cls.suites()}
matching_suites = []
for suite_name in suite_names:
suite = all_suites.get(suite_name.lower())
if suite:
matching_suites.append(suite)
else:
print(
"Ignoring unknown suite name `{}`.".format(suite_name),
file=sys.stderr,
)
return matching_suites
@classmethod
def instantiate(
cls: Type["TestSuite"],
args: argparse.Namespace,
device_type_config: DeviceTypeConfig,
) -> List["TestSuite"]:
# pylint: disable=unused-argument
# unused-argument: device_type_config parameter available for overriding
# methods
"""Generate instances of test suites.
This must be called only on concrete subclasses of TestSuite, as
generated by suites and select_suites. Subclasses may generate one or
multiple instances, depending on the nature and parameters of the test
suite.
Arguments:
args:
Arguments for configuring the suites. args must contain the
required arguments for all selected suites. Use check_args
beforehand to check in a user-friendly way for missing
arguments.
Returns:
List of configured, ready-to-use instances of cls.
"""
return [cls(args=args)]
def __init__(self, args: argparse.Namespace):
"""Initialize a TestSuite instance.
Arguments:
args: Arguments for configuring the test suite.
"""
self.device_type = args.device_type
self.build = args.build
self.android_release = args.android_release
self.git_branch = args.git_branch
self.artifactorial_token = args.artifactorial_token
def render(
self,
template_dir: Path,
output_dir: Path,
extra_job_name_component: Optional[str] = None,
extra_template_kwargs: Optional[Dict[str, Any]] = None,
):
"""Render YAML templates to concrete LAVA test job YAML(s).
Arguments:
template_dir: Path where YAML templates are stored.
output_dir: Path where rendered test jobs will be stored in.
extra_job_name_component: Add an indicator for the job variant
created with extra_template_kwargs to the job name.
extra_template_kwargs: Extra template parameters for the test job,
allowing to render a variant of what the TestJob instance
describes.
"""
env = Environment(
trim_blocks=True,
lstrip_blocks=True,
loader=FileSystemLoader(str(template_dir)),
)
common_vars = yaml.safe_load(
(template_dir / "common-vars.yaml").read_text()
)
template_kwargs = {
**self._template_kwargs(),
**(extra_template_kwargs if extra_template_kwargs else {}),
**common_vars,
}
for template_filename in self.templates:
template_path = Path(template_filename)
template = env.get_template(template_filename)
output_basename = self._output_file_basename(
template_file_base_name=template_path.stem,
extra_job_name_component=extra_job_name_component,
)
output_file = output_dir / "{}{}".format(
output_basename, template_path.suffix
)
with output_file.open("w") as out:
out.write(template.render(**template_kwargs))
def _template_kwargs(self) -> Dict[str, Any]:
"""Define extra variables for YAML template rendering.
Define a key-value-mapping of variables required for rendering
suite-specific YAML templates.
Returns:
Dictionary mapping from names of variables used in YAML templates to
their values.
"""
return {
"device_type": self.device_type,
"build": self.build,
"android_release": self.android_release,
"suite": self.name,
"git_branch": self.git_branch,
"artifactorial_token": self.artifactorial_token,
"resource_host_base_url": RESOURCE_HOST,
}
def _output_file_basename(
self,
template_file_base_name: str,
extra_job_name_component: Optional[str] = None,
) -> str:
"""Build the base name for the output file to write.
This name shall be unique for each test instance across device types,
hardware configurations, test suites and instances of test suites.
Subclasses can override _output_file_job_name to adjust the final,
job-specific part of the name.
Arguments:
template_file_base_name:
Base name of the template file that is to be rendered.
extra_job_name_component:
Part of the job name that indicates a specific variant of the
test.
Returns:
Base name of the rendered output file.
"""
extra_component_formatted = (
"{}-".format(extra_job_name_component)
if extra_job_name_component
else ""
)
return "{}-{}{}-{}".format(
self.device_type,
extra_component_formatted,
self.build,
self._output_file_job_name(template_file_base_name),
)
def _output_file_job_name(self, template_file_base_name: str) -> str:
# pylint: disable=no-self-use
"""Part of the output file name that describes the test job.
Arguments:
template_file_base_name: Base name of the template file that is to
be rendered.
Returns:
Job indicator that will be part of the complete output file name.
"""
return template_file_base_name
class TradefedTestSuite(TestSuite):
"""A Tradefed-based test suite to be run in LAVA.
Class variables:
version: The test suite version.
archive: The absolute path to the test suite archive.
"""
version: ClassVar[str]
archive: ClassVar[str]
def _template_kwargs(self) -> Dict[str, Any]:
return {
**super()._template_kwargs(),
"version": self.version,
"archive": self.archive,
}
class CTS(TradefedTestSuite):
"""LAVA test jobs for the Android Compatibility Test Suite (CTS)."""
name = "CTS"
version = "9.0_r11"
archive = urljoin(
RESOURCE_HOST,
"android/cts/android-cts-{}-linux_x86-arm.zip".format(version),
)
templates = [
"multinode-tradefed-cts-appsecurityhost.yaml",
"multinode-tradefed-cts-deqp.yaml",
"multinode-tradefed-cts-media.yaml",
"multinode-tradefed-cts-net.yaml",
"multinode-tradefed-cts-others.yaml",
]
class GTS(TradefedTestSuite):
"""LAVA test jobs for the Google Mobile Services Test Suite (GTS)."""
name = "GTS"
version = "7.0_r5"
archive = urljoin(RESOURCE_HOST, "android/gts/android-gts-7_r5-6497315.zip")
templates = ["multinode-tradefed-gts.yaml"]
class STS(TradefedTestSuite):
"""LAVA test jobs for Google's Security Test Suite for Android."""
name = "STS"
version = "9.0_202005"
archive = urljoin(
RESOURCE_HOST,
"android/sts/android-sts-{}-linux-arm-no-password.zip".format(version),
)
templates = [
"multinode-tradefed-sts-host.yaml",
"multinode-tradefed-sts-securitybulletinhost.yaml",
"multinode-tradefed-sts-security.yaml",
]
class SanityCheck(TestSuite):
"""The Sanity Check wrapped in a LAVA test job.
Attributes:
viser_scenarios_data_url:
Source URL for downloading viSer scenario data.
"""
name = "SanityCheck"
templates = ["sanity-check.yaml"]
viser_scenarios_data_url: str
def __init__(self, args: argparse.Namespace):
"""Initialize a SanityCheck instance.
Arguments:
args: Arguments for configuring the test suite.
"""
super().__init__(args=args)
self.viser_scenarios_data_url = args.viser_scenarios_data_url
def _template_kwargs(self) -> Dict[str, Any]:
return {
**super()._template_kwargs(),
"viser_scenarios_data_url": self.viser_scenarios_data_url,
}
@classmethod
def check_args(cls, args: argparse.Namespace) -> bool: # noqa: D102
# D102: docstring is inherited
return super().check_args(args) and "viser_scenarios_data_url" in args
class UpgradePathTestSuite(TestSuite):
"""An upgrade path to test.
Attributes:
path:
Tuple of build IDs defining an upgrade path. It starts with the
initial build to be deployed to devices and ends with the final
build under test.
builds:
Dictionary pointing from build IDs to named attributes of those
builds. This must at least contain all builds that are part of
`path`.
otas:
Nested dictionary mapping from source builds to target builds and
finally to attributes of the OTAs between two builds.
viser_scenarios_data_url:
Source URL for downloading viSer scenario data.
execute_sanity:
Boolean indicating whether Sanity Check will be executed for
verifying upgrade steps.
job_timeout_hours:
Total maximum job runtime. Derived from the number of upgrade steps
and whether Sanity Check is executed or not.
"""
name = "UpgradePath"
templates = ["upgrade-path.yaml"]
path: Tuple[str, ...]
builds: Dict[str, Dict[str, str]]
otas: Dict[str, Dict[str, Dict[str, str]]]
viser_scenarios_data_url: Optional[str]
execute_sanity: bool
job_timeout_hours: int
@classmethod
def check_args(cls, args: argparse.Namespace) -> bool: # noqa: D102
# D102: docstring is inherited
if not super().check_args(args):
return False
if args.dev_build:
print(
"DEV_BUILD parameter cannot be combined with UpgradePath test "
'suite (may be implicitly selected with "all" suites).'
)
return False
if "viser_scenarios_data_url" not in args:
print(
"INFO: viser_scenarios_data_url not specified. UpgradePath "
"test suite will run without SanityCheck."
)
return True
@classmethod
def instantiate( # noqa: D102
cls: Type["UpgradePathTestSuite"],
args: argparse.Namespace,
device_type_config: DeviceTypeConfig,
) -> List[TestSuite]:
# D102: docstring is inherited
paths = device_type_config.get_upgrade_paths_for_build(args.build)
return [
cls(
args=args,
path=path,
builds=device_type_config.builds,
otas=device_type_config.otas,
)
for path in paths
]
def _template_kwargs(self) -> Dict[str, Any]:
kwargs = super()._template_kwargs()
if self.viser_scenarios_data_url:
kwargs["viser_scenarios_data_url"] = self.viser_scenarios_data_url
return {**kwargs, "upgrade_path": self.__dict__}
def __init__(
self,
args: argparse.Namespace,
path: Tuple[str, ...],
builds: Dict[str, Dict[str, str]],
otas: Dict[str, Dict[str, Dict[str, str]]],
):
"""Initialize an UpgradePathTestSuite instance.
Arguments:
args: Arguments for configuring the test suite.
path: Tuple of build IDs defining the upgrade path.
builds: Dictionary of available builds and their properties.
ota: Dictionary of available OTAs and their properties.
"""
super().__init__(args=args)
self.path = path
self.builds = builds
self.otas = otas
self.viser_scenarios_data_url = args.viser_scenarios_data_url
self.execute_sanity = bool(self.viser_scenarios_data_url)
def _output_file_job_name(self, template_file_base_name: str) -> str:
return "{}_from_{}".format(
super()._output_file_job_name(template_file_base_name),
"_".join(self.path[0:-1]),
)
class TestSuiteManager:
"""Manager for selecting and rendering TestSuites.
This class allows to select TestSuite classes by name and render them into
YAML LAVA job submissions. It encapsulates the concrete TestSuite subclasses
and their test-suite-specific behavior.
"""
@classmethod
def render_selected_suites(
cls,
template_dir: Path,
output_dir: Path,
args: argparse.Namespace,
device_type_config: DeviceTypeConfig,
):
"""Check arguments, instantiate and render selected TestSuites.
High-level interface for generated TestSuite instances. This function
encapsulates all steps necessary to generate LAVA test jobs based on a
selection of test suites and their required arguments.
Arguments:
template_dir: Path where YAML templates are stored.
output_dir: Path where rendered test jobs will be stored in.
args: Arguments for configuring the test suites.
device_type_config: Configuration of the device type under test.
Returns:
True if all argument are valid and all test suites were rendered
successfully. False otherwise.
"""
selected_suites = cls.select_suites(suite_names=args.suites)
if any(not suite.check_args(args) for suite in selected_suites):
return False
hardware_configs = device_type_config.select_hardware_configs(
args.hardware_configurations
)
instances = list(
itertools.chain.from_iterable(
suite.instantiate(
args=args, device_type_config=device_type_config
)
for suite in selected_suites
)
)
extra_template_kwargs = (
{"dev_build": args.dev_build} if args.dev_build else {}
)
for hardware_config in hardware_configs:
for instance in instances:
instance.render(
template_dir=template_dir,
output_dir=output_dir,
extra_job_name_component=hardware_config.name,
extra_template_kwargs={
**extra_template_kwargs,
"hardware_config": hardware_config,
},
)
return True
@staticmethod
def select_suites(
suite_names: Optional[List[str]] = None
) -> List[Type[TestSuite]]:
"""Select the test suites matching the given names.
Arguments:
suite_names:
A list of suite names. If none are given, all suites are
selected. Name matching is case-insensitive.
Returns:
The list of TestSuite which names match the given names.
"""
if not suite_names:
return TestSuite.suites()
all_suites = {suite.name.lower(): suite for suite in TestSuite.suites()}
matching_suites = []
for suite_name in suite_names:
suite = all_suites.get(suite_name.lower())
if suite:
matching_suites.append(suite)
else:
print(
"Ignoring unknown suite name `{}`.".format(suite_name),
file=sys.stderr,
)
return matching_suites
def parse_args_list_or_all(arg: str) -> Optional[List[str]]:
"""Parse a list argument that allows for "all".
Split a comma-separated list of arguments into a list. If "all" options are
requested, let later steps in the implementation substitute `None` for all
case-specific options.
Arguments:
arg: Comma-separated arguments or "all".
Returns
List of string arguments or `None`, representing all options.
"""
return arg.split(",") if arg != "all" else None
def parse_cmdline_arguments() -> argparse.Namespace:
"""Parse the command line arguments.
Returns:
argparse Namespace containing the parsed command line arguments of this
script.
"""
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"-s",
"--suites",
type=parse_args_list_or_all,
default="all",
help=(
"A comma-separated list of test suites to generate LAVA jobs for."
'Accepts "cts", "gts", "sts", "SanityCheck", "UpgradePath", or the '
'default "all".'
),
)
parser.add_argument(
"build",
type=str,
help=(
'The software build to test, e.g. "18.09.0-gms-0638cf82". When '
"testing an upgrade path, this is the final build on the path."
),
)
parser.add_argument(
"-d",
"--dev-build",
type=str,
help=(
"Generate jobs for a development build: system, boot, recovery and "
"cache images will be downloaded from the `dev/<DEV_BUILD>/` "
"folder on the LAVA data host. All other images will still be "
"downloaded from the `rel/<BUILD>-factory/` folder, assuming that "
"radio images and similar are usually not built during "
"development. Additionally, <DEV_BUILD> will appear in the test "
"jobs names instead of <build>.\n"
"This parameter is only compatible with single-build test suites."
),
)
parser.add_argument(
"-b",
"--git-branch",
type=str,
default="staging/fp",
help=(
"Branch in the qa/test-definitions git repository that test "
"actions are fetched from."
),
)
parser.add_argument(
"-o",
"--output-directory",
type=Path,
default=Path.cwd(),
help=(
"The directory to write the job definitions to. Defaults to the "
"current directory."
),
)
parser.add_argument(
"--device-type",
type=str,
default="fairphone-fp2",
help="LAVA device type to run tests on.",
)
parser.add_argument(
"--hardware-configurations",
type=parse_args_list_or_all,
default="all",
help=(
"Hardware configurations of the selected device type to run tests "
'on. This is "all", "generic" or a comma-separated list of '
"configuration names. A hardware configuration is a selection of "
'specific hardware components or revisions. "generic" lets tests'
"run on any device of the selected device type, regardless of its"
"configuration. Otherwise, all tests suites are instantiated for "
"all selected hardware configurations."
),
)
parser.add_argument(
"--android-release",
type=str,
default="9.0",
help=(
'Android release number of the build to test. Defaults to "9.0".'
),
)
parser.add_argument(
"--viser-scenarios-data-url",
type=str,
help=(
"Source URL for downloading a data archive with files required by "
'Sanity Check. This is required if the "sanity-check" suite is '
"selected via the --suites argument. It will also affect the "
'"upgrade-path" test suite.'
),
)
return parser.parse_args()
def main():
"""Entry point into the generate-lava-jobs command line tool."""
args = parse_cmdline_arguments()
artifactorial_token = os.getenv("ARTIFACTORIAL_TOKEN")
if artifactorial_token:
setattr(args, "artifactorial_token", artifactorial_token)
else:
print(
"WARNING: ARTIFACTORIAL_TOKEN environment variable is not set. "
"artifacts will not be exported and will be lost after test "
"completion."
)
base_dir = Path(sys.path[0])
device_type_config = DeviceTypeConfig(
config_dir=base_dir / "device_types", device_type=args.device_type
)
try:
if not args.output_directory.is_dir():
args.output_directory.mkdir(parents=True)
except FileExistsError:
sys.exit("The output directory already exists, but is not a directory.")
success = TestSuiteManager.render_selected_suites(
template_dir=base_dir / "templates",
output_dir=args.output_directory,
args=args,
device_type_config=device_type_config,
)
if not success:
sys.exit(1)
if __name__ == "__main__":
main()