test: tolerate transient gateway 504 on live premium lane

tests/conftest.py

@@ -5,9 +5,17 @@
 """
 
 import os
+import time
+
 import pytest
 
 from datamaxi import Datamaxi, Telegram, Naver
+from datamaxi.error import ServerError
+
+# Transient gateway statuses: the prod edge returns these when an upstream is
+# briefly slow/unavailable (e.g. the premium endpoint 504-ing under load). They
+# are infra flakiness, not SDK bugs, so the live lane retries then skips.
+_TRANSIENT_STATUS = (502, 503, 504)
 
 # Live-test credentials / target. Resolved once so the keyed lanes
 # (test_call.py, test_integration.py) share a single source of truth; both
@@ -19,6 +27,29 @@
 TIMEOUT = int(os.getenv("DATAMAXI_TIMEOUT") or "30")
 
 
+def live_call(fn, retries=3, backoff=1.0):
+    """Invoke a live-endpoint call, tolerating transient gateway errors.
+
+    Retries ``fn`` on a transient 5xx (``_TRANSIENT_STATUS``) with linear
+    backoff. If every attempt still hits a transient status the call is
+    ``pytest.skip``-ped rather than failed — the non-blocking live lane must
+    not go red on prod infra flakiness. Any other error (real 5xx, 4xx,
+    assertion) propagates unchanged.
+    """
+    for attempt in range(retries):
+        try:
+            return fn()
+        except ServerError as e:
+            if e.status_code not in _TRANSIENT_STATUS:
+                raise
+            if attempt == retries - 1:
+                pytest.skip(
+                    "transient %s from live endpoint after %d attempts"
+                    % (e.status_code, retries)
+                )
+            time.sleep(backoff * (attempt + 1))
+
+
 @pytest.fixture(scope="module")
 def datamaxi():
     """Create Datamaxi client for live tests."""

tests/test_integration.py

@@ -21,7 +21,7 @@
 from datetime import datetime, timedelta
 
 from datamaxi.error import ParameterRequiredError
-from tests.conftest import API_KEY
+from tests.conftest import API_KEY, live_call
 
 # Live integration lane: exercises prod endpoints with every supported param.
 # Skipped without a key and deselected from the keyless CI lane via the
@@ -587,100 +587,106 @@ def test_exchanges(self, datamaxi):
 
     def test_premium_basic(self, datamaxi):
         """Test basic premium data fetch."""
-        result = datamaxi.premium()
+        result = live_call(lambda: datamaxi.premium())
         assert isinstance(result, pd.DataFrame)
         assert len(result) > 0
 
     def test_premium_with_pagination(self, datamaxi):
         """Test premium data with pagination."""
-        result = datamaxi.premium(page=1, limit=10)
+        result = live_call(lambda: datamaxi.premium(page=1, limit=10))
         assert isinstance(result, pd.DataFrame)
         assert len(result) <= 10
 
     def test_premium_source_exchange(self, datamaxi):
         """Test premium data filtered by source_exchange."""
-        result = datamaxi.premium(source_exchange="binance", limit=10)
+        result = live_call(
+            lambda: datamaxi.premium(source_exchange="binance", limit=10)
+        )
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_target_exchange(self, datamaxi):
         """Test premium data filtered by target_exchange."""
-        result = datamaxi.premium(target_exchange="upbit", limit=10)
+        result = live_call(lambda: datamaxi.premium(target_exchange="upbit", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_with_asset(self, datamaxi):
         """Test premium data filtered by asset."""
-        result = datamaxi.premium(asset="BTC", limit=10)
+        result = live_call(lambda: datamaxi.premium(asset="BTC", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_source_quote(self, datamaxi):
         """Test premium data filtered by source_quote."""
-        result = datamaxi.premium(source_quote="USDT", limit=10)
+        result = live_call(lambda: datamaxi.premium(source_quote="USDT", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_target_quote(self, datamaxi):
         """Test premium data filtered by target_quote."""
-        result = datamaxi.premium(target_quote="KRW", limit=10)
+        result = live_call(lambda: datamaxi.premium(target_quote="KRW", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_source_market(self, datamaxi):
         """Test premium data filtered by source_market."""
-        result = datamaxi.premium(source_market="spot", limit=10)
+        result = live_call(lambda: datamaxi.premium(source_market="spot", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_target_market(self, datamaxi):
         """Test premium data filtered by target_market."""
-        result = datamaxi.premium(target_market="spot", limit=10)
+        result = live_call(lambda: datamaxi.premium(target_market="spot", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_both_markets(self, datamaxi):
         """Test premium data filtered by both markets."""
-        result = datamaxi.premium(source_market="spot", target_market="spot", limit=10)
+        result = live_call(
+            lambda: datamaxi.premium(
+                source_market="spot", target_market="spot", limit=10
+            )
+        )
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_sort_asc(self, datamaxi):
         """Test premium data with sort=asc."""
-        result = datamaxi.premium(sort="asc", key="pdp", limit=10)
+        result = live_call(lambda: datamaxi.premium(sort="asc", key="pdp", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_sort_desc(self, datamaxi):
         """Test premium data with sort=desc."""
-        result = datamaxi.premium(sort="desc", key="pdp", limit=10)
+        result = live_call(lambda: datamaxi.premium(sort="desc", key="pdp", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_volume_filters(self, datamaxi):
         """Test premium data with volume filters."""
-        result = datamaxi.premium(min_sv="100000", limit=10)
+        result = live_call(lambda: datamaxi.premium(min_sv="100000", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_only_transferable(self, datamaxi):
         """Test premium data with only_transferable=True."""
-        result = datamaxi.premium(only_transferable=True, limit=10)
+        result = live_call(lambda: datamaxi.premium(only_transferable=True, limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_with_currency(self, datamaxi):
         """Test premium data with currency parameter."""
-        result = datamaxi.premium(currency="KRW", limit=10)
+        result = live_call(lambda: datamaxi.premium(currency="KRW", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_with_conversion_base(self, datamaxi):
         """Test premium data with conversion_base parameter (USD or USDT)."""
-        result = datamaxi.premium(conversion_base="USD", limit=10)
+        result = live_call(lambda: datamaxi.premium(conversion_base="USD", limit=10))
         assert isinstance(result, pd.DataFrame)
         assert len(result) > 0
 
     def test_premium_token_include(self, datamaxi):
         """Test premium data with token_include filter."""
-        result = datamaxi.premium(token_include="bitcoin", limit=10)
+        result = live_call(lambda: datamaxi.premium(token_include="bitcoin", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_token_exclude(self, datamaxi):
         """Test premium data with token_exclude filter."""
-        result = datamaxi.premium(token_exclude="SHIB", limit=10)
+        result = live_call(lambda: datamaxi.premium(token_exclude="SHIB", limit=10))
         assert isinstance(result, pd.DataFrame)
 
     def test_premium_pandas_false(self, datamaxi):
         """Test premium data with pandas=False."""
-        result = datamaxi.premium(pandas=False, limit=10)
+        result = live_call(lambda: datamaxi.premium(pandas=False, limit=10))
         assert isinstance(result, dict)
         assert "data" in result