resultdb.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176
  1. # Copyright 2021 The Chromium Authors. All rights reserved.
  2. # Use of this source code is governed by a BSD-style license that can be
  3. # found in the LICENSE file.
  4. """Functions for querying ResultDB, via the "rdb rpc" subcommand."""
  5. import datetime
  6. import subprocess
  7. import json
  8. import re
  9. from typing import Optional, Tuple
  10. import errors
  11. def get_test_metadata(test_id: str) -> Tuple[str, str]:
  12. """Fetch test metadata from ResultDB.
  13. Args:
  14. test_id: The full ID of the test to fetch. For Chromium tests this will
  15. begin with "ninja://"
  16. Returns:
  17. A tuple of (test name, filename). The test name will for example have the
  18. form SuitName.TestName for GTest tests. The filename is the location in the
  19. source tree where this test is defined.
  20. """
  21. history = get_test_result_history(test_id, 1)
  22. if 'entries' not in history:
  23. raise errors.UserError(
  24. f"ResultDB query couldn't find test with ID: {test_id}")
  25. # TODO: Are there other statuses we need to exclude?
  26. if history['entries'][0]['result'].get('status', '') == 'SKIP':
  27. # Test metadata isn't populated for skipped tests. Ideally this wouldn't be
  28. # the case, but for now we just work around it.
  29. # On the off-chance this test is conditionally disabled, we request a bit
  30. # more history in the hopes of finding a test run where it isn't skipped.
  31. history = get_test_result_history(test_id, 10)
  32. for entry in history['entries'][1:]:
  33. if entry['result'].get('status', '') != 'SKIP':
  34. break
  35. else:
  36. raise errors.UserError(
  37. f"Unable to fetch metadata for test {test_id}, as the last 10 runs " +
  38. "in ResultDB have status 'SKIP'. Is the test already disabled?")
  39. else:
  40. entry = history['entries'][0]
  41. try:
  42. inv_name = entry['result']['name']
  43. except KeyError as e:
  44. raise errors.InternalError(
  45. f"Malformed GetTestResultHistory response: no key {e}") from e
  46. # Ideally GetTestResultHistory would return metadata so we could avoid making
  47. # this second RPC.
  48. result = get_test_result(inv_name)
  49. try:
  50. name = result['testMetadata']['name']
  51. loc = result['testMetadata']['location']
  52. repo, filename = loc['repo'], loc['fileName']
  53. except KeyError as e:
  54. raise errors.InternalError(
  55. f"Malformed GetTestResult response: no key {e}") from e
  56. if repo != 'https://chromium.googlesource.com/chromium/src':
  57. raise errors.UserError(
  58. f"Test is in repo '{repo}', this tool can only disable tests in " +
  59. "chromium/chromium/src")
  60. return name, filename
  61. def get_test_result_history(test_id: str, page_size: int) -> dict:
  62. """Make a GetTestResultHistory RPC call to ResultDB.
  63. Args:
  64. test_id: The full test ID to query. This can be a regex.
  65. page_size: The number of results to return within the first response.
  66. Returns:
  67. The GetTestResultHistoryResponse message, in dict form.
  68. """
  69. now = datetime.datetime.now(datetime.timezone.utc)
  70. request = {
  71. 'realm': 'chromium:ci',
  72. 'testIdRegexp': test_id,
  73. 'timeRange': {
  74. 'earliest': (now - datetime.timedelta(hours=6)).isoformat(),
  75. 'latest': now.isoformat(),
  76. },
  77. 'pageSize': page_size,
  78. }
  79. return rdb_rpc('GetTestResultHistory', request)
  80. def get_test_result(test_name: str) -> dict:
  81. """Make a GetTestResult RPC call to ResultDB.
  82. Args:
  83. test_name: The name of the test result to query. This specifies a result for
  84. a particular test, within a particular test run. As returned by
  85. GetTestResultHistory.
  86. Returns:
  87. The TestResult message, in dict form.
  88. """
  89. return rdb_rpc('GetTestResult', {
  90. 'name': test_name,
  91. })
  92. # Used for caching RPC responses, for development purposes.
  93. CANNED_RESPONSE_FILE: Optional[str] = None
  94. def rdb_rpc(method: str, request: dict) -> dict:
  95. """Call the given RPC method, with the given request.
  96. Args:
  97. method: The method to call. Must be within luci.resultdb.v1.ResultDB.
  98. request: The request, in dict format.
  99. Returns:
  100. The response from ResultDB, in dict format.
  101. """
  102. if CANNED_RESPONSE_FILE is not None:
  103. try:
  104. with open(CANNED_RESPONSE_FILE, 'r') as f:
  105. canned_responses = json.load(f)
  106. except Exception:
  107. canned_responses = {}
  108. # HACK: Strip out timestamps when caching the request. GetTestResultHistory
  109. # includes timestamps based on the current time, which will bust the cache.
  110. # But for e2e testing we just want to cache the result the first time and
  111. # then keep using it.
  112. if 'timeRange' in request:
  113. key_request = dict(request)
  114. del key_request['timeRange']
  115. else:
  116. key_request = request
  117. key = f'{method}/{json.dumps(key_request)}'
  118. if (response_json := canned_responses.get(key, None)) is not None:
  119. return json.loads(response_json)
  120. p = subprocess.Popen(['rdb', 'rpc', 'luci.resultdb.v1.ResultDB', method],
  121. stdin=subprocess.PIPE,
  122. stdout=subprocess.PIPE,
  123. stderr=subprocess.PIPE,
  124. text=True)
  125. stdout, stderr = p.communicate(json.dumps(request))
  126. if p.returncode != 0:
  127. # rdb doesn't return unique status codes for different errors, so we have to
  128. # just match on the output.
  129. if 'interactive login is required' in stderr:
  130. raise errors.UserError(
  131. "Authentication is required to fetch test metadata.\n" +
  132. "Please run:\n\trdb auth-login\nand try again")
  133. raise Exception(f'rdb rpc {method} failed with: {stderr}')
  134. if CANNED_RESPONSE_FILE:
  135. canned_responses[key] = stdout
  136. with open(CANNED_RESPONSE_FILE, 'w') as f:
  137. json.dump(canned_responses, f)
  138. return json.loads(stdout)