""" 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_two(tmp_db): games = get_all_games() assert len(games) == 2 def test_get_all_games_active_only(tmp_db): """active_only=True should return 2 by default (both active).""" games = get_all_games(active_only=True) assert len(games) == 2 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 == {}