Spaces:
Running
Running
File size: 1,780 Bytes
ee37d63 d2d8dd9 ee37d63 d2d8dd9 ee37d63 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | from datetime import date
import pandas as pd
from fastapi import FastAPI
from fastapi.testclient import TestClient
from routers.oil import router
def test_oil_history_uses_ttl_cache_and_force_refresh(monkeypatch):
import routers.oil as oil_router
oil_router._oil_history_cache.clear()
app = FastAPI()
app.include_router(router)
client = TestClient(app)
calls = {"count": 0}
def fake_fetch_history(symbol, months=6):
calls["count"] += 1
return pd.DataFrame(
{
"date": [date(2026, 6, 1), date(2026, 6, 2)],
"open": [80.0, 81.0],
"high": [82.0, 83.0],
"low": [79.0, 80.0],
"close": [81.0, 82.0 + calls["count"]],
"volume": [1000, 1200],
}
)
monkeypatch.setattr("routers.oil.is_oil_ticker", lambda symbol: True)
monkeypatch.setattr("routers.oil.fetch_oil_history", fake_fetch_history)
monkeypatch.setattr("routers.oil.add_all_indicators", lambda df: df)
monkeypatch.setattr("routers.oil._read_route_cache", lambda *args, **kwargs: (None, False))
monkeypatch.setattr("routers.oil._write_route_cache", lambda *args, **kwargs: None)
params = {"months": 6, "blocking": True}
first = client.get("/api/oil/BZ=F/history", params=params)
second = client.get("/api/oil/BZ=F/history", params=params)
forced = client.get("/api/oil/BZ=F/history", params={**params, "force_refresh": True})
assert first.status_code == 200
assert second.status_code == 200
assert forced.status_code == 200
assert calls["count"] == 2
assert first.json()["cache"]["hit"] is False
assert second.json()["cache"]["hit"] is True
assert forced.json()["cache"]["hit"] is False
|