1# Copyright 2024 The Chromium Authors 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5from __future__ import annotations 6 7import abc 8import datetime as dt 9from typing import TYPE_CHECKING, Tuple, Type 10 11from crossbench.action_runner.action.action import ACTION_TIMEOUT, ActionT 12from crossbench.action_runner.action.base_duration import BaseDurationAction 13from crossbench.benchmarks.loading.input_source import InputSource 14 15if TYPE_CHECKING: 16 from crossbench.config import ConfigParser 17 from crossbench.types import JsonDict 18 19 20class InputSourceAction(BaseDurationAction, metaclass=abc.ABCMeta): 21 22 @classmethod 23 def config_parser(cls: Type[ActionT]) -> ConfigParser[ActionT]: 24 parser = super().config_parser() 25 parser.add_argument( 26 "source", type=InputSource.parse, default=InputSource.JS) 27 return parser 28 29 def __init__(self, 30 source: InputSource, 31 duration: dt.timedelta, 32 timeout: dt.timedelta = ACTION_TIMEOUT, 33 index: int = 0) -> None: 34 self._input_source = source 35 super().__init__(duration, timeout, index) 36 37 @property 38 def input_source(self) -> InputSource: 39 return self._input_source 40 41 def validate(self) -> None: 42 super().validate() 43 self.validate_input_source() 44 45 def validate_input_source(self) -> None: 46 if self.input_source not in self.supported_input_sources(): 47 raise ValueError( 48 f"Unsupported input source for {self.__class__.__name__}") 49 50 @abc.abstractmethod 51 def supported_input_sources(self) -> Tuple[InputSource, ...]: 52 pass 53 54 def to_json(self) -> JsonDict: 55 details = super().to_json() 56 details["source"] = self.input_source 57 return details 58