Spaces:
Sleeping
Sleeping
File size: 5,091 Bytes
3f1dfee e213a9c 3f1dfee | 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 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | from __future__ import annotations
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from data.provider_theoddsapi import TheOddsAPIProvider
class _FakeResponse:
def __init__(self, payload, status_code: int = 200, headers: dict | None = None) -> None:
self._payload = payload
self.status_code = status_code
self.headers = headers or {}
self.text = ""
def json(self):
return self._payload
def raise_for_status(self) -> None:
return None
class TestTheOddsAPIProvider(unittest.TestCase):
def test_upcoming_hr_props_passes_requested_bookmakers(self) -> None:
provider = TheOddsAPIProvider()
captured_params: list[dict] = []
def fake_get(url, params=None, timeout=None):
captured_params.append(dict(params or {}))
if url.endswith("/events"):
return _FakeResponse(
[
{
"id": "event-1",
"away_team": "Away",
"home_team": "Home",
"commence_time": "2026-03-25T00:10:00Z",
}
],
headers={"x-requests-remaining": "100"},
)
return _FakeResponse(
{
"bookmakers": [
{
"key": "draftkings",
"markets": [
{
"key": "batter_home_runs",
"outcomes": [
{"description": "Slugger Sam", "price": 400, "point": 0.5}
],
}
],
}
]
},
headers={"x-requests-remaining": "99"},
)
with patch("data.provider_theoddsapi.ODDS_API_KEY", "test-key"), \
patch("data.provider_theoddsapi.requests.get", side_effect=fake_get), \
patch("data.provider_theoddsapi.time.sleep", return_value=None):
df = provider.fetch_all_upcoming_hr_props(
sportsbooks=["draftkings", "fanduel"]
)
self.assertEqual(len(df), 1)
self.assertEqual(captured_params[1]["bookmakers"], "draftkings,fanduel")
def test_upcoming_coverage_probe_reports_market_by_book(self) -> None:
provider = TheOddsAPIProvider()
captured_params: list[dict] = []
def fake_get(url, params=None, timeout=None):
captured_params.append(dict(params or {}))
if url.endswith("/events"):
return _FakeResponse(
[
{
"id": "event-1",
"away_team": "Away",
"home_team": "Home",
"commence_time": "2026-03-25T00:10:00Z",
}
],
headers={"x-requests-remaining": "100"},
)
book_key = params.get("bookmakers")
market_key = params.get("markets")
bookmakers = []
if book_key == "draftkings" and market_key == "batter_hits":
bookmakers = [
{
"key": "draftkings",
"markets": [
{
"key": "batter_hits",
"outcomes": [
{"description": "Slugger Sam", "price": -110, "point": 1.5}
],
}
],
}
]
return _FakeResponse({"bookmakers": bookmakers}, headers={"x-requests-remaining": "99"})
with patch("data.provider_theoddsapi.ODDS_API_KEY", "test-key"), \
patch("data.provider_theoddsapi.requests.get", side_effect=fake_get):
df = provider.fetch_upcoming_market_coverage_probe(
sportsbooks=["draftkings", "fanduel"],
markets=["batter_home_runs", "batter_hits"],
max_events=1,
)
self.assertEqual(len(df), 4)
hits_row = df[
(df["sportsbook_key"] == "draftkings")
& (df["market_key"] == "batter_hits")
].iloc[0]
hr_row = df[
(df["sportsbook_key"] == "fanduel")
& (df["market_key"] == "batter_home_runs")
].iloc[0]
self.assertTrue(bool(hits_row["has_data"]))
self.assertEqual(int(hits_row["outcomes_returned"]), 1)
self.assertFalse(bool(hr_row["has_data"]))
self.assertEqual(
captured_params[1]["bookmakers"],
"draftkings",
)
if __name__ == "__main__":
unittest.main()
|