145 lines
4.5 KiB
Python
145 lines
4.5 KiB
Python
"""
|
|
tests/test_database.py
|
|
----------------------
|
|
Tests for db/database.py:
|
|
- init_db() creates all required tables
|
|
- Default games are seeded (Powerball, Mega Millions)
|
|
- init_db() is idempotent (safe to call multiple times)
|
|
- get_db_stats() returns correct draw counts
|
|
"""
|
|
|
|
import pytest
|
|
from db.database import init_db, get_db_stats, get_connection
|
|
|
|
|
|
def test_tables_created(tmp_db):
|
|
"""All 4 tables must exist after init_db()."""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
|
|
tables = {row["name"] for row in cursor.fetchall()}
|
|
conn.close()
|
|
|
|
assert "games" in tables, "Missing table: games"
|
|
assert "draws" in tables, "Missing table: draws"
|
|
assert "predictions" in tables, "Missing table: predictions"
|
|
assert "fetch_log" in tables, "Missing table: fetch_log"
|
|
|
|
|
|
def test_default_games_seeded(tmp_db):
|
|
"""Powerball and Mega Millions must be seeded after init_db()."""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT name FROM games ORDER BY name")
|
|
names = [row["name"] for row in cursor.fetchall()]
|
|
conn.close()
|
|
|
|
assert "Powerball" in names, "Powerball not seeded"
|
|
assert "Mega Millions" in names, "Mega Millions not seeded"
|
|
|
|
|
|
def test_powerball_config(tmp_db):
|
|
"""Powerball config must match spec: 5 balls 1-69, bonus 1-26."""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM games WHERE name = 'Powerball'")
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
|
|
assert row is not None
|
|
assert row["main_count"] == 5
|
|
assert row["main_max"] == 69
|
|
assert row["bonus_count"] == 1
|
|
assert row["bonus_max"] == 26
|
|
assert row["active"] == 1
|
|
|
|
|
|
def test_megamillions_config(tmp_db):
|
|
"""Mega Millions config must match spec: 5 balls 1-70, bonus 1-25."""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT * FROM games WHERE name = 'Mega Millions'")
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
|
|
assert row is not None
|
|
assert row["main_count"] == 5
|
|
assert row["main_max"] == 70
|
|
assert row["bonus_count"] == 1
|
|
assert row["bonus_max"] == 25
|
|
assert row["active"] == 1
|
|
|
|
|
|
def test_init_db_idempotent(tmp_db):
|
|
"""Calling init_db() multiple times must not raise or duplicate games."""
|
|
init_db()
|
|
init_db()
|
|
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT COUNT(*) as cnt FROM games")
|
|
count = cursor.fetchone()["cnt"]
|
|
conn.close()
|
|
|
|
assert count == 5, f"Expected 5 seeded games, got {count} — possible duplicate seed"
|
|
|
|
|
|
def test_unique_index_on_draws(tmp_db):
|
|
"""Unique index on (game_id, draw_date) must prevent duplicate inserts."""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Get Powerball id
|
|
cursor.execute("SELECT id FROM games WHERE name = 'Powerball'")
|
|
pb_id = cursor.fetchone()["id"]
|
|
|
|
# First insert — should succeed
|
|
cursor.execute("""
|
|
INSERT INTO draws (game_id, draw_date, numbers, bonus, source)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (pb_id, "2024-03-01", "5,12,33,47,65", "8", "test"))
|
|
conn.commit()
|
|
|
|
# Duplicate insert — must raise IntegrityError
|
|
import sqlite3
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
cursor.execute("""
|
|
INSERT INTO draws (game_id, draw_date, numbers, bonus, source)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (pb_id, "2024-03-01", "1,2,3,4,5", "9", "test"))
|
|
conn.commit()
|
|
|
|
conn.close()
|
|
|
|
|
|
def test_get_db_stats_empty(tmp_db):
|
|
"""get_db_stats() returns 0 draw count for all games when DB is empty."""
|
|
stats = get_db_stats()
|
|
assert "Powerball" in stats
|
|
assert "Mega Millions" in stats
|
|
assert stats["Powerball"] == 0
|
|
assert stats["Mega Millions"] == 0
|
|
|
|
|
|
def test_get_db_stats_with_draws(tmp_db):
|
|
"""get_db_stats() returns correct count after inserting draws."""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT id FROM games WHERE name = 'Powerball'")
|
|
pb_id = cursor.fetchone()["id"]
|
|
|
|
cursor.execute("""
|
|
INSERT INTO draws (game_id, draw_date, numbers, bonus)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (pb_id, "2024-01-06", "5,12,33,47,65", "8"))
|
|
cursor.execute("""
|
|
INSERT INTO draws (game_id, draw_date, numbers, bonus)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (pb_id, "2024-01-10", "2,19,30,44,58", "14"))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
stats = get_db_stats()
|
|
assert stats["Powerball"] == 2
|
|
assert stats["Mega Millions"] == 0
|