401 lines
16 KiB
Python
401 lines
16 KiB
Python
"""
|
|
tests/test_fetcher.py
|
|
---------------------
|
|
Tests for core/fetcher.py — all HTTP calls are mocked.
|
|
"""
|
|
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import requests as req_lib
|
|
|
|
from core.fetcher import (
|
|
fetch_all,
|
|
fetch_megamillions_ny,
|
|
fetch_megamillions_tx,
|
|
fetch_powerball_ny,
|
|
fetch_cash5_va,
|
|
fetch_millionaireforlife_va,
|
|
fetch_bankamillion_va,
|
|
)
|
|
from db.models import get_draw_count, get_draws, get_game_by_name, get_last_fetch_log
|
|
|
|
# ── Mock data ─────────────────────────────────────────────────────────────────
|
|
|
|
PB_NY_RECORDS = [
|
|
{"draw_date": "2024-01-01T00:00:00.000", "winning_numbers": "01 13 36 61 69 07", "multiplier": "2"},
|
|
{"draw_date": "2024-01-03T00:00:00.000", "winning_numbers": "05 20 45 60 68 15", "multiplier": "3"},
|
|
]
|
|
|
|
MM_NY_RECORDS = [
|
|
{"draw_date": "2024-02-01T00:00:00.000", "winning_numbers": "07 11 22 29 38", "mega_ball": "04", "multiplier": "2"},
|
|
{"draw_date": "2024-02-05T00:00:00.000", "winning_numbers": "03 18 33 44 67", "mega_ball": "22", "multiplier": "5"},
|
|
]
|
|
|
|
MM_TX_CSV = (
|
|
"Game Name,Month,Day,Year,Num1,Num2,Num3,Num4,Num5,Mega Ball,Megaplier\n"
|
|
"Mega Millions,03,07,2024,10,20,30,40,50,12,3\n"
|
|
"Mega Millions,03,12,2024,15,25,35,45,55,07,4\n"
|
|
)
|
|
|
|
|
|
def _json_resp(data, status=200):
|
|
"""Mock requests.Response returning JSON."""
|
|
m = MagicMock()
|
|
m.status_code = status
|
|
m.raise_for_status = MagicMock()
|
|
m.json.return_value = data
|
|
m.text = json.dumps(data)
|
|
return m
|
|
|
|
|
|
def _text_resp(text, status=200):
|
|
"""Mock requests.Response returning plain text (CSV)."""
|
|
m = MagicMock()
|
|
m.status_code = status
|
|
m.raise_for_status = MagicMock()
|
|
m.text = text
|
|
return m
|
|
|
|
|
|
# ── Powerball NY ──────────────────────────────────────────────────────────────
|
|
|
|
def test_pb_ny_inserts_records(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS), _json_resp([])]):
|
|
result = fetch_powerball_ny()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["added"] == 2
|
|
assert result["skipped"] == 0
|
|
game = get_game_by_name("Powerball")
|
|
assert get_draw_count(game["id"]) == 2
|
|
|
|
|
|
def test_pb_ny_skips_duplicates(tmp_db):
|
|
for _ in range(2):
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS), _json_resp([])]):
|
|
result = fetch_powerball_ny()
|
|
|
|
assert result["added"] == 0
|
|
assert result["skipped"] == 2
|
|
|
|
|
|
def test_pb_ny_parses_numbers_and_bonus(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS[:1]), _json_resp([])]):
|
|
fetch_powerball_ny()
|
|
|
|
game = get_game_by_name("Powerball")
|
|
draws = get_draws(game["id"])
|
|
assert draws[0]["draw_date"] == "2024-01-01"
|
|
assert draws[0]["numbers"] == "1,13,36,61,69"
|
|
assert draws[0]["bonus"] == "7"
|
|
assert draws[0]["multiplier"] == "2"
|
|
assert draws[0]["source"] == "powerball_ny"
|
|
|
|
|
|
def test_pb_ny_network_error(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("timeout")):
|
|
result = fetch_powerball_ny()
|
|
|
|
assert result["status"] == "error"
|
|
assert result["added"] == 0
|
|
assert "timeout" in result["message"]
|
|
log = get_last_fetch_log("powerball_ny")
|
|
assert log["status"] == "error"
|
|
|
|
|
|
def test_pb_ny_malformed_row_skipped(tmp_db):
|
|
"""Row with too few numbers is skipped without crashing."""
|
|
bad_records = [{"draw_date": "2024-01-01T00:00:00.000", "winning_numbers": "01 13 36"}]
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(bad_records), _json_resp([])]):
|
|
result = fetch_powerball_ny()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["added"] == 0
|
|
|
|
|
|
# ── Mega Millions NY ──────────────────────────────────────────────────────────
|
|
|
|
def test_mm_ny_inserts_records(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS), _json_resp([])]):
|
|
result = fetch_megamillions_ny()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["added"] == 2
|
|
assert result["skipped"] == 0
|
|
|
|
|
|
def test_mm_ny_skips_duplicates(tmp_db):
|
|
for _ in range(2):
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS), _json_resp([])]):
|
|
result = fetch_megamillions_ny()
|
|
|
|
assert result["added"] == 0
|
|
assert result["skipped"] == 2
|
|
|
|
|
|
def test_mm_ny_parses_mega_ball(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS[:1]), _json_resp([])]):
|
|
fetch_megamillions_ny()
|
|
|
|
game = get_game_by_name("Mega Millions")
|
|
draws = get_draws(game["id"])
|
|
assert draws[0]["bonus"] == "4"
|
|
assert draws[0]["numbers"] == "7,11,22,29,38"
|
|
|
|
|
|
def test_mm_ny_network_error(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("refused")):
|
|
result = fetch_megamillions_ny()
|
|
|
|
assert result["status"] == "error"
|
|
log = get_last_fetch_log("megamillions_ny")
|
|
assert log["status"] == "error"
|
|
|
|
|
|
# ── Mega Millions TX ──────────────────────────────────────────────────────────
|
|
|
|
def test_mm_tx_inserts_records(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(MM_TX_CSV)):
|
|
result = fetch_megamillions_tx()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["added"] == 2
|
|
assert result["skipped"] == 0
|
|
|
|
|
|
def test_mm_tx_skips_duplicates(tmp_db):
|
|
for _ in range(2):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(MM_TX_CSV)):
|
|
result = fetch_megamillions_tx()
|
|
|
|
assert result["added"] == 0
|
|
assert result["skipped"] == 2
|
|
|
|
|
|
def test_mm_tx_parses_date_and_numbers(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(MM_TX_CSV)):
|
|
fetch_megamillions_tx()
|
|
|
|
game = get_game_by_name("Mega Millions")
|
|
draws = get_draws(game["id"], order="ASC")
|
|
assert draws[0]["draw_date"] == "2024-03-07"
|
|
assert draws[0]["numbers"] == "10,20,30,40,50"
|
|
assert draws[0]["bonus"] == "12"
|
|
assert draws[0]["multiplier"] == "3"
|
|
|
|
|
|
def test_mm_tx_cross_source_dedup(tmp_db):
|
|
"""TX record on same date as an already-inserted NY record is skipped."""
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(MM_NY_RECORDS[:1]), _json_resp([])]):
|
|
fetch_megamillions_ny() # inserts 2024-02-01
|
|
|
|
overlapping_csv = (
|
|
"Game Name,Month,Day,Year,Num1,Num2,Num3,Num4,Num5,Mega Ball,Megaplier\n"
|
|
"Mega Millions,02,01,2024,07,11,22,29,38,4,2\n"
|
|
)
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(overlapping_csv)):
|
|
result = fetch_megamillions_tx()
|
|
|
|
assert result["skipped"] == 1
|
|
assert result["added"] == 0
|
|
|
|
|
|
def test_mm_tx_network_error(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("connect")):
|
|
result = fetch_megamillions_tx()
|
|
|
|
assert result["status"] == "error"
|
|
assert result["added"] == 0
|
|
|
|
|
|
def test_mm_tx_no_header_row(tmp_db):
|
|
"""CSV without a header row still parses correctly."""
|
|
no_header = "Mega Millions,03,07,2024,10,20,30,40,50,12,3\n"
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(no_header)):
|
|
result = fetch_megamillions_tx()
|
|
|
|
assert result["added"] == 1
|
|
|
|
|
|
# ── VA Lottery mock data ──────────────────────────────────────────────────────
|
|
|
|
CASH5_VA_TEXT = (
|
|
"5/22/2026; 15,29,30,34,36\n"
|
|
"5/21/2026; 1,2,5,38,44\n"
|
|
)
|
|
|
|
MILLLIFE_VA_TEXT = (
|
|
"Results for Millionaire for Life\n"
|
|
"5/22/2026; 17,33,36,54,57; Millionaire Ball: 1\n"
|
|
"5/21/2026; 3,15,16,24,28; Millionaire Ball: 4\n"
|
|
)
|
|
|
|
BANKAMIL_VA_TEXT = (
|
|
"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"
|
|
)
|
|
|
|
|
|
# ── Cash 5 VA ─────────────────────────────────────────────────────────────────
|
|
|
|
def test_cash5_va_inserts_records(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT)):
|
|
result = fetch_cash5_va()
|
|
assert result["status"] == "success"
|
|
assert result["added"] == 2
|
|
assert result["skipped"] == 0
|
|
|
|
|
|
def test_cash5_va_skips_duplicates(tmp_db):
|
|
for _ in range(2):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT)):
|
|
result = fetch_cash5_va()
|
|
assert result["added"] == 0
|
|
assert result["skipped"] == 2
|
|
|
|
|
|
def test_cash5_va_parses_date_and_numbers(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(CASH5_VA_TEXT)):
|
|
fetch_cash5_va()
|
|
from db.models import get_game_by_name, get_draws
|
|
game = get_game_by_name("Cash 5")
|
|
draws = get_draws(game["id"], order="ASC")
|
|
assert draws[0]["draw_date"] == "2026-05-21"
|
|
assert draws[0]["numbers"] == "1,2,5,38,44"
|
|
assert draws[0]["bonus"] is None
|
|
assert draws[0]["source"] == "cash5_va"
|
|
|
|
|
|
def test_cash5_va_network_error(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("timeout")):
|
|
result = fetch_cash5_va()
|
|
assert result["status"] == "error"
|
|
assert "timeout" in result["message"]
|
|
|
|
|
|
# ── Millionaire for Life VA ───────────────────────────────────────────────────
|
|
|
|
def test_milllife_va_inserts_records(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(MILLLIFE_VA_TEXT)):
|
|
result = fetch_millionaireforlife_va()
|
|
assert result["status"] == "success"
|
|
assert result["added"] == 2
|
|
|
|
|
|
def test_milllife_va_skips_header_line(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(MILLLIFE_VA_TEXT)):
|
|
result = fetch_millionaireforlife_va()
|
|
assert result["added"] == 2 # header not counted
|
|
|
|
|
|
def test_milllife_va_parses_bonus(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(MILLLIFE_VA_TEXT)):
|
|
fetch_millionaireforlife_va()
|
|
from db.models import get_game_by_name, get_draws
|
|
game = get_game_by_name("Millionaire for Life")
|
|
draws = get_draws(game["id"], order="DESC")
|
|
assert draws[0]["bonus"] == "1"
|
|
assert draws[0]["numbers"] == "17,33,36,54,57"
|
|
|
|
|
|
def test_milllife_va_network_error(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("refused")):
|
|
result = fetch_millionaireforlife_va()
|
|
assert result["status"] == "error"
|
|
|
|
|
|
# ── Bank a Million VA ─────────────────────────────────────────────────────────
|
|
|
|
def test_bankamil_va_inserts_records(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(BANKAMIL_VA_TEXT)):
|
|
result = fetch_bankamillion_va()
|
|
assert result["status"] == "success"
|
|
assert result["added"] == 2
|
|
|
|
|
|
def test_bankamil_va_parses_bonus_ball(tmp_db):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(BANKAMIL_VA_TEXT)):
|
|
fetch_bankamillion_va()
|
|
from db.models import get_game_by_name, get_draws
|
|
game = get_game_by_name("Bank a Million")
|
|
draws = get_draws(game["id"], order="DESC")
|
|
assert draws[0]["bonus"] == "18"
|
|
assert draws[0]["numbers"] == "14,20,21,24,33,35"
|
|
|
|
|
|
def test_bankamil_va_skips_duplicates(tmp_db):
|
|
for _ in range(2):
|
|
with patch("core.fetcher.requests.get", return_value=_text_resp(BANKAMIL_VA_TEXT)):
|
|
result = fetch_bankamillion_va()
|
|
assert result["added"] == 0
|
|
assert result["skipped"] == 2
|
|
|
|
|
|
def test_bankamil_va_network_error(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("connect")):
|
|
result = fetch_bankamillion_va()
|
|
assert result["status"] == "error"
|
|
|
|
|
|
# ── fetch_all ─────────────────────────────────────────────────────────────────
|
|
|
|
def test_fetch_all_returns_six_sources(tmp_db):
|
|
with (
|
|
patch("core.fetcher.fetch_powerball_ny", return_value={"source": "powerball_ny", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_megamillions_ny", return_value={"source": "megamillions_ny", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_megamillions_tx", return_value={"source": "megamillions_tx", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_cash5_va", return_value={"source": "cash5_va", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_millionaireforlife_va", return_value={"source": "millionaireforlife_va", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_bankamillion_va", return_value={"source": "bankamillion_va", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
):
|
|
results = fetch_all()
|
|
|
|
assert len(results) == 6
|
|
sources = {r["source"] for r in results}
|
|
assert sources == {"powerball_ny", "megamillions_ny", "megamillions_tx",
|
|
"cash5_va", "millionaireforlife_va", "bankamillion_va"}
|
|
|
|
|
|
def test_fetch_all_continues_after_one_error(tmp_db):
|
|
"""If one source errors, the remaining sources still complete."""
|
|
with (
|
|
patch("core.fetcher.fetch_powerball_ny", return_value={"source": "powerball_ny", "added": 0, "skipped": 0, "status": "error", "message": "timeout"}),
|
|
patch("core.fetcher.fetch_megamillions_ny", return_value={"source": "megamillions_ny", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_megamillions_tx", return_value={"source": "megamillions_tx", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_cash5_va", return_value={"source": "cash5_va", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_millionaireforlife_va", return_value={"source": "millionaireforlife_va", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
patch("core.fetcher.fetch_bankamillion_va", return_value={"source": "bankamillion_va", "added": 2, "skipped": 0, "status": "success", "message": None}),
|
|
):
|
|
results = fetch_all()
|
|
|
|
assert len(results) == 6
|
|
by_source = {r["source"]: r for r in results}
|
|
assert by_source["powerball_ny"]["status"] == "error"
|
|
assert by_source["megamillions_ny"]["status"] == "success"
|
|
assert by_source["cash5_va"]["status"] == "success"
|
|
|
|
|
|
# ── Fetch log ─────────────────────────────────────────────────────────────────
|
|
|
|
def test_fetch_log_written_on_success(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=[_json_resp(PB_NY_RECORDS), _json_resp([])]):
|
|
fetch_powerball_ny()
|
|
|
|
log = get_last_fetch_log("powerball_ny")
|
|
assert log is not None
|
|
assert log["status"] == "success"
|
|
assert log["added"] == 2
|
|
assert log["skipped"] == 0
|
|
|
|
|
|
def test_fetch_log_written_on_error(tmp_db):
|
|
with patch("core.fetcher.requests.get", side_effect=req_lib.RequestException("fail")):
|
|
fetch_powerball_ny()
|
|
|
|
log = get_last_fetch_log("powerball_ny")
|
|
assert log is not None
|
|
assert log["status"] == "error"
|