tests/test_optimizations.py

"""
Tests for performance optimization fixes:
  #1  datetime.utcnow() deprecation fix
  #3  Cache maxsize (TTLCache)
  #4  History endpoint shares indicator cache
  #5  APP_VERSION constant
  #17 stock_history async (asyncio.to_thread)
  #27 yfinance exchange mapping via twstock.codes
  #30 Dockerfile dual workers
"""

import time
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock

import pytest


# ── #1 datetime.utcnow() deprecation ────────────────────────────────────────

class TestDatetimeUTCNow:
    """Verify no usage of deprecated datetime.utcnow()."""

    def test_main_uses_timezone_utc(self):
        """main.py health endpoint should use datetime.now(timezone.utc)."""
        src = Path(__file__).parent.parent / "main.py"
        code = src.read_text(encoding="utf-8")
        assert "datetime.utcnow()" not in code, "main.py still uses deprecated utcnow()"
        assert "datetime.now(timezone.utc)" in code

    def test_predictor_service_uses_timezone_utc(self):
        """predictor_service.py should use datetime.now(timezone.utc)."""
        src = Path(__file__).parent.parent / "services" / "predictor_service.py"
        code = src.read_text(encoding="utf-8")
        assert "datetime.utcnow()" not in code, "predictor_service still uses utcnow()"
        assert "datetime.now(timezone.utc)" in code


# ── #3 Cache maxsize (TTLCache) ──────────────────────────────────────────────

class TestCacheMaxsize:
    """Verify all four caches use TTLCache with maxsize."""

    def test_stock_router_info_cache_is_ttlcache(self):
        from routers.stock import _info_cache
        from cachetools import TTLCache
        assert isinstance(_info_cache, TTLCache)
        assert _info_cache.maxsize == 200

    def test_stock_router_quote_cache_is_ttlcache(self):
        from routers.stock import _quote_cache
        from cachetools import TTLCache
        assert isinstance(_quote_cache, TTLCache)
        assert _quote_cache.maxsize == 200

    def test_predictor_service_pred_cache_is_ttlcache(self):
        from services.predictor_service import _pred_cache
        from cachetools import TTLCache
        assert isinstance(_pred_cache, TTLCache)
        assert _pred_cache.maxsize == 100

    def test_predictor_service_indicator_cache_is_ttlcache(self):
        from services.predictor_service import _indicator_cache
        from cachetools import TTLCache
        assert isinstance(_indicator_cache, TTLCache)
        assert _indicator_cache.maxsize == 100

    def test_ttlcache_evicts_on_maxsize(self):
        """TTLCache should evict oldest entries when maxsize is reached."""
        from cachetools import TTLCache
        cache = TTLCache(maxsize=3, ttl=60)
        cache["a"] = 1
        cache["b"] = 2
        cache["c"] = 3
        cache["d"] = 4  # should evict "a"
        assert "a" not in cache
        assert len(cache) == 3

    def test_requirements_has_cachetools(self):
        """requirements.txt should include cachetools."""
        req = Path(__file__).parent.parent / "requirements.txt"
        text = req.read_text(encoding="utf-8")
        assert "cachetools" in text


# ── #4 History endpoint shares indicator cache ───────────────────────────────

class TestHistorySharesCache:
    """Verify history endpoint imports and uses predictor_service indicator cache."""

    def test_stock_router_imports_indicator_cache_functions(self):
        """routers/stock.py should import _get_cached_indicators from predictor_service."""
        from routers import stock
        assert hasattr(stock, '_get_cached_indicators')
        assert hasattr(stock, '_set_cached_indicators')
        assert hasattr(stock, '_data_key')

    def test_fetch_history_sync_uses_shared_cache(self):
        """_fetch_history_sync should call _get_cached_indicators."""
        import inspect
        from routers.stock import _fetch_history_sync
        source = inspect.getsource(_fetch_history_sync)
        assert "_get_cached_indicators" in source
        assert "_set_cached_indicators" in source


# ── #5 APP_VERSION constant ──────────────────────────────────────────────────

class TestAppVersion:
    """Verify version is a single constant, not duplicated strings."""

    def test_app_version_constant_exists(self):
        from main import APP_VERSION
        assert isinstance(APP_VERSION, str)
        assert APP_VERSION == "1.1.0"

    def test_no_hardcoded_version_strings(self):
        """main.py should not have any leftover hardcoded version strings."""
        src = Path(__file__).parent.parent / "main.py"
        code = src.read_text(encoding="utf-8")
        # Should not have standalone string "1.0.0" or "1.1.0" outside APP_VERSION
        lines = code.splitlines()
        for line in lines:
            stripped = line.strip()
            if stripped.startswith("APP_VERSION"):
                continue
            assert 'version="1.0.0"' not in stripped, f"Hardcoded 1.0.0: {stripped}"
            assert 'version": "1.0.0"' not in stripped, f"Hardcoded 1.0.0: {stripped}"
            assert '"version": "1.1.0"' not in stripped, f"Hardcoded 1.1.0: {stripped}"

    def test_health_uses_app_version(self):
        """Health endpoint should reference APP_VERSION, not a string literal."""
        src = Path(__file__).parent.parent / "main.py"
        code = src.read_text(encoding="utf-8")
        assert '"version": APP_VERSION' in code or "'version': APP_VERSION" in code


# ── #17 stock_history async (asyncio.to_thread) ─────────────────────────────

class TestHistoryAsync:
    """Verify history endpoint uses asyncio.to_thread."""

    def test_stock_history_uses_to_thread(self):
        """stock_history should wrap sync work in asyncio.to_thread."""
        import inspect
        from routers.stock import stock_history
        source = inspect.getsource(stock_history)
        assert "asyncio.to_thread" in source, "stock_history should use asyncio.to_thread"

    def test_fetch_history_sync_is_sync_function(self):
        """_fetch_history_sync should be a regular (non-async) function."""
        import asyncio
        from routers.stock import _fetch_history_sync
        assert not asyncio.iscoroutinefunction(_fetch_history_sync)


# ── #27 yfinance exchange mapping via twstock.codes ──────────────────────────

class TestExchangeMapping:
    """Verify detect_exchange uses twstock.codes for accurate lookup."""

    def test_detect_exchange_with_suffix(self):
        from data.fetcher import detect_exchange
        assert detect_exchange("2330.TW") == "TWSE"
        assert detect_exchange("7856.TWO") == "TPEX"

    def test_detect_exchange_uses_twstock_codes(self):
        """detect_exchange should use _build_exchange_map for bare codes."""
        from data.fetcher import detect_exchange, _build_exchange_map
        # Mock the exchange map to include a known TPEX stock
        with patch("data.fetcher._exchange_map", {"4169": "TPEX", "2330": "TWSE"}):
            assert detect_exchange("4169") == "TPEX"
            assert detect_exchange("2330") == "TWSE"

    def test_build_exchange_map_returns_dict(self):
        """_build_exchange_map should return a dict."""
        from data.fetcher import _build_exchange_map
        # Reset cache for test
        import data.fetcher as fetcher_mod
        old_map = fetcher_mod._exchange_map
        fetcher_mod._exchange_map = None
        try:
            result = _build_exchange_map()
            assert isinstance(result, dict)
        finally:
            fetcher_mod._exchange_map = old_map

    def test_detect_exchange_fallback_heuristic(self):
        """Unknown stocks should still use the heuristic fallback."""
        from data.fetcher import detect_exchange
        with patch("data.fetcher._exchange_map", {}):
            # 5-digit → TPEX
            assert detect_exchange("12345") == "TPEX"
            # 4-digit 6xxx-9xxx → TPEX
            assert detect_exchange("6000") == "TPEX"
            # 4-digit < 6000 → TWSE
            assert detect_exchange("1234") == "TWSE"


# ── #30 Dockerfile worker count ───────────────────────────────────────────────

class TestDockerfileWorkers:
    """Verify Dockerfile uses a Hugging Face free-plan friendly worker count."""

    def test_dockerfile_has_single_worker(self):
        dockerfile = Path(__file__).parent.parent / "Dockerfile"
        text = dockerfile.read_text(encoding="utf-8")
        assert "--workers" in text, "Dockerfile should specify --workers"
        # Check the final CMD line (not HEALTHCHECK CMD) has one worker.
        for line in text.splitlines():
            stripped = line.strip()
            if stripped.startswith("CMD") and "uvicorn" in stripped:
                assert '"--workers"' in stripped and '"1"' in stripped, \
                    f"CMD should have --workers 1, got: {stripped}"
                break
        else:
            pytest.fail("No uvicorn CMD line found in Dockerfile")


# ── Predictor service cache integration ──────────────────────────────────────

class TestPredictorServiceCache:
    """Test TTLCache-based prediction and indicator caching."""

    def test_pred_cache_set_get(self):
        from services.predictor_service import (
            _pred_cache, _set_cached_prediction, _get_cached_prediction,
        )
        _pred_cache.clear()
        _set_cached_prediction("TEST_STOCK", {"signal": "BUY"})
        result = _get_cached_prediction("TEST_STOCK")
        assert result is not None
        assert result["signal"] == "BUY"
        _pred_cache.clear()

    def test_pred_cache_miss(self):
        from services.predictor_service import _pred_cache, _get_cached_prediction
        _pred_cache.clear()
        assert _get_cached_prediction("NONEXIST") is None

    def test_indicator_cache_set_get(self):
        from services.predictor_service import (
            _indicator_cache, _set_cached_indicators, _get_cached_indicators,
        )
        _indicator_cache.clear()
        _set_cached_indicators("TEST", "2026-04-01", "fake_df")
        result = _get_cached_indicators("TEST", "2026-04-01")
        assert result == "fake_df"
        _indicator_cache.clear()

    def test_indicator_cache_invalidates_on_key_change(self):
        from services.predictor_service import (
            _indicator_cache, _set_cached_indicators, _get_cached_indicators,
        )
        _indicator_cache.clear()
        _set_cached_indicators("TEST", "2026-04-01", "old_df")
        result = _get_cached_indicators("TEST", "2026-04-02")
        assert result is None
        _indicator_cache.clear()


# ── Stock router TTLCache integration ────────────────────────────────────────

class TestStockRouterCache:
    """Test TTLCache-based info and quote caching in stock router."""

    def test_info_cache_stores_and_retrieves(self):
        from routers.stock import _info_cache
        _info_cache.clear()
        _info_cache["2330"] = {"name": "TSMC", "current_price": 600}
        assert "2330" in _info_cache
        assert _info_cache["2330"]["name"] == "TSMC"
        _info_cache.clear()

    def test_quote_cache_stores_and_retrieves(self):
        from routers.stock import _quote_cache
        _quote_cache.clear()
        _quote_cache["2330"] = {"price": 600, "change": 5}
        assert "2330" in _quote_cache
        assert _quote_cache["2330"]["price"] == 600
        _quote_cache.clear()