diff --git a/bitcoinutils/transactions.py b/bitcoinutils/transactions.py index 9591759..dc65c10 100644 --- a/bitcoinutils/transactions.py +++ b/bitcoinutils/transactions.py @@ -678,6 +678,50 @@ def set_witness(self, txin_index: int, witness: TxWitnessInput): raise IndexError("txin_index out of range") self.witnesses[txin_index] = witness + def add_input(self, txin: TxInput) -> None: + """Append a transaction input. + + If the transaction is segwit an empty witness entry is appended so + that the witnesses list stays parallel to inputs. + """ + self.inputs.append(txin) + if self.has_segwit: + self.set_witness(len(self.inputs) - 1, TxWitnessInput([])) + + def add_output(self, txout: TxOutput) -> None: + """Append a transaction output.""" + self.outputs.append(txout) + + def remove_input(self, index: int) -> None: + """Remove the transaction input at *index*. + + If the transaction is segwit the corresponding witness entry is + also removed. Negative indices are rejected. + """ + if index < 0 or index >= len(self.inputs): + raise IndexError("Input index out of range") + del self.inputs[index] + if self.has_segwit and index < len(self.witnesses): + del self.witnesses[index] + + def remove_output(self, index: int) -> None: + """Remove the transaction output at *index*. Negative indices are rejected.""" + if index < 0 or index >= len(self.outputs): + raise IndexError("Output index out of range") + del self.outputs[index] + + def update_input(self, index: int, txin: TxInput) -> None: + """Replace the transaction input at *index*. Negative indices are rejected.""" + if index < 0 or index >= len(self.inputs): + raise IndexError("Input index out of range") + self.inputs[index] = txin + + def update_output(self, index: int, txout: TxOutput) -> None: + """Replace the transaction output at *index*. Negative indices are rejected.""" + if index < 0 or index >= len(self.outputs): + raise IndexError("Output index out of range") + self.outputs[index] = txout + def get_transaction_digest( self, txin_index: int, script: Script, sighash: int = SIGHASH_ALL ): diff --git a/tests/test_transaction_mutation.py b/tests/test_transaction_mutation.py new file mode 100644 index 0000000..ae90222 --- /dev/null +++ b/tests/test_transaction_mutation.py @@ -0,0 +1,173 @@ +# Copyright (C) 2018-2025 The python-bitcoin-utils developers +# +# This file is part of python-bitcoin-utils +# +# It is subject to the license terms in the LICENSE file found in the top-level +# directory of this distribution. +# +# No part of python-bitcoin-utils, including this file, may be copied, +# modified, propagated, or distributed except according to the terms contained +# in the LICENSE file. + +import unittest + +from bitcoinutils.setup import setup +from bitcoinutils.transactions import ( + Transaction, + TxInput, + TxOutput, + TxWitnessInput, +) +from bitcoinutils.script import Script + + +class TestTransactionMutation(unittest.TestCase): + def setUp(self): + setup("testnet") + self.txin1 = TxInput( + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", 0 + ) + self.txin2 = TxInput( + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", 1 + ) + self.txout1 = TxOutput( + 10000, + Script(["OP_DUP", "OP_HASH160", "aa" * 20, "OP_EQUALVERIFY", "OP_CHECKSIG"]), + ) + self.txout2 = TxOutput( + 20000, + Script(["OP_1", "bb" * 32]), + ) + + def txid_index(self, tx): + return [(i.txid, i.txout_index) for i in tx.inputs] + + def amount_list(self, tx): + return [o.amount for o in tx.outputs] + + def test_add_input_preserves_existing_inputs(self): + tx = Transaction([self.txin1], [self.txout1]) + tx.add_input(self.txin2) + self.assertEqual(self.txid_index(tx), [("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", 0), ("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", 1)]) + + def test_add_input_segwit_backfills_witnesses(self): + tx = Transaction([self.txin1], [self.txout1], has_segwit=True) + self.assertEqual(len(tx.witnesses), 0) + tx.add_input(self.txin2) + self.assertEqual( + len(tx.witnesses), len(tx.inputs), + "witnesses must stay parallel with inputs", + ) + self.assertEqual(tx.witnesses[-1].stack, []) + + def test_add_input_non_segwit_no_witness(self): + tx = Transaction([self.txin1], [self.txout1], has_segwit=False) + tx.add_input(self.txin2) + self.assertEqual(len(tx.witnesses), 0) + + def test_add_output_preserves_existing_outputs(self): + tx = Transaction([self.txin1], [self.txout1]) + tx.add_output(self.txout2) + self.assertEqual(self.amount_list(tx), [10000, 20000]) + + def test_remove_input_removes_correct_item_and_witness(self): + tx = Transaction( + [self.txin1, self.txin2], + [self.txout1], + has_segwit=True, + witnesses=[TxWitnessInput(["deadbeef"]), TxWitnessInput([])], + ) + tx.remove_input(0) + self.assertEqual(self.txid_index(tx), [("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", 1)]) + self.assertEqual(len(tx.witnesses), len(tx.inputs)) + self.assertEqual(tx.witnesses[0].stack, []) + + def test_remove_output_removes_correct_item(self): + tx = Transaction([self.txin1], [self.txout1, self.txout2]) + tx.remove_output(0) + self.assertEqual(self.amount_list(tx), [20000]) + + def test_update_input_replaces_item_keeps_length(self): + tx = Transaction([self.txin1], [self.txout1]) + tx.update_input(0, self.txin2) + self.assertEqual(len(tx.inputs), 1) + self.assertEqual(self.txid_index(tx), [("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", 1)]) + + def test_update_output_replaces_item_keeps_length(self): + tx = Transaction([self.txin1], [self.txout1]) + tx.update_output(0, self.txout2) + self.assertEqual(len(tx.outputs), 1) + self.assertEqual(self.amount_list(tx), [20000]) + + def test_negative_and_out_of_bounds_indices_rejected(self): + tx_inputs = Transaction([self.txin1], [self.txout1]) + tx_outputs = Transaction([self.txin1], [self.txout1]) + for index in (-1, 5): + with self.assertRaises(IndexError): + tx_inputs.remove_input(index) + with self.assertRaises(IndexError): + tx_inputs.update_input(index, self.txin2) + with self.assertRaises(IndexError): + tx_outputs.remove_output(index) + with self.assertRaises(IndexError): + tx_outputs.update_output(index, self.txout2) + + def test_mutation_on_empty_transaction_raises(self): + tx = Transaction([], []) + with self.assertRaises(IndexError): + tx.remove_input(0) + with self.assertRaises(IndexError): + tx.remove_output(0) + with self.assertRaises(IndexError): + tx.update_input(0, self.txin1) + with self.assertRaises(IndexError): + tx.update_output(0, self.txout1) + + def test_add_input_then_serialize(self): + tx = Transaction([self.txin1], [self.txout1]) + tx.add_input(self.txin2) + parsed = Transaction.from_raw(tx.to_hex()) + self.assertEqual(len(parsed.inputs), 2) + self.assertEqual(len(parsed.outputs), 1) + + def test_add_output_and_remove_then_serialize(self): + tx = Transaction([self.txin1], [self.txout1]) + tx.add_output(self.txout2) + tx.remove_input(0) + tx.add_input(self.txin2) + parsed = Transaction.from_raw(tx.to_hex()) + self.assertEqual(self.txid_index(parsed), [("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", 1)]) + self.assertEqual(self.amount_list(parsed), [10000, 20000]) + + def test_segwit_add_input_round_trip_nonzero_locktime(self): + tx = Transaction( + [self.txin1], + [self.txout1], + locktime="0000f406", + has_segwit=True, + witnesses=[TxWitnessInput(["deadbeef"])], + ) + tx.add_input(self.txin2) + self.assertEqual(len(tx.witnesses), len(tx.inputs)) + parsed = Transaction.from_raw(tx.to_hex()) + self.assertEqual(parsed.locktime, bytes.fromhex("0000f406")) + self.assertEqual(len(parsed.inputs), 2) + self.assertEqual(parsed.witnesses[0].stack, ["deadbeef"]) + + def test_update_input_changes_txid(self): + tx = Transaction([self.txin1], [self.txout1]) + old_txid = tx.get_txid() + tx.update_input(0, self.txin2) + self.assertNotEqual(tx.get_txid(), old_txid) + + def test_mutation_does_not_affect_copy(self): + tx = Transaction([self.txin1], [self.txout1]) + tx_copy = Transaction.copy(tx) + tx.add_input(self.txin2) + tx.remove_output(0) + self.assertEqual(len(tx_copy.inputs), 1) + self.assertEqual(len(tx_copy.outputs), 1) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file