import collections
import contextlib
import hashlib
import logging
import uuid
from pathlib import PurePosixPath
from typing import (
Collection,
Dict,
FrozenSet,
Iterator,
List,
Optional,
Set,
Tuple,
)
from urllib.parse import urlparse
from blinkpy.common.host import Host
from blinkpy.common.memoized import memoized
from blinkpy.common.path_finder import PathFinder
from blinkpy.web_tests.models.testharness_results import (
is_all_pass_test_result,
is_testharness_output,
is_wdspec_output,
)
from blinkpy.web_tests.models.test_expectations import TestExpectationsCache
from blinkpy.web_tests.models.typ_types import ResultType
from blinkpy.web_tests.port.base import BaselineLocation, Port
_log = logging.getLogger(__name__)
BaselineLocation.ALL_PASS = BaselineLocation(platform='<all-pass>')
BaselineLocation.BLOCK = BaselineLocation(platform='<block>')
SearchPath = List[BaselineLocation]
DigestMap = Dict[BaselineLocation, 'ResultDigest']
PredecessorMap = Dict[BaselineLocation, Set[BaselineLocation]]
class BaselineOptimizer:
"""Deduplicate baselines that are redundant to `run_web_tests.py`.
`run_web_tests.py` has a fallback mechanism to search different locations
for a test's baseline (i.e., expected text/image dump). The optimizer tries
to minimize checkout size by removing baselines that would be redundant.
See also:
Fallback visualization: https://docs.google.com/drawings/d/13l3IUlSE99RoKjDwEWuY1O77simAhhF6Wi0fZdkSaMA/
Details: https://chromium.googlesource.com/chromium/src/+/main/docs/testing/web_test_baseline_fallback.md
"""
def __init__(self,
host: Host,
default_port: Port,
port_names,
exp_cache: Optional[TestExpectationsCache] = None,
check: bool = False):
self._filesystem = host.filesystem
self._finder = PathFinder(self._filesystem)
self._default_port = default_port
self._host = host
self._ports = []
for port_name in port_names:
self._ports.append(self.port(port_name))
flag_spec_options = (
host.builders.flag_specific_options_for_port_name(port_name))
self._ports.extend(
self.port(port_name, flag_specific)
for flag_specific in flag_spec_options)
self._exp_cache = exp_cache or TestExpectationsCache()
self._check = check
@memoized
def port(self,
port_name: str,
flag_specific: Optional[str] = None) -> Port:
port = self._host.port_factory.get(port_name)
port.set_option_default('manifest_update', False)
if flag_specific:
port.set_option_default('flag_specific', flag_specific)
return port
def optimize(self, test_name: str, suffix: str) -> bool:
if self._check:
_log.info('Checking %s (%s)', test_name, suffix)
else:
_log.info('Optimizing %s (%s)', test_name, suffix)
assert not suffix.startswith('.')
extension = '.' + suffix
with _indent_log():
nonvirtual_test, virtual_tests = self.get_tests_to_optimize(
test_name)
paths = list(
self.generate_search_paths(nonvirtual_test, virtual_tests))
baseline_name = self._default_port.output_filename(
nonvirtual_test, self._default_port.BASELINE_SUFFIX, extension)
digests = self._digest(frozenset().union(*paths), baseline_name,
self._is_reftest(nonvirtual_test))
if not digests:
_log.debug('Nothing to optimize.')
return True
predecessors_by_root = _predecessors_by_root(paths)
can_optimize = any([
self._handle_root(root, predecessors, digests, baseline_name)
for root, predecessors in predecessors_by_root.items()
])
redundant_locations = find_redundant_locations(paths, digests)
can_optimize = can_optimize or len(redundant_locations) > 0
for location in redundant_locations:
self._remove(location, baseline_name, 'redundant')
_log.debug('Digests:')
with _indent_log():
for location in sorted(digests):
if location != BaselineLocation.ALL_PASS:
_log.debug('%s -> %s', location, digests[location])
return not self._check or not can_optimize
def get_tests_to_optimize(self, test_name: str) -> Tuple[str, List[str]]:
"""Translate a test name into all associated tests to optimize.
For convenience, we optimize nonvirtual and virtual tests together.
Returns:
A pair: a nonvirtual test name and any virtual tests to optimize.
The nonvirtual test is always optimized. The virtual tests optimized
are either `test_name` alone (if `test_name` is virtual), or every
valid virtual test associated with the nonvirtual `test_name`.
"""
nonvirtual_test = self._virtual_test_base(test_name)
virtual_tests = []
if not nonvirtual_test:
nonvirtual_test = test_name
for suite in self._default_port.virtual_test_suites():
virtual_test = suite.full_prefix + test_name
if self._virtual_test_base(virtual_test):
virtual_tests.append(virtual_test)
else:
virtual_tests.append(test_name)
return nonvirtual_test, virtual_tests
def generate_search_paths(
self,
nonvirtual_test: str,
virtual_tests: List[str],
) -> Iterator[SearchPath]:
"""Generate search paths taken by each port on each provided test.
Note that:
* Most paths are prefixed by others (e.g., win -> <generic> prefixes
linux -> win -> <generic>). For completeness, this method generates
all of them.
* Ports where a test is skipped will not generate corresponding paths.
"""
is_webexposed = 'webexposed' in _test_path(nonvirtual_test).parts
skipped_ports_by_test = collections.defaultdict(list)
for port in self._ports:
search_path = self._baseline_search_path(port)
nonvirtual_locations = [
self.location(path) for path in search_path
]
if not self.skips_test(port, nonvirtual_test):
yield nonvirtual_locations
else:
skipped_ports_by_test[nonvirtual_test].append(port)
for virtual_test in virtual_tests:
if self.skips_test(port, virtual_test):
skipped_ports_by_test[virtual_test].append(port)
continue
virtual_locations = [
self.location(self._filesystem.join(path, virtual_test))
for path in search_path
]
test_abs_path = self._finder.path_from_web_tests(virtual_test)
virtual_suite = self.location(test_abs_path).virtual_suite
assert {
location.virtual_suite
for location in virtual_locations
} == {virtual_suite}, virtual_locations
if virtual_suite == 'stable' and is_webexposed:
virtual_locations.append(BaselineLocation.BLOCK)
yield virtual_locations + nonvirtual_locations
for test, ports in skipped_ports_by_test.items():
port_names = [
port.get_option('flag_specific') or port.name()
for port in ports
]
_log.debug('Excluding ports that skip "%s": %s', test,
', '.join(port_names))
def skips_test(self, port: Port, test: str) -> bool:
expectations = self._exp_cache.load(port)
results = expectations.get_expectations(test).results
return ResultType.Skip in results or port.skips_test(test)
def write_by_directory(self, results_by_directory, writer, indent):
"""Logs results_by_directory in a pretty format."""
for path in sorted(results_by_directory):
writer('%s%s: %s' % (indent, self.location(path).platform
or '(generic)', results_by_directory[path]))
def read_results_by_directory(self, test_name, baseline_name):
"""Reads the baselines with the given file name in all directories.
Returns:
A dict from directory names to the digest of file content.
"""
locations = set()
for port in self._ports.values():
locations.update(
map(self.location, self._baseline_search_path(port)))
digests = self._digest(locations, baseline_name,
self._is_reftest(test_name))
return {
self.path(location, baseline_name): digest
for location, digest in digests.items()
}
def _digest(
self,
locations: FrozenSet[BaselineLocation],
baseline_name: str,
is_reftest: bool = False,
) -> DigestMap:
digests = {}
for location in locations:
path = self.path(location, baseline_name)
if location == BaselineLocation.BLOCK:
digests[location] = random_digest()
elif self._filesystem.exists(path):
digests[location] = ResultDigest.from_file(
self._filesystem, path, is_reftest)
return digests
def _handle_root(
self,
root: BaselineLocation,
predecessors: FrozenSet[BaselineLocation],
digests: DigestMap,
baseline_name: str,
) -> bool:
"""Determine whether to create or remove a generic baseline.
Returns:
Whether a change could be made on disk.
"""
predecessor_digest = _value_if_same(
[digests.get(predecessor) for predecessor in predecessors])
root_digest = digests.get(root)
if predecessor_digest:
self._promote(root, predecessors, baseline_name)
digests[root] = predecessor_digest
return True
elif root_digest and all(
digests.get(predecessor, root_digest) != root_digest
for predecessor in predecessors):
self._remove(root, baseline_name, 'unreachable')
return True
return False
def _promote(
self,
root: BaselineLocation,
predecessors: FrozenSet[BaselineLocation],
baseline_name: str,
) -> None:
predecessor, *_ = predecessors
source = self.path(predecessor, baseline_name)
dest = self.path(root, baseline_name)
if self._check:
_log.info('Can promote %s from %s', dest,
', '.join(map(str, sorted(predecessors))))
else:
self._filesystem.maybe_make_directory(
self._filesystem.dirname(dest))
self._filesystem.copyfile(source, dest)
_log.debug('Promoted %s from %s', root,
', '.join(map(str, sorted(predecessors))))
def _remove(
self,
location: BaselineLocation,
baseline_name: str,
explanation: str,
) -> None:
path = self.path(location, baseline_name)
if self._check:
_log.info('Can remove %s (%s)', path, explanation)
else:
self._filesystem.remove(path)
_log.debug('Removed %s (%s)', location, explanation)
@memoized
def path(self, location: BaselineLocation, baseline_name: str) -> str:
"""Build an absolute path from a baseline location and name.
This is the inverse of `_location(...)`. The path has the format:
/path/to/web_tests/<location>/<baseline-name>
where:
<location> = (platform/*/|flag-specific/*/)?(virtual/*/)?
"""
parts = []
if location.platform:
parts.extend(['platform', location.platform])
elif location.flag_specific:
parts.extend(['flag-specific', location.flag_specific])
if location.virtual_suite:
parts.extend(['virtual', location.virtual_suite])
return self._finder.path_from_web_tests(*parts, baseline_name)
def _is_reftest(self, test_name: str) -> bool:
return bool(self._default_port.reference_files(test_name))
@memoized
def location(self, filename: str) -> BaselineLocation:
"""Guess a baseline location's parameters from a path."""
location, _ = self._default_port.parse_output_filename(filename)
return location
def _baseline_root(self) -> str:
"""Returns the name of the root (generic) baseline directory."""
return self._default_port.web_tests_dir()
def _baseline_search_path(self, port: Port) -> List[str]:
"""Returns the baseline search path (a list of absolute paths) of the
given port."""
return port.baseline_search_path() + [self._baseline_root()]
def _virtual_test_base(self, test_name: str) -> Optional[str]:
"""Returns the base (non-virtual) version of test_name, or None if
test_name is not virtual."""
return self._default_port.lookup_virtual_test_base(test_name)
class ResultDigest:
"""Digest of a result file for fast comparison.
A result file can be any actual or expected output from a web test,
including text and image. SHA1 is used internally to digest the file.
"""
_IMPLICIT_EXTRA_RESULT = '<EXTRA>'
def __init__(self,
sha: str = _IMPLICIT_EXTRA_RESULT,
path: Optional[str] = None,
is_extra_result: bool = False):
self.sha = sha
self.path = path
self.is_extra_result = is_extra_result
@classmethod
def from_file(cls, fs, path, is_reftest=False) -> 'ResultDigest':
"""Constructs the digest for a result.
Args:
fs: An instance of common.system.FileSystem.
path: The path to a result file. If None is provided, the result is
an *implicit* extra result.
is_reftest: Whether the test is a reftest.
"""
if path is None:
return cls(cls._IMPLICIT_EXTRA_RESULT, path, is_extra_result=True)
assert fs.exists(path), f'{path!r} does not exist'
if path.endswith(f'.png') and is_reftest:
return cls('', path, is_extra_result=True)
with fs.open_binary_file_for_reading(path) as baseline_file:
contents = baseline_file.read()
is_extra_result = not contents
if path.endswith('.txt'):
try:
contents_text = contents.decode()
if is_testharness_output(contents_text) or is_wdspec_output(
contents_text):
contents_text = contents_text.strip()
if is_all_pass_test_result(contents_text):
is_extra_result = True
contents = contents_text.encode()
except UnicodeDecodeError:
is_extra_result = False
digest = hashlib.sha1(contents).hexdigest()
return cls(digest, path, is_extra_result)
def __eq__(self, other):
if other is None:
return False
if self.sha == self._IMPLICIT_EXTRA_RESULT or other.sha == self._IMPLICIT_EXTRA_RESULT:
return self.is_extra_result and other.is_extra_result
return self.sha == other.sha
def __ne__(self, other):
return not self.__eq__(other)
def __str__(self):
return self.sha[0:7]
def __repr__(self):
is_extra_result = ' EXTRA' if self.is_extra_result else ''
return '<ResultDigest %s%s %s>' % (self.sha, is_extra_result,
self.path)
ResultDigest.ALL_PASS = ResultDigest(is_extra_result=True)
def find_redundant_locations(paths: List[SearchPath],
digests: DigestMap) -> Set[BaselineLocation]:
"""Find baseline locations that are redundant and can be safely removed.
At a high level, this is done by checking for baselines that, if deleted,
would not affect the resolution of any path (see Example 1). This procedure
repeats for as long as there were new baselines to delete because deleting a
baseline can enable further deletions (see Example 2).
Arguments:
paths: A list of paths to check against.
digests: Maps baseline locations that exist as files to their values.
Example 1:
(generic): a -:-> linux: a -> win: a -> (generic): a
\ : / /
Virtual +---:-+-----------+ Nonvirtual
* Virtual root has nonvirtual 'linux' and 'win' as successors. All have
value 'a', so the virtual root is removed. Virtual 'linux' and 'win'
still resolve to 'a'.
* Nonvirtual 'linux' and 'win' are removed similarly, leaving only the
nonvirtual root.
Example 2:
linux: a -> win: b -:-> linux: a -> win: b
\ : /
Virtual +---:-------------+ Nonvirtual
* Virtual 'win' is removed because it only has nonvirtual 'win' as a
successor (virtual 'linux' provides its own baseline).
* Virtual 'linux' is removed on the next iteration because it now has
nonvirtual 'linux' as a successor.
"""
predecessors = _predecessors(paths)
return max(
_find_redundant_locations_with_order(
paths, digests, list(_visit_postorder(predecessors))),
_find_redundant_locations_with_order(
paths, digests, list(_visit_preorder(predecessors))),
key=len,
)
def _find_redundant_locations_with_order(
paths: List[SearchPath],
digests: DigestMap,
removal_order: List[BaselineLocation],
) -> Set[BaselineLocation]:
redundant_locations, digests = set(), dict(digests)
while digests:
new_redundant_location = _find_new_redundant_location(
paths, digests, removal_order)
if new_redundant_location:
digests.pop(new_redundant_location)
redundant_locations.add(new_redundant_location)
else:
break
return redundant_locations
def _find_new_redundant_location(
paths: List[SearchPath],
digests: DigestMap,
removal_order: List[BaselineLocation],
) -> Optional[BaselineLocation]:
"""Find a baseline location to remove, if available.
Baselines are removed one at a time to avoid incompatible removals.
Consider:
linux: a -> win: a -:-> linux: b -> win: a
:
Virtual : Nonvirtual
Either virtual baseline may be removed, but not both. As soon as one is
removed, the other one will no longer be considered redundant on the next
iteration.
When multiple baselines can be removed, as in this example, remove those
for older OSes before newer OSes and the generic baseline.
"""
digests.setdefault(BaselineLocation.ALL_PASS, ResultDigest.ALL_PASS)
dependencies: Dict[BaselineLocation, Set[BaselineLocation]]
dependencies = collections.defaultdict(set)
for path in paths:
path = [*path, BaselineLocation.ALL_PASS]
assert len(set(path)) == len(path), ('duplicate location in path %s' %
path)
with contextlib.suppress(ValueError):
source, next_source, *_ = filter(digests.get, path)
dependencies[source].add(next_source)
for location in removal_order:
successors = dependencies.get(location)
if successors and _value_if_same(
map(digests.get, [location, *successors])):
return location
return None
def _visit_preorder(
predecessors: PredecessorMap,
current: BaselineLocation = BaselineLocation.ALL_PASS,
visited: Optional[Set[BaselineLocation]] = None,
) -> Iterator[BaselineLocation]:
visited = visited or set()
if current in visited:
return
visited.add(current)
yield current
for predecessor in predecessors[current]:
yield from _visit_preorder(predecessors, predecessor, visited)
def _visit_postorder(
predecessors: PredecessorMap,
current: BaselineLocation = BaselineLocation.ALL_PASS,
visited: Optional[Set[BaselineLocation]] = None,
) -> Iterator[BaselineLocation]:
visited = visited or set()
if current in visited:
return
visited.add(current)
for predecessor in predecessors[current]:
yield from _visit_postorder(predecessors, predecessor, visited)
yield current
def _predecessors(paths: List[SearchPath]) -> PredecessorMap:
"""Map each location to its immediate predecessors."""
predecessors = collections.defaultdict(set)
for path in paths:
path = [*path, BaselineLocation.ALL_PASS]
for predecessor, successor in zip(path[:-1], path[1:]):
predecessors[successor].add(predecessor)
return predecessors
def _predecessors_by_root(paths: List[SearchPath]) -> PredecessorMap:
"""Map baseline roots (virtual or not) to their immediate predecessors."""
return {
maybe_root: predecessors
for maybe_root, predecessors in _predecessors(paths).items()
if maybe_root.root
}
def _value_if_same(digests: Collection[Optional[ResultDigest]]
) -> Optional[ResultDigest]:
"""Get the value of a collection of digests if they are the same."""
if digests:
first, *digests = digests
if all(first == digest for digest in digests):
return first
return None
@contextlib.contextmanager
def _indent_log(prefix: str = ' ' * 2) -> Iterator[None]:
record_factory = logging.getLogRecordFactory()
def make_indented_record(name,
level,
fn,
lno,
msg,
args,
exc_info,
func=None,
sinfo=None,
**kwargs):
return record_factory(name, level, fn, lno, prefix + msg, args,
exc_info, func, sinfo, **kwargs)
try:
logging.setLogRecordFactory(make_indented_record)
yield
finally:
logging.setLogRecordFactory(record_factory)
def _test_path(test: str) -> PurePosixPath:
return PurePosixPath(urlparse(test).path)
def random_digest() -> ResultDigest:
"""Synthesize a digest that is guaranteed to not equal any other.
The purpose of this digest is to simulate a baseline that prevents
predecessors from being removed.
"""
return ResultDigest(uuid.uuid4().hex)
