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