# Copyright 2022 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import collections
import enum
import fnmatch
import json
import logging
import os
import re
import time
from typing import Dict, List, Optional
import dataclasses # Built-in, but pylint gives an ordering false positive.
from gpu_tests import common_browser_args as cba
from gpu_tests import common_typing as ct
from gpu_tests import gpu_integration_test
from gpu_tests.util import host_information
from gpu_tests.util import websocket_server as wss
from gpu_tests.util import websocket_utils as wsu
from typ import expectations_parser
import gpu_path_util
SLOW_TESTS_FILE = os.path.join(gpu_path_util.CHROMIUM_SRC_DIR, 'third_party',
'dawn', 'webgpu-cts', 'slow_tests.txt')
TEST_LIST_FILE = os.path.join(gpu_path_util.CHROMIUM_SRC_DIR, 'third_party',
'dawn', 'third_party', 'gn', 'webgpu-cts',
'test_list.txt')
WORKER_TEST_GLOB_FILE = os.path.join(gpu_path_util.CHROMIUM_SRC_DIR,
'third_party', 'dawn', 'webgpu-cts',
'worker_test_globs.txt')
TEST_RUNS_BETWEEN_CLEANUP = 1000
WEBSOCKET_PORT_TIMEOUT_SECONDS = 10
WEBSOCKET_SETUP_TIMEOUT_SECONDS = 5
DEFAULT_TEST_TIMEOUT = 30
SLOW_MULTIPLIER = 5
ASAN_MULTIPLIER = 4
BACKEND_VALIDATION_MULTIPLIER = 6
FIRST_LOAD_TEST_STARTED_MULTIPLER = 3
MESSAGE_TIMEOUT_CONNECTION_ACK = 5
# In most cases, this should be very fast, but the first test run after a page
# load can be slow.
MESSAGE_TIMEOUT_TEST_STARTED = 10
MESSAGE_TIMEOUT_HEARTBEAT = 10
MESSAGE_TIMEOUT_TEST_LOG = 1
# Thresholds for how slow parts of the test have to be for the test to be
# considered slow overall.
SLOW_HEARTBEAT_THRESHOLD = 0.5
SLOW_GLOBAL_TIMEOUT_THRESHOLD = 0.8
HTML_FILENAME = os.path.join('webgpu-cts', 'test_page.html')
JAVASCRIPT_DURATION = 'javascript_duration'
MAY_EXONERATE = 'may_exonerate'
MESSAGE_TYPE_CONNECTION_ACK = 'CONNECTION_ACK'
MESSAGE_TYPE_INFRA_FAILURE = 'INFRA_FAILURE'
MESSAGE_TYPE_TEST_STARTED = 'TEST_STARTED'
MESSAGE_TYPE_TEST_HEARTBEAT = 'TEST_HEARTBEAT'
MESSAGE_TYPE_TEST_STATUS = 'TEST_STATUS'
MESSAGE_TYPE_TEST_LOG = 'TEST_LOG'
MESSAGE_TYPE_TEST_FINISHED = 'TEST_FINISHED'
TEST_NAME_REGEX = re.compile(r'([^:]+:[^:]+:[^:]+:).*')
# This can be switched to a StrEnum once Python 3.11+ is used.
class WorkerType(enum.Enum):
NONE = 'none'
SERVICE = 'service'
DEDICATED = 'dedicated'
SHARED = 'shared'
@dataclasses.dataclass
class WebGpuTestResult():
"""Struct-like object for holding a single test result."""
status: Optional[str] = None
log_pieces: List[str] = ct.EmptyList()
@dataclasses.dataclass
class WebGpuTestArgs():
"""Struct-like object for holding arguments for a single test."""
query: str
additional_browser_args: Optional[List[str]] = None
class WebGpuCtsIntegrationTestBase(gpu_integration_test.GpuIntegrationTest):
# Whether the test page has already been loaded. Caching this state here is
# faster than checking the URL every time, and given how fast these tests are,
# additional overhead like that can add up quickly.
page_loaded = False
# The first attempt to handle the websocket connection flakily takes
# significantly longer, potentially due to resource contention from all
# parallel browsers starting at the same time. See crbug.com/344009517 for
# more information.
attempted_websocket_connection = False
_test_timeout = DEFAULT_TEST_TIMEOUT
_enable_dawn_backend_validation = False
_use_webgpu_adapter: Optional[str] = None # use the default
_original_environ: Optional[collections.abc.Mapping] = None
_use_webgpu_power_preference: Optional[str] = None
_use_fxc = False
_os_name: Optional[str] = None
_worker_type: Optional[WorkerType] = None
_build_dir: Optional[str] = None
_test_list: Optional[List[str]] = None
_worker_test_globs: Optional[List[str]] = None
total_tests_run = 0
websocket_server: Optional[wss.WebsocketServer] = None
_slow_tests: Optional[expectations_parser.TestExpectations] = None
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._query: Optional[str] = None
self._longest_time_between_heartbeats = 0
self._heartbeat_timeout = 0
self._test_duration = 0
self._global_timeout_without_slow_multiplier = 1
# Only perform the pre/post test cleanup every X tests instead of every test
# to reduce overhead.
def ShouldPerformMinidumpCleanupOnSetUp(self) -> bool:
return (self.total_tests_run % TEST_RUNS_BETWEEN_CLEANUP == 0
and super().ShouldPerformMinidumpCleanupOnSetUp())
def ShouldPerformMinidumpCleanupOnTearDown(self) -> bool:
return self.ShouldPerformMinidumpCleanupOnSetUp()
@classmethod
def _SuiteSupportsParallelTests(cls) -> bool:
return True
@classmethod
def _GetSlowTests(cls) -> expectations_parser.TestExpectations:
if cls._slow_tests is None:
with open(SLOW_TESTS_FILE, 'r') as f:
expectations = expectations_parser.TestExpectations()
expectations.parse_tagged_list(f.read(), f.name)
cls._slow_tests = expectations
return cls._slow_tests
@classmethod
def UseWebGpuCompatMode(cls) -> bool:
raise NotImplementedError()
@classmethod
def AddCommandlineArgs(cls, parser: ct.CmdArgParser) -> None:
super().AddCommandlineArgs(parser)
parser.add_argument('--override-timeout',
type=float,
help='Override the test timeout in seconds')
parser.add_argument(
'--enable-dawn-backend-validation',
action='store_true',
default=False,
help='Runs the browser with Dawn backend validation enabled')
parser.add_argument(
'--use-webgpu-adapter',
help='Runs the browser with a particular WebGPU adapter')
parser.add_argument(
'--use-webgpu-power-preference',
help='Runs the browser with a particular WebGPU power preference')
parser.add_argument(
'--use-fxc',
action='store_true',
default=False,
help=('On Windows, pass --disable-dawn-features=use_dxc to the '
'browser.'))
parser.add_argument(
'--use-worker',
choices=['none', 'service', 'dedicated', 'shared'],
default='none',
help=('Whether to run tests in workers or not. Defaults to running '
'all tests outside of workers. If a worker type is specified, '
'then a subset of tests will be run in the specified worker '
'type.'))
@classmethod
def StartBrowser(cls) -> None:
cls.page_loaded = False
super().StartBrowser()
cls._os_name = cls.browser.platform.GetOSName()
# Set up the slow tests expectations' tags to match the test runner
# expectations' tags
cls._GetSlowTests().set_tags(cls.child.expectations.tags)
@classmethod
def GenerateBrowserArgs(cls, additional_args: List[str]) -> List[str]:
"""Adds default arguments to |additional_args|.
See the parent class' method documentation for additional information.
"""
browser_args = super().GenerateBrowserArgs(additional_args)
enable_dawn_features = ['allow_unsafe_apis']
disable_dawn_features = []
if host_information.IsWindows():
if cls._use_fxc:
disable_dawn_features.append('use_dxc')
else:
enable_dawn_features.append('use_dxc')
if enable_dawn_features:
browser_args.append('--enable-dawn-features=%s' %
','.join(enable_dawn_features))
if disable_dawn_features:
browser_args.append('--disable-dawn-features=%s' %
','.join(disable_dawn_features))
browser_args.extend(cba.ENABLE_WEBGPU_FOR_TESTING)
if cls._use_webgpu_adapter:
browser_args.append('--use-webgpu-adapter=%s' % cls._use_webgpu_adapter)
if cls._use_webgpu_power_preference:
browser_args.append('--use-webgpu-power-preference=%s' %
cls._use_webgpu_power_preference)
if cls._enable_dawn_backend_validation:
if host_information.IsWindows():
browser_args.append('--enable-dawn-backend-validation=partial')
else:
browser_args.append('--enable-dawn-backend-validation')
return browser_args
@classmethod
def SetUpProcess(cls) -> None:
super().SetUpProcess()
cls.websocket_server = wss.WebsocketServer()
cls.websocket_server.StartServer()
cls.CustomizeBrowserArgs([])
cls.StartBrowser()
# Shared workers are not supported on Android and test suites should be
# removed accordingly, but add a runtime check in case tests are still run
# for some reason.
if cls._os_name == 'android' and cls._worker_type == WorkerType.SHARED:
raise RuntimeError('Shared workers are not supported on Android')
# pylint:disable=protected-access
cls._build_dir = cls.browser._browser_backend.build_dir
# pylint:enable=protected-access
cls.SetStaticServerDirs([
os.path.join(cls._build_dir, 'gen', 'third_party', 'dawn'),
])
@classmethod
def TearDownProcess(cls) -> None:
cls.websocket_server.StopServer()
cls.websocket_server = None
super().TearDownProcess()
@classmethod
def _SetClassVariablesFromOptions(cls, options: ct.ParsedCmdArgs) -> None:
super()._SetClassVariablesFromOptions(options)
if options.override_timeout:
cls._test_timeout = options.override_timeout
cls._enable_dawn_backend_validation = options.enable_dawn_backend_validation
cls._use_webgpu_adapter = options.use_webgpu_adapter
cls._use_webgpu_power_preference = options.use_webgpu_power_preference
cls._use_fxc = options.use_fxc
cls._worker_type = WorkerType(options.use_worker)
@classmethod
def _ModifyBrowserEnvironment(cls) -> None:
super()._ModifyBrowserEnvironment()
if host_information.IsMac() and cls._enable_dawn_backend_validation:
if cls._original_environ is None:
cls._original_environ = os.environ.copy()
os.environ['MTL_DEBUG_LAYER'] = '1'
os.environ['MTL_DEBUG_LAYER_VALIDATE_LOAD_ACTIONS'] = '1'
# TODO(crbug.com/40275874) Re-enable when Apple fixes the validation
# os.environ['MTL_DEBUG_LAYER_VALIDATE_STORE_ACTIONS'] = '1'
os.environ['MTL_DEBUG_LAYER_VALIDATE_UNRETAINED_RESOURCES'] = '4'
@classmethod
def _RestoreBrowserEnvironment(cls) -> None:
if cls._original_environ is not None:
os.environ.clear()
os.environ.update(cls._original_environ)
super()._RestoreBrowserEnvironment()
@classmethod
def _GetAdditionalBrowserArgsForQuery(cls, query: str) -> Optional[List[str]]:
"""Returns additional browser args for a given query.
Should be overridden by child class to actually return args when necessary.
"""
del query
@classmethod
def GenerateGpuTests(cls, options: ct.ParsedCmdArgs) -> ct.TestGenerator:
cls._SetClassVariablesFromOptions(options)
if cls._test_list is None:
with open(TEST_LIST_FILE) as f:
cls._test_list = [l for l in f.read().splitlines() if l]
if cls._worker_type != WorkerType.NONE:
if cls._worker_test_globs is None:
with open(WORKER_TEST_GLOB_FILE) as f:
contents = f.read()
cls._worker_test_globs = [l for l in contents.splitlines() if l]
# Iterate through all valid test names. Generate a test for each one if not
# running worker tests, or generate a test for each if we are running worker
# tests and the test name matches a worker glob.
for line in cls._test_list: # pylint:disable=not-an-iterable
if cls._worker_type != WorkerType.NONE:
for wg in cls._worker_test_globs: # pylint:disable=not-an-iterable
if fnmatch.fnmatch(line, wg):
break
else:
continue
# At this point, we're either not running worker tests or we are running
# worker tests and the current test is a worker test.
additional_browser_args = cls._GetAdditionalBrowserArgsForQuery(line)
test_args = WebGpuTestArgs(
query=line, additional_browser_args=additional_browser_args)
yield test_args.query, HTML_FILENAME, [test_args]
def _DetermineRetryWorkaround(self, exception: Exception) -> bool:
# Instances of WebGpuMessageTimeoutError:
# https://luci-analysis.appspot.com/p/chromium/rules/b9130da14f0fcab5d6ee415d209bf71b
# https://luci-analysis.appspot.com/p/chromium/rules/6aea7ea7df4734c677f4fc5944f3b66a
if not isinstance(exception, WebGpuMessageTimeoutError):
return False
if self.browser.GetAllUnsymbolizedMinidumpPaths():
# There were some crashes, potentially expected. Don't automatically
# retry the test.
return False
# Otherwise, there were no crashes and we got a `WebGpuMessageTimeoutError`.
# Retry in case the timeout was a flake.
return True
def _TestWasSlow(self) -> bool:
# Consider the test slow if it either had a relatively long time between
# heartbeats or took up a significant chunk of the global timeout.
heartbeat_was_slow = False
if self._heartbeat_timeout > 0:
heartbeat_fraction = (self._longest_time_between_heartbeats /
self._heartbeat_timeout)
heartbeat_was_slow = heartbeat_fraction > SLOW_HEARTBEAT_THRESHOLD
global_fraction = (self._test_duration /
self._global_timeout_without_slow_multiplier)
global_was_slow = global_fraction > SLOW_GLOBAL_TIMEOUT_THRESHOLD
return heartbeat_was_slow or global_was_slow
def RunActualGpuTest(self, test_path: str, args: ct.TestArgs) -> None:
cls = self.__class__
test_args = args[0]
self._query = test_args.query
additional_browser_args = test_args.additional_browser_args
# Some CTS tests require non-standard browser arguments so we need to
# verify before running each case.
self.RestartBrowserIfNecessaryWithArgs(additional_browser_args)
# Only a single instance is used to run tests despite a number of instances
# (~2x the number of total tests) being initialized, so make sure to clear
# this state so we don't accidentally keep it around from a previous test.
if JAVASCRIPT_DURATION in self.additionalTags:
del self.additionalTags[JAVASCRIPT_DURATION]
if MAY_EXONERATE in self.additionalTags:
del self.additionalTags[MAY_EXONERATE]
if self.UseWebGpuCompatMode():
self._query += '&compatibility=1'
if self._worker_type != WorkerType.NONE:
self._query += f'&worker={self._worker_type.value}'
try:
first_load = self._NavigateIfNecessary(test_path)
cls.websocket_server.Send(json.dumps({'q': self._query}))
result = self.HandleMessageLoop(first_load)
log_str = ''.join(result.log_pieces)
if result.status in ['skip', 'fail']:
log_summary, *log_rest = log_str.split('\n', maxsplit=1)
if len(log_rest):
log_details = log_rest[0]
logging.log(logging.ERROR, log_details)
if result.status == 'skip':
self.skipTest('WebGPU CTS JavaScript reported test skip\n' +
log_summary)
elif result.status == 'fail':
self.fail(
TEST_NAME_REGEX.match(self._query).group(1) + ' failed\n' +
log_summary)
except wss.ClientClosedConnectionError as e:
raise RuntimeError(
'Detected closed websocket - likely caused by renderer crash') from e
except WebGpuMessageTimeoutError as e:
self.additionalTags[MAY_EXONERATE] = str(
self._DetermineRetryWorkaround(e))
raise
finally:
cls.total_tests_run += 1
def GetBrowserTimeoutMultipler(self) -> float:
"""Compute the timeout multiplier to account for overall browser slowness.
Returns:
A float.
"""
# Parallel jobs can cause heavier tests to flakily time out, so increase the
# timeout based on the number of parallel jobs. 2x the timeout with 4 jobs
# seemed to work well, so target that.
# This multiplier will be applied for all message events.
browser_timeout_multiplier = 1 + (self.child.jobs - 1) / 3.0
# Scale up all timeouts with ASAN.
if self._is_asan:
browser_timeout_multiplier *= ASAN_MULTIPLIER
return browser_timeout_multiplier
def GetTestExecutionTimeoutMultiplier(self) -> float:
"""Compute the timeout multiplier to account for slow test execution.
Returns:
A float.
"""
return (self._GetTestExecutionTimeoutMultiplierWithoutSlowMultiplier() *
self._GetSlowTestMultiplier())
def _GetTestExecutionTimeoutMultiplierWithoutSlowMultiplier(self) -> float:
test_execution_timeout_multiplier = 1
if self._enable_dawn_backend_validation:
test_execution_timeout_multiplier *= BACKEND_VALIDATION_MULTIPLIER
return test_execution_timeout_multiplier
def _GetSlowTestMultiplier(self) -> float:
if self._IsSlowTest():
return SLOW_MULTIPLIER
return 1
#pylint: disable=too-many-branches
def HandleMessageLoop(self, first_load) -> WebGpuTestResult:
"""Helper function to handle the loop for the message protocol.
See //docs/gpu/webgpu_cts_harness_message_protocol.md for more information
on the message format.
Args:
first_load: A bool denoting whether this is the first test run after a
page load. Increases the timeout for the TEST_STARTED step.
Returns:
A filled WebGpuTestResult instance.
"""
result = WebGpuTestResult()
message_state = {
MESSAGE_TYPE_TEST_STARTED: False,
MESSAGE_TYPE_TEST_STATUS: False,
MESSAGE_TYPE_TEST_LOG: False,
}
step_timeout = MESSAGE_TIMEOUT_TEST_STARTED
if first_load:
step_timeout *= FIRST_LOAD_TEST_STARTED_MULTIPLER
browser_timeout_multiplier = self.GetBrowserTimeoutMultipler()
test_execution_timeout_multiplier = self.GetTestExecutionTimeoutMultiplier()
global_timeout = (self._test_timeout * test_execution_timeout_multiplier *
browser_timeout_multiplier)
self._global_timeout_without_slow_multiplier = (
self._test_timeout *
self._GetTestExecutionTimeoutMultiplierWithoutSlowMultiplier() *
browser_timeout_multiplier)
start_time = time.time()
# Loop until one of the following happens:
# * The test sends MESSAGE_TYPE_TEST_FINISHED, in which case we report the
# result normally.
# * We hit the timeout for waiting for a message, e.g. if the JavaScript
# code fails to send a heartbeat, in which case we report a timeout.
# * We hit the global timeout, e.g. if the JavaScript code somehow got
# stuck in a loop, in which case we report a timeout.
# * We receive a message in the wrong order or an unknown message, in
# which case we report a message protocol error.
while True:
timeout = step_timeout * browser_timeout_multiplier
self._heartbeat_timeout = timeout
try:
response_start_time = time.time()
response = self.__class__.websocket_server.Receive(timeout)
self._longest_time_between_heartbeats = max(
self._longest_time_between_heartbeats,
time.time() - response_start_time)
response = json.loads(response)
response_type = response['type']
if time.time() - start_time > global_timeout:
self.HandleDurationTagOnFailure(message_state, global_timeout)
raise WebGpuTestTimeoutError(
'%s hit %.3f second global timeout. Message state: %s' %
(self._query, global_timeout, message_state))
if response_type == MESSAGE_TYPE_INFRA_FAILURE:
self.fail(response['message'])
if response_type == MESSAGE_TYPE_TEST_STARTED:
# If we ever want the adapter information from WebGPU, we would
# retrieve it from the message here. However, to avoid pylint
# complaining about unused variables, don't grab it until we actually
# need it.
VerifyMessageOrderTestStarted(message_state)
step_timeout = (MESSAGE_TIMEOUT_HEARTBEAT *
test_execution_timeout_multiplier)
elif response_type == MESSAGE_TYPE_TEST_HEARTBEAT:
VerifyMessageOrderTestHeartbeat(message_state)
continue
elif response_type == MESSAGE_TYPE_TEST_STATUS:
VerifyMessageOrderTestStatus(message_state)
result.status = response['status']
js_duration = response['js_duration_ms'] / 1000
# Specify the precision to avoid scientific notation. Nanoseconds
# should be more precision than we need anyways.
self.additionalTags[JAVASCRIPT_DURATION] = '%.9fs' % js_duration
step_timeout = MESSAGE_TIMEOUT_TEST_LOG
elif response_type == MESSAGE_TYPE_TEST_LOG:
VerifyMessageOrderTestLog(message_state)
result.log_pieces.append(response['log'])
elif response_type == MESSAGE_TYPE_TEST_FINISHED:
VerifyMessageOrderTestFinished(message_state)
break
else:
raise WebGpuMessageProtocolError(
'%s received unknown message type %s' % self._query,
response_type)
except wss.WebsocketReceiveMessageTimeoutError as e:
self.HandleDurationTagOnFailure(message_state, global_timeout)
raise WebGpuMessageTimeoutError(
'%s timed out waiting %.3f seconds for a message. Message state: %s'
% (self._query, timeout, message_state)) from e
finally:
self._test_duration = time.time() - start_time
return result
# pylint: enable=too-many-branches
def HandleDurationTagOnFailure(self, message_state: Dict[str, bool],
test_timeout: float) -> None:
"""Handles setting the JAVASCRIPT_DURATION tag on failure.
Args:
message_state: A map from message type to a boolean denoting whether a
message of that type has been received before.
test_timeout: A float denoting the number of seconds to wait for the test
to finish before timing out.
"""
# Report the max timeout if the JavaScript code actually timed out (i.e. we
# were between TEST_STARTED and TEST_STATUS), otherwise don't modify
# anything.
if (message_state[MESSAGE_TYPE_TEST_STARTED]
and not message_state[MESSAGE_TYPE_TEST_STATUS]
and JAVASCRIPT_DURATION not in self.additionalTags):
self.additionalTags[JAVASCRIPT_DURATION] = '%.9fs' % test_timeout
def _NavigateIfNecessary(self, path: str) -> bool:
cls = self.__class__
if cls.page_loaded:
return False
cls.websocket_server.ClearCurrentConnection()
url = self.UrlOfStaticFilePath(path)
self.tab.Navigate(url)
self.tab.action_runner.WaitForJavaScriptCondition(
'window.setupWebsocket != undefined')
self.tab.action_runner.ExecuteJavaScript('window.setupWebsocket("%s")' %
cls.websocket_server.server_port)
timeout_multiplier = 1
if not cls.attempted_websocket_connection:
cls.attempted_websocket_connection = True
timeout_multiplier = 2
cls.websocket_server.WaitForConnection(
timeout_multiplier * wsu.GetScaledConnectionTimeout(self.child.jobs))
# Wait for the page to set up the websocket.
response = cls.websocket_server.Receive(MESSAGE_TIMEOUT_CONNECTION_ACK)
assert json.loads(response)['type'] == MESSAGE_TYPE_CONNECTION_ACK
cls.page_loaded = True
return True
def _IsSlowTest(self) -> bool:
# We access the expectations directly instead of using
# self.GetExpectationsForTest since we need the raw results, but that method
# only returns the parsed results and whether the test should be retried.
expectation = self._GetSlowTests().expectations_for(self._query)
return 'Slow' in expectation.raw_results
@classmethod
def GetPlatformTags(cls, browser: ct.Browser) -> List[str]:
tags = super().GetPlatformTags(browser)
if cls._enable_dawn_backend_validation:
tags.append('dawn-backend-validation')
else:
tags.append('dawn-no-backend-validation')
if cls._use_webgpu_adapter:
tags.append('webgpu-adapter-' + cls._use_webgpu_adapter)
else:
tags.append('webgpu-adapter-default')
if host_information.IsWindows():
if cls._use_fxc:
tags.append('webgpu-dxc-disabled')
else:
tags.append('webgpu-dxc-enabled')
tags.append(_GetWorkerTag(cls._worker_type))
# No need to tag _use_webgpu_power_preference here,
# since Telemetry already reports the GPU vendorID
return tags
@classmethod
def GetExpectationsFilesRepoPath(cls) -> str:
return os.path.join(gpu_path_util.CHROMIUM_SRC_DIR, 'third_party', 'dawn')
class WebGpuMessageProtocolError(RuntimeError):
pass
class WebGpuMessageTimeoutError(RuntimeError):
pass
class WebGpuTestTimeoutError(RuntimeError):
pass
def VerifyMessageOrderTestStarted(message_state: Dict[str, bool]) -> None:
"""Helper function to verify that messages are ordered correctly.
Handles MESSAGE_TYPE_TEST_STARTED messages.
Split out to reduce the number of branches within a single function.
Args:
message_state: A map from message type to a boolean denoting whether a
message of that type has been received before.
"""
if message_state[MESSAGE_TYPE_TEST_STARTED]:
raise WebGpuMessageProtocolError(
'Received multiple start messages for one test')
message_state[MESSAGE_TYPE_TEST_STARTED] = True
def VerifyMessageOrderTestHeartbeat(message_state: Dict[str, bool]) -> None:
"""Helper function to verify that messages are ordered correctly.
Handles MESSAGE_TYPE_TEST_HEARTBEAT messages.
Split out to reduce the number of branches within a single function.
Args:
message_state: A map from message type to a boolean denoting whether a
message of that type has been received before.
"""
if not message_state[MESSAGE_TYPE_TEST_STARTED]:
raise WebGpuMessageProtocolError('Received heartbeat before test start')
if message_state[MESSAGE_TYPE_TEST_STATUS]:
raise WebGpuMessageProtocolError(
'Received heartbeat after test supposedly done')
def VerifyMessageOrderTestStatus(message_state: Dict[str, bool]) -> None:
"""Helper function to verify that messages are ordered correctly.
Handles MESSAGE_TYPE_TEST_STATUS messages.
Split out to reduce the number of branches within a single function.
Args:
message_state: A map from message type to a boolean denoting whether a
message of that type has been received before.
"""
if not message_state[MESSAGE_TYPE_TEST_STARTED]:
raise WebGpuMessageProtocolError(
'Received test status message before test start')
if message_state[MESSAGE_TYPE_TEST_STATUS]:
raise WebGpuMessageProtocolError(
'Received multiple status messages for one test')
message_state[MESSAGE_TYPE_TEST_STATUS] = True
def VerifyMessageOrderTestLog(message_state: Dict[str, bool]) -> None:
"""Helper function to verify that messages are ordered correctly.
Handles MESSAGE_TYPE_TEST_LOG messages.
Split out to reduce the number of branches within a single function.
Args:
message_state: A map from message type to a boolean denoting whether a
message of that type has been received before.
"""
if not message_state[MESSAGE_TYPE_TEST_STATUS]:
raise WebGpuMessageProtocolError(
'Received log message before status message')
message_state[MESSAGE_TYPE_TEST_LOG] = True
def VerifyMessageOrderTestFinished(message_state: Dict[str, bool]) -> None:
"""Helper function to verify that messages are ordered correctly.
Handles MESSAGE_TYPE_TEST_FINISHED messages.
Split out to reduce the number of branches within a single function.
Args:
message_state: A map from message type to a boolean denoting whether a
message of that type has been received before.
"""
if not message_state[MESSAGE_TYPE_TEST_LOG]:
raise WebGpuMessageProtocolError(
'Received finish message before log message')
def _GetWorkerTag(worker_type: WorkerType) -> str:
"""Helper function to get worker type tags."""
if worker_type == WorkerType.NONE:
return 'webgpu-no-worker'
if worker_type == WorkerType.SERVICE:
return 'webgpu-service-worker'
if worker_type == WorkerType.DEDICATED:
return 'webgpu-dedicated-worker'
if worker_type == WorkerType.SHARED:
return 'webgpu-shared-worker'
raise RuntimeError(f'Unhandled worker type {worker_type.name}')
Codebase Browser
chromium