diff --git a/backend/app/market/stream.py b/backend/app/market/stream.py index 7fd974b7c..f7938fd1b 100644 --- a/backend/app/market/stream.py +++ b/backend/app/market/stream.py @@ -14,14 +14,14 @@ logger = logging.getLogger(__name__) -router = APIRouter(prefix="/api/stream", tags=["streaming"]) - - def create_stream_router(price_cache: PriceCache) -> APIRouter: """Create the SSE streaming router with a reference to the price cache. This factory pattern lets us inject the PriceCache without globals. + The router is created per call, so calling the factory twice never + registers a duplicate route bound to the first cache. """ + router = APIRouter(prefix="/api/stream", tags=["streaming"]) @router.get("/prices") async def stream_prices(request: Request) -> StreamingResponse: diff --git a/backend/tests/market/test_massive.py b/backend/tests/market/test_massive.py index cdd7dbd24..8254ba1bf 100644 --- a/backend/tests/market/test_massive.py +++ b/backend/tests/market/test_massive.py @@ -1,8 +1,10 @@ """Tests for MassiveDataSource (mocked).""" +import asyncio from unittest.mock import MagicMock, patch import pytest +from massive.rest.models import SnapshotMarketType from app.market.cache import PriceCache from app.market.massive_client import MassiveDataSource @@ -199,3 +201,85 @@ async def test_start_immediate_poll(self): assert cache.get_price("AAPL") == 190.50 await source.stop() + + async def test_poll_loop_polls_repeatedly(self): + """Test that the background loop keeps polling after start().""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=0.01) + + first = [_make_snapshot("AAPL", 190.00, 1707580800000)] + later = [_make_snapshot("AAPL", 191.00, 1707580801000)] + + with patch("app.market.massive_client.RESTClient"): + with patch.object(source, "_fetch_snapshots", side_effect=[first] + [later] * 50): + await source.start(["AAPL"]) + assert cache.get_price("AAPL") == 190.00 + await asyncio.sleep(0.1) + await source.stop() + + assert cache.get_price("AAPL") == 191.00 + + async def test_poll_loop_survives_errors(self): + """Test that a failed poll doesn't kill the loop.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache, poll_interval=0.01) + good = [_make_snapshot("AAPL", 192.00, 1707580800000)] + + with patch("app.market.massive_client.RESTClient"): + with patch.object( + source, + "_fetch_snapshots", + side_effect=[[], Exception("429")] + [good] * 50, + ): + await source.start(["AAPL"]) + await asyncio.sleep(0.1) + await source.stop() + + assert cache.get_price("AAPL") == 192.00 + + async def test_fetch_snapshots_calls_client(self): + """Test that _fetch_snapshots passes tracked tickers to the REST client.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache) + source._tickers = ["AAPL", "MSFT"] + source._client = MagicMock() + source._client.get_snapshot_all.return_value = ["snap"] + + result = source._fetch_snapshots() + + assert result == ["snap"] + source._client.get_snapshot_all.assert_called_once_with( + market_type=SnapshotMarketType.STOCKS, + tickers=["AAPL", "MSFT"], + ) + + async def test_poll_without_client_is_noop(self): + """Test that polling before start() (no client) does nothing.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache) + source._tickers = ["AAPL"] + + with patch.object(source, "_fetch_snapshots") as mock_fetch: + await source._poll_once() + mock_fetch.assert_not_called() + + async def test_add_duplicate_ticker_ignored(self): + """Test that adding an existing ticker doesn't duplicate it.""" + cache = PriceCache() + source = MassiveDataSource(api_key="test-key", price_cache=cache) + + await source.add_ticker("AAPL") + await source.add_ticker("aapl") + assert source.get_tickers() == ["AAPL"] + + async def test_start_creates_client_with_api_key(self): + """Test that start() builds the REST client with the configured key.""" + cache = PriceCache() + source = MassiveDataSource(api_key="secret", price_cache=cache, poll_interval=60.0) + + with patch("app.market.massive_client.RESTClient") as mock_client: + with patch.object(source, "_fetch_snapshots", return_value=[]): + await source.start(["AAPL"]) + await source.stop() + + mock_client.assert_called_once_with(api_key="secret") diff --git a/backend/tests/market/test_stream.py b/backend/tests/market/test_stream.py new file mode 100644 index 000000000..260c60848 --- /dev/null +++ b/backend/tests/market/test_stream.py @@ -0,0 +1,116 @@ +"""Tests for the SSE streaming module.""" + +import json +from unittest.mock import MagicMock + +import pytest + +from app.market.cache import PriceCache +from app.market.stream import _generate_events, create_stream_router + +DATA_PREFIX = "data: " + + +def _request(disconnect_after: int) -> MagicMock: + """Fake Request whose is_disconnected() turns True after N checks.""" + calls = {"n": 0} + + async def is_disconnected() -> bool: + calls["n"] += 1 + return calls["n"] > disconnect_after + + request = MagicMock() + request.client.host = "127.0.0.1" + request.is_disconnected = is_disconnected + return request + + +async def _collect(cache: PriceCache, request: MagicMock) -> list[str]: + return [chunk async for chunk in _generate_events(cache, request, interval=0.001)] + + +def _data_events(events: list[str]) -> list[dict]: + return [json.loads(e[len(DATA_PREFIX):]) for e in events if e.startswith(DATA_PREFIX)] + + +class TestGenerateEvents: + @pytest.mark.asyncio + async def test_first_message_is_retry(self): + events = await _collect(PriceCache(), _request(disconnect_after=0)) + assert events == ["retry: 1000\n\n"] + + @pytest.mark.asyncio + async def test_emits_all_tickers_in_one_event(self): + cache = PriceCache() + cache.update("AAPL", 190.0) + cache.update("GOOGL", 175.0) + + events = await _collect(cache, _request(disconnect_after=1)) + + assert len(events) == 2 + assert events[1].endswith("\n\n") + payload = _data_events(events)[0] + assert set(payload) == {"AAPL", "GOOGL"} + assert payload["AAPL"]["price"] == 190.0 + assert payload["AAPL"]["ticker"] == "AAPL" + + @pytest.mark.asyncio + async def test_no_event_when_unchanged(self): + cache = PriceCache() + cache.update("AAPL", 190.0) + + # Several loop iterations, but the cache version never changes + events = await _collect(cache, _request(disconnect_after=5)) + + assert len(_data_events(events)) == 1 + + @pytest.mark.asyncio + async def test_no_event_when_cache_empty(self): + events = await _collect(PriceCache(), _request(disconnect_after=3)) + assert events == ["retry: 1000\n\n"] + + @pytest.mark.asyncio + async def test_new_event_after_price_change(self): + cache = PriceCache() + cache.update("AAPL", 190.0) + + events = [] + async for chunk in _generate_events(cache, _request(disconnect_after=4), interval=0.001): + events.append(chunk) + if len(events) == 2: + cache.update("AAPL", 191.0) + + data = _data_events(events) + assert [d["AAPL"]["price"] for d in data] == [190.0, 191.0] + assert data[1]["AAPL"]["direction"] == "up" + + @pytest.mark.asyncio + async def test_missing_client_info(self): + request = _request(disconnect_after=0) + request.client = None + events = await _collect(PriceCache(), request) + assert events == ["retry: 1000\n\n"] + + +class TestCreateStreamRouter: + def test_route_registered(self): + router = create_stream_router(PriceCache()) + assert [r.path for r in router.routes] == ["/api/stream/prices"] + + def test_factory_returns_independent_routers(self): + r1 = create_stream_router(PriceCache()) + r2 = create_stream_router(PriceCache()) + assert r1 is not r2 + assert len(r1.routes) == 1 + assert len(r2.routes) == 1 + + @pytest.mark.asyncio + async def test_endpoint_response_headers(self): + router = create_stream_router(PriceCache()) + endpoint = router.routes[0].endpoint + + response = await endpoint(_request(disconnect_after=0)) + + assert response.media_type == "text/event-stream" + assert response.headers["cache-control"] == "no-cache" + assert response.headers["x-accel-buffering"] == "no"