tests/test_cached_props_bundle.py

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()