| # Copyright 2023 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 argparse |
| import datetime as dt |
| from typing import TYPE_CHECKING, Any, ClassVar, Mapping, Optional, Sequence, \ |
| Type |
| |
| from typing_extensions import override |
| |
| from crossbench.benchmarks.base import StoryFilter, SubStoryBenchmark |
| from crossbench.benchmarks.loading.config.pages import \ |
| DevToolsRecorderPagesConfig, ListPagesConfig, PageConfig, PagesConfig |
| from crossbench.benchmarks.loading.page.base import DEFAULT_DURATION, Page |
| from crossbench.benchmarks.loading.page.combined import CombinedPage |
| from crossbench.benchmarks.loading.page.interactive import InteractivePage |
| from crossbench.benchmarks.loading.page.live import PAGE_LIST, \ |
| PAGE_LIST_SMALL, PAGES, LivePage |
| from crossbench.benchmarks.loading.playback_controller import \ |
| PlaybackController |
| from crossbench.benchmarks.loading.tab_controller import TabController |
| from crossbench.parse import DurationParser, ObjectParser |
| |
| if TYPE_CHECKING: |
| from crossbench.action_runner.config import ActionRunnerConfig |
| from crossbench.cli.parser import CrossBenchArgumentParser |
| from crossbench.cli.types import Subparsers |
| from crossbench.stories.story import Story |
| |
| |
| class LoadingPageFilter(StoryFilter[Page]): |
| """ |
| Filter / create loading stories |
| |
| Syntax: |
| "name" Include LivePage with the given name from predefined list. |
| "name", 10 Include predefined page with given 10s timeout. |
| "http://..." Include custom page at the given URL with a default |
| timeout of 15 seconds. |
| "http://...", 12 Include custom page at the given URL with a 12s timeout |
| |
| These patterns can be combined: |
| ["http://foo.com", 5, "http://bar.co.jp", "amazon"] |
| """ |
| stories: Sequence[Page] |
| |
| @classmethod |
| @override |
| def add_cli_arguments( |
| cls, parser: argparse.ArgumentParser) -> argparse.ArgumentParser: |
| parser = super().add_cli_arguments(parser) |
| cls.add_page_config_parser(parser) |
| tab_group = parser.add_mutually_exclusive_group() |
| tab_group.add_argument( |
| "--single-tab", |
| dest="tabs", |
| const=TabController.single(), |
| default=TabController.default(), |
| action="store_const", |
| help="Open given urls in a single tab.") |
| tab_group.add_argument( |
| "--multiple-tab", |
| dest="tabs", |
| nargs="?", |
| type=TabController.parse, |
| const=TabController.multiple(), |
| help="Open given urls in separate tabs " |
| "(optional value for number of tabs for each url).") |
| tab_group.add_argument( |
| "--infinite-tab", |
| dest="tabs", |
| const=TabController.forever(), |
| action="store_const", |
| help="Open given urls in separate tabs infinitely.") |
| |
| playback_group = parser.add_mutually_exclusive_group() |
| playback_group.add_argument( |
| "--playback", |
| "--cycle", |
| type=PlaybackController.parse, |
| default=PlaybackController.default(), |
| help="Set limit on looping through/repeating the selected stories. " |
| "Default is once. " |
| "Valid values are: 'once', 'forever', number, time. " |
| "Cycle 10 times: '--playback=10x'. " |
| "Repeat for 1.5 hours: '--playback=1.5h'. " |
| "Run 5 times, once every 10 seconds: '--playback=\"5x every 10s\"'.") |
| playback_group.add_argument( |
| "--forever", |
| dest="playback", |
| const=PlaybackController.forever(), |
| action="store_const", |
| help="Equivalent to --playback=infinity") |
| |
| parser.add_argument( |
| "--about-blank-duration", |
| "--about-blank", |
| type=DurationParser.positive_or_zero_duration, |
| default=dt.timedelta(), |
| help="If non-zero, navigate to about:blank after every page.") |
| |
| block_modifier_group = parser.add_argument_group("Action Block Options") |
| block_modifier_group.add_argument( |
| "--skip-login", |
| dest="run_login", |
| default=True, |
| action="store_const", |
| const=False, |
| help="Skip the login block, useful for replaying " |
| "archive that filtered already all login requests " |
| "to hide potential secrets. " |
| "The login block is run by default.") |
| block_modifier_group.add_argument( |
| "--skip-setup", |
| dest="run_setup", |
| default=True, |
| action="store_const", |
| const=False, |
| help="Skip the setup block, useful for replaying " |
| "archive that filtered already all login requests " |
| "to hide potential secrets. " |
| "The setup block is run by default.") |
| |
| return parser |
| |
| @classmethod |
| def add_page_config_arguments(cls, group: argparse._ArgumentGroup) -> None: |
| group.add_argument( |
| "--page-config", |
| "--pages-config", |
| dest="pages_config", |
| type=PagesConfig.parse, |
| help="Stories we want to perform in the benchmark run following a " |
| "specified scenario. For a reference on how to build scenarios and " |
| "possible actions check config/doc/pages.config.hjson") |
| |
| @classmethod |
| def add_page_config_parser(cls, parser: argparse.ArgumentParser) -> None: |
| page_config_group = parser.add_mutually_exclusive_group() |
| # TODO: move --stories into mutually exclusive group as well |
| page_config_group.add_argument( |
| "--urls", |
| "--url", |
| dest="urls", |
| help="List of urls and durations to load: url,seconds,...") |
| cls.add_page_config_arguments(page_config_group) |
| page_config_group.add_argument( |
| "--url-file", |
| "--urls-file", |
| dest="pages_config", |
| type=ListPagesConfig.parse, |
| help=("List of urls and durations in a line-by-line file. " |
| "Each line has the same format as --url for a single Page.")) |
| page_config_group.add_argument( |
| "--devtools-recorder", |
| dest="pages_config", |
| type=DevToolsRecorderPagesConfig.parse, |
| help="Run a single story from a serialized DevTools recorder session. " |
| "See https://developer.chrome.com/docs/devtools/recorder/ " |
| "for more details.") |
| |
| @classmethod |
| @override |
| def kwargs_from_cli(cls, args: argparse.Namespace) -> dict[str, Any]: |
| kwargs = super().kwargs_from_cli(args) |
| kwargs["separate"] = args.separate |
| return kwargs |
| |
| @override |
| def process_all(self, patterns: Sequence[str]) -> None: |
| name_or_url_list = patterns |
| if len(name_or_url_list) == 1: |
| if name_or_url_list[0] == "all": |
| self.stories = self.all_stories() |
| return |
| if name_or_url_list[0] == "default": |
| self.stories = self.default_stories() |
| return |
| # Let the PageConfig handle the arg splitting again: |
| config = PagesConfig.parse(",".join(patterns)) |
| self.stories = self.stories_from_config(self.args, config) |
| |
| @classmethod |
| def all_stories(cls) -> tuple[Page, ...]: |
| return tuple(PAGE_LIST) |
| |
| @classmethod |
| def default_stories(cls) -> tuple[Page, ...]: |
| return PAGE_LIST_SMALL |
| |
| @classmethod |
| def stories_from_config(cls, args: argparse.Namespace, |
| config: PagesConfig) -> Sequence[Page]: |
| labels: set[str |
| | None] = {page_config.label for page_config in config.pages} |
| use_labels = len(labels) == len(config.pages) |
| |
| stories: list[Page] = [] |
| for page_config in config.pages: |
| stories.append(cls._story_from_config(args, page_config, use_labels)) |
| |
| if not use_labels: |
| # Double check that the urls are unique |
| urls: set[str] = {page_config.first_url for page_config in config.pages} |
| if len(urls) != len(config.pages): |
| raise argparse.ArgumentTypeError( |
| "Got non-unique story labels and urls.") |
| return stories |
| |
| @classmethod |
| def _story_from_config(cls, args: argparse.Namespace, config: PageConfig, |
| use_labels: bool) -> Page: |
| playback: PlaybackController = args.playback |
| tabs: TabController = args.tabs |
| if config.playback: |
| # TODO: support custom config playback |
| playback = config.playback |
| duration: dt.timedelta = config.duration |
| if config.label in PAGES: |
| page = PAGES[config.label] |
| duration = duration or page.duration |
| return LivePage(page.name, page.url, duration, playback, tabs, |
| args.about_blank_duration) |
| |
| label: str = config.any_label if use_labels else config.first_url |
| duration = duration or DEFAULT_DURATION |
| |
| if not config.blocks: |
| return LivePage(label, config.first_url, duration, playback, tabs, |
| args.about_blank_duration) |
| return InteractivePage( |
| name=label, |
| blocks=config.blocks, |
| login=config.login, |
| setup=config.setup, |
| teardown=config.teardown, |
| secrets=config.secrets, |
| playback=playback, |
| tabs=tabs, |
| about_blank_duration=args.about_blank_duration, |
| run_login=args.run_login, |
| run_setup=args.run_setup) |
| |
| @override |
| def create_stories(self, separate: bool) -> Sequence[Page]: |
| if not separate and len(self.stories) > 1: |
| combined_name = "_".join(page.name for page in self.stories) |
| args = self.args |
| self.stories = (CombinedPage(self.stories, combined_name, args.playback, |
| args.tabs),) |
| return self.stories |
| |
| |
| class LoadingBenchmark(SubStoryBenchmark): |
| """ |
| Benchmark runner for loading pages with complex interactions. |
| |
| Use --urls/--stories to either choose from an existing set of pages, or direct |
| URLs. After each page you can also specify a custom wait/load duration in |
| seconds. Multiple URLs/page names can be provided as a comma-separated list. |
| |
| Use --separate to load each page individually. |
| |
| Example: |
| --urls=amazon |
| --urls=http://cnn.com,10s |
| --urls=http://twitter.com,5s,http://cnn.com,10s |
| """ |
| NAME: ClassVar = "loading" |
| DEFAULT_STORY_CLS: ClassVar = Page |
| STORY_FILTER_CLS: ClassVar[Type[LoadingPageFilter]] = LoadingPageFilter |
| |
| @classmethod |
| @override |
| def add_cli_parser(cls, subparsers: Subparsers) -> CrossBenchArgumentParser: |
| parser = super().add_cli_parser(subparsers) |
| cls.STORY_FILTER_CLS.add_cli_arguments(parser) |
| return parser |
| |
| @classmethod |
| @override |
| def stories_from_cli_args(cls, args: argparse.Namespace) -> Sequence[Story]: |
| has_default_stories: bool = ( |
| args.stories and args.stories == LoadingPageFilter.DEFAULT_STORY_NAME) |
| if config := cls.get_pages_config(args): |
| # TODO: make stories and page_config mutually exclusive. |
| if not has_default_stories: |
| raise argparse.ArgumentTypeError( |
| f"Cannot specify --stories={repr(args.stories)} " |
| "with any other page config option.") |
| pages = cls.STORY_FILTER_CLS.stories_from_config(args, config) |
| if args.separate: |
| return pages |
| if len(pages) == 1: |
| return pages |
| return (CombinedPage(pages, "Page Scenarios - Combined", args.playback, |
| args.tabs),) |
| |
| if args.urls: |
| # TODO: make urls and stories mutually exclusive. |
| if not has_default_stories: |
| raise argparse.ArgumentTypeError( |
| "Cannot specify --urls and --stories at the same time.") |
| args.stories = args.urls |
| |
| # Fall back to story filter class. |
| return super().stories_from_cli_args(args) |
| |
| @classmethod |
| def get_pages_config(cls, |
| args: Optional[argparse.Namespace] = None |
| ) -> Optional[PagesConfig]: |
| if not args: |
| raise ValueError("Missing args") |
| if global_config := args.config: |
| # TODO: migrate --config to an already parsed hjson/json dict |
| config_file = global_config |
| config_data = ObjectParser.hjson_file(config_file) |
| if pages_config_dict := config_data.get("pages"): |
| if args.pages_config: |
| raise argparse.ArgumentTypeError( |
| "Conflicting arguments: " |
| "either specify a --config file without a 'pages' property " |
| "or remove the --page-config argument.") |
| # TODO: PagesConfig.parse_dict should be able to parse the inner dict. |
| return PagesConfig.parse_dict({"pages": pages_config_dict}) |
| return args.pages_config |
| |
| @classmethod |
| @override |
| def aliases(cls) -> tuple[str, ...]: |
| return ("load", "ld") |
| |
| @classmethod |
| @override |
| def describe_stories(cls) -> Mapping[str, str]: |
| result: dict[str, str] = {} |
| for story in cls.all_stories(): |
| story_help = story.help() |
| if story_help == story.name: |
| story_help = "" |
| result[story.name] = story_help |
| return result |
| |
| @classmethod |
| @override |
| def all_story_names(cls) -> Sequence[str]: |
| # TODO: Use StoryFilter for listing stories everywhere. |
| return sorted(story.name for story in cls.STORY_FILTER_CLS.all_stories()) |
| |
| def __init__( |
| self, |
| stories: Sequence[Page], |
| action_runner_config: Optional[ActionRunnerConfig] = None) -> None: |
| for story in stories: |
| assert isinstance(story, Page) |
| super().__init__(stories, action_runner_config) |
