"""Unit tests for the S/R levels router — zone integration.""" from datetime import datetime from unittest.mock import AsyncMock, patch from fastapi import FastAPI from fastapi.testclient import TestClient from app.middleware import register_exception_handlers from app.routers.sr_levels import router # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- class _FakeLevel: """Mimics an SRLevel ORM model.""" def __init__(self, id, price_level, type, strength, detection_method): self.id = id self.price_level = price_level self.type = type self.strength = strength self.detection_method = detection_method self.created_at = datetime(2024, 1, 1) class _FakeOHLCV: """Mimics an OHLCVRecord with price attributes.""" def __init__(self, close: float, high: float | None = None, low: float | None = None): self.close = close self.high = high if high is not None else close + 1.0 self.low = low if low is not None else close - 1.0 def _make_app() -> FastAPI: app = FastAPI() register_exception_handlers(app) app.include_router(router, prefix="/api/v1") # Override auth dependency to no-op from app.dependencies import require_access, get_db app.dependency_overrides[require_access] = lambda: None app.dependency_overrides[get_db] = lambda: AsyncMock() return app # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- SAMPLE_LEVELS = [ _FakeLevel(1, 95.0, "support", 60, "volume_profile"), _FakeLevel(2, 96.0, "support", 40, "pivot_point"), _FakeLevel(3, 110.0, "resistance", 80, "merged"), ] SAMPLE_OHLCV = [_FakeOHLCV(100.0)] class TestGateTargetLadderRouter: @patch("app.routers.sr_levels.detect_gate_target_ladder") @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) def test_returns_transient_price_traffic_proposals(self, mock_ohlcv, mock_detect): mock_ohlcv.return_value = [ _FakeOHLCV(100.0, high=101.0, low=99.0), _FakeOHLCV(102.0, high=103.0, low=100.0), ] mock_detect.return_value = [ { "price_level": 105.0, "type": "resistance", "strength": 80, "detection_method": "merged", "sources": ["pivot_point", "range_grid"], "rejection_count": 7, }, { "price_level": 95.0, "type": "support", "strength": 60, "detection_method": "range_grid", "sources": ["range_grid"], "rejection_count": 4, }, ] response = TestClient(_make_app()).get("/api/v1/gate-target-ladder/aapl") assert response.status_code == 200 data = response.json()["data"] assert data["symbol"] == "AAPL" assert data["lookback_bars"] == 2 assert [level["price_level"] for level in data["levels"]] == [95.0, 105.0] assert data["levels"][1] == { "price_level": 105.0, "type": "resistance", "strength": 80, "detection_method": "merged", "sources": ["pivot_point", "range_grid"], "traffic_count": 7, } mock_detect.assert_called_once_with( [101.0, 103.0], [99.0, 100.0], [100.0, 102.0], ) @patch("app.routers.sr_levels.detect_gate_target_ladder") @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) def test_empty_history_returns_empty_ladder(self, mock_ohlcv, mock_detect): mock_ohlcv.return_value = [] response = TestClient(_make_app()).get("/api/v1/gate-target-ladder/AAPL") assert response.status_code == 200 assert response.json()["data"] == { "symbol": "AAPL", "levels": [], "count": 0, "lookback_bars": 0, } mock_detect.assert_not_called() class TestSRLevelsRouterZones: """Tests for max_zones parameter and zone inclusion in response.""" @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_default_max_zones_returns_zones(self, mock_get_sr, mock_ohlcv): mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL") assert resp.status_code == 200 body = resp.json() assert body["status"] == "success" data = body["data"] assert "zones" in data assert isinstance(data["zones"], list) # With default max_zones=6, we should get zones assert len(data["zones"]) > 0 @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_max_zones_zero_returns_empty_zones(self, mock_get_sr, mock_ohlcv): mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL?max_zones=0") assert resp.status_code == 200 data = resp.json()["data"] assert data["zones"] == [] @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_max_zones_limits_zone_count(self, mock_get_sr, mock_ohlcv): mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL?max_zones=1") assert resp.status_code == 200 data = resp.json()["data"] assert len(data["zones"]) <= 1 @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_no_ohlcv_data_returns_empty_zones(self, mock_get_sr, mock_ohlcv): mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = [] # No OHLCV data app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL") assert resp.status_code == 200 data = resp.json()["data"] assert data["zones"] == [] # Levels should still be present assert len(data["levels"]) == 3 @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_no_levels_returns_empty_zones(self, mock_get_sr, mock_ohlcv): mock_get_sr.return_value = [] mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL") assert resp.status_code == 200 data = resp.json()["data"] assert data["zones"] == [] assert data["levels"] == [] assert data["count"] == 0 @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_zone_fields_present(self, mock_get_sr, mock_ohlcv): mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL") data = resp.json()["data"] for zone in data["zones"]: assert "low" in zone assert "high" in zone assert "midpoint" in zone assert "strength" in zone assert "type" in zone assert "level_count" in zone assert zone["type"] in ("support", "resistance") class TestSRLevelsRouterVisibleLevels: """Tests for visible_levels filtering in the SR levels response.""" @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_visible_levels_present_in_response(self, mock_get_sr, mock_ohlcv): """visible_levels field is always present in the API response.""" mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL") assert resp.status_code == 200 data = resp.json()["data"] assert "visible_levels" in data assert isinstance(data["visible_levels"], list) @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_visible_levels_within_zone_bounds(self, mock_get_sr, mock_ohlcv): """Every visible level has a price within at least one zone's [low, high] range.""" mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL") data = resp.json()["data"] zones = data["zones"] visible = data["visible_levels"] # When zones exist, each visible level must fall within a zone for lvl in visible: price = lvl["price_level"] assert any( z["low"] <= price <= z["high"] for z in zones ), f"visible level price {price} not within any zone bounds" # visible_levels must be a subset of levels (by id) level_ids = {level["id"] for level in data["levels"]} for lvl in visible: assert lvl["id"] in level_ids @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_visible_levels_empty_when_no_ohlcv(self, mock_get_sr, mock_ohlcv): """visible_levels is empty when no OHLCV data exists (zones are empty).""" mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = [] app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL") data = resp.json()["data"] assert data["zones"] == [] assert data["visible_levels"] == [] @patch("app.routers.sr_levels.query_ohlcv", new_callable=AsyncMock) @patch("app.routers.sr_levels.get_sr_levels", new_callable=AsyncMock) def test_visible_levels_empty_when_max_zones_zero(self, mock_get_sr, mock_ohlcv): """visible_levels is empty when max_zones=0 (zones are empty).""" mock_get_sr.return_value = SAMPLE_LEVELS mock_ohlcv.return_value = SAMPLE_OHLCV app = _make_app() client = TestClient(app) resp = client.get("/api/v1/sr-levels/AAPL?max_zones=0") data = resp.json()["data"] assert data["zones"] == [] assert data["visible_levels"] == []