tests/test_arch_deploy.py

"""
Tests for architecture & deployment optimizations:
  #2   In-memory model cache (ephemeral disk fix)
  #6   CORS includes production domains
  #16  Startup warmup event
  #22  line_webhook.py Flex Message builder extraction
  #26  twstock realtime non-trading-hours cache
  #29  MoneyDJ SSL verification with certifi
  #31  render.yaml completeness
  #32  API 404 catch-all guard
"""

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

import pytest


# ── #2 In-memory model cache ──────────────────────────────────────────────────

class TestInMemoryModelCache:
    """Verify the in-memory model cache layer in predictor_service."""

    def test_mem_cache_functions_exist(self):
        from services.predictor_service import (
            _get_mem_cached_model,
            _set_mem_cached_model,
            _evict_oldest_model,
            _model_mem_cache,
            _MEM_CACHE_MAX_SIZE,
        )
        assert callable(_get_mem_cached_model)
        assert callable(_set_mem_cached_model)
        assert callable(_evict_oldest_model)
        assert isinstance(_model_mem_cache, dict)
        assert _MEM_CACHE_MAX_SIZE > 0

    def test_set_and_get_mem_cache(self):
        from services.predictor_service import (
            _model_mem_cache, _set_mem_cached_model, _get_mem_cached_model,
        )
        _model_mem_cache.clear()
        mock_predictor = MagicMock()
        _set_mem_cached_model("TEST_STOCK", "2026-04-01", mock_predictor)
        result = _get_mem_cached_model("TEST_STOCK", "2026-04-01")
        assert result is mock_predictor
        _model_mem_cache.clear()

    def test_mem_cache_miss_on_key_mismatch(self):
        from services.predictor_service import (
            _model_mem_cache, _set_mem_cached_model, _get_mem_cached_model,
        )
        _model_mem_cache.clear()
        mock_predictor = MagicMock()
        _set_mem_cached_model("TEST_STOCK", "2026-04-01", mock_predictor)
        result = _get_mem_cached_model("TEST_STOCK", "2026-04-02")
        assert result is None
        _model_mem_cache.clear()

    def test_mem_cache_miss_on_absent_stock(self):
        from services.predictor_service import _model_mem_cache, _get_mem_cached_model
        _model_mem_cache.clear()
        result = _get_mem_cached_model("NONEXIST", "2026-04-01")
        assert result is None

    def test_eviction_at_max_size(self):
        from services.predictor_service import (
            _model_mem_cache, _set_mem_cached_model, _MEM_CACHE_MAX_SIZE,
        )
        _model_mem_cache.clear()
        # Fill to max
        for i in range(_MEM_CACHE_MAX_SIZE):
            _set_mem_cached_model(f"STOCK_{i}", "key", MagicMock())
            time.sleep(0.001)  # ensure different timestamps
        assert len(_model_mem_cache) == _MEM_CACHE_MAX_SIZE
        # Add one more — should evict the oldest
        _set_mem_cached_model("STOCK_NEW", "key", MagicMock())
        assert len(_model_mem_cache) == _MEM_CACHE_MAX_SIZE
        assert "STOCK_0" not in _model_mem_cache
        assert "STOCK_NEW" in _model_mem_cache
        _model_mem_cache.clear()

    def test_load_predictor_checks_mem_cache_first(self):
        """load_predictor should return from memory cache without touching disk."""
        import inspect
        from services.predictor_service import load_predictor
        source = inspect.getsource(load_predictor)
        assert "_get_mem_cached_model" in source
        assert "Memory cache hit" in source


# ── #6 CORS includes production domains ───────────────────────────────────────

class TestCORSConfig:
    """Verify CORS allowed origins include production Render domain."""

    def test_allowed_origins_includes_render(self):
        from main import ALLOWED_ORIGINS
        render_found = any("onrender.com" in o for o in ALLOWED_ORIGINS)
        assert render_found, f"No onrender.com in ALLOWED_ORIGINS: {ALLOWED_ORIGINS}"

    def test_allowed_origins_includes_localhost(self):
        from main import ALLOWED_ORIGINS
        assert "http://localhost:5173" in ALLOWED_ORIGINS

    def test_render_external_url_env_support(self):
        """main.py should read RENDER_EXTERNAL_URL env var."""
        src = Path(__file__).parent.parent / "main.py"
        code = src.read_text(encoding="utf-8")
        assert "RENDER_EXTERNAL_URL" in code


# ── #16 Startup warmup ────────────────────────────────────────────────────────

class TestStartupWarmup:
    """Verify FastAPI startup event for model pre-training."""

    def test_warmup_function_exists(self):
        from main import _warmup_models
        import inspect
        assert inspect.iscoroutinefunction(_warmup_models)

    def test_lifespan_registered(self):
        from main import app
        # FastAPI stores lifespan in app.router.lifespan_context
        assert app.router.lifespan_context is not None, "No lifespan handler registered"

    def test_warmup_stocks_configurable(self):
        """WARMUP_STOCKS should be configurable via env var."""
        src = Path(__file__).parent.parent / "main.py"
        code = src.read_text(encoding="utf-8")
        assert "WARMUP_STOCKS" in code

    def test_warmup_stocks_default_includes_popular(self):
        from main import _WARMUP_STOCKS
        assert "2330" in _WARMUP_STOCKS


# ── #22 line_webhook.py Flex Message builder extraction ───────────────────────

class TestFlexMessageExtraction:
    """Verify Flex Message builder is extracted to services/line_message_builder.py."""

    def test_builder_module_exists(self):
        builder_path = Path(__file__).parent.parent / "services" / "line_message_builder.py"
        assert builder_path.exists(), "services/line_message_builder.py not found"

    def test_builder_has_key_functions(self):
        from services.line_message_builder import (
            build_flex_prediction,
            build_quick_reply_menu,
            build_plain_text_prediction,
            prob_bar,
            FLEX_AVAILABLE,
        )
        assert callable(build_flex_prediction)
        assert callable(build_quick_reply_menu)
        assert callable(build_plain_text_prediction)
        assert callable(prob_bar)

    def test_prob_bar_output(self):
        from services.line_message_builder import prob_bar
        result = prob_bar(0.72)
        assert "72%" in result
        assert "█" in result
        assert "░" in result

    def test_prob_bar_boundaries(self):
        from services.line_message_builder import prob_bar
        # 0% — all empty
        result_zero = prob_bar(0.0)
        assert "0%" in result_zero
        # 100% — all filled
        result_full = prob_bar(1.0)
        assert "100%" in result_full

    def test_plain_text_prediction_output(self):
        from services.line_message_builder import build_plain_text_prediction
        result = build_plain_text_prediction(
            name="台積電", stock_no="2330", exchange="TWSE",
            price_str="NT$ 600.00", signal="BUY", confidence="HIGH",
            buy_prob=0.75, sell_prob=0.25, change_pct=2.5,
            today_change_pct=1.2,
        )
        assert "台積電" in result
        assert "2330" in result
        assert "買進" in result
        assert "+2.50%" in result
        assert "+1.20%" in result

    def test_plain_text_prediction_no_today_change(self):
        from services.line_message_builder import build_plain_text_prediction
        result = build_plain_text_prediction(
            name="Test", stock_no="1234", exchange="",
            price_str="N/A", signal="HOLD", confidence="LOW",
            buy_prob=0.5, sell_prob=0.5, change_pct=0.0,
        )
        assert "今日漲跌" not in result
        assert "觀望" in result

    def test_webhook_imports_from_builder(self):
        """line_webhook.py should import from services.line_message_builder."""
        src = Path(__file__).parent.parent / "routers" / "line_webhook.py"
        code = src.read_text(encoding="utf-8")
        assert "from services.line_message_builder import" in code

    def test_webhook_no_inline_flex_builder(self):
        """line_webhook.py should NOT have the old inline _build_flex_prediction."""
        src = Path(__file__).parent.parent / "routers" / "line_webhook.py"
        code = src.read_text(encoding="utf-8")
        # The old function had a docstring mentioning "Flex Message bubble"
        assert "def _prob_bar(" not in code, "Old _prob_bar still in line_webhook.py"

    def test_webhook_line_count_reduced(self):
        """line_webhook.py should be significantly shorter after extraction."""
        src = Path(__file__).parent.parent / "routers" / "line_webhook.py"
        lines = src.read_text(encoding="utf-8").splitlines()
        # Original was 595 lines; after extraction should be under 520
        # (grew from ~420 → ~500 with chart image feature)
        assert len(lines) < 520, f"line_webhook.py still has {len(lines)} lines (expected < 520)"


# ── #26 twstock realtime non-trading-hours cache ──────────────────────────────

class TestRealtimeQuoteCache:
    """Verify stale quote cache for non-trading hours."""

    def test_realtime_cache_exists(self):
        from data.fetcher import _realtime_cache
        from collections.abc import MutableMapping
        assert isinstance(_realtime_cache, MutableMapping)

    def test_cache_and_return_stores_data(self):
        """get_realtime_quote should store results in _realtime_cache."""
        import inspect
        from data.fetcher import get_realtime_quote
        source = inspect.getsource(get_realtime_quote)
        assert "_cache_and_return" in source
        assert "_get_stale" in source

    def test_is_realtime_field_in_source(self):
        """Results should include is_realtime field."""
        import inspect
        from data.fetcher import get_realtime_quote
        source = inspect.getsource(get_realtime_quote)
        assert "is_realtime" in source

    def test_stale_cache_returns_on_empty(self):
        """_get_stale should return empty dict when cache is empty."""
        from data.fetcher import _realtime_cache
        _realtime_cache.clear()
        from data.fetcher import get_realtime_quote
        # Extract _get_stale from the function's closure-free approach
        # We test the concept: no crash on empty cache
        src = Path(__file__).parent.parent / "data" / "fetcher.py"
        code = src.read_text(encoding="utf-8")
        assert 'def _get_stale(key: str)' in code


# ── #29 MoneyDJ SSL verification ─────────────────────────────────────────────

class TestMoneyDJSSL:
    """Verify MoneyDJ uses proper SSL verification with certifi fallback."""

    def test_no_global_verify_false(self):
        """fund_fetcher.py should not have global s.verify = False."""
        src = Path(__file__).parent.parent / "data" / "fund_fetcher.py"
        code = src.read_text(encoding="utf-8")
        # The main _session() should use certifi, not verify=False
        lines = code.splitlines()
        for i, line in enumerate(lines):
            if "def _session():" in line:
                # Check next 10 lines for verify setting
                block = "\n".join(lines[i:i+10])
                assert "certifi.where()" in block, "_session() should use certifi.where()"
                break

    def test_insecure_fallback_exists(self):
        """_session_insecure() should exist as a fallback."""
        from data.fund_fetcher import _session_insecure
        assert callable(_session_insecure)

    def test_certifi_in_requirements(self):
        req = Path(__file__).parent.parent / "requirements.txt"
        text = req.read_text(encoding="utf-8")
        assert "certifi" in text

    def test_certifi_import(self):
        """fund_fetcher should import certifi."""
        src = Path(__file__).parent.parent / "data" / "fund_fetcher.py"
        code = src.read_text(encoding="utf-8")
        assert "import certifi" in code

    def test_no_global_disable_warnings(self):
        """Global urllib3.disable_warnings should be removed (only in _session_insecure)."""
        src = Path(__file__).parent.parent / "data" / "fund_fetcher.py"
        code = src.read_text(encoding="utf-8")
        # Count occurrences — should only appear inside _session_insecure
        lines = code.splitlines()
        global_disable = False
        for line in lines:
            stripped = line.strip()
            if stripped == "urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)":
                # Check if it's at module level (no indentation) vs inside a function
                if not line.startswith(" ") and not line.startswith("\t"):
                    global_disable = True
        assert not global_disable, "Global disable_warnings should be removed"


# ── #31 render.yaml completeness ──────────────────────────────────────────────

class TestRenderYaml:
    """Verify render.yaml has all required fields."""

    def test_render_yaml_exists(self):
        path = Path(__file__).parent.parent / "render.yaml"
        assert path.exists()

    def test_render_yaml_has_required_fields(self):
        import yaml
        path = Path(__file__).parent.parent / "render.yaml"
        data = yaml.safe_load(path.read_text(encoding="utf-8"))
        service = data["services"][0]
        assert service.get("healthCheckPath") == "/health"
        assert service.get("region") is not None
        assert "envVars" in service

    def test_render_yaml_env_vars(self):
        import yaml
        path = Path(__file__).parent.parent / "render.yaml"
        data = yaml.safe_load(path.read_text(encoding="utf-8"))
        service = data["services"][0]
        env_keys = {e["key"] for e in service["envVars"]}
        assert "LINE_CHANNEL_SECRET" in env_keys
        assert "LINE_CHANNEL_ACCESS_TOKEN" in env_keys
        assert "WARMUP_STOCKS" in env_keys
        assert "LIGHTWEIGHT_MODE" in env_keys


# ── HF Space local-architecture integration ─────────────────────────────────

class TestHuggingFaceSpaceIntegration:
    """Verify HF Space stays lightweight and uses Mac-precomputed artifacts."""

    def test_dockerfile_disables_slow_hf_enrichments(self):
        dockerfile = Path(__file__).parent.parent / "Dockerfile"
        text = dockerfile.read_text(encoding="utf-8")
        for required in [
            "LIGHTWEIGHT_MODE=1",
            "ENABLE_OPTUNA=0",
            "ENABLE_NEWS_OVERLAY=0",
            "ENABLE_SECURITIES_LENDING=0",
            "ENABLE_BLOCK_TRADE_FEATURES=0",
            "ENABLE_LARGE_HOLDER_FEATURES=0",
            "LOCAL_PRECOMPUTE_FRESH_HOURS=96",
            "WARMUP_STOCKS=\"\"",
        ]:
            assert required in text

    def test_dockerignore_excludes_local_only_runtime_state(self):
        dockerignore = Path(__file__).parent.parent / ".dockerignore"
        text = dockerignore.read_text(encoding="utf-8")
        for required in [
            "logs/",
            "model_cache/",
            ".hf_token",
            "frontend/node_modules/",
            "data/cache/",
            "prompt-optimizer-local/",
        ]:
            assert required in text

    def test_predictor_service_skips_background_ml_on_hf(self):
        src = Path(__file__).parent.parent / "services" / "predictor_service.py"
        text = src.read_text(encoding="utf-8")
        assert "HF free CPU should not train durable ML models" in text
        assert "await asyncio.to_thread(enqueue, stock_no)" in text


# ── #32 API 404 handler ──────────────────────────────────────────────────────

class TestAPI404Handler:
    """Verify /api/* typos return 404 instead of being caught by SPA."""

    def test_api_not_found_route_exists(self):
        from main import app
        routes = [r.path for r in app.routes if hasattr(r, 'path')]
        assert "/api/{rest_of_path:path}" in routes, \
            f"API 404 catch-all route not found. Routes: {routes}"

    def test_api_not_found_returns_404(self):
        from fastapi.testclient import TestClient
        from main import app
        client = TestClient(app, raise_server_exceptions=False)
        resp = client.get("/api/nonexistent/endpoint")
        assert resp.status_code == 404
        data = resp.json()
        assert "detail" in data
        assert "not found" in data["detail"].lower()

    def test_real_api_endpoints_still_work(self):
        """Existing /api/stock/* routes should NOT be intercepted by the 404 handler."""
        from main import app
        routes = [r.path for r in app.routes if hasattr(r, 'path')]
        # Check that real stock API routes exist and are registered BEFORE the catch-all
        assert "/api/stock/search" in routes or any("/api/stock/" in r for r in routes)

    def test_api_404_before_spa_catchall(self):
        """The API 404 handler should be defined BEFORE the SPA catch-all in source."""
        src = Path(__file__).parent.parent / "main.py"
        code = src.read_text(encoding="utf-8")
        api_404_pos = code.find("api_not_found")
        spa_pos = code.find("serve_spa")
        if spa_pos > 0:
            assert api_404_pos < spa_pos, "API 404 handler must appear before SPA catch-all"