278 lines
8.3 KiB
Python
278 lines
8.3 KiB
Python
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
|