blob: 281830dc1e6b897747cf3f00d025034c91af392c [file] [log] [blame]
# Copyright 2024 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
from __future__ import annotations
import abc
import contextlib
import logging
from typing import TYPE_CHECKING, Iterable, Iterator, Tuple
from crossbench import plt
from crossbench.helper import DurationMeasureContext, Durations
from crossbench.probes.result_location import ResultLocation
if TYPE_CHECKING:
from crossbench.browsers.browser import Browser
from crossbench.exception import (Annotator, ExceptionAnnotationScope,
TExceptionTypes)
from crossbench.path import LocalPath, RemotePath
from crossbench.probes.probe import Probe
from crossbench.runner.runner import Runner
class ResultOrigin(abc.ABC):
"""Base class for Run and BrowserSession, both places where
probe results can be placed."""
@property
def is_local(self) -> bool:
return self.browser_platform.is_local
@property
def is_remote(self) -> bool:
return self.browser_platform.is_remote
@property
@abc.abstractmethod
def browser_tmp_dir(self) -> RemotePath:
pass
@property
@abc.abstractmethod
def out_dir(self) -> LocalPath:
pass
@property
@abc.abstractmethod
def exceptions(self) -> Annotator:
pass
@property
@abc.abstractmethod
def durations(self) -> Durations:
pass
@property
@abc.abstractmethod
def browser(self) -> Browser:
pass
@property
@abc.abstractmethod
def runner(self) -> Runner:
pass
@property
def runner_platform(self) -> plt.Platform:
return self.runner.platform
@property
def browser_platform(self) -> plt.Platform:
return self.browser.platform
@property
def probes(self) -> Iterable[Probe]:
return self.runner.probes
@contextlib.contextmanager
def measure(
self, label: str
) -> Iterator[Tuple[ExceptionAnnotationScope, DurationMeasureContext]]:
# Return a combined context manager that adds an named exception info
# and measures the time during the with-scope.
with self.exceptions.info(label) as stack, self.durations.measure(
label) as timer:
yield (stack, timer)
def exception_info(self, *stack_entries: str) -> ExceptionAnnotationScope:
return self.exceptions.info(*stack_entries)
def exception_handler(
self, *stack_entries: str, exceptions: TExceptionTypes = (Exception,)
) -> ExceptionAnnotationScope:
return self.exceptions.capture(*stack_entries, exceptions=exceptions)
def get_default_probe_result_path(self, probe: Probe) -> RemotePath:
"""Return a local or remote/browser-based result path depending on the
Probe default RESULT_LOCATION."""
if probe.RESULT_LOCATION == ResultLocation.BROWSER:
return self.get_browser_probe_result_path(probe)
if probe.RESULT_LOCATION == ResultLocation.LOCAL:
return self.get_local_probe_result_path(probe)
raise ValueError(f"Invalid probe.RESULT_LOCATION {probe.RESULT_LOCATION} "
f"for probe {probe}")
@abc.abstractmethod
def get_local_probe_result_path(self, probe: Probe) -> LocalPath:
pass
def get_browser_probe_result_path(self, probe: Probe) -> RemotePath:
local_path = self.get_local_probe_result_path(probe)
if self.is_local:
return local_path
# Create a temp file relative to the remote browser tmp dir.
relative_path = local_path.relative_to(self.out_dir)
path = self.browser_tmp_dir / relative_path
logging.debug("Creating remote result dir=%s on platform=%s", path.parent,
self.browser_platform)
self.browser_platform.mkdir(path.parent)
return path