tool_utils.py raw
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()
134