diff --git a/blockapi/test/v2/api/debank/test_debank.py b/blockapi/test/v2/api/debank/test_debank.py index 0e2dbdd..781900c 100644 --- a/blockapi/test/v2/api/debank/test_debank.py +++ b/blockapi/test/v2/api/debank/test_debank.py @@ -2,6 +2,7 @@ from blockapi.utils.address import make_checksum_address from blockapi.v2.api import DebankApi +from blockapi.v2.models import FetchResult def test_build_balance_request_url(debank_api): @@ -57,6 +58,79 @@ def test_build_portfolio_request_url(debank_api): ) +def test_build_token_list_for_chain_request_url(debank_api): + url = debank_api._build_request_url( + 'get_token_list_for_chain', + address='0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca', + chain_id='eth', + is_all=True, + ) + assert ( + url + == 'https://pro-openapi.debank.com/v1/user/token_list?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&chain_id=eth&is_all=True' + ) + + +def test_build_protocol_for_address_request_url(debank_api): + url = debank_api._build_request_url( + 'get_protocol_for_address', + address='0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca', + protocol_id='yflink', + ) + assert ( + url + == 'https://pro-openapi.debank.com/v1/user/protocol?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&protocol_id=yflink' + ) + + +def test_parse_pool_for_address_wraps_single_object( + debank_api, yflink_protocol_response_raw, portfolio_response, requests_mock +): + # /v1/user/protocol returns a single protocol object (dict, not list). + # parse_pool_for_address must wrap it so DebankPortfolioParser can iterate. + requests_mock.get( + 'https://pro-openapi.debank.com/v1/protocol/all_list', + text=yflink_protocol_response_raw, + ) + requests_mock.get( + 'https://pro-openapi.debank.com/v1/user/protocol?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&protocol_id=avax_traderjoexyz_lending', + json=portfolio_response, + ) + debank_api._protocol_cache.invalidate() + fetched = debank_api.fetch_protocol_for_address( + '0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca', + 'avax_traderjoexyz_lending', + ) + parsed = debank_api.parse_pool_for_address(fetched) + assert parsed.errors is None + assert len(parsed.data) > 0 + assert ( + parsed.data[0].pool_info.pool_id == '0xdc13687554205e5b89ac783db14bb5bba4a1edac' + ) + + +def test_parse_pool_for_address_handles_empty_response(debank_api, protocol_cache): + # When Debank returns null/None for an address with no positions in this protocol. + protocol_cache.update({}) + parsed = debank_api.parse_pool_for_address(FetchResult(status_code=200, data=None)) + assert parsed.errors is None + assert parsed.data == [] + + +def test_fetch_token_list_for_chain_uses_api_key( + debank_api, protocol_cache, requests_mock +): + req = requests_mock.get( + 'https://pro-openapi.debank.com/v1/user/token_list?id=0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca&chain_id=eth&is_all=True', + text='[]', + ) + protocol_cache.update({}) + debank_api.fetch_token_list_for_chain( + '0xca8fa8f0b631ecdb18cda619c4fc9d197c8affca', 'eth' + ) + assert req.last_request.headers.get('AccessKey') == 'dummy-key' + + def test_error_response_returns_empty_balances( debank_api, protocol_cache, error_response_raw, requests_mock ): diff --git a/blockapi/v2/api/debank.py b/blockapi/v2/api/debank.py index 1ea68fb..63a8e1d 100644 --- a/blockapi/v2/api/debank.py +++ b/blockapi/v2/api/debank.py @@ -735,6 +735,12 @@ class DebankApi(CustomizableBlockchainApi, BalanceMixin, IPortfolio): 'get_protocols': '/v1/protocol/all_list', 'usage': '/v1/account/units', 'get_complex_app_list': '/v1/user/complex_app_list?id={address}', + 'get_token_list_for_chain': ( + '/v1/user/token_list?id={address}&chain_id={chain_id}&is_all={is_all}' + ), + 'get_protocol_for_address': ( + '/v1/user/protocol?id={address}&protocol_id={protocol_id}' + ), } default_protocol_cache = DebankProtocolCache() @@ -783,6 +789,23 @@ def fetch_debank_apps(self, address: str) -> FetchResult: address=address, ) + def fetch_token_list_for_chain(self, address: str, chain_id: str) -> FetchResult: + return self.get_data( + 'get_token_list_for_chain', + headers=self._headers, + address=address, + chain_id=chain_id, + is_all=self._is_all, + ) + + def fetch_protocol_for_address(self, address: str, protocol_id: str) -> FetchResult: + return self.get_data( + 'get_protocol_for_address', + headers=self._headers, + address=address, + protocol_id=protocol_id, + ) + def fetch_protocols(self) -> FetchResult: return self.get_data( 'get_protocols', @@ -809,6 +832,17 @@ def parse_pools(self, fetch_result: FetchResult) -> ParseResult: self._maybe_update_protocols() return ParseResult(data=self._portfolio_parser.parse(fetch_result.data)) + def parse_pool_for_address(self, fetch_result: FetchResult) -> ParseResult: + if error := self._get_error(fetch_result.data): + return ParseResult(errors=[error]) + + self._maybe_update_protocols() + data = fetch_result.data + # /v1/user/protocol returns a single protocol object; wrap in a list + # so the portfolio parser (which iterates) handles it uniformly. + wrapped = [data] if isinstance(data, dict) else (data or []) + return ParseResult(data=self._portfolio_parser.parse(wrapped)) + def get_protocols(self) -> Dict[str, Protocol]: response = self.get('get_protocols', headers=self._headers) if self._has_error(response):