1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from functools import reduce
from typing import Dict, List, Optional, Set
class FailedPlatform:
"""
Stores all failures on different build types and test variants for a single platform.
This allows us to detect when a platform failed on all build types or all test variants to
generate a simpler skip-if condition.
"""
def __init__(
self,
# Keys are build types, values are test variants for this build type
# Tests variants can be composite by using the "+" character
# eg: a11y_checks+swgl
# each build_type[test_variant] has a {'pass': x, 'fail': y}
# x and y represent number of times this was run in the last 30 days
# See examples in
# https://firefox-ci-tc.services.mozilla.com/api/index/v1/task/gecko.v2.mozilla-central.latest.source.test-info-all/artifacts/public%2Ftest-info-testrun-matrix.json
oop_permutations: Optional[
Dict[
str, # Build type
Dict[str, Dict[str, int]], # Test Variant # {'pass': x, 'fail': y}
]
],
) -> None:
# Contains all test variants for each build type the task failed on
self.failures: Dict[str, Set[str]] = {}
self.oop_permutations = oop_permutations
def get_possible_build_types(self) -> List[str]:
return (
list(self.oop_permutations.keys())
if self.oop_permutations is not None
else []
)
def get_possible_test_variants(self, build_type: str) -> List[str]:
permutations = (
self.oop_permutations.get(build_type, {})
if self.oop_permutations is not None
else []
)
return [tv for tv in permutations]
def is_full_fail(self) -> bool:
"""
Test if failed on every test variant of every build type
"""
build_types = set(self.failures.keys())
possible_build_types = self.get_possible_build_types()
# If we do not have information on possible build types, do not consider it a full fail
# This avoids creating a too broad skip-if condition
if len(possible_build_types) == 0:
return False
return all(
[
bt in build_types and self.is_full_test_variants_fail(bt)
for bt in possible_build_types
]
)
def is_full_test_variants_fail(self, build_type: str) -> bool:
"""
Test if failed on every test variant of given build type
"""
failed_variants = self.failures.get(build_type, [])
possible_test_variants = self.get_possible_test_variants(build_type)
# If we do not have information on possible test variants, do not consider it a full fail
# This avoids creating a too broad skip-if condition
if len(possible_test_variants) == 0:
return False
return all([t in failed_variants for t in possible_test_variants])
def get_negated_variant(self, test_variant: str):
if not test_variant.startswith("!"):
return "!" + test_variant
return test_variant.replace("!", "", 1)
def get_no_variant_conditions(self, and_str: str, build_type: str):
"""
The no_variant test variant does not really exist and is only internal.
This function gets all available test variants for the given build type
and negates them to create a skip-if that handle tasks without test variants
"""
variants = [
tv
for tv in self.get_possible_test_variants(build_type)
if tv != "no_variant"
]
return_str = ""
for tv in variants:
return_str += and_str + self.get_negated_variant(tv)
return return_str
def get_test_variant_condition(
self, and_str: str, build_type: str, test_variant: str
):
"""
If the given test variant is part of another composite test variant, then add negations matching that composite
variant to prevent overlapping in skips.
eg: test variant "a11y_checks" is to be added while "a11y_checks+swgl" exists
the resulting condition will be "a11y_checks && !swgl"
"""
all_test_variants_parts = [
tv.split("+")
for tv in self.get_possible_test_variants(build_type)
if tv not in ["no_variant", test_variant]
]
test_variant_parts = test_variant.split("+")
# List of composite test variants more specific than the current one
matching_variants_parts = [
tv_parts
for tv_parts in all_test_variants_parts
if all(x in tv_parts for x in test_variant_parts)
]
variants_to_negate = [
part
for tv_parts in matching_variants_parts
for part in tv_parts
if part not in test_variant_parts
]
return_str = reduce((lambda x, y: x + and_str + y), test_variant_parts, "")
return_str = reduce(
(lambda x, y: x + and_str + self.get_negated_variant(y)),
variants_to_negate,
return_str,
)
return return_str
def get_test_variant_string(self, test_variant: str):
"""
Some test variants strings need to be updated to match what is given in oop_permutations
"""
if test_variant == "no-fission":
return "!fission"
if test_variant == "1proc":
return "!e10s"
return test_variant
def get_skip_string(self, and_str: str, build_type: str, test_variant: str) -> str:
if self.failures.get(build_type) is None:
self.failures[build_type] = {test_variant}
else:
self.failures[build_type].add(test_variant)
return_str = ""
# If every test variant of every build type failed, do not add anything
if not self.is_full_fail():
return_str += and_str + build_type
if not self.is_full_test_variants_fail(build_type):
if test_variant == "no_variant":
return_str += self.get_no_variant_conditions(and_str, build_type)
else:
return_str += self.get_test_variant_condition(
and_str, build_type, test_variant
)
return return_str
|