96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
from datetime import datetime
|
|
import pytest
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from httpx import ConnectError
|
|
|
|
from conftest import MegaMillionsDraw
|
|
from util.drawing import get_mega_millions_object
|
|
from util.scrape import scrape_game_history
|
|
|
|
|
|
def test_create_mega_millions_draw(db_session) -> None:
|
|
"""add a record"""
|
|
date = datetime(2010, 2, 2).date() # 6 7 26 27 49 9 4x
|
|
new_record = MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26, main_ball4=27,
|
|
main_ball5=49, mega_ball=9, megaplier=4)
|
|
db_session.add(new_record)
|
|
db_session.commit()
|
|
|
|
# verify the record exists
|
|
record = db_session.query(MegaMillionsDraw).filter_by(draw_date=date).first()
|
|
assert record is not None
|
|
assert record == new_record
|
|
|
|
|
|
def test_mega_millions_draw_isolation(db_session) -> None:
|
|
"""query for all records in the database"""
|
|
draws = db_session.query(MegaMillionsDraw).all()
|
|
assert len(draws) == 0
|
|
|
|
|
|
def test_mega_millions_draw_by_date_blank(db_session) -> None:
|
|
"""query for all mega millions records with a blank date"""
|
|
blank_results = db_session.query(MegaMillionsDraw).filter_by(draw_date='').all()
|
|
assert len(blank_results) == 0
|
|
|
|
|
|
def test_mega_millions_draw_integrity(db_session) -> None:
|
|
"""ensure that duplicate records cannot be inserted"""
|
|
date = datetime(2010, 2, 2).date() # 6 7 26 27 49 9 4x
|
|
new_record = MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26, main_ball4=27,
|
|
main_ball5=49, mega_ball=9, megaplier=4)
|
|
|
|
try:
|
|
# insert a new record
|
|
db_session.add(new_record)
|
|
db_session.commit()
|
|
|
|
# try inserting the same record again
|
|
db_session.add(new_record)
|
|
db_session.commit()
|
|
except IntegrityError:
|
|
assert True
|
|
|
|
|
|
def test_mega_millions_draw_eq_repr_and_str(db_session) -> None:
|
|
"""test the __eq__, __repr__ and __str__ methods of MegaMillionsDraw"""
|
|
date = datetime(2010, 2, 2).date()
|
|
new_record = MegaMillionsDraw(draw_date=date, main_ball1=6, main_ball2=7, main_ball3=26, main_ball4=27,
|
|
main_ball5=49, mega_ball=9, megaplier=4)
|
|
|
|
# insert a new record
|
|
db_session.add(new_record)
|
|
db_session.commit()
|
|
|
|
# query the record
|
|
record = db_session.query(MegaMillionsDraw).filter_by(draw_date=date).first()
|
|
|
|
# compare the str and repr values of each using __eq__
|
|
assert str(record) == str(new_record)
|
|
assert repr(record) == repr(new_record)
|
|
|
|
|
|
def test_mega_millions_scrape_and_insert(db_session) -> None:
|
|
"""test scraping game data and inserting it into the database"""
|
|
# scrape the current year history for mega millions
|
|
try:
|
|
raw_data = scrape_game_history(game='mega-millions')
|
|
|
|
print(len(raw_data))
|
|
# loop over the records returned
|
|
for key, value in raw_data.items():
|
|
# convert each record to a MegaMillionsDraw object
|
|
record = get_mega_millions_object(draw_date=key, draw_result=value)
|
|
# add the record to the database
|
|
db_session.add(record)
|
|
|
|
# commit the changes
|
|
db_session.commit()
|
|
|
|
# get all the records from the table
|
|
records = db_session.query(MegaMillionsDraw).all()
|
|
assert len(records) == len(raw_data)
|
|
except ConnectError as ce:
|
|
assert pytest.fail(f"Target scrape site is unavailable: {ce}")
|