devplacepy/tests/api/game/mutations.py

# retoor <retoor@molodetz.nl>

import time
from datetime import datetime, timezone, timedelta

import requests

from devplacepy.database import get_table, refresh_snapshot
from devplacepy.services.game import store
from tests.conftest import BASE_URL

JSON = {"Accept": "application/json"}
_counter_gm = [0]


def _signup():
    _counter_gm[0] += 1
    name = f"gm{int(time.time() * 1000)}{_counter_gm[0]}"
    session = requests.Session()
    session.post(
        f"{BASE_URL}/auth/signup",
        data={
            "username": name,
            "email": f"{name}@t.dev",
            "password": "secret123",
            "confirm_password": "secret123",
        },
        allow_redirects=True,
    )
    return session, name


def _reset_farm(username, coins=100000):
    refresh_snapshot()
    user = get_table("users").find_one(username=username)
    farm = store.get_farm(user["uid"])
    if farm:
        get_table("game_plots").delete(farm_uid=farm["uid"])
        get_table("game_farms").delete(uid=farm["uid"])
    get_table("game_market_ticks").delete()
    farm = store.ensure_farm(user["uid"])
    get_table("game_farms").update({"uid": farm["uid"], "coins": coins}, ["uid"])
    refresh_snapshot()
    return user, farm


def test_fertilize_requires_auth(app_server, seeded_db):
    r = requests.post(f"{BASE_URL}/game/fertilize", data={"slot": 0}, headers=JSON)
    assert r.status_code in (401, 303)


def test_fertilize_empty_plot_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    r = session.post(f"{BASE_URL}/game/fertilize", data={"slot": 0}, headers=JSON)
    assert r.status_code == 400


def test_claim_quest_requires_auth(app_server, seeded_db):
    r = requests.post(f"{BASE_URL}/game/quests/claim", data={"quest": "first_harvest"}, headers=JSON)
    assert r.status_code in (401, 303)


def test_claim_unearned_quest_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    r = session.post(
        f"{BASE_URL}/game/quests/claim", data={"quest": "first_harvest"}, headers=JSON
    )
    assert r.status_code == 400


def test_steal_requires_auth(app_server, seeded_db):
    r = requests.post(
        f"{BASE_URL}/game/farm/alice_test/steal", data={"slot": 0}, headers=JSON
    )
    assert r.status_code in (401, 303)


def test_steal_unknown_farm_404(app_server, seeded_db):
    session, name = _signup()
    r = session.post(
        f"{BASE_URL}/game/farm/nobody_xyz_zzz/steal", data={"slot": 0}, headers=JSON
    )
    assert r.status_code == 404


def test_steal_own_farm_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    r = session.post(
        f"{BASE_URL}/game/farm/{name}/steal", data={"slot": 0}, headers=JSON
    )
    assert r.status_code == 400


def test_steal_empty_plot_returns_400(app_server, seeded_db):
    owner_session, owner = _signup()
    _reset_farm(owner)
    thief_session, thief = _signup()
    r = thief_session.post(
        f"{BASE_URL}/game/farm/{owner}/steal", data={"slot": 0}, headers=JSON
    )
    assert r.status_code == 400


def _set_stars(username, stars):
    refresh_snapshot()
    user = get_table("users").find_one(username=username)
    farm = store.get_farm(user["uid"])
    get_table("game_farms").update({"uid": farm["uid"], "stars": stars}, ["uid"])
    refresh_snapshot()


def test_legacy_requires_auth(app_server, seeded_db):
    r = requests.post(f"{BASE_URL}/game/legacy", data={"key": "multiplier"}, headers=JSON)
    assert r.status_code in (401, 303)


def test_legacy_upgrade_spends_stars(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    _set_stars(name, 1000)
    r = session.post(f"{BASE_URL}/game/legacy", data={"key": "multiplier"}, headers=JSON)
    assert r.status_code == 200, r.text[:200]
    refresh_snapshot()
    user = get_table("users").find_one(username=name)
    farm = store.get_farm(user["uid"])
    assert int(farm.get("legacy_multiplier") or 0) == 1
    assert int(farm.get("stars") or 0) < 1000


def test_legacy_insufficient_stars_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    _set_stars(name, 0)
    r = session.post(f"{BASE_URL}/game/legacy", data={"key": "multiplier"}, headers=JSON)
    assert r.status_code == 400


def test_legacy_unknown_key_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    _set_stars(name, 1000)
    r = session.post(f"{BASE_URL}/game/legacy", data={"key": "telekinesis"}, headers=JSON)
    assert r.status_code == 400


def test_steal_ready_unprotected_crop_transfers_coins(app_server, seeded_db):
    owner_session, owner = _signup()
    _reset_farm(owner, coins=100000)
    plant = owner_session.post(
        f"{BASE_URL}/game/plant", data={"slot": 0, "crop": "shell"}, headers=JSON
    )
    assert plant.status_code == 200, plant.text[:200]
    refresh_snapshot()
    owner_user = get_table("users").find_one(username=owner)
    farm = store.get_farm(owner_user["uid"])
    plot = get_table("game_plots").find_one(farm_uid=farm["uid"], slot_index=0)
    assert plot is not None
    past = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
    get_table("game_plots").update(
        {"uid": plot["uid"], "planted_at": past, "ready_at": past}, ["uid"]
    )
    refresh_snapshot()

    thief_session, thief = _signup()
    _reset_farm(thief, coins=0)
    r = thief_session.post(
        f"{BASE_URL}/game/farm/{owner}/steal", data={"slot": 0}, headers=JSON
    )
    assert r.status_code == 200, r.text[:200]
    refresh_snapshot()
    thief_user = get_table("users").find_one(username=thief)
    thief_farm = store.get_farm(thief_user["uid"])
    assert int(thief_farm.get("coins", 0)) > 0
    assert (get_table("game_plots").find_one(uid=plot["uid"]).get("crop_key") or "") == ""


def _ripen_owner_plot(owner, slot=0):
    refresh_snapshot()
    owner_user = get_table("users").find_one(username=owner)
    farm = store.get_farm(owner_user["uid"])
    plot = get_table("game_plots").find_one(farm_uid=farm["uid"], slot_index=slot)
    past = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
    get_table("game_plots").update(
        {"uid": plot["uid"], "planted_at": past, "ready_at": past}, ["uid"]
    )
    refresh_snapshot()
    return plot


def _set_farm(username, **fields):
    refresh_snapshot()
    user = get_table("users").find_one(username=username)
    farm = store.get_farm(user["uid"])
    get_table("game_farms").update({"uid": farm["uid"], **fields}, ["uid"])
    refresh_snapshot()


def test_defense_upgrade_requires_auth(app_server, seeded_db):
    r = requests.post(f"{BASE_URL}/game/defense/upgrade", headers=JSON)
    assert r.status_code in (401, 303)


def test_defense_upgrade_success(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    r = session.post(f"{BASE_URL}/game/defense/upgrade", headers=JSON)
    assert r.status_code == 200, r.text[:200]
    assert r.json()["farm"]["defense_level"] == 1


def test_defense_upgrade_insufficient_coins_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name, coins=0)
    r = session.post(f"{BASE_URL}/game/defense/upgrade", headers=JSON)
    assert r.status_code == 400


def test_infrastructure_buy_requires_prestige(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name, coins=100_000_000)
    r = session.post(
        f"{BASE_URL}/game/infrastructure/buy", data={"key": "registry"}, headers=JSON
    )
    assert r.status_code == 400


def test_infrastructure_buy_success(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name, coins=100_000_000)
    _set_farm(name, prestige=5)
    r = session.post(
        f"{BASE_URL}/game/infrastructure/buy", data={"key": "registry"}, headers=JSON
    )
    assert r.status_code == 200, r.text[:200]
    farm = r.json()["farm"]
    owned = next(i for i in farm["infrastructure"] if i["key"] == "registry")
    assert owned["owned"] is True


def test_infrastructure_buy_unknown_key_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    r = session.post(
        f"{BASE_URL}/game/infrastructure/buy", data={"key": "does_not_exist"}, headers=JSON
    )
    assert r.status_code == 400


def test_cosmetics_buy_requires_auth(app_server, seeded_db):
    r = requests.post(
        f"{BASE_URL}/game/cosmetics/buy", data={"key": "title_architect"}, headers=JSON
    )
    assert r.status_code in (401, 303)


def test_cosmetics_buy_and_equip(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name, coins=1_000_000)
    r = session.post(
        f"{BASE_URL}/game/cosmetics/buy", data={"key": "title_architect"}, headers=JSON
    )
    assert r.status_code == 200, r.text[:200]
    r = session.post(
        f"{BASE_URL}/game/cosmetics/equip", data={"key": "title_architect"}, headers=JSON
    )
    assert r.status_code == 200, r.text[:200]
    assert r.json()["farm"]["active_title"] == "title_architect"


def test_cosmetics_buy_insufficient_coins_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name, coins=0)
    r = session.post(
        f"{BASE_URL}/game/cosmetics/buy", data={"key": "title_architect"}, headers=JSON
    )
    assert r.status_code == 400


def test_mastery_upgrade_requires_auth(app_server, seeded_db):
    r = requests.post(
        f"{BASE_URL}/game/mastery", data={"key": "autoreplant"}, headers=JSON
    )
    assert r.status_code in (401, 303)


def test_mastery_upgrade_insufficient_points_returns_400(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    r = session.post(
        f"{BASE_URL}/game/mastery", data={"key": "autoreplant"}, headers=JSON
    )
    assert r.status_code == 400


def test_mastery_upgrade_success(app_server, seeded_db):
    session, name = _signup()
    _reset_farm(name)
    _set_farm(name, mastery_points=1000)
    r = session.post(
        f"{BASE_URL}/game/mastery", data={"key": "autoreplant"}, headers=JSON
    )
    assert r.status_code == 200, r.text[:200]
    upgrade = next(m for m in r.json()["farm"]["mastery"] if m["key"] == "autoreplant")
    assert upgrade["level"] == 1


def test_leaderboard_boards_all_return_200(app_server, seeded_db):
    for board in ("score", "prestige", "harvests", "raids", "time_to_kernel", "fair_play", "era"):
        r = requests.get(f"{BASE_URL}/game/leaderboard", params={"board": board}, headers=JSON)
        assert r.status_code == 200, (board, r.text[:200])
        assert "entries" in r.json()


def test_steal_cooldown_blocks_second_raid(app_server, seeded_db):
    owner_session, owner = _signup()
    _reset_farm(owner, coins=100000)
    thief_session, thief = _signup()
    _reset_farm(thief, coins=0)

    owner_session.post(
        f"{BASE_URL}/game/plant", data={"slot": 0, "crop": "shell"}, headers=JSON
    )
    _ripen_owner_plot(owner)
    first = thief_session.post(
        f"{BASE_URL}/game/farm/{owner}/steal", data={"slot": 0}, headers=JSON
    )
    assert first.status_code == 200, first.text[:200]

    owner_session.post(
        f"{BASE_URL}/game/plant", data={"slot": 0, "crop": "shell"}, headers=JSON
    )
    _ripen_owner_plot(owner)
    second = thief_session.post(
        f"{BASE_URL}/game/farm/{owner}/steal", data={"slot": 0}, headers=JSON
    )
    assert second.status_code == 400