test: refactor: replace unnecessary BytesIO uses

Rather than needing to create intermediate stream variables, we can use
helper functions like `tx_from_hex` instead or access the result
directly, leading both to increased readability and less code.
This commit is contained in:
Sebastian Falbesoner 2023-04-01 14:10:01 +02:00
parent a0d37d1d23
commit f842ed9a40
3 changed files with 9 additions and 15 deletions

View File

@ -19,6 +19,7 @@ from test_framework.messages import (
CTxInWitness, CTxInWitness,
CTxOut, CTxOut,
SEQUENCE_FINAL, SEQUENCE_FINAL,
tx_from_hex,
) )
from test_framework.script import ( from test_framework.script import (
ANNEX_TAG, ANNEX_TAG,
@ -109,7 +110,6 @@ from test_framework.address import (
program_to_witness, program_to_witness,
) )
from collections import OrderedDict, namedtuple from collections import OrderedDict, namedtuple
from io import BytesIO
import json import json
import hashlib import hashlib
import os import os
@ -1386,8 +1386,7 @@ class TaprootTest(BitcoinTestFramework):
# Add change # Add change
fund_tx.vout.append(CTxOut(balance - 10000, random.choice(host_spks))) fund_tx.vout.append(CTxOut(balance - 10000, random.choice(host_spks)))
# Ask the wallet to sign # Ask the wallet to sign
ss = BytesIO(bytes.fromhex(node.signrawtransactionwithwallet(fund_tx.serialize().hex())["hex"])) fund_tx = tx_from_hex(node.signrawtransactionwithwallet(fund_tx.serialize().hex())["hex"])
fund_tx.deserialize(ss)
# Construct UTXOData entries # Construct UTXOData entries
fund_tx.rehash() fund_tx.rehash()
for i in range(count_this_tx): for i in range(count_this_tx):

View File

@ -7,9 +7,7 @@
from decimal import Decimal from decimal import Decimal
from enum import Enum from enum import Enum
import http.client import http.client
from io import BytesIO
import json import json
from struct import pack, unpack
import typing import typing
import urllib.parse import urllib.parse
@ -160,12 +158,11 @@ class RESTTest (BitcoinTestFramework):
bin_request = b'\x01\x02' bin_request = b'\x01\x02'
for txid, n in [spending, spent]: for txid, n in [spending, spent]:
bin_request += bytes.fromhex(txid) bin_request += bytes.fromhex(txid)
bin_request += pack("i", n) bin_request += n.to_bytes(4, 'little')
bin_response = self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body=bin_request, ret_type=RetType.BYTES) bin_response = self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body=bin_request, ret_type=RetType.BYTES)
output = BytesIO(bin_response) chain_height = int.from_bytes(bin_response[0:4], 'little')
chain_height, = unpack("<i", output.read(4)) response_hash = bin_response[4:36][::-1].hex()
response_hash = output.read(32)[::-1].hex()
assert_equal(bb_hash, response_hash) # check if getutxo's chaintip during calculation was fine assert_equal(bb_hash, response_hash) # check if getutxo's chaintip during calculation was fine
assert_equal(chain_height, 201) # chain height must be 201 (pre-mined chain [200] + generated block [1]) assert_equal(chain_height, 201) # chain height must be 201 (pre-mined chain [200] + generated block [1])

View File

@ -4,6 +4,7 @@
# file COPYING or http://www.opensource.org/licenses/mit-license.php. # file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test the ZMQ notification interface.""" """Test the ZMQ notification interface."""
import struct import struct
from time import sleep
from test_framework.address import ( from test_framework.address import (
ADDRESS_BCRT1_P2WSH_OP_TRUE, ADDRESS_BCRT1_P2WSH_OP_TRUE,
@ -16,8 +17,8 @@ from test_framework.blocktools import (
) )
from test_framework.test_framework import BitcoinTestFramework from test_framework.test_framework import BitcoinTestFramework
from test_framework.messages import ( from test_framework.messages import (
CTransaction,
hash256, hash256,
tx_from_hex,
) )
from test_framework.util import ( from test_framework.util import (
assert_equal, assert_equal,
@ -28,8 +29,7 @@ from test_framework.wallet import (
MiniWallet, MiniWallet,
) )
from test_framework.netutil import test_ipv6_local from test_framework.netutil import test_ipv6_local
from io import BytesIO
from time import sleep
# Test may be skipped and not have zmq installed # Test may be skipped and not have zmq installed
try: try:
@ -198,9 +198,7 @@ class ZMQTest (BitcoinTestFramework):
txid = hashtx.receive() txid = hashtx.receive()
# Should receive the coinbase raw transaction. # Should receive the coinbase raw transaction.
hex = rawtx.receive() tx = tx_from_hex(rawtx.receive().hex())
tx = CTransaction()
tx.deserialize(BytesIO(hex))
tx.calc_sha256() tx.calc_sha256()
assert_equal(tx.hash, txid.hex()) assert_equal(tx.hash, txid.hex())