You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
397 lines
14 KiB
Python
397 lines
14 KiB
Python
"""
|
|
Overlay API tests — Phase 5
|
|
|
|
Tests cover:
|
|
1. Cache utility unit tests (no DB, no Redis required)
|
|
2. Overlay API endpoint integration tests using the real SQLite DB
|
|
(Redis is not required — graceful degradation is expected)
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.utils.cache import build_cache_key, compute_etag, _serialize, _deserialize
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared test client (uses real stock_oracle.db — no table creation needed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
|
|
# ===========================================================================
|
|
# 1. Cache utility unit tests
|
|
# ===========================================================================
|
|
|
|
class TestCacheUtils:
|
|
def test_build_cache_key_basic(self):
|
|
key = build_cache_key("overlay:score", "AAPL")
|
|
assert key == "overlay:score:AAPL"
|
|
|
|
def test_build_cache_key_multiple_parts(self):
|
|
key = build_cache_key("overlay:bulk", "AAPL,MSFT", "true")
|
|
assert key == "overlay:bulk:AAPL,MSFT:true"
|
|
|
|
def test_build_cache_key_skips_none(self):
|
|
key = build_cache_key("overlay:headlines", "TSLA", None)
|
|
assert key == "overlay:headlines:TSLA"
|
|
|
|
def test_build_cache_key_skips_empty_string(self):
|
|
key = build_cache_key("overlay:wiki", "AAPL", "")
|
|
assert key == "overlay:wiki:AAPL"
|
|
|
|
def test_compute_etag_deterministic(self):
|
|
data = {"symbol": "AAPL", "score": 1.23}
|
|
b = _serialize(data)
|
|
assert compute_etag(b) == compute_etag(b)
|
|
|
|
def test_compute_etag_different_data(self):
|
|
a = _serialize({"a": 1})
|
|
b = _serialize({"a": 2})
|
|
assert compute_etag(a) != compute_etag(b)
|
|
|
|
def test_compute_etag_is_hex_string(self):
|
|
b = _serialize({"x": "y"})
|
|
etag = compute_etag(b)
|
|
assert isinstance(etag, str)
|
|
assert len(etag) == 64 # sha256 hex = 64 chars
|
|
|
|
def test_serialize_deserialize_roundtrip(self):
|
|
payload = {"symbol": "AAPL", "overlay_score": 0.75, "items": [1, 2, 3]}
|
|
raw = _serialize(payload)
|
|
result = _deserialize(raw)
|
|
assert result == payload
|
|
|
|
def test_deserialize_none_returns_none(self):
|
|
assert _deserialize(None) is None
|
|
|
|
def test_deserialize_invalid_bytes_returns_none(self):
|
|
assert _deserialize(b"not-json{{{") is None
|
|
|
|
|
|
# ===========================================================================
|
|
# 2. Redis config fix verification
|
|
# ===========================================================================
|
|
|
|
class TestRedisConfig:
|
|
def test_redis_url_uses_port_16380(self):
|
|
from app.core.config import settings
|
|
assert "16380" in settings.REDIS_URL, (
|
|
f"REDIS_URL should use port 16380 (docker-compose), got: {settings.REDIS_URL}"
|
|
)
|
|
|
|
def test_redis_port_default_matches_docker_compose(self):
|
|
from app.core.config import settings
|
|
assert settings.REDIS_PORT == 16380
|
|
|
|
def test_cache_graceful_degradation_no_redis(self):
|
|
"""With no Redis running, get_cached_response returns None without error."""
|
|
import asyncio
|
|
from app.utils.cache import get_cached_response, set_cached_response
|
|
|
|
async def _run():
|
|
# set_cached_response should return an etag even without Redis
|
|
etag = await set_cached_response("test:key", {"a": 1}, ttl_seconds=60)
|
|
assert isinstance(etag, str)
|
|
assert len(etag) == 64
|
|
# get_cached_response should return None without Redis
|
|
result = await get_cached_response("test:key")
|
|
# Either None (no Redis) or a tuple (Redis connected)
|
|
assert result is None or isinstance(result, tuple)
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
# ===========================================================================
|
|
# 3. Overlay endpoint integration tests
|
|
# ===========================================================================
|
|
|
|
class TestAdminHealth:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/admin/health")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
r = client.get("/api/v1/overlay/admin/health")
|
|
data = r.json()
|
|
assert "overlay_enabled" in data
|
|
assert "sources" in data
|
|
assert isinstance(data["sources"], list)
|
|
assert len(data["sources"]) > 0
|
|
|
|
def test_sources_have_required_fields(self, client):
|
|
r = client.get("/api/v1/overlay/admin/health")
|
|
for src in r.json()["sources"]:
|
|
assert "source" in src
|
|
assert "status" in src
|
|
assert "records_24h" in src
|
|
|
|
def test_overlay_is_enabled(self, client):
|
|
r = client.get("/api/v1/overlay/admin/health")
|
|
assert r.json()["overlay_enabled"] is True
|
|
|
|
|
|
class TestAdminJobLog:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/admin/job-log")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/admin/job-log").json()
|
|
assert "logs" in data
|
|
assert "total_count" in data
|
|
assert isinstance(data["logs"], list)
|
|
|
|
def test_limit_param(self, client):
|
|
r = client.get("/api/v1/overlay/admin/job-log?limit=5")
|
|
assert r.status_code == 200
|
|
assert len(r.json()["logs"]) <= 5
|
|
|
|
def test_limit_too_large_is_capped(self, client):
|
|
r = client.get("/api/v1/overlay/admin/job-log?limit=600")
|
|
assert r.status_code == 422 # exceeds max 500
|
|
|
|
|
|
class TestTopMovers:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/top-movers")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/top-movers").json()
|
|
assert "top_movers" in data
|
|
assert "total_count" in data
|
|
assert isinstance(data["top_movers"], list)
|
|
|
|
def test_limit_param(self, client):
|
|
r = client.get("/api/v1/overlay/top-movers?limit=5")
|
|
assert r.status_code == 200
|
|
assert len(r.json()["top_movers"]) <= 5
|
|
|
|
def test_limit_below_min_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/top-movers?limit=0")
|
|
assert r.status_code == 422
|
|
|
|
def test_limit_above_max_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/top-movers?limit=101")
|
|
assert r.status_code == 422
|
|
|
|
def test_x_cache_header_present(self, client):
|
|
r = client.get("/api/v1/overlay/top-movers")
|
|
# X-Cache should be HIT or MISS (Redis may or may not be running)
|
|
assert "X-Cache" in r.headers
|
|
assert r.headers["X-Cache"] in ("HIT", "MISS")
|
|
|
|
|
|
class TestBulkOverlay:
|
|
def test_returns_200_with_known_symbol(self, client):
|
|
r = client.get("/api/v1/overlay/bulk?symbols=AAPL")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/bulk?symbols=AAPL,MSFT").json()
|
|
assert "results" in data
|
|
assert "total_count" in data
|
|
assert isinstance(data["results"], list)
|
|
|
|
def test_too_many_symbols_rejected(self, client):
|
|
symbols = ",".join([f"S{i:03d}" for i in range(51)])
|
|
r = client.get(f"/api/v1/overlay/bulk?symbols={symbols}")
|
|
assert r.status_code == 400
|
|
assert "50" in r.json()["detail"]
|
|
|
|
def test_empty_symbols_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/bulk?symbols=,,,")
|
|
assert r.status_code == 400
|
|
|
|
def test_symbols_uppercased(self, client):
|
|
r = client.get("/api/v1/overlay/bulk?symbols=aapl")
|
|
assert r.status_code == 200
|
|
|
|
def test_x_cache_header_present(self, client):
|
|
r = client.get("/api/v1/overlay/bulk?symbols=AAPL")
|
|
assert "X-Cache" in r.headers
|
|
|
|
|
|
class TestHeadlines:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/headlines")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/headlines").json()
|
|
assert "symbol" in data
|
|
assert data["symbol"] == "AAPL"
|
|
assert "headlines" in data
|
|
assert "headline_count_6h" in data
|
|
assert "headline_count_24h" in data
|
|
assert "publisher_breadth_24h" in data
|
|
|
|
def test_hours_param_valid(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/headlines?hours=48")
|
|
assert r.status_code == 200
|
|
|
|
def test_hours_below_min_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/headlines?hours=0")
|
|
assert r.status_code == 422
|
|
|
|
def test_hours_above_max_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/headlines?hours=200")
|
|
assert r.status_code == 422
|
|
|
|
def test_symbol_uppercased(self, client):
|
|
r = client.get("/api/v1/overlay/aapl/headlines")
|
|
assert r.status_code == 200
|
|
assert r.json()["symbol"] == "AAPL"
|
|
|
|
def test_headline_counts_non_negative(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/headlines").json()
|
|
assert data["headline_count_6h"] >= 0
|
|
assert data["headline_count_24h"] >= 0
|
|
assert data["publisher_breadth_24h"] >= 0
|
|
|
|
|
|
class TestYouTube:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/youtube")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/youtube").json()
|
|
assert "symbol" in data
|
|
assert data["symbol"] == "AAPL"
|
|
assert "videos" in data
|
|
assert "mentions_24h" in data
|
|
assert "weighted_views_24h" in data
|
|
|
|
def test_mentions_non_negative(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/youtube").json()
|
|
assert data["mentions_24h"] >= 0
|
|
assert data["weighted_views_24h"] >= 0.0
|
|
|
|
|
|
class TestWiki:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/wiki")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/wiki").json()
|
|
assert "symbol" in data
|
|
assert data["symbol"] == "AAPL"
|
|
assert "pageviews" in data
|
|
assert isinstance(data["pageviews"], list)
|
|
|
|
def test_days_param(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/wiki?days=7")
|
|
assert r.status_code == 200
|
|
|
|
def test_days_below_min_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/wiki?days=0")
|
|
assert r.status_code == 422
|
|
|
|
def test_days_above_max_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/wiki?days=91")
|
|
assert r.status_code == 422
|
|
|
|
|
|
class TestCrowding:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/crowding")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/crowding").json()
|
|
assert "symbol" in data
|
|
assert data["symbol"] == "AAPL"
|
|
assert "short_volume_ratio" in data
|
|
assert "crowding_stress_z" in data
|
|
|
|
|
|
class TestTrends:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/trends")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/trends").json()
|
|
assert "symbol" in data
|
|
assert data["symbol"] == "AAPL"
|
|
assert "trends" in data
|
|
assert isinstance(data["trends"], list)
|
|
|
|
def test_no_topics_returns_empty_trends(self, client):
|
|
# Symbol with no topic mapping → empty trends list
|
|
data = client.get("/api/v1/overlay/AAPL/trends").json()
|
|
# Either empty (no mappings) or populated — both are valid
|
|
assert isinstance(data["trends"], list)
|
|
|
|
|
|
class TestHistory:
|
|
def test_returns_200(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/history")
|
|
assert r.status_code == 200
|
|
|
|
def test_response_schema(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/history").json()
|
|
assert "symbol" in data
|
|
assert data["symbol"] == "AAPL"
|
|
assert "history" in data
|
|
assert "metadata" in data
|
|
|
|
def test_days_param(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/history?days=7")
|
|
assert r.status_code == 200
|
|
|
|
def test_days_above_max_rejected(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL/history?days=366")
|
|
assert r.status_code == 422
|
|
|
|
def test_metadata_includes_data_points(self, client):
|
|
data = client.get("/api/v1/overlay/AAPL/history").json()
|
|
assert "data_points" in data["metadata"]
|
|
|
|
|
|
class TestOverlayScore:
|
|
def test_unknown_symbol_returns_404_or_200(self, client):
|
|
# If no data exists for an obscure symbol → 404
|
|
# If get_or_build succeeds → 200
|
|
r = client.get("/api/v1/overlay/ZZZZZ")
|
|
assert r.status_code in (200, 404)
|
|
|
|
def test_404_detail_message(self, client):
|
|
r = client.get("/api/v1/overlay/ZZZZZ")
|
|
if r.status_code == 404:
|
|
assert "ZZZZZ" in r.json()["detail"]
|
|
|
|
def test_200_response_schema(self, client):
|
|
r = client.get("/api/v1/overlay/AAPL")
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
assert "symbol" in data
|
|
assert "overlay_score" in data
|
|
assert "overlay_confidence" in data
|
|
assert "features" in data
|
|
assert "source_presence" in data
|
|
|
|
|
|
class TestRouteOrdering:
|
|
"""Verify static paths aren't shadowed by /{symbol}."""
|
|
|
|
def test_bulk_not_treated_as_symbol(self, client):
|
|
r = client.get("/api/v1/overlay/bulk?symbols=AAPL")
|
|
assert r.status_code in (200, 400) # Not 404 from symbol route
|
|
|
|
def test_top_movers_not_treated_as_symbol(self, client):
|
|
r = client.get("/api/v1/overlay/top-movers")
|
|
assert r.status_code == 200
|
|
|
|
def test_admin_health_not_treated_as_symbol(self, client):
|
|
r = client.get("/api/v1/overlay/admin/health")
|
|
assert r.status_code == 200
|