test: Add unit and integration tests for backend

master
John Montagu, the 4th Earl of Sandvich 2025-05-06 10:16:54 -07:00
parent 3f15a89d57
commit f3f0189da0
Signed by: sandvich
GPG Key ID: 9A39BE37E602B22D
9 changed files with 664 additions and 111 deletions

View File

@ -1,7 +0,0 @@
import unittest
import flask_testing
if __name__ == "__main__":
suite = unittest.TestLoader().discover("tests")
unittest.TextTestRunner(verbosity=1).run(suite)

View File

@ -0,0 +1,126 @@
import datetime
from flask import Blueprint
import pytest
import app_db
from unittest.mock import patch
from models.auth_session import AuthSession
from models.event import Event
from models.player import Player
from models.player_event import PlayerEvent
from models.player_team import PlayerTeam
from models.player_team_role import PlayerTeamRole
from models.team import Team
from models.team_integration import TeamLogsTfIntegration
@pytest.fixture()
def app():
flask_app = app_db.create_app()
flask_app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
db = app_db.db
import login
import schedule
import team
import user
import events
import match
api = Blueprint("api", __name__, url_prefix="/api")
api.register_blueprint(login.api_login)
api.register_blueprint(schedule.api_schedule)
api.register_blueprint(team.api_team)
api.register_blueprint(user.api_user)
api.register_blueprint(events.api_events)
api.register_blueprint(match.api_match)
flask_app.register_blueprint(api)
db.init_app(flask_app)
with flask_app.app_context():
db.create_all()
populate_db(db)
yield flask_app
db.session.remove()
db.drop_all()
@pytest.fixture()
def client(app):
return app.test_client()
@pytest.fixture
def mock_get():
with patch("requests.get") as _mock_get:
yield _mock_get
@pytest.fixture
def mock_post():
with patch("requests.post") as _mock_post:
yield _mock_post
@pytest.fixture
def headers():
return {
"Content-Type": "application/json",
"Accept": "application/json",
"Cookie": "auth=test_key",
}
def populate_db(db):
player = Player(steam_id=76561198248436608, username="pyro from csgo")
team = Team(team_name="Team Pepeja", tz_timezone="America/New_York", minute_offset=30)
db.session.add(player)
db.session.add(team)
db.session.flush()
player_team = PlayerTeam(
player_id=player.steam_id,
team_id=team.id,
team_role=PlayerTeam.TeamRole.Player,
is_team_leader=True,
)
logs_tf_integration = TeamLogsTfIntegration(
team_id=team.id,
min_team_member_count=2,
)
db.session.add(player_team)
db.session.add(logs_tf_integration)
auth_session = AuthSession(
player_id=player.steam_id,
key="test_key",
)
db.session.add(auth_session)
event = Event(
team_id=team.id,
name="Test event",
description="Test description",
start_time=datetime.datetime.now(datetime.timezone.utc),
)
db.session.add(event)
db.session.flush()
player_event = PlayerEvent(
event_id=event.id,
player_id=76561198248436608,
player_team_role_id=1,
)
ptr = PlayerTeamRole(
player_team_id=player_team.id,
role=PlayerTeamRole.Role.PocketSoldier,
is_main=True,
)
db.session.add(player_event)
db.session.add(ptr)
db.session.commit()

View File

@ -0,0 +1,76 @@
import datetime
from models.event import Event
from app_db import db
from models.player_event import PlayerEvent
def test_get_event(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/events/1",
headers=headers)
assert response.json["name"] == "Test event"
def test_get_team_events(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/events/team/id/1",
headers=headers)
assert len(response.json) == 1
def test_create_event(client, headers):
client.set_cookie("auth", "test_key")
response = client.post(
"/api/events/team/id/1",
json={
"name": "New Event",
"description": "Test event description",
"startTime": 0,
"playerRoles": [
{
"player": {
"steamId": "76561198248436608",
"username": "pyro from csgo",
},
"role": {
"role": "Pyro",
"isMain": False,
},
}
]
},
headers=headers)
assert response.json["name"] == "New Event"
def test_update_event(client, headers):
client.set_cookie("auth", "test_key")
response = client.patch(
"/api/events/1",
json={
"name": "Updated Event",
"description": "Updated description",
"startTime": 0,
"playerRoles": [],
},
headers=headers)
print(response)
assert response.json["name"] == "Updated Event"
def test_delete_event(client, headers):
client.set_cookie("auth", "test_key")
response = client.delete(
"/api/events/1",
headers=headers)
assert db.session.query(Event).where(Event.id == 1).one_or_none() is None
def test_get_maximum_matching_1_player(app):
event = db.session.query(Event).first()
assert event.get_maximum_matching() == 1
def test_get_maximum_matching_no_players(app):
event = Event(
team_id=1,
name="New Event",
start_time=datetime.datetime.now(datetime.timezone.utc),
)
assert event.get_maximum_matching() == 0

View File

@ -1,104 +0,0 @@
from datetime import timedelta
from base_test_case import BaseTestCase
from app_db import db
from models.match import Match, RawLogDetails
from models.player import Player
from models.player_match import PlayerMatch
from models.player_team import PlayerTeam
from models.team_match import TeamMatch
class TestLogstfJob(BaseTestCase):
def populate_db(self):
from app_db import db
super().populate_db()
wesker_u = Player(steam_id=76561198024482308, username="Wesker U")
wesker_u_pt = PlayerTeam(
player_id=wesker_u.steam_id,
team_id=1,
team_role=PlayerTeam.TeamRole.Player,
is_team_leader=True,
)
db.session.add(wesker_u)
db.session.add(wesker_u_pt)
db.session.commit()
def test_get_common_teams(self):
from jobs.fetch_logstf import get_common_teams
rows = get_common_teams([76561198248436608, 76561198024482308])
assert len(rows) == 1
assert rows == [(1, 2, 2)]
def test_transform(self):
from jobs.fetch_logstf import transform
details: RawLogDetails = {
"players": {
"[U:1:288170880]": {
"team": "Red",
"kills": 1,
"deaths": 2,
"assists": 3,
"dmg": 4,
"dt": 5,
},
"[U:1:64216580]": {
"team": "Red",
"kills": 6,
"deaths": 7,
"assists": 8,
"dmg": 9,
"dt": 10,
},
"[U:1:64216581]": {
"team": "Blue",
"kills": 6,
"deaths": 7,
"assists": 8,
"dmg": 9,
"dt": 10,
},
},
"info": {
"title": "I LOVE DUSTBOWL",
"map": "cp_dustbowl",
"date": 1614547200,
},
"teams": {
"Blue": {
"score": 1
},
"Red": {
"score": 2
}
},
"length": 3025,
}
for instance in transform(1, details):
db.session.add(instance)
db.session.commit()
assert db.session.query(Player).count() == 2
assert db.session.query(PlayerMatch).count() == 2
assert db.session.query(TeamMatch).count() == 1
assert db.session.query(Match).count() == 1
assert db.session.query(PlayerTeam).count() == 2
player_team = db.session.query(PlayerTeam).first()
assert player_team is not None
print(player_team.playtime)
assert player_team.playtime == 3025
def test_steam3_to_steam64(self):
from jobs.fetch_logstf import steam3_to_steam64
assert steam3_to_steam64("[U:1:123456]") == 76561197960265728 + 123456
def test_steam64_to_steam3(self):
from jobs.fetch_logstf import steam64_to_steam3
assert steam64_to_steam3(76561197960265728 + 123456) == "[U:1:123456]"

View File

@ -0,0 +1,277 @@
import datetime
import pytest
import app_db
from models.auth_session import AuthSession
from models.match import RawLogDetails
from models.player import Player
from models.player_team import PlayerTeam
from models.player_team_availability import PlayerTeamAvailability
from models.team import Team
from unittest.mock import Mock, patch
from requests.models import Response
from requests import Request
from models.team_match import TeamMatch
## Integration test 1: team creation
def test_create_team(client, headers):
client.set_cookie("auth", "test_key")
client.post(
"/api/team/",
json={
"teamName": "Test Team",
"leagueTimezone": "America/New_York",
"minuteOffset": 30,
},
headers=headers)
assert app_db.db.session.query(Team).where(Team.team_name == "Test Team").one_or_none() is not None
def test_create_team_player_added_as_tl(client, headers):
client.set_cookie("auth", "test_key")
client.post(
"/api/team/",
json={
"teamName": "Test Team",
"leagueTimezone": "America/New_York",
"minuteOffset": 30,
},
headers=headers)
team_id = app_db.db.session.query(Team).where(Team.team_name == "Test Team").one().id
player_team = app_db.db.session.query(
PlayerTeam
).where(
PlayerTeam.team_id == team_id
).one()
assert player_team.is_team_leader
## Integration test 2: leaving team
def test_leaving_team_deletes_team(client, headers):
client.set_cookie("auth", "test_key")
response = client.delete(
"/api/team/id/1/player/76561198248436608/",
headers=headers)
assert app_db.db.session.query(Team).where(Team.id == 1).one_or_none() is None
## Integration test 3: availability scheduling
def test_availability_scheduling(client, headers):
client.set_cookie("auth", "test_key")
response = client.put(
"/api/schedule/",
json={
"teamId": 1,
"availability": [1] * 168,
"windowStart": "2024-10-01T00:30:00Z",
},
headers=headers,
)
pta = app_db.db.session.query(
PlayerTeamAvailability
).where(
PlayerTeamAvailability.player_team_id == 1,
).one()
assert pta.end_time == datetime.datetime(2024, 10, 8, 0, 30, tzinfo=datetime.timezone.utc)
def test_availability_merge(client, headers):
client.set_cookie("auth", "test_key")
client.put(
"/api/schedule/",
json={
"teamId": 1,
"availability": [1] * 168,
"windowStart": "2024-10-01T00:30:00Z",
},
headers=headers,
)
client.put(
"/api/schedule/",
json={
"teamId": 1,
"availability": [1] * 168,
"windowStart": "2024-10-08T00:30:00Z",
},
headers=headers,
)
ptas = app_db.db.session.query(
PlayerTeamAvailability
).where(
PlayerTeamAvailability.player_team_id == 1,
).all()
assert len(ptas) == 1
def test_availability_split(client, headers):
client.set_cookie("auth", "test_key")
client.put(
"/api/schedule/",
json={
"teamId": 1,
"availability": [1] * 168,
"windowStart": "2024-10-01T00:30:00Z",
},
headers=headers,
)
client.put(
"/api/schedule/",
json={
"teamId": 1,
"availability": [0] * 4 + [1] * 164,
"windowStart": "2024-10-01T04:30:00Z",
},
headers=headers,
)
pta = app_db.db.session.query(
PlayerTeamAvailability
).where(
PlayerTeamAvailability.player_team_id == 1,
).first()
assert pta is not None
assert pta.end_time == datetime.datetime(2024, 10, 1, 4, 30, tzinfo=datetime.timezone.utc)
## Integration test 4: ETL job
@pytest.fixture
def mock_example_log() -> RawLogDetails:
return {
"teams": {
"Blue": {"score": 1},
"Red": {"score": 2},
},
"players": {
"[U:1:288170880]": {
"team": "Blue",
"kills": 0,
"deaths": 1,
"assists": 0,
"dmg": 0,
"dt": 0,
},
},
"info": {
"title": "Test Match",
"date": int(datetime.datetime.now(datetime.timezone.utc).timestamp()),
"map": "cp_process_f12",
},
"length": 3600,
}
def test_transform_load(client, app, mock_example_log):
from jobs.fetch_logstf import transform
team_id = 1
# patch celery task to avoid sending a task to the queue
with patch("jobs.fetch_logstf.update_playtime.delay", return_value=None):
for instance in transform(1, mock_example_log, None, team_id):
app_db.db.session.add(instance)
app_db.db.session.commit()
team_match = app_db.db.session.query(
TeamMatch
).where(
TeamMatch.team_id == team_id,
).one()
assert len(team_match.match.players) == 1
def test_transform_load_no_team(client, app, mock_example_log):
from jobs.fetch_logstf import transform
team_id = 1
# patch celery task to avoid sending a task to the queue
with patch("jobs.fetch_logstf.update_playtime.delay", return_value=None):
for instance in transform(1, mock_example_log, None, None):
app_db.db.session.add(instance)
app_db.db.session.commit()
team_match = app_db.db.session.query(
TeamMatch
).where(
TeamMatch.team_id == team_id,
).one_or_none()
assert team_match is None
## Integration test 5: OpenID
@pytest.fixture
def mock_openid():
return {
"openid.ns": "http://specs.openid.net/auth/2.0",
"openid.mode": "id_res",
"openid.op_endpoint": "https://steamcommunity.com/openid/login",
"openid.claimed_id": "https://steamcommunity.com/openid/id/76561198248436608",
"openid.identity": "https://steamcommunity.com/openid/id/76561198248436608",
"openid.return_to": "https://availabili-tf.sandvich.xyz/login",
"openid.response_nonce": "2025-05-06T16:58:39ZPoXodsvJwAB/SEAs6xwz25rZvmU=",
"openid.assoc_handle": "1234567890",
"openid.signed": "signed,op_endpoint,claimed_id,identity,return_to,response_nonce,assoc_handle",
"openid.sig": "GI2sIIWma7SR0Jz/tQfTKzUie/o="
}
def test_steam_authenticate_new_auth_token(client, headers, mock_openid):
client.set_cookie("auth", "test_key")
# patch requests.get
mock = Mock()
mock.status_code = 200
mock.headers = {"Content-Type": "application/json"}
mock.text = "openid.ns:http://specs.openid.net/auth/2.0\nis_valid:true\n"
auth_session_count = app_db.db.session.query(AuthSession).count()
with patch("requests.post", return_value=mock):
client.post(
"/api/login/authenticate",
json=mock_openid,
headers=headers)
assert app_db.db.session.query(AuthSession).count() == auth_session_count + 1
def test_steam_authenticate_new_user(client, headers, mock_openid):
client.set_cookie("auth", "test_key")
# patch requests.get
mock = Mock()
mock.status_code = 200
mock_openid["openid.claimed_id"] = "https://steamcommunity.com/openid/id/123"
mock_openid["openid.identity"] = "https://steamcommunity.com/openid/id/123"
mock.headers = {"Content-Type": "application/json"}
mock.text = "openid.ns:http://specs.openid.net/auth/2.0\nis_valid:true\n"
with patch("requests.post", return_value=mock):
client.post(
"/api/login/authenticate",
json=mock_openid,
headers=headers)
assert app_db.db.session.query(Player).where(Player.steam_id == 123).one() is not None
def test_steam_authenticate_fail(client, headers, mock_openid):
client.set_cookie("auth", "test_key")
# patch requests.get
mock = Mock()
mock.status_code = 401
mock.headers = {"Content-Type": "application/json"}
mock.text = "openid.ns:http://specs.openid.net/auth/2.0\nis_valid:false\n"
auth_session_count = app_db.db.session.query(AuthSession).count()
with patch("requests.post", return_value=mock):
client.post(
"/api/login/authenticate",
json=mock_openid,
headers=headers)
assert app_db.db.session.query(AuthSession).count() == auth_session_count

View File

@ -0,0 +1,24 @@
import app_db
from login import extract_steam_id_from_response, generate_base36
def test_initial_state(client):
from models.auth_session import AuthSession
assert app_db.db.session.query(AuthSession).count() == 1
def test_get_user(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/login/get-user",
headers=headers
)
assert response.status_code == 200
def test_generate_base36():
string = generate_base36(8)
assert len(string) == 8
def test_extract_steam_id_from_response():
response = "http://steamcommunity.com/openid/id/76561198248436608"
steam_id = extract_steam_id_from_response(response)
assert steam_id == "76561198248436608"

View File

@ -0,0 +1,71 @@
import datetime
import app_db
from models.player_team_availability import PlayerTeamAvailability
from models.player_team_role import PlayerTeamRole
def test_get_schedule_7days_168elements(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/schedule/team?windowStart=0&teamId=1&windowSizeDays=7",
headers=headers)
assert len(response.json["playerAvailability"]["76561198248436608"]["availability"]) == 168
def test_find_consecutive_blocks_len_1():
from schedule import find_consecutive_blocks
blocks = find_consecutive_blocks([0, 1, 1, 1, 0])
assert len(blocks) == 1
def test_find_consecutive_blocks_len_2():
from schedule import find_consecutive_blocks
blocks = find_consecutive_blocks([0, 1, 0, 1, 0])
assert len(blocks) == 2
def test_find_consecutive_blocks_size_4():
from schedule import find_consecutive_blocks
blocks = find_consecutive_blocks([0, 2, 2, 2, 2])
print(blocks)
assert blocks[0][2] - blocks[0][1] == 4
def test_get_team_availability(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/schedule/team?windowStart=0&teamId=1&windowSizeDays=7",
headers=headers)
assert len(response.json["playerAvailability"]) == 1
def test_view_available_at_time_not_available(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/schedule/view-available?teamId=1&startTime=2024-10-01T00:00:00Z",
headers=headers)
assert len(response.json["players"]) == 0
def test_view_available_at_time_is_available(client, headers):
client.set_cookie("auth", "test_key")
pta = PlayerTeamAvailability(
player_team_id=1,
start_time=datetime.datetime(2024, 9, 1, 0, 0, tzinfo=datetime.timezone.utc),
end_time=datetime.datetime(2024, 10, 5, 2, 0, tzinfo=datetime.timezone.utc),
)
app_db.db.session.add(pta)
ptr = PlayerTeamRole(
player_team_id=1,
role=PlayerTeamRole.Role.Pyro,
is_main=True,
)
app_db.db.session.add(ptr)
app_db.db.session.commit()
response = client.get(
"/api/schedule/view-available?teamId=1&startTime=1727740800",
headers=headers)
print(response.json)
assert len(response.json["players"]) == 1

View File

@ -0,0 +1,81 @@
import pytest
import app_db
from models.team import Team
def test_create_team(client, headers):
client.set_cookie("auth", "test_key")
response = client.post(
"/api/team/",
json={
"teamName": "Test Team",
"leagueTimezone": "America/New_York",
"minuteOffset": 30,
},
headers=headers)
assert response.json["team"]["teamName"] == "Test Team"
def test_create_team_invalid(client, headers):
client.set_cookie("auth", "test_key")
response = client.post(
"/api/team/",
json={ },
headers=headers)
assert response.status_code == 422
#def test_update_team(client, headers):
# client.set_cookie("auth", "test_key")
# response = client.patch(
# "/api/team/id/1/",
# json={
# "teamName": "Updated Team Name",
# "leagueTimezone": "America/New_York",
# "minuteOffset": 30,
# },
# headers=headers)
# assert response.json["teamName"] == "Updated Team Name"
#def test_remove_team_member(client, headers):
# client.set_cookie("auth", "test_key")
# response = client.delete(
# "/api/team/id/1/player/76561198248436608/",
# headers=headers)
# assert response.status_code == 200
def test_view_teams(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/team/all/",
headers=headers)
assert len(response.json["teams"]) > 0
def test_view_team_by_id(client, headers):
client.set_cookie("auth", "test_key")
response = client.get(
"/api/team/id/1/",
headers=headers)
assert response.json["team"]["teamName"] == "Team Pepeja"
def test_edit_member_roles(client, headers):
client.set_cookie("auth", "test_key")
response = client.patch(
"/api/team/id/1/edit-player/76561198248436608",
json={
"roles": [
{
"role": "Pyro",
"isMain": False,
}
],
},
headers=headers)
assert response.status_code == 204
def test_make_player_team_leader(client, headers):
client.set_cookie("auth", "test_key")
response = client.put(
"/api/team/id/1/player/76561198248436608/",
json={},
headers=headers)
assert response.status_code == 500 # not implemented

View File

@ -0,0 +1,9 @@
def test_set_username(client, headers):
client.set_cookie("auth", "test_key")
response = client.post(
"/api/user/username",
json={
"username": "NewUsername"
},
headers=headers)
assert response.json["username"] == "NewUsername"