""" 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()