Spaces:
Running
Running
File size: 3,052 Bytes
e5071ce | 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 | from __future__ import annotations
import os
import sys
import unittest
import pandas as pd
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from database.db import _build_hr_snapshot_state, _preserve_last_known_good_hr_rows
def _row(
*,
market: str,
sportsbook_key: str,
player_name: str,
event_id: str = "event-1",
) -> dict:
return {
"event_id": event_id,
"commence_time": "2026-03-25T00:10:00Z",
"away_team": "Away",
"home_team": "Home",
"sportsbook": sportsbook_key.title(),
"sportsbook_key": sportsbook_key,
"market": market,
"market_family": market,
"player_name": player_name,
}
class TestCachedPropsBundle(unittest.TestCase):
def test_preserves_existing_hr_rows_when_incoming_has_none(self) -> None:
incoming = pd.DataFrame(
[
_row(market="k", sportsbook_key="draftkings", player_name="ace pitcher"),
]
)
existing = pd.DataFrame(
[
_row(market="hr", sportsbook_key="williamhill_us", player_name="slugger sam"),
_row(market="k", sportsbook_key="fanduel", player_name="other pitcher", event_id="event-2"),
]
)
merged, prevented = _preserve_last_known_good_hr_rows(incoming, existing)
self.assertTrue(prevented)
self.assertEqual(set(merged["market_family"]), {"hr", "k"})
hr_rows = merged[merged["market_family"] == "hr"]
self.assertEqual(len(hr_rows), 1)
self.assertEqual(hr_rows.iloc[0]["player_name"], "slugger sam")
def test_does_not_preserve_when_incoming_has_hr_rows(self) -> None:
incoming = pd.DataFrame(
[
_row(market="hr", sportsbook_key="draftkings", player_name="new slugger"),
]
)
existing = pd.DataFrame(
[
_row(market="hr", sportsbook_key="williamhill_us", player_name="old slugger"),
]
)
merged, prevented = _preserve_last_known_good_hr_rows(incoming, existing)
self.assertFalse(prevented)
self.assertEqual(len(merged), 1)
self.assertEqual(merged.iloc[0]["player_name"], "new slugger")
def test_hr_snapshot_state_values(self) -> None:
self.assertEqual(
_build_hr_snapshot_state(current_hr_row_count=0, is_complete=False, overwrite_prevented=False),
"empty",
)
self.assertEqual(
_build_hr_snapshot_state(current_hr_row_count=5, is_complete=True, overwrite_prevented=False),
"usable_complete",
)
self.assertEqual(
_build_hr_snapshot_state(current_hr_row_count=5, is_complete=False, overwrite_prevented=False),
"usable_partial",
)
self.assertEqual(
_build_hr_snapshot_state(current_hr_row_count=5, is_complete=False, overwrite_prevented=True),
"stale_degraded",
)
if __name__ == "__main__":
unittest.main()
|