feat: add support for Litecoin balance fetching via SoChain API

blockapi/test/v2/api/data/chainso_balance_response.json

@@ -0,0 +1,7 @@
+{
+  "status": "success",
+  "data": {
+    "confirmed": "262.99968329",
+    "unconfirmed": "0.00000000"
+  }
+}

blockapi/test/v2/api/test_chainso.py

@@ -0,0 +1,76 @@
+from decimal import Decimal
+from unittest.mock import Mock
+
+import pytest
+from requests.exceptions import ReadTimeout
+
+from blockapi.test.v2.api.conftest import read_file
+from blockapi.test.v2.test_data import ltc_test_address
+from blockapi.v2.api import ChainSoLitecoinApi
+from blockapi.v2.base import ApiException
+from blockapi.v2.models import FetchResult
+
+URL = f'https://sochain.com/api/v3/balance/LTC/{ltc_test_address}'
+
+
+def test_fetch_balances(requests_mock):
+    requests_mock.get(URL, text=read_file('data/chainso_balance_response.json'))
+    api = ChainSoLitecoinApi()
+    balances = api.get_balance(ltc_test_address)
+    assert len(balances) == 1
+    assert balances[0].balance == Decimal('262.99968329')
+    assert balances[0].balance_raw == Decimal('26299968329')
+    assert balances[0].coin == api.coin
+    assert requests_mock.last_request.timeout == (3.0, 10.0)
+
+
+@pytest.mark.parametrize('confirmed', ['0.00000000', '0.00000001'])
+def test_confirmed_balance_only(requests_mock, confirmed):
+    requests_mock.get(
+        URL,
+        json={
+            'status': 'success',
+            'data': {'confirmed': confirmed, 'unconfirmed': '12.00000000'},
+        },
+    )
+    balances = ChainSoLitecoinApi().get_balance(ltc_test_address)
+    if Decimal(confirmed):
+        assert balances[0].balance == Decimal(confirmed)
+        assert balances[0].balance_raw == 1
+    else:
+        assert balances == []
+
+
+@pytest.mark.parametrize(
+    'response',
+    [
+        None,
+        {},
+        {'status': 'fail', 'data': {'confirmed': '0'}},
+        {'status': 'success', 'data': {}},
+        *[
+            {'status': 'success', 'data': {'confirmed': value}}
+            for value in [None, 'bad', 'NaN', 'Infinity', '-1', '0.000000001']
+        ],
+    ],
+)
+def test_invalid_response_is_not_zero(response):
+    with pytest.raises(ApiException):
+        ChainSoLitecoinApi().parse_balances(FetchResult(data=response))
+
+
+@pytest.mark.parametrize('status', [429, 502])
+def test_http_failure_does_not_retry_or_return_zero(requests_mock, status):
+    requests_mock.get(URL, status_code=status, json={'status': 'fail'})
+    sleep_provider = Mock()
+    api = ChainSoLitecoinApi(sleep_provider=sleep_provider)
+    with pytest.raises(ApiException):
+        api.get_balance(ltc_test_address)
+    assert requests_mock.call_count == 1
+    sleep_provider.sleep.assert_not_called()
+
+
+def test_read_timeout_is_not_zero(requests_mock):
+    requests_mock.get(URL, exc=ReadTimeout('stalled'))
+    with pytest.raises(ApiException):
+        ChainSoLitecoinApi().get_balance(ltc_test_address)

blockapi/v2/api/__init__.py

@@ -7,6 +7,7 @@
     BlockchairDogecoinApi,
     BlockchairLitecoinApi,
 )
+from blockapi.v2.api.chainso import ChainSoLitecoinApi
 from blockapi.v2.api.debank import DebankApi, DebankApp, DebankPrediction
 from blockapi.v2.api.ethplorer import EthplorerApi
 from blockapi.v2.api.haskoin import HaskoinApi

blockapi/v2/api/chainso.py

@@ -0,0 +1,69 @@
+from decimal import Decimal, InvalidOperation
+
+from blockapi.v2.base import ApiException, BalanceMixin, BlockchainApi
+from blockapi.v2.coins import COIN_LTC
+from blockapi.v2.models import (
+    ApiOptions,
+    AssetType,
+    BalanceItem,
+    Blockchain,
+    FetchResult,
+    ParseResult,
+)
+
+
+class ChainSoLitecoinApi(BlockchainApi, BalanceMixin):
+    """Litecoin confirmed balances: https://sochain.com/docs."""
+
+    coin = COIN_LTC
+    api_options = ApiOptions(
+        blockchain=Blockchain.LITECOIN,
+        base_url='https://sochain.com',
+        rate_limit=1.0,
+    )
+    supported_requests = {
+        'get_balance': '/api/v3/balance/LTC/{address}',
+    }
+    http_timeout = (3.0, 10.0)
+    max_rate_limit_retries = 1
+
+    def fetch_balances(self, address: str) -> FetchResult:
+        return self.get_data('get_balance', address=address)
+
+    def _get_response(self, request_method, headers, params, req_args):
+        url = self._build_request_url(request_method, **req_args)
+        return self._session.get(
+            url, headers=headers, params=params, timeout=self.http_timeout
+        )
+
+    def parse_balances(self, fetch_result: FetchResult) -> ParseResult:
+        response = fetch_result.data
+        if not isinstance(response, dict) or response.get('status') != 'success':
+            raise ApiException('SoChain balance request was not successful')
+
+        try:
+            # SoChain returns LTC, whereas balance_raw uses litoshis.
+            balance_raw = (
+                Decimal(response['data']['confirmed']) * 10**self.coin.decimals
+            )
+        except (KeyError, TypeError, ValueError, InvalidOperation) as exc:
+            raise ApiException('Invalid SoChain confirmed balance') from exc
+        if (
+            not balance_raw.is_finite()
+            or balance_raw < 0
+            or balance_raw != balance_raw.to_integral_value()
+        ):
+            raise ApiException('Invalid SoChain confirmed balance')
+        if not balance_raw:
+            return ParseResult()
+
+        return ParseResult(
+            data=[
+                BalanceItem.from_api(
+                    balance_raw=balance_raw,
+                    coin=self.coin,
+                    asset_type=AssetType.AVAILABLE,
+                    raw=response,
+                )
+            ]
+        )