Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions allways/chain_providers/bitcoin.py
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ def check_connection(self, require_send: bool = True) -> None:
raise ConnectionError(f'Cannot reach Bitcoin RPC at {self.rpc_url}')
bt.logging.success(f'BTC RPC connected: chain={result.get("chain")}, blocks={result.get("blocks")}')

def rpc_call(self, method: str, params: Optional[list] = None) -> Optional[dict]:
def rpc_call(self, method: str, params: Optional[list] = None) -> Optional[Any]:
"""Generic JSON-RPC helper for BTC Core."""
if self.mode == 'lightweight':
return None
Expand Down Expand Up @@ -303,9 +303,9 @@ def api_verify_transaction(

def get_balance(self, address: str) -> int:
"""Get balance for a Bitcoin address in satoshis via RPC with Esplora fallback."""
result = self.rpc_call('getreceivedbyaddress', [address, 0])
if result is not None:
return int(round(result * BTC_TO_SAT))
utxos = self.rpc_call('listunspent', [0, 9999999, [address]])
if isinstance(utxos, list):
return sum(int(round(float(utxo.get('amount', 0)) * BTC_TO_SAT)) for utxo in utxos)
return self.api_get_balance(address)

def btc_api_bases(self) -> Tuple[str, ...]:
Expand Down
54 changes: 54 additions & 0 deletions tests/test_bitcoin_signing.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,60 @@ def make_lightweight_provider() -> BitcoinProvider:
return BitcoinProvider()


def make_node_provider() -> BitcoinProvider:
with patch.dict(os.environ, {'BTC_MODE': 'node', 'BTC_NETWORK': 'mainnet'}, clear=False):
return BitcoinProvider()


class TestBitcoinProviderGetBalance:
def test_node_mode_sums_current_utxos(self):
provider = make_node_provider()
address = 'bc1q6tvmnmetj8vfz98vuetpvtuplqtj4uvvwjgxxc'
rpc = MagicMock(
return_value=[
{'amount': 0.25},
{'amount': 0.00000001},
]
)

with patch.object(provider, 'rpc_call', rpc):
assert provider.get_balance(address) == 25_000_001

rpc.assert_called_once_with('listunspent', [0, 9999999, [address]])

def test_node_mode_spent_address_reports_zero(self):
provider = make_node_provider()
address = 'bc1q6tvmnmetj8vfz98vuetpvtuplqtj4uvvwjgxxc'

def rpc_call(method, _params):
if method == 'listunspent':
return []
if method == 'getreceivedbyaddress':
return 1.0
return None

with (
patch.object(provider, 'rpc_call', side_effect=rpc_call) as rpc,
patch.object(provider, 'api_get_balance') as api_get_balance,
):
assert provider.get_balance(address) == 0

rpc.assert_called_once_with('listunspent', [0, 9999999, [address]])
api_get_balance.assert_not_called()

def test_node_mode_falls_back_to_esplora_when_rpc_unavailable(self):
provider = make_node_provider()
address = 'bc1q6tvmnmetj8vfz98vuetpvtuplqtj4uvvwjgxxc'

with (
patch.object(provider, 'rpc_call', return_value=None),
patch.object(provider, 'api_get_balance', return_value=12345) as api_get_balance,
):
assert provider.get_balance(address) == 12345

api_get_balance.assert_called_once_with(address)


class TestBitcoinProviderSignFromProof:
"""Direct coverage of BitcoinProvider.sign_from_proof — the wrapper our
validator/CLI actually invoke, not the underlying library."""
Expand Down