311 lines
9.5 KiB
Python
311 lines
9.5 KiB
Python
import sqlite3
|
|
|
|
import requests as req
|
|
|
|
from app.standings import (
|
|
create_standings_table,
|
|
fetch_standings,
|
|
insert_standings,
|
|
migrate_standings_table,
|
|
refresh_standings,
|
|
truncate_standings_table,
|
|
)
|
|
|
|
SAMPLE_API_RESPONSE = {
|
|
"standings": [
|
|
{
|
|
"teamCommonName": {"default": "Bruins"},
|
|
"leagueSequence": 1,
|
|
"leagueL10Sequence": 2,
|
|
"divisionAbbrev": "ATL",
|
|
"conferenceAbbrev": "E",
|
|
"gamesPlayed": 60,
|
|
"wildcardSequence": 5,
|
|
},
|
|
{
|
|
"teamCommonName": {"default": "Maple Leafs"},
|
|
"leagueSequence": 5,
|
|
"leagueL10Sequence": 3,
|
|
"divisionAbbrev": "ATL",
|
|
"conferenceAbbrev": "E",
|
|
"gamesPlayed": 61,
|
|
"wildcardSequence": 8,
|
|
},
|
|
]
|
|
}
|
|
|
|
|
|
class TestFetchStandings:
|
|
def test_returns_parsed_standings(self, mocker):
|
|
mock_get = mocker.patch("app.standings.requests.get")
|
|
mock_get.return_value.json.return_value = SAMPLE_API_RESPONSE
|
|
|
|
result = fetch_standings()
|
|
|
|
assert len(result) == 2
|
|
assert result[0] == {
|
|
"team_common_name": "Bruins",
|
|
"league_sequence": 1,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 60,
|
|
"wildcard_sequence": 5,
|
|
}
|
|
assert result[1]["team_common_name"] == "Maple Leafs"
|
|
|
|
def test_returns_none_on_request_exception(self, mocker):
|
|
mocker.patch(
|
|
"app.standings.requests.get", side_effect=req.RequestException("err")
|
|
)
|
|
|
|
result = fetch_standings()
|
|
|
|
assert result is None
|
|
|
|
def test_returns_none_on_bad_status(self, mocker):
|
|
mock_get = mocker.patch("app.standings.requests.get")
|
|
mock_get.return_value.raise_for_status.side_effect = req.HTTPError("503")
|
|
|
|
result = fetch_standings()
|
|
|
|
assert result is None
|
|
|
|
def test_returns_empty_list_when_no_standings_key(self, mocker):
|
|
mock_get = mocker.patch("app.standings.requests.get")
|
|
mock_get.return_value.json.return_value = {}
|
|
|
|
result = fetch_standings()
|
|
|
|
assert result == []
|
|
|
|
|
|
class TestCreateStandingsTable:
|
|
def test_creates_table(self, tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
create_standings_table(conn)
|
|
|
|
row = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='standings'"
|
|
).fetchone()
|
|
|
|
assert row is not None
|
|
conn.close()
|
|
|
|
def test_is_idempotent(self, tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
create_standings_table(conn)
|
|
create_standings_table(conn) # should not raise
|
|
|
|
conn.close()
|
|
|
|
|
|
class TestTruncateStandingsTable:
|
|
def test_removes_all_rows(self, tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
create_standings_table(conn)
|
|
insert_standings(
|
|
conn,
|
|
[
|
|
{
|
|
"team_common_name": "Bruins",
|
|
"league_sequence": 1,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 60,
|
|
"wildcard_sequence": 5,
|
|
}
|
|
],
|
|
)
|
|
|
|
truncate_standings_table(conn)
|
|
|
|
count = conn.execute("SELECT COUNT(*) FROM standings").fetchone()[0]
|
|
assert count == 0
|
|
conn.close()
|
|
|
|
|
|
class TestInsertStandings:
|
|
def test_inserts_all_rows(self, tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
create_standings_table(conn)
|
|
|
|
data = [
|
|
{
|
|
"team_common_name": "Bruins",
|
|
"league_sequence": 1,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 60,
|
|
"wildcard_sequence": 5,
|
|
},
|
|
{
|
|
"team_common_name": "Maple Leafs",
|
|
"league_sequence": 5,
|
|
"league_l10_sequence": 3,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 61,
|
|
"wildcard_sequence": 8,
|
|
},
|
|
]
|
|
insert_standings(conn, data)
|
|
|
|
count = conn.execute("SELECT COUNT(*) FROM standings").fetchone()[0]
|
|
assert count == 2
|
|
conn.close()
|
|
|
|
def test_data_is_queryable_after_insert(self, tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
create_standings_table(conn)
|
|
insert_standings(
|
|
conn,
|
|
[
|
|
{
|
|
"team_common_name": "Bruins",
|
|
"league_sequence": 1,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 60,
|
|
"wildcard_sequence": 5,
|
|
}
|
|
],
|
|
)
|
|
|
|
row = conn.execute(
|
|
"SELECT league_sequence FROM standings WHERE team_common_name = ?",
|
|
("Bruins",),
|
|
).fetchone()
|
|
|
|
assert row[0] == 1
|
|
conn.close()
|
|
|
|
|
|
class TestRefreshStandings:
|
|
def test_populates_db_from_api(self, mocker, tmp_path):
|
|
standings = [
|
|
{
|
|
"team_common_name": "Bruins",
|
|
"league_sequence": 1,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 60,
|
|
"wildcard_sequence": 5,
|
|
}
|
|
]
|
|
mocker.patch("app.standings.fetch_standings", return_value=standings)
|
|
mocker.patch("app.standings.DB_PATH", str(tmp_path / "test.db"))
|
|
|
|
refresh_standings()
|
|
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
count = conn.execute("SELECT COUNT(*) FROM standings").fetchone()[0]
|
|
conn.close()
|
|
assert count == 1
|
|
|
|
def test_clears_old_data_before_inserting(self, mocker, tmp_path):
|
|
db_path = str(tmp_path / "test.db")
|
|
mocker.patch("app.standings.DB_PATH", db_path)
|
|
|
|
first = [
|
|
{
|
|
"team_common_name": "Bruins",
|
|
"league_sequence": 1,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 60,
|
|
"wildcard_sequence": 5,
|
|
}
|
|
]
|
|
mocker.patch("app.standings.fetch_standings", return_value=first)
|
|
refresh_standings()
|
|
|
|
second = [
|
|
{
|
|
"team_common_name": "Oilers",
|
|
"league_sequence": 3,
|
|
"league_l10_sequence": 1,
|
|
"division_abbrev": "PAC",
|
|
"conference_abbrev": "W",
|
|
"games_played": 62,
|
|
"wildcard_sequence": 3,
|
|
},
|
|
{
|
|
"team_common_name": "Jets",
|
|
"league_sequence": 4,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "CEN",
|
|
"conference_abbrev": "W",
|
|
"games_played": 61,
|
|
"wildcard_sequence": 4,
|
|
},
|
|
]
|
|
mocker.patch("app.standings.fetch_standings", return_value=second)
|
|
refresh_standings()
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
count = conn.execute("SELECT COUNT(*) FROM standings").fetchone()[0]
|
|
conn.close()
|
|
assert count == 2
|
|
|
|
def test_does_not_insert_when_fetch_fails(self, mocker, tmp_path):
|
|
db_path = str(tmp_path / "test.db")
|
|
mocker.patch("app.standings.DB_PATH", db_path)
|
|
|
|
# Seed with existing data before the failed refresh
|
|
seed = [
|
|
{
|
|
"team_common_name": "Bruins",
|
|
"league_sequence": 1,
|
|
"league_l10_sequence": 2,
|
|
"division_abbrev": "ATL",
|
|
"conference_abbrev": "E",
|
|
"games_played": 60,
|
|
"wildcard_sequence": 5,
|
|
}
|
|
]
|
|
mocker.patch("app.standings.fetch_standings", return_value=seed)
|
|
refresh_standings()
|
|
|
|
# Now simulate a fetch failure — existing data must be preserved
|
|
mocker.patch("app.standings.fetch_standings", return_value=None)
|
|
refresh_standings()
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
count = conn.execute("SELECT COUNT(*) FROM standings").fetchone()[0]
|
|
conn.close()
|
|
assert count == 1
|
|
|
|
|
|
class TestMigrateStandingsTable:
|
|
def test_adds_missing_columns_to_existing_table(self, tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
conn.execute(
|
|
"CREATE TABLE standings "
|
|
"(team_common_name TEXT, league_sequence INTEGER, league_l10_sequence INTEGER)"
|
|
)
|
|
conn.commit()
|
|
|
|
migrate_standings_table(conn)
|
|
|
|
cols = [
|
|
row[1] for row in conn.execute("PRAGMA table_info(standings)").fetchall()
|
|
]
|
|
assert "division_abbrev" in cols
|
|
assert "conference_abbrev" in cols
|
|
assert "games_played" in cols
|
|
assert "wildcard_sequence" in cols
|
|
conn.close()
|
|
|
|
def test_is_idempotent(self, tmp_path):
|
|
conn = sqlite3.connect(str(tmp_path / "test.db"))
|
|
create_standings_table(conn)
|
|
migrate_standings_table(conn)
|
|
migrate_standings_table(conn) # must not raise
|
|
conn.close()
|