|
# retoor <retoor@molodetz.nl>
|
|
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import requests
|
|
|
|
from devplacepy.database import get_table, refresh_snapshot
|
|
from devplacepy.services.opinionwar import rules
|
|
from tests.api.battles._helpers import (
|
|
JSON,
|
|
_create_war_post,
|
|
_fight,
|
|
_fighter_row,
|
|
_join,
|
|
_session_battles,
|
|
_user_row,
|
|
_war_row,
|
|
)
|
|
from tests.conftest import BASE_URL
|
|
|
|
|
|
def _warp_cooldown(war_uid, user_uid):
|
|
fighter = _fighter_row(war_uid, user_uid)
|
|
warped = (datetime.now(timezone.utc) - timedelta(hours=25)).isoformat()
|
|
get_table("opinion_war_fighters").update(
|
|
{"uid": fighter["uid"], "last_fight_at": warped}, ["uid"]
|
|
)
|
|
|
|
|
|
def test_join_and_switch(app_server):
|
|
s, name = _session_battles()
|
|
_, war = _create_war_post(s, f"join-{int(time.time() * 1000)}")
|
|
r = _join(s, war["uid"], "a")
|
|
assert r.status_code == 200
|
|
payload = r.json()
|
|
assert payload["ok"] is True
|
|
assert payload["data"]["war"]["viewer"]["faction"] == "a"
|
|
fighter = _fighter_row(war["uid"], _user_row(name)["uid"])
|
|
assert fighter["faction"] == "a"
|
|
r = _join(s, war["uid"], "b")
|
|
assert r.json()["data"]["war"]["viewer"]["faction"] == "b"
|
|
fighter = _fighter_row(war["uid"], _user_row(name)["uid"])
|
|
assert fighter["faction"] == "b"
|
|
|
|
|
|
def test_join_invalid_faction_422(app_server):
|
|
s, _ = _session_battles()
|
|
_, war = _create_war_post(s, f"badfaction-{int(time.time() * 1000)}")
|
|
r = s.post(
|
|
f"{BASE_URL}/battles/{war['uid']}/join",
|
|
data={"faction": "c"},
|
|
headers=JSON,
|
|
)
|
|
assert r.status_code in (400, 422)
|
|
|
|
|
|
def test_join_requires_login(app_server):
|
|
s, _ = _session_battles()
|
|
_, war = _create_war_post(s, f"anon-{int(time.time() * 1000)}")
|
|
r = requests.post(
|
|
f"{BASE_URL}/battles/{war['uid']}/join",
|
|
data={"faction": "a"},
|
|
allow_redirects=False,
|
|
)
|
|
assert r.status_code in (303, 401)
|
|
|
|
|
|
def test_fight_spends_coins_and_deals_level_damage(app_server):
|
|
s, name = _session_battles()
|
|
_, war = _create_war_post(s, f"fight-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "a")
|
|
r = _fight(s, war["uid"])
|
|
assert r.status_code == 200
|
|
payload = r.json()
|
|
user = _user_row(name)
|
|
expected = rules.damage_for(user.get("level"))
|
|
assert payload["data"]["damage"] == expected
|
|
assert payload["data"]["war"]["hp_a"] == expected
|
|
war_row = _war_row(war["uid"])
|
|
assert int(war_row["hp_a"]) == expected
|
|
refresh_snapshot()
|
|
farm = get_table("game_farms").find_one(user_uid=user["uid"])
|
|
from devplacepy.services.game import economy
|
|
|
|
assert int(farm["coins"]) == economy.STARTING_COINS - rules.FIGHT_COST_COINS
|
|
|
|
|
|
def test_fight_cooldown_refused(app_server):
|
|
s, _ = _session_battles()
|
|
_, war = _create_war_post(s, f"cooldown-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "a")
|
|
assert _fight(s, war["uid"]).status_code == 200
|
|
r = _fight(s, war["uid"])
|
|
assert r.status_code == 400
|
|
assert "24 hours" in r.json()["error"]["message"]
|
|
|
|
|
|
def test_fight_after_warped_cooldown_lands(app_server):
|
|
s, name = _session_battles()
|
|
_, war = _create_war_post(s, f"warp-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "b")
|
|
assert _fight(s, war["uid"]).status_code == 200
|
|
_warp_cooldown(war["uid"], _user_row(name)["uid"])
|
|
r = _fight(s, war["uid"])
|
|
assert r.status_code == 200
|
|
fighter = _fighter_row(war["uid"], _user_row(name)["uid"])
|
|
assert int(fighter["fight_count"]) == 2
|
|
|
|
|
|
def test_fight_without_coins_refused_and_cooldown_kept(app_server):
|
|
s, name = _session_battles()
|
|
_, war = _create_war_post(s, f"broke-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "a")
|
|
assert _fight(s, war["uid"]).status_code == 200
|
|
user = _user_row(name)
|
|
_warp_cooldown(war["uid"], user["uid"])
|
|
refresh_snapshot()
|
|
farm = get_table("game_farms").find_one(user_uid=user["uid"])
|
|
get_table("game_farms").update({"uid": farm["uid"], "coins": 0}, ["uid"])
|
|
r = _fight(s, war["uid"])
|
|
assert r.status_code == 400
|
|
assert "coins" in r.json()["error"]["message"]
|
|
fighter = _fighter_row(war["uid"], user["uid"])
|
|
assert int(fighter["fight_count"]) == 1
|
|
r = _fight(s, war["uid"])
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_fight_before_join_refused(app_server):
|
|
s, _ = _session_battles()
|
|
other, _ = _session_battles()
|
|
_, war = _create_war_post(s, f"nojoin-{int(time.time() * 1000)}")
|
|
r = _fight(other, war["uid"])
|
|
assert r.status_code == 400
|
|
assert "Join" in r.json()["error"]["message"]
|
|
|
|
|
|
def test_switch_keeps_dealt_damage(app_server):
|
|
s, name = _session_battles()
|
|
_, war = _create_war_post(s, f"defect-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "a")
|
|
assert _fight(s, war["uid"]).status_code == 200
|
|
r = _join(s, war["uid"], "b")
|
|
war_state = r.json()["data"]["war"]
|
|
user = _user_row(name)
|
|
expected = rules.damage_for(user.get("level"))
|
|
assert war_state["hp_a"] == expected
|
|
assert war_state["hp_b"] == 0
|
|
fighter = _fighter_row(war["uid"], user["uid"])
|
|
assert fighter["faction"] == "b"
|
|
assert int(fighter["hp_a"]) == expected
|
|
|
|
|
|
def test_actions_on_resolved_war_refused(app_server):
|
|
s, _ = _session_battles()
|
|
other, _ = _session_battles()
|
|
_, war = _create_war_post(s, f"over-{int(time.time() * 1000)}")
|
|
_join(s, war["uid"], "a")
|
|
past = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat()
|
|
get_table("opinion_wars").update({"uid": war["uid"], "ends_at": past}, ["uid"])
|
|
r = _fight(s, war["uid"])
|
|
assert r.status_code == 400
|
|
r = _join(other, war["uid"], "b")
|
|
assert r.status_code == 400
|