1  #!/usr/bin/env python3
  2  # Copyright 2014 BitPay Inc.
  3  # Copyright 2016-present The Bitcoin Core developers
  4  # Distributed under the MIT software license, see the accompanying
  5  # file COPYING or https://opensource.org/license/mit.
  6  """Exercise the utils via json-defined tests."""
  7  
  8  from test_framework.test_framework import BitcoinTestFramework
  9  
 10  import difflib
 11  import json
 12  import os
 13  import subprocess
 14  from pathlib import Path
 15  
 16  
 17  class ToolUtils(BitcoinTestFramework):
 18      def set_test_params(self):
 19          self.num_nodes = 0  # No node/datadir needed
 20  
 21      def setup_network(self):
 22          pass
 23  
 24      def skip_test_if_missing_module(self):
 25          self.skip_if_no_bitcoin_tx()
 26          self.skip_if_no_bitcoin_util()
 27  
 28      def run_test(self):
 29          self.testcase_dir = Path(self.config["environment"]["SRCDIR"]) / "test" / "functional" / "data" / "util"
 30          self.bins = self.get_binaries()
 31          with open(self.testcase_dir / "bitcoin-util-test.json") as f:
 32              input_data = json.loads(f.read())
 33  
 34          for i, test_obj in enumerate(input_data):
 35              self.log.debug(f"Running [{i}]: " + test_obj["description"])
 36              self.test_one(test_obj)
 37  
 38      def test_one(self, testObj):
 39          """Runs a single test, comparing output and RC to expected output and RC.
 40  
 41          Raises an error if input can't be read, executable fails, or output/RC
 42          are not as expected. Error is caught by bctester() and reported.
 43          """
 44          # Get the exec names and arguments
 45          if testObj["exec"] == "./bitcoin-util":
 46              execrun = self.bins.util_argv() + testObj["args"]
 47          elif testObj["exec"] == "./bitcoin-tx":
 48              execrun = self.bins.tx_argv() + testObj["args"]
 49  
 50          # Read the input data (if there is any)
 51          inputData = None
 52          if "input" in testObj:
 53              with open(self.testcase_dir / testObj["input"]) as f:
 54                  inputData = f.read()
 55  
 56          # Read the expected output data (if there is any)
 57          outputFn = None
 58          outputData = None
 59          outputType = None
 60          if "output_cmp" in testObj:
 61              outputFn = testObj['output_cmp']
 62              outputType = os.path.splitext(outputFn)[1][1:]  # output type from file extension (determines how to compare)
 63              with open(self.testcase_dir / outputFn) as f:
 64                  outputData = f.read()
 65              if not outputData:
 66                  raise Exception(f"Output data missing for {outputFn}")
 67              if not outputType:
 68                  raise Exception(f"Output file {outputFn} does not have a file extension")
 69  
 70          # Run the test
 71          res = subprocess.run(execrun, capture_output=True, text=True, input=inputData)
 72  
 73          if outputData:
 74              data_mismatch, formatting_mismatch = False, False
 75              # Parse command output and expected output
 76              try:
 77                  a_parsed = parse_output(res.stdout, outputType)
 78              except Exception as e:
 79                  self.log.error(f"Error parsing command output as {outputType}: '{str(e)}'; res: {str(res)}")
 80                  raise
 81              try:
 82                  b_parsed = parse_output(outputData, outputType)
 83              except Exception as e:
 84                  self.log.error('Error parsing expected output %s as %s: %s' % (outputFn, outputType, e))
 85                  raise
 86              # Compare data
 87              if a_parsed != b_parsed:
 88                  self.log.error(f"Output data mismatch for {outputFn} (format {outputType}); res: {str(res)}")
 89                  data_mismatch = True
 90              # Compare formatting
 91              if res.stdout != outputData:
 92                  error_message = f"Output formatting mismatch for {outputFn}:\nres: {str(res)}\n"
 93                  error_message += "".join(difflib.context_diff(outputData.splitlines(True),
 94                                                                res.stdout.splitlines(True),
 95                                                                fromfile=outputFn,
 96                                                                tofile="returned"))
 97                  self.log.error(error_message)
 98                  formatting_mismatch = True
 99  
100              assert not data_mismatch and not formatting_mismatch
101  
102          # Compare the return code to the expected return code
103          wantRC = 0
104          if "return_code" in testObj:
105              wantRC = testObj['return_code']
106          if res.returncode != wantRC:
107              raise Exception(f"Return code mismatch for {outputFn}; res: {str(res)}")
108  
109          if "error_txt" in testObj:
110              want_error = testObj["error_txt"]
111              # A partial match instead of an exact match makes writing tests easier
112              # and should be sufficient.
113              if want_error not in res.stderr:
114                  raise Exception(f"Error mismatch:\nExpected: {want_error}\nReceived: {res.stderr.rstrip()}\nres: {str(res)}")
115          else:
116              if res.stderr:
117                  raise Exception(f"Unexpected error received: {res.stderr.rstrip()}\nres: {str(res)}")
118  
119  
120  def parse_output(a, fmt):
121      """Parse the output according to specified format.
122  
123      Raise an error if the output can't be parsed."""
124      if fmt == 'json':  # json: compare parsed data
125          return json.loads(a)
126      elif fmt == 'hex':  # hex: parse and compare binary data
127          return bytes.fromhex(a.strip())
128      else:
129          raise NotImplementedError("Don't know how to compare %s" % fmt)
130  
131  
132  if __name__ == "__main__":
133      ToolUtils(__file__).main()