diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index c6b77de..4cd20d4 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -36,10 +36,10 @@ jobs: run: | flake8 . --statistics # Offline mocked tests run keyless (no DATAMAXI_API_KEY, no network). - # `-k "not integration"` excludes the live test_integration.py; the - # keyed test_call.py tests skip cleanly when no key is set. + # `-m "not integration"` deselects the live integration lane; the keyed + # smoke lane (test_call.py) skips cleanly when no key is set. - name: Run offline mocked tests - run: python -m pytest tests/ -k "not integration" -q - # The live test_integration.py hits prod endpoints and depends on prod - # data availability, so it is not run in CI. Run it locally with + run: python -m pytest tests/ -m "not integration" -q + # The live integration + smoke lanes hit prod endpoints and depend on prod + # data availability, so they are not run in CI. Run them locally with # `DATAMAXI_API_KEY=... python -m pytest tests/`. diff --git a/README.md b/README.md index beac37c..79de814 100644 --- a/README.md +++ b/README.md @@ -432,8 +432,8 @@ uv pip install -r requirements/requirements-dev.txt # Install test dependencies (skip if you already ran the dev install above) uv pip install -r requirements/requirements-test.txt -# Run unit tests (no API key required) -uv run pytest tests/test_api.py -v +# Run keyless tests (no API key required) — this is the lane CI runs on every push +uv run pytest tests/ -m "not integration" -v # Run integration tests (requires API key) export DATAMAXI_API_KEY="your_api_key" @@ -446,6 +446,7 @@ uv run pytest tests/test_integration.py -m "premium" -v uv run pytest tests/test_integration.py -m "forex" -v uv run pytest tests/test_integration.py -m "telegram" -v uv run pytest tests/test_integration.py -m "naver" -v +uv run pytest tests/test_integration.py -m "types" -v # Run all tests uv run pytest tests/ -v diff --git a/setup.cfg b/setup.cfg index 14b906e..caea9d0 100644 --- a/setup.cfg +++ b/setup.cfg @@ -2,6 +2,7 @@ description_file = README.md [tool:pytest] +addopts = --strict-markers markers = cex: tests for CEX endpoints funding: tests for funding rate endpoints @@ -9,8 +10,9 @@ markers = forex: tests for forex endpoints telegram: tests for telegram endpoints naver: tests for naver endpoints - errors: tests for error handling types: tests for response types + integration: live tests that hit prod endpoints (need DATAMAXI_API_KEY) + smoke: live alive-check lane pinging each endpoint once (need DATAMAXI_API_KEY) [flake8] exclude = diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..d84c976 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,49 @@ +"""Shared test fixtures, constants, and markers. + +Centralizes the API key / base URL resolution, the live client fixtures +(``datamaxi`` / ``telegram`` / ``naver``), and the ``_FLAKY_PROD_DATA_XFAIL`` +marker that ``test_call.py`` and ``test_integration.py`` previously +copy-pasted verbatim. +""" + +import os +import pytest + +from datamaxi import Datamaxi, Telegram, Naver + +# Live-test credentials / target. Resolved once so the keyed lanes +# (test_call.py, test_integration.py) share a single source of truth; both +# honor DATAMAXI_API_KEY (preferred) and the legacy API_KEY. +API_KEY = os.getenv("DATAMAXI_API_KEY") or os.getenv("API_KEY") +BASE_URL = os.getenv("BASE_URL") or "https://api.datamaxiplus.com" + +# Shared xfail marker for tests whose outcome depends on prod-data +# availability — funding-rate / naver-trend state are NATS-warmed in-memory +# caches on the API pods, so any cold-start of the API fleet leaves them +# temporarily empty and the smoke-style tests raise ServerError(500, "no data +# found"). Marked strict=False so they pass cleanly once the cache is hot. +_FLAKY_PROD_DATA_XFAIL = pytest.mark.xfail( + reason=( + "Depends on prod NATS-warmed state; intermittent 500 'no data found' " + "on cold pods. Pre-existing flakiness — unrelated to SDK regen." + ), + strict=False, +) + + +@pytest.fixture(scope="module") +def datamaxi(): + """Create Datamaxi client for live tests.""" + return Datamaxi(api_key=API_KEY, base_url=BASE_URL) + + +@pytest.fixture(scope="module") +def telegram(): + """Create Telegram client for live tests.""" + return Telegram(api_key=API_KEY, base_url=BASE_URL) + + +@pytest.fixture(scope="module") +def naver(): + """Create Naver client for live tests.""" + return Naver(api_key=API_KEY, base_url=BASE_URL) diff --git a/tests/test_call.py b/tests/test_call.py index e3e123f..0c4ab30 100644 --- a/tests/test_call.py +++ b/tests/test_call.py @@ -9,45 +9,21 @@ python -m pytest tests/test_call.py -v """ -import os import pytest -from datamaxi import Datamaxi, Telegram, Naver - -# Skip all tests if no API key is provided -API_KEY = os.getenv("DATAMAXI_API_KEY") or os.getenv("API_KEY") -BASE_URL = os.getenv("BASE_URL") or "https://api.datamaxiplus.com" - -pytestmark = pytest.mark.skipif( - not API_KEY, - reason="API key not provided. Set DATAMAXI_API_KEY environment variable.", -) - -# Shared xfail marker for tests whose outcome depends on prod-data -# availability (funding-rate / naver-trend rely on NATS-warmed -# in-memory caches; cold pods → 500 'no data found' until events -# arrive). strict=False so they pass cleanly when the cache is hot. -_FLAKY_PROD_DATA_XFAIL = pytest.mark.xfail( - reason=( - "Depends on prod NATS-warmed state; intermittent 500 'no data found' " - "on cold pods. Pre-existing flakiness — unrelated to SDK regen." - ), - strict=False, -) - - -@pytest.fixture(scope="module") -def datamaxi(): - return Datamaxi(api_key=API_KEY, base_url=BASE_URL) - -@pytest.fixture(scope="module") -def telegram(): - return Telegram(api_key=API_KEY, base_url=BASE_URL) - - -@pytest.fixture(scope="module") -def naver(): - return Naver(api_key=API_KEY, base_url=BASE_URL) +from tests.conftest import API_KEY, _FLAKY_PROD_DATA_XFAIL + +# Live alive-check / smoke lane: a thin subset of test_integration.py that +# pings each endpoint once. Skipped without a key and deselected from the +# keyless CI lane via the `smoke` marker. Client fixtures and the +# flaky-prod-data marker come from tests/conftest.py. +pytestmark = [ + pytest.mark.smoke, + pytest.mark.skipif( + not API_KEY, + reason="API key not provided. Set DATAMAXI_API_KEY environment variable.", + ), +] def test_cex_candle(datamaxi): diff --git a/tests/test_error.py b/tests/test_error.py new file mode 100644 index 0000000..2abfe77 --- /dev/null +++ b/tests/test_error.py @@ -0,0 +1,41 @@ +"""Unit tests for datamaxi.error exception classes. No API key / network.""" + +from datamaxi.error import ( + Error, + ClientError, + ServerError, + ParameterRequiredError, + AtLeastOneParameterRequiredError, +) + + +def test_client_error_stores_fields(): + err = ClientError(400, "bad request", {"h": "1"}, error_data={"code": -1}) + assert isinstance(err, Error) + assert err.status_code == 400 + assert err.error_message == "bad request" + assert err.header == {"h": "1"} + assert err.error_data == {"code": -1} + + +def test_client_error_error_data_defaults_none(): + err = ClientError(404, "not found", {}) + assert err.error_data is None + + +def test_server_error_stores_fields(): + err = ServerError(500, "no data found") + assert isinstance(err, Error) + assert err.status_code == 500 + assert err.message == "no data found" + + +def test_parameter_required_error_message(): + err = ParameterRequiredError(["exchange", "symbol"]) + assert str(err) == "exchange, symbol is mandatory, but received empty." + + +def test_at_least_one_parameter_required_error_message(): + assert ( + str(AtLeastOneParameterRequiredError()) == "At least one parameter is required." + ) diff --git a/tests/test_integration.py b/tests/test_integration.py index 3da291d..90c4e5d 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -16,56 +16,24 @@ python -m pytest tests/test_integration.py -m "cex" -v """ -import os import pytest import pandas as pd from datetime import datetime, timedelta -from datamaxi import Datamaxi, Telegram, Naver from datamaxi.error import ClientError - -# Skip all tests if no API key is provided -API_KEY = os.getenv("DATAMAXI_API_KEY") or os.getenv("API_KEY") -BASE_URL = os.getenv("BASE_URL") or "https://api.datamaxiplus.com" - -pytestmark = pytest.mark.skipif( - not API_KEY, - reason="API key not provided. Set DATAMAXI_API_KEY environment variable.", -) - -# Shared xfail marker for tests whose outcome depends on prod-data -# availability — funding-rate / naver-trend state are NATS-warmed -# in-memory caches on the API pods, so any cold-start of the API -# fleet leaves them temporarily empty and the smoke-style tests below -# raise ServerError(500, "no data found"). Marked strict=False so -# they pass cleanly if the cache has filled by test time. Replacing -# with a "skip if upstream empty" precheck would be cleaner — left -# for a follow-up since it's orthogonal to SDK regen. -_FLAKY_PROD_DATA_XFAIL = pytest.mark.xfail( - reason=( - "Depends on prod NATS-warmed state; intermittent 500 'no data found' " - "on cold pods. Pre-existing flakiness — unrelated to SDK regen." +from tests.conftest import API_KEY, _FLAKY_PROD_DATA_XFAIL + +# Live integration lane: exercises prod endpoints with every supported param. +# Skipped without a key and deselected from the keyless CI lane via the +# `integration` marker. Client fixtures (datamaxi / telegram / naver) and the +# flaky-prod-data marker come from tests/conftest.py. +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not API_KEY, + reason="API key not provided. Set DATAMAXI_API_KEY environment variable.", ), - strict=False, -) - - -@pytest.fixture(scope="module") -def datamaxi(): - """Create Datamaxi client for tests.""" - return Datamaxi(api_key=API_KEY, base_url=BASE_URL) - - -@pytest.fixture(scope="module") -def telegram(): - """Create Telegram client for tests.""" - return Telegram(api_key=API_KEY, base_url=BASE_URL) - - -@pytest.fixture(scope="module") -def naver(): - """Create Naver client for tests.""" - return Naver(api_key=API_KEY, base_url=BASE_URL) +] # ============================================================================= @@ -170,22 +138,6 @@ def test_candle_with_currency_krw(self, datamaxi): assert isinstance(result, pd.DataFrame) assert len(result) > 0 - def test_candle_with_from_unix(self, datamaxi): - """Test candle data with from_unix timestamp (requires to_unix as well).""" - # Note: API works best with both from and to specified - from_ts = int((datetime.now() - timedelta(days=30)).timestamp()) - to_ts = int((datetime.now() - timedelta(days=1)).timestamp()) - result = datamaxi.cex.candle( - exchange="binance", - symbol="BTC-USDT", - interval="1d", - market="spot", - from_unix=str(from_ts), - to_unix=str(to_ts), - ) - assert isinstance(result, pd.DataFrame) - assert len(result) > 0 - def test_candle_with_from_and_to_unix(self, datamaxi): """Test candle data with both from_unix and to_unix.""" from_ts = int((datetime.now() - timedelta(days=30)).timestamp()) @@ -777,7 +729,7 @@ def test_premium_with_conversion_base(self, datamaxi): def test_premium_token_include(self, datamaxi): """Test premium data with token_include filter.""" - result = datamaxi.premium(token_include="BTC", limit=10) + result = datamaxi.premium(token_include="bitcoin", limit=10) assert isinstance(result, pd.DataFrame) @pytest.mark.xfail( @@ -971,43 +923,6 @@ def test_trend_pandas_false(self, naver): assert isinstance(result, list) -# ============================================================================= -# Error Handling Tests -# ============================================================================= -@pytest.mark.errors -class TestErrorHandling: - """Test error handling across endpoints.""" - - def test_invalid_exchange(self, datamaxi): - """Test that invalid exchange returns error or empty.""" - # This might raise an error or return empty depending on API behavior - try: - result = datamaxi.cex.candle( - exchange="nonexistent_exchange", - symbol="BTC-USDT", - interval="1d", - market="spot", - ) - # If it doesn't raise, result should be empty - assert len(result) == 0 or result is None - except (ValueError, Exception): - # Expected - invalid exchange should raise error - pass - - def test_invalid_symbol(self, datamaxi): - """Test that invalid symbol returns error or empty.""" - try: - result = datamaxi.cex.candle( - exchange="binance", - symbol="INVALID-SYMBOL", - interval="1d", - market="spot", - ) - assert len(result) == 0 or result is None - except (ValueError, Exception): - pass - - # ============================================================================= # Response Type Tests # =============================================================================