"""Utility to support parsing a TestStat.""" from collections import defaultdict from dataclasses import dataclass from datetime import datetime from itertools import chain from typing import NamedTuple, List, Callable, Optional from evergreen import EvergreenApi, TestStats from buildscripts.util.testname import split_test_hook_name, is_resmoke_hook, get_short_name_from_test_file TASK_LEVEL_HOOKS = {"CleanEveryN"} class TestRuntime(NamedTuple): """ Container for the runtime of a test. test_name: Name of test. runtime: Average of runtime of test. """ test_name: str runtime: float @dataclass class _RuntimeHistory: """ History of runtime results. duration: Average duration of test runtime. num_runs: Number of test runs seen. """ duration: float num_runs: int @classmethod def empty(cls) -> "_RuntimeHistory": """Create an empty runtime entry.""" return cls(duration=0.0, num_runs=0) def add_runtimes(self, duration: float, num_runs: int) -> None: """ Add the given duration number this history. :param duration: Average duration to include. :param num_runs: Number of runs to include. """ self.duration = _average(self.duration, self.num_runs, duration, num_runs) self.num_runs += num_runs def normalize_test_name(test_name: str) -> str: """Normalize test names that may have been run on windows or unix.""" return test_name.replace("\\", "/") def _average(value_a: float, num_a: int, value_b: float, num_b: int) -> float: """Compute a weighted average of 2 values with associated numbers.""" divisor = num_a + num_b if divisor == 0: return 0 else: return float(value_a * num_a + value_b * num_b) / divisor class HistoricHookInfo(NamedTuple): """Historic information about a test hook.""" hook_id: str num_pass: int avg_duration: float @classmethod def from_test_stats(cls, test_stats: TestStats) -> "HistoricHookInfo": """Create an instance from a test_stats object.""" return cls(hook_id=test_stats.test_file, num_pass=test_stats.num_pass, avg_duration=test_stats.avg_duration_pass) def test_name(self) -> str: """Get the name of the test associated with this hook.""" return split_test_hook_name(self.hook_id)[0] def hook_name(self) -> str: """Get the name of this hook.""" return split_test_hook_name(self.hook_id)[-1] def is_task_level_hook(self) -> bool: """Determine if this hook should be counted against the task not the test.""" return self.hook_name() in TASK_LEVEL_HOOKS class HistoricTestInfo(NamedTuple): """Historic information about a test.""" test_name: str num_pass: int avg_duration: float hooks: List[HistoricHookInfo] @classmethod def from_test_stats(cls, test_stats: TestStats, hooks: List[HistoricHookInfo]) -> "HistoricTestInfo": """Create an instance from a test_stats object.""" return cls(test_name=test_stats.test_file, num_pass=test_stats.num_pass, avg_duration=test_stats.avg_duration_pass, hooks=hooks) def normalized_test_name(self) -> str: """Get the normalized version of the test name.""" return normalize_test_name(self.test_name) def total_hook_runtime(self, predicate: Optional[Callable[[HistoricHookInfo], bool]] = None) -> float: """Get the average runtime of all the hooks associated with this test.""" if not predicate: predicate = lambda _: True return sum([hook.avg_duration for hook in self.hooks if predicate(hook)]) def total_test_runtime(self) -> float: """Get the average runtime of this test and it's non-task level hooks.""" return self.avg_duration + self.total_hook_runtime(lambda h: not h.is_task_level_hook()) def get_hook_overhead(self) -> float: """Get the average runtime of this test and it's non-task level hooks.""" return self.total_hook_runtime(lambda h: h.is_task_level_hook()) class HistoricTaskData(object): """Represent the test statistics for the task that is being analyzed.""" def __init__(self, historic_test_results: List[HistoricTestInfo]) -> None: """Initialize the TestStats with raw results from the Evergreen API.""" self.historic_test_results = historic_test_results # pylint: disable=too-many-arguments @classmethod def from_evg(cls, evg_api: EvergreenApi, project: str, start_date: datetime, end_date: datetime, task: str, variant: str) -> "HistoricTaskData": """ Retrieve test stats from evergreen for a given task. :param evg_api: Evergreen API client. :param project: Project to query. :param start_date: Start date to query. :param end_date: End date to query. :param task: Task to query. :param variant: Build variant to query. :return: Test stats for the specified task. """ days = (end_date - start_date).days historic_stats = evg_api.test_stats_by_project( project, after_date=start_date, before_date=end_date, tasks=[task], variants=[variant], group_by="test", group_num_days=days) return cls.from_stats_list(historic_stats) @classmethod def from_stats_list(cls, historic_stats: List[TestStats]) -> "HistoricTaskData": """ Build historic task data from a list of historic stats. :param historic_stats: List of historic stats to build from. :return: Historic task data from the list of stats. """ hooks = defaultdict(list) for hook in [stat for stat in historic_stats if is_resmoke_hook(stat.test_file)]: historical_hook = HistoricHookInfo.from_test_stats(hook) hooks[historical_hook.test_name()].append(historical_hook) return cls([ HistoricTestInfo.from_test_stats(stat, hooks[get_short_name_from_test_file(stat.test_file)]) for stat in historic_stats if not is_resmoke_hook(stat.test_file) ]) def get_tests_runtimes(self) -> List[TestRuntime]: """Return the list of (test_file, runtime_in_secs) tuples ordered by decreasing runtime.""" tests = [ TestRuntime(test_name=test_stats.normalized_test_name(), runtime=test_stats.total_test_runtime()) for test_stats in self.historic_test_results ] return sorted(tests, key=lambda x: x.runtime, reverse=True) def get_avg_hook_runtime(self, hook_name: str) -> float: """Get the average runtime for the specified hook.""" hook_instances = list( chain.from_iterable([[hook for hook in test.hooks if hook.hook_name() == hook_name] for test in self.historic_test_results])) if not hook_instances: return 0 return sum([hook.avg_duration for hook in hook_instances]) / len(hook_instances) def __len__(self) -> int: """Get the number of historical entries.""" return len(self.historic_test_results)