244 lines
8.7 KiB
Python
244 lines
8.7 KiB
Python
"""Unit tests for the S/R levels router — zone integration."""
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
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 a close attribute."""
|
|
|
|
def __init__(self, close: float):
|
|
self.close = close
|
|
|
|
|
|
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 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 = {l["id"] for l 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"] == []
|
|
|