| # Copyright 2019 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import attr |
| import collections |
| |
| from recipe_engine import recipe_api |
| from recipe_engine.config_types import Path |
| |
| from RECIPE_MODULES.fuchsia.swarming_retry import api as swarming_retry_api |
| from RECIPE_MODULES.fuchsia.testing_requests import api as testing_requests_api |
| from RECIPE_MODULES.fuchsia.utils import cached_property, nice_duration, pluralize |
| |
| # Name of the file produced by a testing task that contains results and outputs |
| # of the tests that were run. |
| TEST_SUMMARY_JSON = "summary.json" |
| |
| # What to name the file that contains the swarming task output. |
| TEST_TASK_OUTPUT_FILE = "infra_and_test_std_and_klog.txt" |
| |
| # Flag names from |
| # https://fuchsia.googlesource.com/fuchsia/+/master/tools/testing/tefmocheck/cmd/main.go |
| LOG_NAME_TO_TEFMOCHECK_FLAG = { |
| TEST_TASK_OUTPUT_FILE: "-swarming-output", |
| testing_requests_api.SERIAL_LOG_NAME: "-serial-log", |
| testing_requests_api.SYSLOG_NAME: "-syslog", |
| } |
| |
| |
| @attr.s |
| class FuchsiaTestResults(object): |
| """Represents the result of testing of a Fuchsia build. |
| |
| Attributes: |
| from_fuchsia (bool): Whether the tests ran on Fuchsia. |
| results_dir (Path): The directory that the test results have been moved |
| into (out of task_output_dir). |
| task_output_dir (Path): A directory containing the outputs of the swarming |
| task that ran these tests. Anything that's in this directory will be |
| uploaded to GCS when upload_results() is called. |
| swarming_task_id (str): The ID of the task that ran these tests. |
| env_name (str): The name of the environment that these tags ran in. |
| shard_name (str): The name of the task that ran these tests. |
| api (RecipeApi): The api to use for accessing recipe modules from this |
| object. |
| summary_bytes (bytes): The contents of TEST_SUMMARY_JSON. |
| is_multiplied (bool): Whether the test results are for a multiplier |
| shard. |
| """ |
| |
| from_fuchsia = attr.ib(type=bool) |
| results_dir = attr.ib(type=Path) |
| task_output_dir = attr.ib(type=Path) |
| swarming_task_id = attr.ib(type=str) |
| env_name = attr.ib(type=str) |
| _shard_name = attr.ib(type=str) |
| _api = attr.ib(type=recipe_api.RecipeApi) |
| _summary_bytes = attr.ib(type=bytes) |
| _is_multiplied = attr.ib(type=bool) |
| |
| # A mapping of relative paths to files in the results_dir containing |
| # stdout+stderr data to strings containing those contents. |
| _outputs = attr.ib(factory=dict, init=False) |
| |
| # Set lazily by `_separate_tests_by_result()`, not a parameter to __init__. |
| _pass_name_to_tests = attr.ib(factory=dict, init=False) |
| _fail_name_to_tests = attr.ib(factory=dict, init=False) |
| |
| # Constants representing the result of running a test. These enumerate the |
| # values of the 'results' field of the entries in the summary.json file |
| # obtained from the target device. |
| _TEST_RESULT_PASS = "PASS" |
| _TEST_RESULT_FAIL = "FAIL" |
| |
| @cached_property |
| def summary(self): |
| """The parsed summary file as a Dict or {} if missing.""" |
| if not self._summary_bytes: |
| return {} |
| |
| try: |
| return self._api.json.loads(self._summary_bytes) |
| except ValueError as e: # pragma: no cover |
| # TODO(olivernewman): JSONDecodeError in python >=3.5 |
| raise self._api.step.StepFailure( |
| "Invalid %s: %s" % (TEST_SUMMARY_JSON, e.args[0]) |
| ) |
| |
| @property |
| def passed_tests(self): |
| """All entries in |self._outputs| for tests that passed.""" |
| self._separate_tests_by_result() |
| passed_tests = [] |
| for tests in self._pass_name_to_tests.itervalues(): |
| passed_tests.extend(tests) |
| return passed_tests |
| |
| @property |
| def failed_tests(self): |
| """All entries in |self._outputs| for tests that failed.""" |
| self._separate_tests_by_result() |
| failed_tests = [] |
| for name, tests in self._fail_name_to_tests.iteritems(): |
| if self._is_multiplied or name not in self._pass_name_to_tests: |
| failed_tests.extend(tests) |
| return failed_tests |
| |
| @property |
| def flaked_tests(self): |
| """All entries in |self._outputs| for tests that flaked.""" |
| self._separate_tests_by_result() |
| flaked_tests = [] |
| for name, tests in self._fail_name_to_tests.iteritems(): |
| if not self._is_multiplied and name in self._pass_name_to_tests: |
| flaked_tests.extend(tests) |
| return flaked_tests |
| |
| def _separate_tests_by_result(self): |
| """Separates entries in |self._outputs| into maps based on test result. |
| |
| Passed tests will be stored in self._pass_name_to_tests and failed |
| tests will be stored in self._fail_name_to_tests. |
| """ |
| if not self._pass_name_to_tests or not self._fail_name_to_tests: |
| # Using OrderedDict() will keep the tests in the same order as they |
| # ran in. This doesn't really matter for displaying results, but it |
| # keeps the expectation files from changing by maintaining the same |
| # order of tests. |
| self._pass_name_to_tests = collections.OrderedDict() |
| self._fail_name_to_tests = collections.OrderedDict() |
| for test in self.summary.get("tests") or (): |
| if test["result"] == self._TEST_RESULT_PASS: |
| tests = self._pass_name_to_tests.get(test["name"], []) |
| tests.append(test) |
| self._pass_name_to_tests[test["name"]] = tests |
| else: |
| tests = self._fail_name_to_tests.get(test["name"], []) |
| tests.append(test) |
| self._fail_name_to_tests[test["name"]] = tests |
| |
| def present_tests(self, show_failures_in_red, show_passed): |
| def show_tests(tests, status): |
| for test in tests: |
| name = "%s: %s" % (status, test["name"]) |
| stdio_file = test["output_file"] |
| if stdio_file and stdio_file not in self._outputs: |
| self._outputs[stdio_file] = self._api.file.read_text( |
| name, self.results_dir.join(stdio_file), test_data="output" |
| ) |
| step = self._api.step.active_result |
| # file.read_text creates a step that displays a log of the file contents |
| # with the name as the basename of the file read. We want to remove this |
| # log and rename it to `stdio`. |
| step.presentation.logs.pop(self._api.path.basename(stdio_file)) |
| else: |
| step = self._api.step(name, None) |
| if stdio_file: |
| step.presentation.logs["stdio"] = self._outputs[ |
| stdio_file |
| ].splitlines() |
| if show_failures_in_red: |
| step.presentation.status = self._api.step.FAILURE |
| |
| show_tests(self.failed_tests, "failed") |
| show_tests(self.flaked_tests, "failed") |
| |
| # There's recipe overhead that makes step creation slow, which we mitigate |
| # by cramming all the passed tests into a single step. We also skip |
| # presenting stdio since it's generally only useful if the test failed. |
| with self._api.step.nest("all passed tests") as passed_tests_step: |
| passed_tests_step.presentation.step_summary_text = pluralize( |
| "passed test", self.passed_tests |
| ) |
| if show_passed: |
| # Start with a newline to prevent the first test from showing up on |
| # the same line as the step name. |
| passed_tests_step.presentation.step_text = "".join( |
| "\n" + test["name"] for test in self.passed_tests |
| ) |
| |
| def get_output(self, output_path): |
| """Returns the contents of the file at output_path. |
| |
| The output_path should be a relative path to the results_dir. |
| """ |
| if output_path not in self._outputs: |
| self._outputs[output_path] = self._api.file.read_text( |
| "read %s" % output_path, |
| self.results_dir.join(output_path), |
| test_data="output", |
| ) |
| return self._outputs[output_path] |
| |
| def upload_results( |
| self, |
| gcs_bucket, |
| upload_to_catapult, |
| orchestration_inputs, |
| resultdb_base_variant, |
| ): |
| """Upload select test results (e.g., coverage data) to a given GCS bucket.""" |
| assert gcs_bucket |
| with self._api.step.nest("upload %s test results" % self._shard_name): |
| if self.summary: |
| # Save the summary JSON to the test shard output dir so it gets |
| # uploaded to GCS for easy access by e.g. Dachsiaboard. |
| summary_path = self.task_output_dir.join(TEST_SUMMARY_JSON) |
| assert not self._api.path.exists(summary_path), ( |
| "test output files should not be named %s" % TEST_SUMMARY_JSON |
| ) |
| self._api.file.write_json( |
| "write %s" % TEST_SUMMARY_JSON, summary_path, self.summary |
| ) |
| self._api.upload.test_results_to_resultdb( |
| resultdb_bin_path=orchestration_inputs.resultdb, |
| outputs_dir=self.results_dir, |
| summary_filepaths=[summary_path], |
| base_variant=resultdb_base_variant, |
| ) |
| snapshot_zip = self.results_dir.join(testing_requests_api.SNAPSHOT_NAME) |
| self._api.path.mock_add_paths(snapshot_zip) |
| if self._api.path.exists(snapshot_zip): |
| self._api.file.move( |
| "move snapshot to output dir", |
| snapshot_zip, |
| self.task_output_dir.join(testing_requests_api.SNAPSHOT_NAME), |
| ) |
| |
| # If an output was important enough to present (meaning it's in _outputs), |
| # we should upload it so it can be shown in Sponge, etc. |
| with self._api.step.nest("move test outputs"): |
| for output in self._outputs: |
| dest = self.task_output_dir.join(output) |
| self._api.file.ensure_directory( |
| "mkdir", self._api.path.dirname(dest) |
| ) |
| self._api.file.move( |
| "mv", self.results_dir.join(output), dest, |
| ) |
| self._upload_outputs(gcs_bucket) |
| |
| if upload_to_catapult: |
| self._api.upload.test_outputs_to_catapult(self.results_dir) |
| |
| def _upload_outputs(self, gcs_bucket): |
| self._api.upload.directory_to_gcs( |
| source=self.task_output_dir, |
| bucket=gcs_bucket, |
| # Namespace to avoid collision across shards and attempts. |
| subpath="%s/%s" % (self._shard_name, self.swarming_task_id), |
| # Internal gsutil retries of transient failures can cause rsync to |
| # fail when run by a service account that doesn't have permission |
| # to delete objects. |
| rsync=False, |
| ) |
| |
| def run_triage(self, triage_tool, triage_sources): |
| snapshot_zip = self.results_dir.join(testing_requests_api.SNAPSHOT_NAME) |
| self._api.path.mock_add_paths(snapshot_zip) |
| if not self._api.path.exists(snapshot_zip): # pragma: no cover |
| return |
| with self._api.step.nest( |
| "run triage for %s" % self._shard_name |
| ) as presentation: |
| snapshot_dir = self._api.path.mkdtemp("snapshot") |
| self._api.tar.extract( |
| "extract snapshot", path=snapshot_zip, directory=snapshot_dir |
| ) |
| cmd = ( |
| [triage_tool,] |
| + ["--config=%s" % f for f in triage_sources] |
| + ["--data", snapshot_dir] |
| ) |
| triage_output = self._api.path.mkstemp() |
| triage_step = self._api.step( |
| "triage", |
| cmd, |
| # The triage tool returns a non-zero exit code when it detects |
| # any violations or errors. We don't want to fail in this case, |
| # but just write the output to a file for analysis. |
| ok_ret="any", |
| stdout=self._api.raw_io.output(leak_to=triage_output), |
| ) |
| presentation.logs["triage_output"] = ( |
| triage_step.stdout.splitlines() if triage_step.stdout else [] |
| ) |
| self._api.file.move( |
| "move triage output file to output dir", |
| triage_output, |
| self.task_output_dir.join("triage_output"), |
| ) |
| |
| |
| class Task(swarming_retry_api.TriggeredTask): |
| """Task processes and presents results of testing Swarming tasks.""" |
| |
| def __init__( |
| self, |
| request, |
| api, |
| symbolize_tool, |
| llvm_symbolizer, |
| debug_symbol_url, |
| tefmocheck, |
| minfs=None, |
| retry_task_on_test_failure=False, |
| **kwargs |
| ): |
| super(Task, self).__init__(request, api, **kwargs) |
| self._symbolize_tool = symbolize_tool |
| self._llvm_symbolizer = llvm_symbolizer |
| self._debug_symbol_url = debug_symbol_url |
| self._tefmocheck = tefmocheck |
| self._minfs = minfs |
| self._retry_task_on_test_failure = retry_task_on_test_failure |
| # Test shards with the 'multiplied:' prefix come from |
| # tools/integration/testsharder/shard.go in fuchsia.git. They were |
| # specifically created to run a test or set of tests many times to look |
| # for flakes. It doesn't make sense to retry these when they fail--the |
| # goal is to see if they fail not to get them to pass. |
| if self.name.startswith("multiplied:"): |
| self.max_attempts = 1 |
| self._is_multiplied = True |
| else: |
| self._is_multiplied = False |
| # Test shards with the 'affected:' prefix run affected tests. |
| # If one of these tasks fail, we should abort early without waiting for |
| # the rest of the tasks to finish. |
| self.abort_early_if_failed = self.name.startswith("affected:") |
| |
| def launch(self, priority_boost_amount): |
| """Launches a swarming task attempt. |
| |
| It also initializes the test_results field on the returned |
| swarming_retry.Attempt to None so that the field will be set even if |
| the attempt is not completed or the result processing fails. |
| """ |
| attempt = super(Task, self).launch(priority_boost_amount) |
| attempt.test_results = None |
| return attempt |
| |
| def process_result(self, attempt): |
| """Processes the results produced by a test shard.""" |
| assert attempt.result |
| result = attempt.result |
| |
| if result.isolated_outputs: |
| attempt.task_outputs_link = result.isolated_outputs.url |
| |
| @attr.s |
| class _LogToProcess(object): |
| name = attr.ib(type=str) |
| data = attr.ib(type=str) |
| path = attr.ib(type=Path) |
| |
| swarming_summary_path = self._api.path.mkstemp() |
| self._api.file.write_raw( |
| "write swarming summary JSON", |
| swarming_summary_path, |
| self._api.json.dumps(result.raw), |
| ) |
| |
| tefmocheck_cmd = [ |
| self._tefmocheck, |
| "-swarming-summary-json", |
| swarming_summary_path, |
| "-swarming-host", |
| attempt.host, |
| ] |
| |
| to_process = [ |
| _LogToProcess( |
| name=TEST_TASK_OUTPUT_FILE, |
| path=result.output_dir.join(TEST_TASK_OUTPUT_FILE), |
| data=result.output, |
| ) |
| ] |
| for log_name in ( |
| testing_requests_api.SYSLOG_NAME, |
| testing_requests_api.SERIAL_LOG_NAME, |
| ): |
| if log_name in result.outputs: |
| to_process.append( |
| _LogToProcess( |
| name=log_name, |
| path=result.outputs[log_name], |
| data=self._api.file.read_text( |
| "read %s" % log_name, |
| result.outputs[log_name], |
| test_data="extra log contents", |
| ), |
| ) |
| ) |
| |
| for log in to_process: |
| tefmocheck_flag_name = LOG_NAME_TO_TEFMOCHECK_FLAG.get(log.name) |
| if tefmocheck_flag_name: |
| tefmocheck_cmd.extend((tefmocheck_flag_name, log.path)) |
| # Symbolize and overwrite so that tefmocheck and logdog see the |
| # symbolized versions. |
| if self._api.testing.task_targets_fuchsia(result): |
| # Non-Fuchsia should already be symbolized, and attempting to use |
| # the symbolizer may fail, if e.g. it was built on Mac and this is |
| # running on Linux. |
| attempt.logs[log.name] = self._api.symbolize( |
| symbolize_tool=self._symbolize_tool, |
| llvm_symbolizer=self._llvm_symbolizer, |
| data=log.data, |
| name="symbolize %s" % log.name, |
| debug_symbol_url=self._debug_symbol_url, |
| symbolizer_output=log.path, |
| ) |
| else: |
| attempt.logs[log.name] = log.data |
| if not self._api.path.exists(log.path): |
| # Ensure it exists on file system even if we didn't symbolize |
| # so that it gets uploaded to GCS later. |
| self._api.file.write_raw( |
| "write %s" % log.name, log.path, log.data, include_log=False |
| ) |
| self._api.path.mock_add_paths(log.path) |
| |
| test_results_dir = self._api.testing.results_dir_on_host.join(result.id) |
| tefmocheck_cmd.extend(("-outputs-dir", test_results_dir)) |
| summary_json_path = test_results_dir.join(TEST_SUMMARY_JSON) |
| # If the swarming task failed, extracting test results will likely fail. |
| if result.success: |
| self._api.testing.extract_test_results( |
| step_name="extract results", |
| task_result=result, |
| directory=test_results_dir, |
| minfs_path=self._minfs, |
| ) |
| |
| if self._api.path.exists(summary_json_path): |
| tefmocheck_cmd.extend(("-test-summary-json", summary_json_path)) |
| |
| summary_bytes = self._api.step( |
| "tefmocheck", tefmocheck_cmd, stdout=self._api.raw_io.output() |
| ).stdout |
| attempt.logs[TEST_SUMMARY_JSON] = summary_bytes |
| |
| tags = {} |
| for tag in self._request.tags: |
| k, v = tag.split(":", 1) |
| tags[k] = v |
| |
| test_results = FuchsiaTestResults( |
| from_fuchsia=self._api.testing.task_targets_fuchsia(result), |
| results_dir=test_results_dir, |
| swarming_task_id=result.id, |
| shard_name=result.name, |
| env_name=tags[self._api.testing_requests.TEST_ENVIRONMENT_TAG_NAME], |
| api=self._api, |
| task_output_dir=result.output_dir, |
| summary_bytes=summary_bytes, |
| is_multiplied=self._is_multiplied, |
| ) |
| attempt.test_results = test_results |
| flaked_tests = test_results.flaked_tests |
| failed_tests = test_results.failed_tests |
| if not result.success: |
| if result.state == self._api.swarming.TaskState.COMPLETED: |
| # The task will have state COMPLETED even if it had a non-zero |
| # exit code, so show a more helpful message than "completed". |
| attempt.failure_reason = "task failed" |
| else: |
| # "BOT_DIED" -> "bot died" |
| attempt.failure_reason = result.state.name.lower().replace("_", " ") |
| else: |
| if flaked_tests: |
| attempt.has_flakes = True |
| if failed_tests: |
| attempt.failure_reason = "%s failed" % pluralize("test", failed_tests) |
| # Only retry if all the failures are testing_failure_mode checks. |
| # Sometimes, a problem can occur in the system that causes all tests |
| # to fail after that point, in which case we should also retry. |
| # The number of failed tests to check for in this case should match |
| # that of the mass_test_failure_check here: |
| # https://fuchsia.googlesource.com/fuchsia/+/e2f3065a52c5bd69090d0c2dcd8cbc76a75d601c/tools/testing/tefmocheck/cmd/main.go#125 |
| should_retry = ( |
| all(t.get("is_testing_failure_mode") for t in failed_tests) |
| or len(failed_tests) > 5 |
| ) |
| if not self._retry_task_on_test_failure and not should_retry: |
| self.max_attempts = 1 |
| # Log the contents of each output file mentioned in the summary. |
| # Note this assumes the outputs are all valid UTF-8 (See fxb/9500). |
| for output_name, output_path in test_results.summary.get( |
| "outputs", {} |
| ).iteritems(): |
| attempt.logs[output_name] = test_results.get_output(output_path) |
| |
| def present_attempt(self, _, attempt, category=None): |
| """Present an Attempt when summarizing results at the end of the run. |
| |
| Args: |
| attempt (Attempt): the Attempt to present |
| category (str): the group of tasks ('passes', 'failures', or |
| 'flakes') that this attempt should be presented under |
| """ |
| show_failures_in_red = True |
| # The 'passes' category includes all attempts of all tasks that |
| # eventually passed, so it includes some failures. Show those in |
| # green so people don't get confused and think the overall task |
| # failed. |
| # TODO(fxb/36647) after this bug is fixed show these steps in |
| # red, but show parent steps of those in green. |
| if category == "passes" or category == "incomplete": |
| show_failures_in_red = False |
| |
| attempt_status = "fail" |
| if attempt.success: |
| if attempt.has_flakes: |
| attempt_status = "flake" |
| else: |
| attempt_status = "pass" |
| name = "%s (%s)" % (attempt.name, attempt_status) |
| with self._api.step.nest(name) as presentation: |
| if show_failures_in_red and (not attempt.success or attempt.has_flakes): |
| presentation.status = self._api.step.FAILURE |
| self._present( |
| presentation, |
| attempt, |
| show_failures_in_red=show_failures_in_red, |
| show_passed=True, |
| ) |
| |
| def present_status(self, parent_step, attempt): |
| """Present an Attempt while showing progress in launch/collect step. |
| |
| Args: |
| parent_step (Step): will always be 'passed tasks' or 'failed tasks' |
| attempt (Attempt): the Attempt to present |
| """ |
| del parent_step # Unused. |
| with self._api.step.nest("%s (%s)" % (self.name, attempt.name)) as presentation: |
| self._present( |
| presentation, attempt, show_failures_in_red=False, show_passed=False |
| ) |
| |
| def _present(self, presentation, attempt, show_failures_in_red, show_passed): |
| """Present an Attempt. |
| |
| Choosing to do largely the same thing for both kinds of presentations. |
| |
| Args: |
| presentation (StepPresentation): where to present the attempt info |
| attempt (api.swarming_retry.Attempt): object to present |
| show_failures_in_red (bool): show failures in red (for final |
| 'flakes' and 'failures' steps) or not (for 'launch/collect' |
| progress and 'passes' steps) |
| show_passed (bool): show the names of passed tests (only done for |
| the end) |
| |
| Note: the 'passes' step can have failures underneath it because the |
| first attempt can fail but the retry passed. |
| """ |
| if attempt.result and attempt.result.duration_secs: |
| presentation.step_text = nice_duration(attempt.result.duration_secs) |
| |
| presentation.presentation.links["swarming task"] = attempt.task_ui_link |
| if attempt.task_outputs_link: |
| presentation.links["task outputs"] = attempt.task_outputs_link |
| |
| if attempt.failure_reason: |
| presentation.step_summary_text = attempt.failure_reason |
| |
| for log, data in attempt.logs.iteritems(): |
| presentation.logs[log] = data |
| |
| if attempt.test_results: |
| test_results = attempt.test_results |
| |
| # Log the contents of each output file mentioned in the summary. |
| # Note this assumes the outputs are all valid UTF-8 (See fxb/9500). |
| for name, path in test_results.summary.get("outputs", {}).iteritems(): |
| presentation.logs[name] = test_results.get_output(path).split("\n") |
| |
| test_results.present_tests( |
| show_failures_in_red=show_failures_in_red, show_passed=show_passed |
| ) |
| |
| |
| class FuchsiaTestApi(recipe_api.RecipeApi): |
| """API for running tests and processing test results.""" |
| |
| FuchsiaTestResults = FuchsiaTestResults |
| |
| TEST_SUMMARY_JSON = TEST_SUMMARY_JSON |
| |
| def task_targets_fuchsia(self, task_result): |
| # Sharded tasks will have the OS environment in the task name. |
| # Deprecated tasks do not have the OS in the name, but they always target |
| # fuchsia, so this function should always return true. |
| name = task_result.name.lower() |
| return "linux" not in name and "mac" not in name |
| |
| @property |
| def results_dir_on_host(self): |
| """The directory on host to which host and target test results will be written. |
| |
| Target test results will be copied over to this location and |
| host test results will be written here. Host and target tests on |
| should write to separate subdirectories so as not to collide. |
| """ |
| return self.m.path["cleanup"].join("test_results") |
| |
| def extract_test_results(self, step_name, task_result, directory, minfs_path=None): |
| """Extracts test results from task_result. |
| |
| Moves test results from the task_result's outputs into directory. |
| We do this because we later want to be able to archive everything in the |
| swarming task result's outputs, but we don't want to archive most of the |
| test result files. |
| |
| Args: |
| step_name (str): The name of the step. |
| task_result (api.swarming.TaskResult): The task result from which to extract |
| the test results. |
| directory (Path): The directory to extract the test results into. |
| minfs_path (Path): The path to the minfs tool. |
| """ |
| if self.m.testing_requests.TEST_RESULTS_MINFS_NAME in task_result.outputs: |
| assert minfs_path, "minfs path not set" |
| self.m.minfs.minfs_path = minfs_path |
| test_results_path = task_result.outputs[ |
| self.m.testing_requests.TEST_RESULTS_MINFS_NAME |
| ] |
| self.m.file.ensure_directory("create test results dir", directory) |
| self.m.minfs.copy_image( |
| step_name=step_name, image_path=test_results_path, out_dir=directory, |
| ) |
| # Delete the archive so it doesn't get uploaded with the other files in |
| # the swarming task's output directory. |
| self.m.file.remove( |
| "remove %s" % self.m.path.basename(test_results_path), test_results_path |
| ) |
| else: |
| # The outputs should be in an out directory defined by |
| # TEST_RESULTS_DIR_NAME. |
| test_results_path = None |
| for relative_path in sorted(task_result.outputs.keys()): |
| if relative_path.startswith( |
| self.m.testing_requests.TEST_RESULTS_DIR_NAME + "/" |
| ): |
| test_results_path = str( |
| task_result.output_dir.join( |
| self.m.testing_requests.TEST_RESULTS_DIR_NAME |
| ) |
| ) |
| break |
| assert test_results_path, ( |
| "test results not found amongst outputs of task %s: %s" |
| % (task_result.name, task_result.outputs) |
| ) |
| self.m.file.move(step_name, source=test_results_path, dest=directory) |
| |
| # This is only needed for the recipe tests. file.listdir() doesn't mock the |
| # existence of the paths it returns, so we must add it separately. |
| # We add summary_path because we check for its existence in |
| # FuchsiaTestResults._parse_summary(). |
| summary_path = directory.join(TEST_SUMMARY_JSON) |
| outputs = self.m.file.listdir("get extracted files", directory) |
| if summary_path in outputs: |
| self.m.path.mock_add_paths(summary_path) |
| |
| def run_test_tasks( |
| self, |
| debug_symbol_url, |
| max_attempts, |
| orchestration_inputs, |
| rerun_budget_secs=None, |
| runs_per_shard=1, |
| retry_task_on_test_failure=False, |
| ): |
| """Tests a Fuchsia build by sharding. |
| |
| Expects the build and artifacts to be at the same place they were at |
| the end of the build. |
| |
| Args: |
| debug_symbol_url (str): A GCS URL hosting debug symbols. |
| max_attempts (int): Maximum number of attempts before marking a shard |
| as failed. |
| orchestration_inputs (TestOrchestrationInputs): Build artifacts needed |
| for testing. |
| rerun_budget_secs (int): Will run tests repeatedly until this budget is. |
| consumed. If set, max_attempts is ignored. |
| runs_per_shard (int): Number of times to run each shard. |
| retry_task_on_test_failure (bool): Retry tasks on all test failures. |
| If false, retry whole tasks on tefmocheck failures but not on |
| regular test failures. |
| |
| Returns: |
| A list of swarming_retry.Tasks representing the completed test |
| tasks that were not subject to an infra failure. |
| """ |
| |
| # If no shards have been provided, then we have successfully run the empty |
| # set of tests. |
| if not orchestration_inputs.task_requests: |
| return [] |
| |
| launch_deadline_time = None |
| present = True |
| if rerun_budget_secs: |
| # If we have a rerun deadline, we want to run the tests as many times as |
| # possible within that time regardless of whether the tasks pass or fail, |
| # so we don't care about setting a max number of attempts. |
| max_attempts = float("inf") |
| # Presenting results is currently slow due to inefficiencies in |
| # the recipe engine. For builds that have rerun_budget_secs enabled, |
| # we only care about consuming the data after it's been uploaded, not |
| # from the Milo build page, so skip presentation. |
| present = False |
| launch_deadline_time = self.m.time.time() + rerun_budget_secs |
| |
| tasks = [] |
| for task_request in orchestration_inputs.task_requests: |
| if rerun_budget_secs: |
| # Rerun tasks should not block other tasks. |
| task_request = task_request.with_priority(task_request.priority + 1) |
| tasks.append( |
| Task( |
| request=task_request, |
| api=self.m, |
| symbolize_tool=orchestration_inputs.symbolize_tool, |
| llvm_symbolizer=orchestration_inputs.llvm_symbolizer, |
| debug_symbol_url=debug_symbol_url, |
| tefmocheck=orchestration_inputs.tefmocheck, |
| minfs=orchestration_inputs.minfs, |
| retry_task_on_test_failure=retry_task_on_test_failure, |
| ) |
| ) |
| |
| tasks = self.m.swarming_retry.run_tasks( |
| tasks=tasks, |
| collect_output_dir=self.m.path.mkdtemp("swarming"), |
| max_attempts=max_attempts, |
| launch_deadline_time=launch_deadline_time, |
| present=present, |
| run_count=runs_per_shard, |
| ) |
| |
| if present: |
| self.m.swarming_retry.present_tasks(tasks=tasks) |
| |
| return tasks |
| |
| def raise_failures(self, tasks): |
| failed_tests = [] |
| failed_tests_no_tefmo = [] |
| tasks_without_results = [] |
| for task in tasks: |
| if task.failed: |
| # Report the test failures from the last failed attempt. |
| attempt = task.get_failed_attempts()[-1] |
| if attempt.test_results: |
| failed_tests.extend( |
| t["name"] for t in attempt.test_results.failed_tests |
| ) |
| failed_tests_no_tefmo.extend( |
| t["name"] |
| for t in attempt.test_results.failed_tests |
| if not t.get("is_testing_failure_mode") |
| ) |
| else: |
| tasks_without_results.append(task.name) |
| |
| if tasks_without_results: |
| raise self.m.step.StepFailure( |
| "failed to retrieve test results for %s: %s" |
| % ( |
| pluralize("task", tasks_without_results), |
| ", ".join(sorted(tasks_without_results)), |
| ) |
| ) |
| |
| # Expose these so tools can be used to re-run these failed tests. |
| with self.m.step.nest("record failed_test_names") as presentation: |
| presentation.properties["failed_test_names"] = failed_tests_no_tefmo |
| |
| if not failed_tests: |
| return |
| |
| def _test_description(test, failure_count): |
| count_info = "" |
| if failure_count > 1: |
| count_info = " (%d failures)" % failure_count |
| return "%s%s" % (test, count_info) |
| |
| failure_counts = collections.Counter(failed_tests) |
| header = "%s failed:" % pluralize("test", failure_counts) |
| if len(failure_counts) == 1: |
| test, failure_count = failure_counts.popitem() |
| raise self.m.step.StepFailure( |
| "%s %s" % (header, _test_description(test, failure_count)) |
| ) |
| |
| error_lines = [ |
| header, |
| # Milo's markdown renderer requires a blank line before the |
| # start of a bulleted list. Otherwise the list lines will all |
| # be rendered on one hyphen-separated line. |
| "", |
| ] |
| |
| # Present up to 10 tests and show a count of the rest. But if there are |
| # 11-15 failed tests, still show all of the tests because it's not |
| # worth saving lines by hiding only a couple tests. |
| max_tests_to_present = 15 |
| min_tests_to_hide = 5 |
| if len(failure_counts) > max_tests_to_present: |
| tests_to_present = max_tests_to_present - min_tests_to_hide |
| else: |
| tests_to_present = len(failure_counts) |
| |
| counts_to_present = sorted(failure_counts.items())[:tests_to_present] |
| for test, failure_count in counts_to_present: |
| error_lines.append("- %s" % _test_description(test, failure_count)) |
| |
| hidden_tests = len(failure_counts) - tests_to_present |
| if hidden_tests > 0: |
| error_lines.append("- (%d more)" % hidden_tests) |
| |
| raise self.m.step.StepFailure("\n".join(error_lines)) |
