coverage.py raw

   1  #!/usr/bin/env python3
   2  # Copyright (c) 2015-present The Bitcoin Core developers
   3  # Distributed under the MIT software license, see the accompanying
   4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5  """Utilities for doing coverage analysis on the RPC interface.
   6  
   7  Provides a way to track which RPC commands are exercised during
   8  testing.
   9  """
  10  
  11  import os
  12  
  13  from .authproxy import AuthServiceProxy
  14  from typing import Optional
  15  
  16  REFERENCE_FILENAME = 'rpc_interface.txt'
  17  
  18  
  19  class AuthServiceProxyWrapper():
  20      """
  21      An object that wraps AuthServiceProxy to record specific RPC calls.
  22  
  23      """
  24      def __init__(self, auth_service_proxy_instance: AuthServiceProxy, rpc_url: str, coverage_logfile: Optional[str]=None):
  25          """
  26          Kwargs:
  27              auth_service_proxy_instance: the instance being wrapped.
  28              rpc_url: url of the RPC instance being wrapped
  29              coverage_logfile: if specified, write each service_name
  30                  out to a file when called.
  31  
  32          """
  33          self.auth_service_proxy_instance = auth_service_proxy_instance
  34          self.rpc_url = rpc_url
  35          self.coverage_logfile = coverage_logfile
  36  
  37      def __getattr__(self, name):
  38          return_val = getattr(self.auth_service_proxy_instance, name)
  39          if not isinstance(return_val, type(self.auth_service_proxy_instance)):
  40              # If proxy getattr returned an unwrapped value, do the same here.
  41              return return_val
  42          return AuthServiceProxyWrapper(return_val, self.rpc_url, self.coverage_logfile)
  43  
  44      def __call__(self, *args, **kwargs):
  45          """
  46          Delegates to AuthServiceProxy, then writes the particular RPC method
  47          called to a file.
  48  
  49          """
  50          return_val = self.auth_service_proxy_instance.__call__(*args, **kwargs)
  51          self._log_call()
  52          return return_val
  53  
  54      def _log_call(self):
  55          rpc_method = self.auth_service_proxy_instance._service_name
  56  
  57          if self.coverage_logfile:
  58              with open(self.coverage_logfile, 'a+') as f:
  59                  f.write("%s\n" % rpc_method)
  60  
  61      def __truediv__(self, relative_uri):
  62          return AuthServiceProxyWrapper(self.auth_service_proxy_instance / relative_uri,
  63                                         self.rpc_url,
  64                                         self.coverage_logfile)
  65  
  66      def get_request(self, *args, **kwargs):
  67          self._log_call()
  68          return self.auth_service_proxy_instance.get_request(*args, **kwargs)
  69  
  70  def get_filename(dirname, n_node):
  71      """
  72      Get a filename unique to the test process ID and node.
  73  
  74      This file will contain a list of RPC commands covered.
  75      """
  76      pid = str(os.getpid())
  77      return os.path.join(
  78          dirname, "coverage.pid%s.node%s.txt" % (pid, str(n_node)))
  79  
  80  
  81  def write_all_rpc_commands(dirname: str, node: AuthServiceProxy) -> bool:
  82      """
  83      Write out a list of all RPC functions available in `bitcoin-cli` for
  84      coverage comparison. This will only happen once per coverage
  85      directory.
  86  
  87      Args:
  88          dirname: temporary test dir
  89          node: client
  90  
  91      Returns:
  92          if the RPC interface file was written.
  93  
  94      """
  95      filename = os.path.join(dirname, REFERENCE_FILENAME)
  96  
  97      if os.path.isfile(filename):
  98          return False
  99  
 100      help_output = node.help().split('\n')
 101      commands = set()
 102  
 103      for line in help_output:
 104          line = line.strip()
 105  
 106          # Ignore blanks and headers
 107          if line and not line.startswith('='):
 108              commands.add("%s\n" % line.split()[0])
 109  
 110      with open(filename, 'w') as f:
 111          f.writelines(list(commands))
 112  
 113      return True
 114