From 34293847062705c699b6159c0e81a3c319edae1a Mon Sep 17 00:00:00 2001 From: "marek.galvanek" Date: Wed, 25 Mar 2026 18:42:59 +0100 Subject: [PATCH] fix: refactor Terra API using CosmosApiBase, add tests for balance and IBC denom handling --- blockapi/test/v2/api/test_terra.py | 305 +++++++++++++++++++++++++++ blockapi/v2/api/terra.py | 319 +++-------------------------- blockapi/v2/coins.py | 4 +- 3 files changed, 337 insertions(+), 291 deletions(-) create mode 100644 blockapi/test/v2/api/test_terra.py diff --git a/blockapi/test/v2/api/test_terra.py b/blockapi/test/v2/api/test_terra.py new file mode 100644 index 0000000..1ab58e4 --- /dev/null +++ b/blockapi/test/v2/api/test_terra.py @@ -0,0 +1,305 @@ +from decimal import Decimal + +import pytest + +from blockapi.v2.api.terra import TerraApi +from blockapi.v2.models import AssetType, Blockchain + + +@pytest.fixture() +def terra_api(requests_mock): + """TerraApi with token mapping disabled to avoid external calls.""" + requests_mock.get( + 'https://raw.githubusercontent.com/PulsarDefi/IBC-Token-Data-Cosmos/main/native_token_data.min.json', + json={ + 'uluna__terra': { + 'name': 'Terra Classic', + 'chain': 'terra', + 'denom': 'uluna', + 'symbol': 'LUNC', + 'decimals': 6, + 'coingecko_id': 'terra-luna', + 'bridge_asset': None, + 'logos': {}, + }, + }, + ) + requests_mock.get( + 'https://raw.githubusercontent.com/PulsarDefi/IBC-Token-Data-Cosmos/main/ibc_data.min.json', + json={}, + ) + return TerraApi() + + +@pytest.fixture() +def balances_response(): + return { + 'balances': [ + {'denom': 'uluna', 'amount': '23068009633'}, + {'denom': 'uusd', 'amount': '85997844'}, + ], + 'pagination': {'next_key': None, 'total': '2'}, + } + + +@pytest.fixture() +def staking_response(): + return { + 'delegation_responses': [ + { + 'delegation': { + 'delegator_address': 'terra1test', + 'validator_address': 'terravaloper1test', + 'shares': '100000000.000000000000000000', + }, + 'balance': {'denom': 'uluna', 'amount': '100000000'}, + } + ], + 'pagination': {'next_key': None, 'total': '1'}, + } + + +@pytest.fixture() +def unbonding_response(): + return { + 'unbonding_responses': [], + 'pagination': {'next_key': None, 'total': '0'}, + } + + +@pytest.fixture() +def rewards_response(): + return { + 'rewards': [ + { + 'validator_address': 'terravaloper1test', + 'reward': [ + {'denom': 'uluna', 'amount': '5000000.000000000000000000'}, + {'denom': 'uusd', 'amount': '1000000.000000000000000000'}, + ], + } + ], + 'total': [ + {'denom': 'uluna', 'amount': '5000000.000000000000000000'}, + {'denom': 'uusd', 'amount': '1000000.000000000000000000'}, + ], + } + + +@pytest.fixture() +def ibc_denom_trace_response(): + return { + 'denom_trace': { + 'path': 'transfer/channel-7', + 'base_denom': 'xrowan', + } + } + + +ADDRESS = 'terra1yltenl48mhl370ldpyt83werd9x3s645509gaf' +BASE_URL = 'https://terra-classic-fcd.publicnode.com' + + +def test_terra_api_options(): + api = TerraApi(enable_token_mapping=False) + assert api.api_options.blockchain == Blockchain.TERRA + assert api.coin.symbol == 'LUNC' + assert api.TOKENS_MAP_BLOCKCHAIN_KEY == 'terra' + + +def test_get_available_balances(terra_api, balances_response, requests_mock): + requests_mock.get( + f'{BASE_URL}/cosmos/bank/v1beta1/balances/{ADDRESS}', + json=balances_response, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegations/{ADDRESS}', + json={'delegation_responses': [], 'pagination': {}}, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegators/{ADDRESS}/unbonding_delegations', + json={'unbonding_responses': [], 'pagination': {}}, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/distribution/v1beta1/delegators/{ADDRESS}/rewards', + json={'rewards': [], 'total': []}, + ) + + balances = terra_api.get_balance(ADDRESS) + + available = [b for b in balances if b.asset_type == AssetType.AVAILABLE] + assert len(available) == 2 + + luna = next(b for b in available if b.coin.symbol == 'LUNC') + assert luna.balance == Decimal('23068.009633') + + usd = next(b for b in available if b.coin.address == 'uusd') + assert usd.balance == Decimal('85.997844') + + +def test_get_staking_balances( + terra_api, + balances_response, + staking_response, + unbonding_response, + rewards_response, + requests_mock, +): + requests_mock.get( + f'{BASE_URL}/cosmos/bank/v1beta1/balances/{ADDRESS}', + json=balances_response, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegations/{ADDRESS}', + json=staking_response, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegators/{ADDRESS}/unbonding_delegations', + json=unbonding_response, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/distribution/v1beta1/delegators/{ADDRESS}/rewards', + json=rewards_response, + ) + + balances = terra_api.get_balance(ADDRESS) + + staked = [b for b in balances if b.asset_type == AssetType.STAKED] + assert len(staked) == 1 + assert staked[0].balance == Decimal('100') + assert staked[0].coin.symbol == 'LUNC' + + rewards = [b for b in balances if b.asset_type == AssetType.REWARDS] + assert len(rewards) == 2 + + luna_reward = next(b for b in rewards if b.coin.address == 'uluna') + assert luna_reward.balance == Decimal('5') + + +def test_resolve_ibc_denom(terra_api, ibc_denom_trace_response, requests_mock): + ibc_hash = '0A866A7A214C42CEF84430C8A4C7210C8C7A980548A9B9BE64316D1610A87C6C' + ibc_denom = f'ibc/{ibc_hash}' + + requests_mock.get( + f'{BASE_URL}/ibc/apps/transfer/v1/denom_traces/{ibc_hash}', + json=ibc_denom_trace_response, + ) + + coin = terra_api.create_default_coin(ibc_denom) + assert coin.symbol == 'ROWAN' + assert coin.address == ibc_denom + assert 'ibc' in coin.standards + + +def test_resolve_ibc_denom_fallback_on_error(terra_api, requests_mock): + ibc_hash = 'DEADBEEF' + ibc_denom = f'ibc/{ibc_hash}' + + requests_mock.get( + f'{BASE_URL}/ibc/apps/transfer/v1/denom_traces/{ibc_hash}', + status_code=404, + ) + + coin = terra_api.create_default_coin(ibc_denom) + assert coin.address == ibc_denom + assert coin.blockchain == Blockchain.TERRA + + +def test_non_ibc_denom_uses_default(terra_api): + coin = terra_api.create_default_coin('ufoo') + assert coin.address == 'ufoo' + assert coin.blockchain == Blockchain.TERRA + assert coin.decimals == 6 + + +def test_get_balance_with_ibc_tokens( + terra_api, + staking_response, + unbonding_response, + rewards_response, + ibc_denom_trace_response, + requests_mock, +): + ibc_hash = '0A866A7A214C42CEF84430C8A4C7210C8C7A980548A9B9BE64316D1610A87C6C' + + requests_mock.get( + f'{BASE_URL}/cosmos/bank/v1beta1/balances/{ADDRESS}', + json={ + 'balances': [ + {'denom': 'uluna', 'amount': '1000000'}, + {'denom': f'ibc/{ibc_hash}', 'amount': '500000'}, + ], + 'pagination': {'next_key': None, 'total': '2'}, + }, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegations/{ADDRESS}', + json={'delegation_responses': [], 'pagination': {}}, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegators/{ADDRESS}/unbonding_delegations', + json={'unbonding_responses': [], 'pagination': {}}, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/distribution/v1beta1/delegators/{ADDRESS}/rewards', + json={'rewards': [], 'total': []}, + ) + requests_mock.get( + f'{BASE_URL}/ibc/apps/transfer/v1/denom_traces/{ibc_hash}', + json=ibc_denom_trace_response, + ) + + balances = terra_api.get_balance(ADDRESS) + assert len(balances) == 2 + + rowan = next(b for b in balances if b.coin.symbol == 'ROWAN') + assert rowan.balance == Decimal('0.5') + assert 'ibc' in rowan.coin.standards + + +def test_unbonding_included_in_staked(terra_api, requests_mock): + requests_mock.get( + f'{BASE_URL}/cosmos/bank/v1beta1/balances/{ADDRESS}', + json={'balances': [], 'pagination': {}}, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegations/{ADDRESS}', + json={ + 'delegation_responses': [ + { + 'delegation': { + 'delegator_address': ADDRESS, + 'validator_address': 'terravaloper1test', + 'shares': '50000000', + }, + 'balance': {'denom': 'uluna', 'amount': '50000000'}, + } + ], + 'pagination': {}, + }, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/staking/v1beta1/delegators/{ADDRESS}/unbonding_delegations', + json={ + 'unbonding_responses': [ + { + 'delegator_address': ADDRESS, + 'validator_address': 'terravaloper1test2', + 'entries': [ + {'balance': '25000000', 'completion_time': '2026-04-01'}, + ], + } + ], + 'pagination': {}, + }, + ) + requests_mock.get( + f'{BASE_URL}/cosmos/distribution/v1beta1/delegators/{ADDRESS}/rewards', + json={'rewards': [], 'total': []}, + ) + + balances = terra_api.get_balance(ADDRESS) + staked = [b for b in balances if b.asset_type == AssetType.STAKED] + assert len(staked) == 1 + # 50 delegated + 25 unbonding = 75 + assert staked[0].balance == Decimal('75') diff --git a/blockapi/v2/api/terra.py b/blockapi/v2/api/terra.py index 6240337..a3b313b 100644 --- a/blockapi/v2/api/terra.py +++ b/blockapi/v2/api/terra.py @@ -1,318 +1,59 @@ -import json +import logging from functools import lru_cache -from typing import Dict, List, Optional, Sequence, Tuple -from cytoolz import concatv -from requests import Response - -from blockapi.v2.base import ( - ApiException, - ApiOptions, - BalanceMixin, - BlockchainApi, - InvalidAddressException, -) +from blockapi.v2.api.cosmos import CosmosApiBase +from blockapi.v2.base import ApiException, ApiOptions from blockapi.v2.coins import COIN_TERRA -from blockapi.v2.models import ( - AssetType, - BalanceItem, - Blockchain, - Coin, - CoinInfo, - FetchResult, - ParseResult, -) +from blockapi.v2.models import Blockchain, Coin +logger = logging.getLogger(__name__) -class TerraApi(BalanceMixin): - """ - Terra Money, implemented by multiple api providers. - Explorer: https://finder.terra.money - """ - coin = COIN_TERRA - - def __init__(self): - self.mantle = TerraMantleApi() - self.fcd = TerraFcdApi() - - def fetch_balances(self, address: str) -> FetchResult: - status, balances, balance_errors = self.fcd.fetch_native_balances(address) - _, staking_balances, staking_errors = self.fcd.fetch_staking_balances(address) - cw20_balances = self.mantle.fetch_cw20_balances(address) - - return FetchResult( - status_code=status, - data=dict( - balances=balances, - raw_staking_balances=staking_balances, - raw_cw20_balances=cw20_balances, - ), - errors=list(concatv(balance_errors, staking_errors)), - ) - - def parse_balances(self, fetch_result: FetchResult) -> ParseResult: - native_balances = self.fcd.parse_native_balances( - fetch_result.data.get('balances') - ) - staking_balances = self.fcd.parse_staking_balances( - fetch_result.data.get('raw_staking_balances') - ) - cw20_balances = self.mantle.parse_cw20_balances( - fetch_result.data.get('raw_cw20_balances') - ) - return ParseResult( - data=list(concatv(native_balances, staking_balances, cw20_balances)) - ) - - -class TerraFcdApi(BlockchainApi): +class TerraApi(CosmosApiBase): """ - Terra Money FCD - API docs: https://fcd.terra.dev/swagger + Terra Classic (LUNC) via standard Cosmos LCD endpoints. + Explorer: https://finder.terra.money """ coin = COIN_TERRA + TOKENS_MAP_BLOCKCHAIN_KEY = 'terra' api_options = ApiOptions( blockchain=Blockchain.TERRA, - base_url='https://fcd.terra.dev/', - rate_limit=1, + base_url='https://terra-classic-fcd.publicnode.com/', + rate_limit=CosmosApiBase.API_BASE_RATE_LIMIT, ) supported_requests = { - 'get_native_balances': '/v1/bank/{address}', + **CosmosApiBase.supported_requests, 'get_ibc_denom_trace': '/ibc/apps/transfer/v1/denom_traces/{hash}', - 'get_staking_data': '/v1/staking/{address}', } - def fetch_native_balances(self, address: str) -> FetchResult: - return self.get_data('get_native_balances', address=address) + def create_default_coin(self, denom: str) -> Coin: + if denom.startswith('ibc/'): + return self._resolve_ibc_denom(denom) - def parse_native_balances(self, response: dict) -> List[BalanceItem]: - balances = [] - for b in response['balance']: - if int(b['available']) == 0: - continue - - coin = ( - self._get_terra_token_by_denom(b['denom']) - if b['denom'].startswith('u') - else self._get_ibc_token_by_denom(b['denom']) - ) - - balances.append( - BalanceItem.from_api(balance_raw=b['available'], coin=coin, raw=b) - ) - - return balances - - def get_native_balances(self, address: str) -> List[BalanceItem]: - _, response, _ = self.fetch_native_balances(address) - return self.parse_native_balances(response) - - def fetch_staking_balances(self, address: str) -> FetchResult: - return self.get_data('get_staking_data', address=address) - - def get_staking_balances(self, address: str) -> List[BalanceItem]: - _, response, _ = self.fetch_staking_balances(address) - return self.parse_staking_balances(response) - - def parse_staking_balances(self, response: dict) -> List[BalanceItem]: - total_staked = 0 - balances = [] - - # active stake - if int(response['delegationTotal']) > 0: - total_staked += int(response['delegationTotal']) - # undelegated stake - if response['undelegations']: - total_staked += sum(int(u['amount']) for u in response['undelegations']) - # total stake - sum of staked and undelegated - # add redelegations? - if total_staked: - balances.append( - BalanceItem.from_api( - balance_raw=total_staked, - coin=self.coin, - asset_type=AssetType.STAKED, - raw=response, - ) - ) - - # staking rewards - for d in response['rewards']['denoms']: - balances.append( - BalanceItem.from_api( - balance_raw=d['amount'], - coin=self._get_terra_token_by_denom(d['denom']), - asset_type=AssetType.CLAIMABLE, - raw=d, - ) - ) - - return balances - - # It's possible to get cw20 balances, but it needs to be done one by one. - # Use .terra_mantle.py for that - # def get_cw20_balances(self): - - @staticmethod - def _get_terra_token_by_denom(denom: str) -> Coin: - if denom == 'uluna': - return COIN_TERRA - else: - symbol = f'{denom[1:3].upper()}TC' - return Coin.from_api( - symbol=symbol, - name=symbol, - decimals=6, - blockchain=Blockchain.TERRA, - address=denom, - standards=['terra-native'], - ) + return super().create_default_coin(denom) - @lru_cache(maxsize=8) - def _get_ibc_token_by_denom(self, denom: str) -> Coin: + @lru_cache(maxsize=64) + def _resolve_ibc_denom(self, denom: str) -> Coin: hash_ = denom.split('/')[1] try: response = self.get('get_ibc_denom_trace', hash=hash_) - except ApiException: - # add log - symbol = None - else: - denom = response['denom_trace']['base_denom'] - symbol = denom[1:].upper() + base_denom = response['denom_trace']['base_denom'] + symbol = ( + base_denom.lstrip('ux').upper() + if base_denom.startswith(('u', 'x')) + else base_denom.upper() + ) + except (ApiException, KeyError) as e: + logger.warning(f'Failed to resolve IBC denom {denom}: {e}') + return super().create_default_coin(denom) return Coin.from_api( symbol=symbol, name=symbol, - decimals=6, - blockchain=Blockchain.TERRA, - address=hash_, + decimals=self.coin.decimals, + blockchain=self.api_options.blockchain, + address=denom, standards=['ibc'], ) - - -class TerraMantleApi(BlockchainApi): - """ - Terra Money Subgraph API - API docs: https://mantle.terra.dev - """ - - coin = COIN_TERRA - api_options = ApiOptions( - blockchain=Blockchain.TERRA, - base_url='https://mantle.terra.dev', - rate_limit=1, - ) - - # API uses post requests - supported_requests = {} - _post_requests = { - 'wasm_contract_address_store': """ - WasmContractsContractAddressStore( - ContractAddress: "$CONTRACT_ADDRESS", - QueryMsg: "$QUERY_MSG" - ){ - Result - } - """ - } - - _tokens_map: Optional[Dict[str, Dict]] = None - - @property - def tokens_map(self) -> Dict[str, Dict]: - if self._tokens_map is None: - response = self._session.get('https://assets.terra.money/cw20/tokens.json') - token_list = response.json() - self._tokens_map = token_list['classic'] - - return self._tokens_map - - def fetch_cw20_balances(self, address) -> dict: - return self._get_raw_balances(address) - - def get_cw20_balances(self, address: str): - raw_balances = self._get_raw_balances(address) - return self.parse_cw20_balances(raw_balances) - - def parse_cw20_balances(self, raw_balances): - balances = [] - for contract, result_raw in raw_balances['data'].items(): - if not result_raw: - # should be error in response, TODO add log - continue - - data_raw = json.loads(result_raw['Result']) - balance_raw = data_raw['balance'] - if int(balance_raw) == 0: - continue - - balances.append( - BalanceItem.from_api( - balance_raw=balance_raw, - coin=self._get_token_data(contract), - raw=result_raw, - ) - ) - - return balances - - def _get_token_data(self, address: str) -> Coin: - raw_token = self.tokens_map[address] - return Coin( - symbol=raw_token['symbol'], - name=raw_token['name'] if raw_token.get('name') else raw_token['symbol'], - decimals=6, - blockchain=Blockchain.TERRA, - address=address, - standards=['CW20'], - protocol_id=raw_token.get('protocol'), - info=CoinInfo.from_api(logo_url=raw_token.get('icon')), - ) - - def _get_raw_balances(self, address: str) -> Dict: - cw20_contracts = list(self.tokens_map.keys()) - message = '{\\"balance\\": {\\"address\\": \\"$ADDR\\"}}'.replace( - '$ADDR', address - ) - - key_queries = [ - self._create_key_query( - key=contract, - query=self._build_query( - method='wasm_contract_address_store', - params={'$CONTRACT_ADDRESS': contract, '$QUERY_MSG': message}, - ), - ) - for contract in cw20_contracts - ] - query = self._concat_key_queries(key_queries) - return self.post(json={'query': query}) - - def _build_query(self, method: str, params: Optional[Dict[str, str]] = None) -> str: - query = self._post_requests.get(method) - if params: - for k, v in params.items(): - query = query.replace(k, v) - return query - - @staticmethod - def _create_key_query(key: str, query: str) -> str: - return f'{key}: {query}' - - @staticmethod - def _concat_key_queries(key_queries: Sequence[str]) -> str: - return '{' + ',\n'.join(key_queries) + '}' - - def _opt_raise_on_other_error(self, response: Response) -> None: - json_response = response.json() - if json_response.get('errors') is None: - return - - # pick first message - err = json_response['errors'][0] - - if 'addr_canonicalize' in err['message']: - raise InvalidAddressException(f'Invalid address format.') diff --git a/blockapi/v2/coins.py b/blockapi/v2/coins.py index ee14027..3ac9e19 100644 --- a/blockapi/v2/coins.py +++ b/blockapi/v2/coins.py @@ -34,8 +34,8 @@ COIN_TERRA = Coin( - symbol='LUNA', - name='Terra', + symbol='LUNC', + name='Terra Classic', decimals=6, blockchain=Blockchain.TERRA, address='uluna',