crypto-mcp/tests.py

"""
Tests for indicators and cache logic.
Run with: python -m pytest tests.py -v
"""

import sys, os
sys.path.insert(0, os.path.dirname(__file__))

import pytest
import time
from fastapi.testclient import TestClient
from server_fastmcp import app
from indicators import rsi, ema, macd, sma, compute_indicator
from cache import TTLCache
from errors import InsufficientDataError, UnsupportedIndicatorError


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

def make_candles(closes: list[float]) -> list:
    """Create minimal candles from a close price list."""
    return [[i * 3600, c * 0.99, c * 1.01, c * 0.98, c, 1000.0] for i, c in enumerate(closes)]


# Realistic-ish BTC close prices (50 candles)
SAMPLE_CLOSES = [
    67000, 67200, 66800, 67500, 68000, 67800, 68200, 68500, 68300, 68700,
    69000, 68900, 69200, 69500, 69100, 68800, 68600, 68400, 68200, 68000,
    67800, 67600, 67400, 67200, 67000, 66800, 67200, 67500, 67800, 68100,
    68400, 68700, 69000, 69300, 69600, 69900, 70200, 70500, 70800, 71000,
    71200, 71000, 70800, 70600, 70400, 70200, 70000, 69800, 69600, 69400,
]
CANDLES = make_candles(SAMPLE_CLOSES)
client = TestClient(app)


# ---------------------------------------------------------------------------
# EMA
# ---------------------------------------------------------------------------

class TestEMA:
    def test_basic(self):
        val = ema(CANDLES, period=20)
        assert isinstance(val, float)
        assert 60000 < val < 80000

    def test_period_1(self):
        val = ema(CANDLES, period=1)
        # EMA(1) == last close
        assert val == pytest.approx(SAMPLE_CLOSES[-1], rel=1e-4)

    def test_insufficient_data(self):
        with pytest.raises(InsufficientDataError):
            ema(make_candles([100.0] * 5), period=10)

    def test_ema_50_needs_more_than_20(self):
        val_20 = ema(CANDLES, period=20)
        val_50 = ema(CANDLES, period=50)
        # With uptrend ending, EMA50 should be lower than EMA20
        assert isinstance(val_50, float)


# ---------------------------------------------------------------------------
# RSI
# ---------------------------------------------------------------------------

class TestRSI:
    def test_range(self):
        val = rsi(CANDLES, period=14)
        assert 0 <= val <= 100

    def test_all_up(self):
        # Consistently rising prices → RSI near 100
        candles = make_candles([float(i) for i in range(1, 50)])
        val = rsi(candles, period=14)
        assert val > 80

    def test_all_down(self):
        # Consistently falling prices → RSI near 0
        candles = make_candles([float(50 - i) for i in range(50)])
        val = rsi(candles, period=14)
        assert val < 20

    def test_insufficient_data(self):
        with pytest.raises(InsufficientDataError):
            rsi(make_candles([100.0] * 5), period=14)


# ---------------------------------------------------------------------------
# MACD
# ---------------------------------------------------------------------------

class TestMACD:
    def test_structure(self):
        result = macd(CANDLES)
        assert "macd" in result
        assert "signal" in result
        assert "histogram" in result

    def test_histogram_math(self):
        result = macd(CANDLES)
        assert result["histogram"] == pytest.approx(
            result["macd"] - result["signal"], rel=1e-5
        )

    def test_insufficient_data(self):
        with pytest.raises(InsufficientDataError):
            macd(make_candles([100.0] * 10))


# ---------------------------------------------------------------------------
# SMA
# ---------------------------------------------------------------------------

class TestSMA:
    def test_basic(self):
        # SMA of constant values == that value
        candles = make_candles([500.0] * 30)
        val = sma(candles, period=20)
        assert val == pytest.approx(500.0, rel=1e-6)

    def test_insufficient_data(self):
        with pytest.raises(InsufficientDataError):
            sma(make_candles([100.0] * 5), period=20)


# ---------------------------------------------------------------------------
# Compute Indicator Dispatcher
# ---------------------------------------------------------------------------

class TestDispatcher:
    def test_rsi(self):
        result = compute_indicator(CANDLES, "rsi", {"period": 14})
        assert result["indicator"] == "rsi"
        assert 0 <= result["value"] <= 100

    def test_ema(self):
        result = compute_indicator(CANDLES, "ema", {"period": 20})
        assert result["indicator"] == "ema"

    def test_macd(self):
        result = compute_indicator(CANDLES, "macd", {})
        assert result["indicator"] == "macd"
        assert "histogram" in result["value"]

    def test_unsupported(self):
        with pytest.raises(UnsupportedIndicatorError):
            compute_indicator(CANDLES, "bollinger", {})

    def test_case_insensitive(self):
        result = compute_indicator(CANDLES, "RSI", {"period": 14})
        assert result["indicator"] == "rsi"


# ---------------------------------------------------------------------------
# TTL Cache
# ---------------------------------------------------------------------------

class TestTTLCache:
    def test_set_get(self):
        cache = TTLCache()
        cache.set("k", {"x": 1}, ttl=60)
        assert cache.get("k") == {"x": 1}

    def test_expiry(self):
        cache = TTLCache()
        cache.set("k", "val", ttl=1)
        time.sleep(1.1)
        assert cache.get("k") is None

    def test_missing_key(self):
        cache = TTLCache()
        assert cache.get("nonexistent") is None

    def test_overwrite(self):
        cache = TTLCache()
        cache.set("k", "a", ttl=60)
        cache.set("k", "b", ttl=60)
        assert cache.get("k") == "b"

    def test_delete(self):
        cache = TTLCache()
        cache.set("k", "v", ttl=60)
        cache.delete("k")
        assert cache.get("k") is None

    def test_stats(self):
        cache = TTLCache()
        cache.set("a", 1, ttl=60)
        cache.set("b", 2, ttl=60)
        stats = cache.stats()
        assert stats["alive_keys"] == 2


# ---------------------------------------------------------------------------
# HTTP / MCP surface
# ---------------------------------------------------------------------------

class TestHTTPMCP:
    def test_root_returns_tools(self):
        resp = client.get("/")
        assert resp.status_code == 200
        body = resp.json()
        assert body["status"] == "ok"
        assert "tools" in body

    def test_health(self):
        resp = client.get("/health")
        assert resp.status_code == 200
        body = resp.json()
        assert body["status"] == "ok"
        assert "cache" in body

    def test_mcp_mount_path(self):
        resp = client.get("/")
        assert resp.status_code == 200
        assert resp.json().get("mount") == "/mcp"


if __name__ == "__main__":
    pytest.main([__file__, "-v"])