• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 datetime as dt
8import logging
9from typing import TYPE_CHECKING, Any, Dict, Optional, Type
10
11from crossbench import exception
12from crossbench import path as pth
13from crossbench.action_runner.action.action import (ACTION_TIMEOUT, Action,
14                                                    ActionT)
15from crossbench.action_runner.action.action_type import ActionType
16from crossbench.parse import ObjectParser, PathParser
17
18if TYPE_CHECKING:
19  from crossbench.action_runner.base import ActionRunner
20  from crossbench.config import ConfigParser
21  from crossbench.runner.run import Run
22  from crossbench.types import JsonDict
23
24
25def parse_replacement_dict(value: Any) -> Dict[str, str]:
26  dict_value = ObjectParser.dict(value)
27  for replace_key, replace_value in dict_value.items():
28    with exception.annotate_argparsing(
29        f"Parsing ...[{repr(replace_key)}] = {repr(value)}"):
30      ObjectParser.non_empty_str(replace_key, "replacement key")
31      ObjectParser.any_str(replace_value, "replacement value")
32  return dict_value
33
34
35class JsAction(Action):
36  TYPE: ActionType = ActionType.JS
37
38  @classmethod
39  def config_parser(cls: Type[ActionT]) -> ConfigParser[ActionT]:
40    parser = super().config_parser()
41    parser.add_argument("script", type=ObjectParser.non_empty_str)
42    parser.add_argument(
43        "script_path", aliases=("path",), type=PathParser.existing_file_path)
44    parser.add_argument(
45        "replacements", aliases=("replace",), type=parse_replacement_dict)
46    return parser
47
48  def __init__(self,
49               script: Optional[str],
50               script_path: Optional[pth.LocalPath],
51               replacements: Optional[Dict[str, str]] = None,
52               timeout: dt.timedelta = ACTION_TIMEOUT,
53               index: int = 0) -> None:
54    self._original_script = script
55    self._script_path = script_path
56    self._script = ""
57    if bool(script) == bool(script_path):
58      raise ValueError(
59          f"One of {self}.script or {self}.script_path, but not both, "
60          "have to specified. ")
61    if script:
62      self._script = script
63    elif script_path:
64      self._script = script_path.read_text()
65      logging.debug("Loading script from %s: %s", script_path, script)
66      # TODO: Support argument injection into shared file script.
67    self._replacements = replacements
68    if replacements:
69      for key, value in replacements.items():
70        self._script = self._script.replace(key, value)
71    super().__init__(timeout, index)
72
73  @property
74  def script(self) -> str:
75    return self._script
76
77  def run_with(self, run: Run, action_runner: ActionRunner) -> None:
78    action_runner.js(run, self)
79
80  def validate(self) -> None:
81    super().validate()
82    if not self.script:
83      raise ValueError(
84          f"{self}.script is missing or the provided script file is empty.")
85
86  def to_json(self) -> JsonDict:
87    details = super().to_json()
88    if self._original_script:
89      details["script"] = self._original_script
90    if self._script_path:
91      details["script_path"] = str(self._script_path)
92    if self._replacements:
93      details["replacements"] = self._replacements
94    return details
95