386 lines
13 KiB
Python
386 lines
13 KiB
Python
"""
|
|
tests/test_models.py
|
|
--------------------
|
|
Tests for db/models.py covering all CRUD functions:
|
|
Games — get_all_games, get_game_by_name, get_game_by_id,
|
|
set_game_active, add_custom_game
|
|
Draws — insert_draw, draw_exists, get_draws, get_last_draw,
|
|
get_draw_count, get_all_draws_numbers
|
|
Predict — insert_prediction, get_predictions
|
|
FetchLog — insert_fetch_log, get_last_fetch_log,
|
|
get_fetch_logs, get_last_fetch_per_source
|
|
"""
|
|
|
|
import pytest
|
|
from db.models import (
|
|
# Games
|
|
get_all_games, get_game_by_name, get_game_by_id,
|
|
set_game_active, add_custom_game,
|
|
# Draws
|
|
insert_draw, draw_exists, get_draws, get_last_draw,
|
|
get_draw_count, get_all_draws_numbers,
|
|
# Predictions
|
|
insert_prediction, get_predictions,
|
|
# Fetch log
|
|
insert_fetch_log, get_last_fetch_log,
|
|
get_fetch_logs, get_last_fetch_per_source,
|
|
)
|
|
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────
|
|
|
|
def _pb_id(tmp_db):
|
|
"""Return Powerball game id."""
|
|
return get_game_by_name("Powerball")["id"]
|
|
|
|
def _mm_id(tmp_db):
|
|
"""Return Mega Millions game id."""
|
|
return get_game_by_name("Mega Millions")["id"]
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# GAMES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def test_get_all_games_returns_five(tmp_db):
|
|
games = get_all_games()
|
|
assert len(games) == 5
|
|
|
|
|
|
def test_get_all_games_active_only(tmp_db):
|
|
"""active_only=True should return all 5 seeded games (all active by default)."""
|
|
games = get_all_games(active_only=True)
|
|
assert len(games) == 5
|
|
|
|
|
|
def test_get_game_by_name_powerball(tmp_db):
|
|
row = get_game_by_name("Powerball")
|
|
assert row is not None
|
|
assert row["name"] == "Powerball"
|
|
|
|
|
|
def test_get_game_by_name_not_found(tmp_db):
|
|
row = get_game_by_name("NonExistentGame")
|
|
assert row is None
|
|
|
|
|
|
def test_get_game_by_id(tmp_db):
|
|
pb = get_game_by_name("Powerball")
|
|
row = get_game_by_id(pb["id"])
|
|
assert row is not None
|
|
assert row["name"] == "Powerball"
|
|
|
|
|
|
def test_get_game_by_id_not_found(tmp_db):
|
|
row = get_game_by_id(9999)
|
|
assert row is None
|
|
|
|
|
|
def test_set_game_active_disable(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
set_game_active(pb_id, False)
|
|
row = get_game_by_id(pb_id)
|
|
assert row["active"] == 0
|
|
|
|
|
|
def test_set_game_active_enable(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
set_game_active(pb_id, False)
|
|
set_game_active(pb_id, True)
|
|
row = get_game_by_id(pb_id)
|
|
assert row["active"] == 1
|
|
|
|
|
|
def test_active_only_filters_disabled(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
set_game_active(pb_id, False)
|
|
active_games = get_all_games(active_only=True)
|
|
names = [g["name"] for g in active_games]
|
|
assert "Powerball" not in names
|
|
assert "Mega Millions" in names
|
|
|
|
|
|
def test_add_custom_game(tmp_db):
|
|
new_id = add_custom_game("Pick 3", 3, 9, bonus_count=0, bonus_max=0)
|
|
assert new_id is not None
|
|
row = get_game_by_id(new_id)
|
|
assert row["name"] == "Pick 3"
|
|
assert row["main_count"] == 3
|
|
assert row["main_max"] == 9
|
|
assert row["active"] == 1
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# DRAWS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def test_insert_draw_returns_inserted(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
result = insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65],
|
|
bonus=8, source="test")
|
|
assert result == "inserted"
|
|
|
|
|
|
def test_insert_draw_duplicate_returns_skipped(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
|
|
result = insert_draw(pb_id, "2024-01-06", [1, 2, 3, 4, 5], bonus=9)
|
|
assert result == "skipped"
|
|
|
|
|
|
def test_insert_draw_accepts_string_numbers(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
result = insert_draw(pb_id, "2024-02-01", "10,20,30,40,50", bonus="5")
|
|
assert result == "inserted"
|
|
|
|
|
|
def test_draw_exists_true(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
|
|
assert draw_exists(pb_id, "2024-01-06") is True
|
|
|
|
|
|
def test_draw_exists_false(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
assert draw_exists(pb_id, "2099-12-31") is False
|
|
|
|
|
|
def test_get_draws_returns_correct_game(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
mm_id = _mm_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
|
|
insert_draw(mm_id, "2024-01-05", [3, 17, 28, 41, 60], bonus=12)
|
|
|
|
pb_draws = get_draws(pb_id)
|
|
mm_draws = get_draws(mm_id)
|
|
assert len(pb_draws) == 1
|
|
assert len(mm_draws) == 1
|
|
|
|
|
|
def test_get_draws_order_desc(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
|
|
insert_draw(pb_id, "2024-01-10", [6, 7, 8, 9, 10], bonus=2)
|
|
insert_draw(pb_id, "2024-01-20", [11, 12, 13, 14, 15], bonus=3)
|
|
|
|
draws = get_draws(pb_id, order="DESC")
|
|
dates = [d["draw_date"] for d in draws]
|
|
assert dates == sorted(dates, reverse=True)
|
|
|
|
|
|
def test_get_draws_order_asc(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
|
|
insert_draw(pb_id, "2024-01-10", [6, 7, 8, 9, 10], bonus=2)
|
|
|
|
draws = get_draws(pb_id, order="ASC")
|
|
dates = [d["draw_date"] for d in draws]
|
|
assert dates == sorted(dates)
|
|
|
|
|
|
def test_get_draws_with_limit(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
for i in range(1, 6):
|
|
insert_draw(pb_id, f"2024-01-{i:02d}", [i, i+1, i+2, i+3, i+4], bonus=i)
|
|
|
|
draws = get_draws(pb_id, limit=3)
|
|
assert len(draws) == 3
|
|
|
|
|
|
def test_get_draws_date_filter(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
|
|
insert_draw(pb_id, "2024-06-15", [6, 7, 8, 9, 10], bonus=2)
|
|
insert_draw(pb_id, "2024-12-31", [11, 12, 13, 14, 15], bonus=3)
|
|
|
|
draws = get_draws(pb_id, date_from="2024-06-01", date_to="2024-12-01")
|
|
assert len(draws) == 1
|
|
assert draws[0]["draw_date"] == "2024-06-15"
|
|
|
|
|
|
def test_get_last_draw(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
|
|
insert_draw(pb_id, "2024-01-20", [6, 7, 8, 9, 10], bonus=2)
|
|
|
|
last = get_last_draw(pb_id)
|
|
assert last["draw_date"] == "2024-01-20"
|
|
|
|
|
|
def test_get_last_draw_empty(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
assert get_last_draw(pb_id) is None
|
|
|
|
|
|
def test_get_draw_count(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
assert get_draw_count(pb_id) == 0
|
|
insert_draw(pb_id, "2024-01-01", [1, 2, 3, 4, 5], bonus=1)
|
|
insert_draw(pb_id, "2024-01-10", [6, 7, 8, 9, 10], bonus=2)
|
|
assert get_draw_count(pb_id) == 2
|
|
|
|
|
|
def test_get_all_draws_numbers_parses_correctly(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
insert_draw(pb_id, "2024-01-06", [5, 12, 33, 47, 65], bonus=8)
|
|
insert_draw(pb_id, "2024-01-10", [2, 19, 30, 44, 58], bonus=14)
|
|
|
|
draws = get_all_draws_numbers(pb_id)
|
|
assert len(draws) == 2
|
|
|
|
# Oldest first (ASC)
|
|
assert draws[0]["draw_date"] == "2024-01-06"
|
|
assert draws[0]["numbers"] == [5, 12, 33, 47, 65]
|
|
assert draws[0]["bonus"] == 8
|
|
|
|
assert draws[1]["draw_date"] == "2024-01-10"
|
|
assert draws[1]["numbers"] == [2, 19, 30, 44, 58]
|
|
assert draws[1]["bonus"] == 14
|
|
|
|
|
|
def test_get_all_draws_numbers_empty(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
draws = get_all_draws_numbers(pb_id)
|
|
assert draws == []
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# PREDICTIONS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def test_insert_prediction_returns_id(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
new_id = insert_prediction(pb_id, "Hot Numbers", [7, 14, 22, 36, 55], bonus=18)
|
|
assert isinstance(new_id, int)
|
|
assert new_id > 0
|
|
|
|
|
|
def test_insert_prediction_list_and_string(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
id1 = insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5], bonus=6)
|
|
id2 = insert_prediction(pb_id, "Due Numbers", "10,20,30,40,50", bonus=None)
|
|
assert id1 != id2
|
|
|
|
|
|
def test_get_predictions_by_game(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
mm_id = _mm_id(tmp_db)
|
|
insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5], bonus=6)
|
|
insert_prediction(mm_id, "Monte Carlo", [10, 20, 30, 40, 50], bonus=7)
|
|
|
|
pb_preds = get_predictions(game_id=pb_id)
|
|
mm_preds = get_predictions(game_id=mm_id)
|
|
assert len(pb_preds) == 1
|
|
assert len(mm_preds) == 1
|
|
assert pb_preds[0]["strategy"] == "Hot Numbers"
|
|
assert mm_preds[0]["strategy"] == "Monte Carlo"
|
|
|
|
|
|
def test_get_predictions_all_games(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
mm_id = _mm_id(tmp_db)
|
|
insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5])
|
|
insert_prediction(mm_id, "Due Numbers", [6, 7, 8, 9, 10])
|
|
|
|
all_preds = get_predictions()
|
|
assert len(all_preds) == 2
|
|
|
|
|
|
def test_get_predictions_respects_limit(tmp_db):
|
|
pb_id = _pb_id(tmp_db)
|
|
for i in range(10):
|
|
insert_prediction(pb_id, "Weighted Random", [i+1, i+2, i+3, i+4, i+5])
|
|
|
|
preds = get_predictions(game_id=pb_id, limit=5)
|
|
assert len(preds) == 5
|
|
|
|
|
|
def test_get_predictions_newest_first(tmp_db):
|
|
"""
|
|
Verify get_predictions returns DESC order by id (newest first).
|
|
We can't rely on created_at within the same second in SQLite,
|
|
so compare by id: higher id = inserted later = should appear first.
|
|
"""
|
|
pb_id = _pb_id(tmp_db)
|
|
id1 = insert_prediction(pb_id, "Hot Numbers", [1, 2, 3, 4, 5])
|
|
id2 = insert_prediction(pb_id, "Due Numbers", [6, 7, 8, 9, 10])
|
|
|
|
preds = get_predictions(game_id=pb_id)
|
|
ids = [p["id"] for p in preds]
|
|
# Should be descending: id2 before id1
|
|
assert ids.index(id2) < ids.index(id1), (
|
|
"Predictions not returned newest-first by id"
|
|
)
|
|
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# FETCH LOG
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def test_insert_fetch_log_returns_id(tmp_db):
|
|
new_id = insert_fetch_log("NY Powerball", added=10, skipped=2)
|
|
assert isinstance(new_id, int)
|
|
assert new_id > 0
|
|
|
|
|
|
def test_insert_fetch_log_error_status(tmp_db):
|
|
new_id = insert_fetch_log("TX Mega Millions", added=0, skipped=0,
|
|
status="error", message="Connection timeout")
|
|
assert new_id > 0
|
|
|
|
|
|
def test_get_last_fetch_log_any(tmp_db):
|
|
"""
|
|
get_last_fetch_log() with no source filter returns the most recent entry.
|
|
Within the same second, SQLite order is undefined — use id to verify.
|
|
"""
|
|
id1 = insert_fetch_log("NY Powerball", added=5, skipped=1)
|
|
id2 = insert_fetch_log("NY Mega Millions", added=3, skipped=0)
|
|
row = get_last_fetch_log()
|
|
assert row is not None
|
|
# Most recent = highest id
|
|
assert row["id"] == max(id1, id2)
|
|
|
|
|
|
def test_get_last_fetch_log_by_source(tmp_db):
|
|
insert_fetch_log("NY Powerball", added=5, skipped=1)
|
|
insert_fetch_log("NY Mega Millions", added=3, skipped=0)
|
|
row = get_last_fetch_log(source="NY Powerball")
|
|
assert row is not None
|
|
assert row["source"] == "NY Powerball"
|
|
assert row["added"] == 5
|
|
assert row["skipped"] == 1
|
|
|
|
|
|
def test_get_last_fetch_log_none_when_empty(tmp_db):
|
|
row = get_last_fetch_log()
|
|
assert row is None
|
|
|
|
|
|
def test_get_fetch_logs_limit(tmp_db):
|
|
for i in range(10):
|
|
insert_fetch_log(f"source_{i}", added=i, skipped=0)
|
|
logs = get_fetch_logs(limit=5)
|
|
assert len(logs) == 5
|
|
|
|
|
|
def test_get_last_fetch_per_source(tmp_db):
|
|
insert_fetch_log("NY Powerball", added=5, skipped=1)
|
|
insert_fetch_log("NY Mega Millions", added=3, skipped=0)
|
|
insert_fetch_log("TX Mega Millions", added=2, skipped=1)
|
|
# Second Powerball fetch — should be the "latest" for that source
|
|
insert_fetch_log("NY Powerball", added=1, skipped=4)
|
|
|
|
per_source = get_last_fetch_per_source()
|
|
assert "NY Powerball" in per_source
|
|
assert "NY Mega Millions" in per_source
|
|
assert "TX Mega Millions" in per_source
|
|
|
|
# Latest Powerball fetch had added=1
|
|
assert per_source["NY Powerball"]["added"] == 1
|
|
|
|
|
|
def test_get_last_fetch_per_source_empty(tmp_db):
|
|
per_source = get_last_fetch_per_source()
|
|
assert per_source == {}
|