tools/disable_tests/resultdb.py
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:
    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)
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.
  
Call the given RPC method, with the given request.
Call the given RPC method with the given request .
def rdb_rpc(method: str, request: dict) -> dict:
  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 = {}
    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:
    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)
def rdb_rpc ( method : str , request : dict ) -> dict : 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 = { } 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 : 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 )
Call the given RPC method, with the given request.
Call the given RPC method with the given request .
"""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.
  """ # 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.
The response from ResultDB, in dict format.
The response from ResultDB in dict format .
method
str
The method to call. Must be within luci.resultdb.v1.ResultDB.
The method to call . Must be within luci . resultdb . v1 . ResultDB .
request
dict
The request, in dict format.
The request in dict format .