209 lines
7.2 KiB
Python
209 lines
7.2 KiB
Python
"""
|
|
tests/test_phase22.py
|
|
----------------------
|
|
Tests for Phase 22 features:
|
|
- Incremental VA fetch (break on already-stored dates)
|
|
- top_prize column in games table
|
|
- Dashboard game filter
|
|
- Auto-check predictions after fetch (match detection logic)
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from db.database import get_db_stats, init_db
|
|
from db.models import (
|
|
get_all_games,
|
|
get_game_by_name,
|
|
get_last_draw,
|
|
insert_draw,
|
|
insert_prediction,
|
|
get_predictions,
|
|
)
|
|
from core.fetcher import fetch_cash5_va, fetch_bankamillion_va
|
|
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────────
|
|
|
|
def _text_resp(text, status=200):
|
|
m = MagicMock()
|
|
m.status_code = status
|
|
m.raise_for_status = MagicMock()
|
|
m.text = text
|
|
return m
|
|
|
|
|
|
CASH5_VA_TEXT_OLD = (
|
|
"5/22/2026; 15,29,30,34,36\n"
|
|
"5/21/2026; 1,2,5,38,44\n"
|
|
)
|
|
|
|
CASH5_VA_TEXT_NEWER = (
|
|
"5/23/2026; 10,11,12,13,14\n" # new record
|
|
"5/22/2026; 15,29,30,34,36\n" # already in DB
|
|
"5/21/2026; 1,2,5,38,44\n" # already in DB
|
|
)
|
|
|
|
|
|
# ── top_prize column ───────────────────────────────────────────────────────────
|
|
|
|
def test_top_prize_seeded_powerball(tmp_db):
|
|
row = get_game_by_name("Powerball")
|
|
assert row["top_prize"] == "Jackpot (variable)"
|
|
|
|
|
|
def test_top_prize_seeded_mega_millions(tmp_db):
|
|
row = get_game_by_name("Mega Millions")
|
|
assert row["top_prize"] == "Jackpot (variable)"
|
|
|
|
|
|
def test_top_prize_seeded_cash5(tmp_db):
|
|
row = get_game_by_name("Cash 5")
|
|
assert row["top_prize"] == "Jackpot from $200K"
|
|
|
|
|
|
def test_top_prize_seeded_millionaire_for_life(tmp_db):
|
|
row = get_game_by_name("Millionaire for Life")
|
|
assert row["top_prize"] == "$1M/yr for life"
|
|
|
|
|
|
def test_top_prize_seeded_bank_a_million(tmp_db):
|
|
row = get_game_by_name("Bank a Million")
|
|
assert row["top_prize"] == "$1M after taxes"
|
|
|
|
|
|
def test_top_prize_all_games_non_empty(tmp_db):
|
|
games = get_all_games()
|
|
for g in games:
|
|
# Builtin games must have a top_prize; custom games may be empty
|
|
assert "top_prize" in g.keys()
|
|
|
|
|
|
# ── Incremental VA fetch ───────────────────────────────────────────────────────
|
|
|
|
def test_incremental_fetch_only_new_records_added(tmp_db):
|
|
"""Second fetch with a new leading record only inserts that one record."""
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT_OLD)):
|
|
r1 = fetch_cash5_va()
|
|
assert r1["added"] == 2
|
|
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT_NEWER)):
|
|
r2 = fetch_cash5_va()
|
|
assert r2["added"] == 1
|
|
assert r2["skipped"] == 0 # stopped before re-attempting stored dates
|
|
|
|
|
|
def test_incremental_fetch_no_new_data(tmp_db):
|
|
"""Second fetch of identical data adds nothing and doesn't error."""
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT_OLD)):
|
|
fetch_cash5_va()
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT_OLD)):
|
|
r = fetch_cash5_va()
|
|
assert r["status"] == "success"
|
|
assert r["added"] == 0
|
|
|
|
|
|
def test_incremental_fetch_db_count_correct(tmp_db):
|
|
"""Total draws in DB after incremental fetch equals unique dates only."""
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT_OLD)):
|
|
fetch_cash5_va()
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT_NEWER)):
|
|
fetch_cash5_va()
|
|
|
|
game = get_game_by_name("Cash 5")
|
|
stats = get_db_stats()
|
|
assert stats["Cash 5"] == 3 # 2026-05-21, 05-22, 05-23
|
|
|
|
|
|
def test_incremental_fetch_bank_a_million(tmp_db):
|
|
"""Bank a Million incremental fetch behaves the same as Cash 5."""
|
|
bam_old = (
|
|
"Results for Bank a Million\n"
|
|
"5/20/2026; 14,20,21,24,33,35; Bonus Ball: 18\n"
|
|
"5/16/2026; 6,8,17,20,29,39; Bonus Ball: 38\n"
|
|
)
|
|
bam_newer = (
|
|
"Results for Bank a Million\n"
|
|
"5/23/2026; 1,5,10,20,35,38; Bonus Ball: 7\n"
|
|
"5/20/2026; 14,20,21,24,33,35; Bonus Ball: 18\n"
|
|
"5/16/2026; 6,8,17,20,29,39; Bonus Ball: 38\n"
|
|
)
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(bam_old)):
|
|
r1 = fetch_bankamillion_va()
|
|
assert r1["added"] == 2
|
|
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(bam_newer)):
|
|
r2 = fetch_bankamillion_va()
|
|
assert r2["added"] == 1
|
|
assert r2["skipped"] == 0
|
|
|
|
|
|
# ── Match detection logic (unit-level) ────────────────────────────────────────
|
|
|
|
def _count_matches(pred_numbers: str, draw_numbers: str) -> int:
|
|
pred = {int(n) for n in pred_numbers.split(",") if n.strip().isdigit()}
|
|
draw = {int(n) for n in draw_numbers.split(",") if n.strip().isdigit()}
|
|
return len(pred & draw)
|
|
|
|
|
|
def test_match_detection_exact(tmp_db):
|
|
pb = get_game_by_name("Powerball")
|
|
insert_draw(pb["id"], "2026-05-22", [1, 13, 36, 61, 69], bonus=7)
|
|
last = get_last_draw(pb["id"])
|
|
|
|
pred_id = insert_prediction(pb["id"], "Hot Numbers", [1, 13, 36, 61, 69], bonus=7)
|
|
preds = get_predictions(game_id=pb["id"])
|
|
pred = preds[0]
|
|
|
|
matches = _count_matches(pred["numbers"], last["numbers"])
|
|
assert matches == 5
|
|
|
|
|
|
def test_match_detection_partial(tmp_db):
|
|
pb = get_game_by_name("Powerball")
|
|
insert_draw(pb["id"], "2026-05-22", [1, 13, 36, 61, 69], bonus=7)
|
|
last = get_last_draw(pb["id"])
|
|
|
|
insert_prediction(pb["id"], "Hot Numbers", [1, 13, 5, 6, 7], bonus=99)
|
|
preds = get_predictions(game_id=pb["id"])
|
|
|
|
matches = _count_matches(preds[0]["numbers"], last["numbers"])
|
|
assert matches == 2
|
|
|
|
|
|
def test_match_detection_no_match(tmp_db):
|
|
pb = get_game_by_name("Powerball")
|
|
insert_draw(pb["id"], "2026-05-22", [1, 13, 36, 61, 69], bonus=7)
|
|
last = get_last_draw(pb["id"])
|
|
|
|
insert_prediction(pb["id"], "Due Numbers", [2, 4, 6, 8, 10], bonus=99)
|
|
preds = get_predictions(game_id=pb["id"])
|
|
|
|
matches = _count_matches(preds[0]["numbers"], last["numbers"])
|
|
assert matches == 0
|
|
|
|
|
|
def test_match_alert_threshold_is_two(tmp_db):
|
|
"""Only predictions with ≥2 main matches (or bonus hit) count toward alert."""
|
|
pb = get_game_by_name("Powerball")
|
|
insert_draw(pb["id"], "2026-05-22", [1, 13, 36, 61, 69], bonus=7)
|
|
last = get_last_draw(pb["id"])
|
|
|
|
preds_data = [
|
|
([1, 2, 3, 4, 5], 99), # 1 match — below threshold
|
|
([1, 13, 2, 3, 4], 99), # 2 matches — at threshold
|
|
([1, 13, 36, 2, 3], 99), # 3 matches — above threshold
|
|
]
|
|
for nums, bonus in preds_data:
|
|
insert_prediction(pb["id"], "Test", nums, bonus=bonus)
|
|
|
|
preds = get_predictions(game_id=pb["id"])
|
|
draw_nums = {int(n) for n in last["numbers"].split(",") if n.strip().isdigit()}
|
|
|
|
qualifying = [
|
|
p for p in preds
|
|
if len({int(n) for n in p["numbers"].split(",") if n.strip().isdigit()} & draw_nums) >= 2
|
|
]
|
|
assert len(qualifying) == 2
|