123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176 |
- # Copyright 2021 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- """Functions for querying ResultDB, via the "rdb rpc" subcommand."""
- import datetime
- import subprocess
- import json
- import re
- from typing import Optional, Tuple
- import errors
- def get_test_metadata(test_id: str) -> Tuple[str, str]:
- """Fetch test metadata from ResultDB.
- Args:
- test_id: The full ID of the test to fetch. For Chromium tests this will
- begin with "ninja://"
- Returns:
- A tuple of (test name, filename). The test name will for example have the
- form SuitName.TestName for GTest tests. The filename is the location in the
- source tree where this test is defined.
- """
- history = get_test_result_history(test_id, 1)
- if 'entries' not in history:
- raise errors.UserError(
- f"ResultDB query couldn't find test with ID: {test_id}")
- # TODO: Are there other statuses we need to exclude?
- if history['entries'][0]['result'].get('status', '') == 'SKIP':
- # Test metadata isn't populated for skipped tests. Ideally this wouldn't be
- # the case, but for now we just work around it.
- # On the off-chance this test is conditionally disabled, we request a bit
- # more history in the hopes of finding a test run where it isn't skipped.
- history = get_test_result_history(test_id, 10)
- for entry in history['entries'][1:]:
- if entry['result'].get('status', '') != 'SKIP':
- break
- else:
- raise errors.UserError(
- f"Unable to fetch metadata for test {test_id}, as the last 10 runs " +
- "in ResultDB have status 'SKIP'. Is the test already disabled?")
- else:
- entry = history['entries'][0]
- try:
- inv_name = entry['result']['name']
- except KeyError as e:
- raise errors.InternalError(
- f"Malformed GetTestResultHistory response: no key {e}") from e
- # Ideally GetTestResultHistory would return metadata so we could avoid making
- # this second RPC.
- result = get_test_result(inv_name)
- try:
- name = result['testMetadata']['name']
- loc = result['testMetadata']['location']
- repo, filename = loc['repo'], loc['fileName']
- except KeyError as e:
- raise errors.InternalError(
- f"Malformed GetTestResult response: no key {e}") from e
- if repo != 'https://chromium.googlesource.com/chromium/src':
- raise errors.UserError(
- f"Test is in repo '{repo}', this tool can only disable tests in " +
- "chromium/chromium/src")
- return name, filename
- def get_test_result_history(test_id: str, page_size: int) -> dict:
- """Make a GetTestResultHistory RPC call to ResultDB.
- Args:
- test_id: The full test ID to query. This can be a regex.
- page_size: The number of results to return within the first response.
- Returns:
- The GetTestResultHistoryResponse message, in dict form.
- """
- now = datetime.datetime.now(datetime.timezone.utc)
- request = {
- 'realm': 'chromium:ci',
- 'testIdRegexp': test_id,
- 'timeRange': {
- 'earliest': (now - datetime.timedelta(hours=6)).isoformat(),
- 'latest': now.isoformat(),
- },
- 'pageSize': page_size,
- }
- return rdb_rpc('GetTestResultHistory', request)
- def get_test_result(test_name: str) -> dict:
- """Make a GetTestResult RPC call to ResultDB.
- Args:
- test_name: The name of the test result to query. This specifies a result for
- a particular test, within a particular test run. As returned by
- GetTestResultHistory.
- Returns:
- The TestResult message, in dict form.
- """
- return rdb_rpc('GetTestResult', {
- 'name': test_name,
- })
- # Used for caching RPC responses, for development purposes.
- CANNED_RESPONSE_FILE: Optional[str] = None
- def rdb_rpc(method: str, request: dict) -> dict:
- """Call the given RPC method, with the given request.
- Args:
- method: The method to call. Must be within luci.resultdb.v1.ResultDB.
- request: The request, in dict format.
- Returns:
- The response from ResultDB, in dict format.
- """
- if CANNED_RESPONSE_FILE is not None:
- try:
- with open(CANNED_RESPONSE_FILE, 'r') as f:
- canned_responses = json.load(f)
- except Exception:
- canned_responses = {}
- # HACK: Strip out timestamps when caching the request. GetTestResultHistory
- # includes timestamps based on the current time, which will bust the cache.
- # But for e2e testing we just want to cache the result the first time and
- # then keep using it.
- if 'timeRange' in request:
- key_request = dict(request)
- del key_request['timeRange']
- else:
- key_request = request
- key = f'{method}/{json.dumps(key_request)}'
- if (response_json := canned_responses.get(key, None)) is not None:
- return json.loads(response_json)
- p = subprocess.Popen(['rdb', 'rpc', 'luci.resultdb.v1.ResultDB', method],
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- text=True)
- stdout, stderr = p.communicate(json.dumps(request))
- if p.returncode != 0:
- # rdb doesn't return unique status codes for different errors, so we have to
- # just match on the output.
- if 'interactive login is required' in stderr:
- raise errors.UserError(
- "Authentication is required to fetch test metadata.\n" +
- "Please run:\n\trdb auth-login\nand try again")
- raise Exception(f'rdb rpc {method} failed with: {stderr}')
- if CANNED_RESPONSE_FILE:
- canned_responses[key] = stdout
- with open(CANNED_RESPONSE_FILE, 'w') as f:
- json.dump(canned_responses, f)
- return json.loads(stdout)
|