Market Data backend: raise test coverage and fix stream router

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:

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

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"