OpenFUT/fifa17-recon/docker/fifa17-python/tools/test_match_rewards.py

#!/usr/bin/env python3
"""Unit tests for the FUT match core loop — PURE, no server, no state, no profile.

Why separate from test_fut_contract.py: that suite is read-only by design (it hits
a live server and must never mutate the save), but the match loop credits coins and
bumps the W/D/L record. So the two pure pieces — result detection and the reward
body — are tested here instead of making the HTTP suite stateful.

Guards the two things that would silently break the loop:
  * `_match_result()` mis-reading a scoreline (wrong result -> wrong payout)
  * `destroy_match_body()` drifting from FutDestroyMatchServerResponse
    (deser 0x180121b60): a non-scalar there is the freeze class at 0x1801c7f1a,
    and a renamed key is silently SKIP'd, i.e. the reward vanishes with no error.
  * the shared base `/match` path distinguishing CREATEMATCH from PLAYGAME by
    the body-level matchId that CardsDLL serializes for subsequent operations

Run:  python3 tools/test_match_rewards.py     (exit 0 = pass)
"""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
os.environ.setdefault("FUT_PROFILE", "/tmp/openfut_unittest_profile.json")

import utas_server as U                                            # noqa: E402

_fail = []
_pass = 0


def check(name, cond, detail=""):
    global _pass
    if cond:
        _pass += 1
    else:
        _fail.append("%s: %s" % (name, detail))


# ---- _match_result: scoreline -> outcome ------------------------------------
def test_result_detection():
    cases = [
        ({"goals": 3, "opponentGoals": 1}, "won"),
        ({"goals": 0, "opponentGoals": 2}, "loss"),
        ({"goals": 1, "opponentGoals": 1}, "draw"),
        ({"score": 2, "opponentScore": 0}, "won"),
        ({"homeGoals": 0, "awayGoals": 4}, "loss"),
        ({"match": {"goals": 5, "opponentGoals": 0}}, "won"),   # nested
        ({"stats": {"score": 0, "opponentScore": 3}}, "loss"),  # nested
        ({"result": "WIN"}, "won"),
        ({"outcome": "defeat"}, "loss"),
        ({"result": "tie"}, "draw"),
        ({}, "draw"),                    # unknown -> neutral fallback
        (None, "draw"),                  # malformed body -> neutral fallback
        ({"goals": "2", "opponentGoals": 1}, "draw"),   # non-int -> no guess
    ]
    for body, expect in cases:
        got, _ = U._match_result(body)
        check("result %r -> %s" % (body, expect), got == expect, "got %s" % got)

    # a 0-0 draw must not be mistaken for "no data"
    r, s = U._match_result({"goals": 0, "opponentGoals": 0})
    check("0-0 is a draw with a score", r == "draw" and s == (0, 0), "%s %s" % (r, s))


# ---- destroy_match_body: the reward record ----------------------------------
# These expectations were REWRITTEN on 2026-08-04 after reversing deser 0x180121b60
# properly. The previous version asserted a top-level `coins` and a
# `qualifiedChampionEventId`, and it passed happily while the server shipped a body
# whose reward field the client never read. A test that encodes the wrong schema is
# worse than no test: it converts a bug into a guarantee.
REQUIRED_INT = ("allCoins", "matchCoins", "seasonCoins", "tournamentCoins",
                "boostConis", "participationAward")


def test_reward_body():
    b = U.destroy_match_body("won", 400, 13000)
    for k in REQUIRED_INT:
        check("reward.%s present" % k, k in b)
        check("reward.%s is int (scalar, not nested)" % k,
              isinstance(b.get(k), int) and not isinstance(b.get(k), bool), repr(b.get(k)))
    check("reward.teamOfTournamentWinner is bool",
          isinstance(b.get("teamOfTournamentWinner"), bool), repr(b.get("teamOfTournamentWinner")))
    check("allCoins is the NEW balance", b["allCoins"] == 13000, repr(b["allCoins"]))
    # EA's typo is load-bearing: the atom is 96 == "boostConis", not "boostCoins".
    check("key is EA's misspelled boostConis", "boostConis" in b and "boostCoins" not in b,
          repr(sorted(b)))
    # userData and matchCoinMultipliers stay OUT (SKIP-safe; userData is freeze-risk)
    for k in ("userData", "matchCoinMultipliers"):
        check("reward omits nested %s" % k, k not in b)

    if U.MATCH_END:
        # `coins` (atom 149) is read ONLY inside gameModeAward. A top-level one is
        # silently skipped, which is what the server used to send.
        check("coins is NOT top level", "coins" not in b, repr(sorted(b)))
        gma = b.get("gameModeAward")
        check("gameModeAward is an object", isinstance(gma, dict), repr(gma))
        if isinstance(gma, dict):
            check("gameModeAward.coins echoes the credited amount",
                  gma.get("coins") == 400, repr(gma.get("coins")))
            # atom 89 is MATCHED inside gameModeAward and then handled by nothing:
            # not read, not skip-routed. Its value token is left unconsumed, which is
            # the precondition for the desync spin. This check is the guard rail.
            check("gameModeAward NEVER contains bidTokens (freeze trap)",
                  "bidTokens" not in gma, repr(sorted(gma)))
            for k, v in gma.items():
                check("gameModeAward.%s is scalar" % k,
                      isinstance(v, (int, bool, str)), repr(v))
        # side-effecting atom 0x269: its branch calls through a manager vtable
        check("qualifiedChampionEventId omitted (it has a side effect)",
              "qualifiedChampionEventId" not in b, repr(sorted(b)))
    else:
        check("legacy body keeps top-level coins", b.get("coins") == 400, repr(b.get("coins")))

    # nothing non-scalar may sneak in at the top level except gameModeAward
    for k, v in b.items():
        if k == "gameModeAward":
            continue
        check("reward.%s is scalar" % k, isinstance(v, (int, bool, str)), repr(v))


def test_end_reason_is_authoritative():
    """endReason, not a score comparison, is how the client reports the outcome.

    Reversed from the DestroyMatch serializer: atom 260, a string enum with nine
    values. Both stats objects are OMITTED by the client when endReason is DNF or
    QUIT, so a result parser must not require them.
    """
    for reason, want in (("WIN", "won"), ("DNF_WIN", "won"),
                         ("DRAW", "draw"), ("DNF_DRAW", "draw"), ("NO_CONTEST", "draw"),
                         ("LOSS", "loss"), ("DNF_LOSS", "loss"),
                         ("DNF", "loss"), ("QUIT", "loss")):
        r, _ = U._match_result({"endReason": reason})
        check("endReason %s -> %s" % (reason, want), r == want, r)
    # score comes from goals in the two stats objects, first field of each
    r, s = U._match_result({"endReason": "WIN",
                            "myMatchStats": {"goals": 3},
                            "opponentMatchStats": {"goals": 1}})
    check("goals are read from myMatchStats/opponentMatchStats", s == (3, 1), repr(s))
    # a DNF with no stats objects must still resolve, not crash or fall back blindly
    r, s = U._match_result({"endReason": "DNF"})
    check("DNF with no stats resolves to loss with no score", r == "loss" and s is None,
          "%s %s" % (r, s))
    # endReason must WIN over a contradictory score probe
    r, _ = U._match_result({"endReason": "LOSS", "goals": 5, "opponentGoals": 0})
    check("endReason beats the legacy score probe", r == "loss", r)


def test_payout_table():
    for res in ("won", "draw", "loss"):
        b = U.destroy_match_body(res, U.MATCH_COINS[res], 0)
        check("matchCoins matches the %s payout" % res,
              b["matchCoins"] == U.MATCH_COINS[res], repr(b["matchCoins"]))
    check("win pays >= draw", U.MATCH_COINS["won"] >= U.MATCH_COINS["draw"])
    check("draw pays >= loss", U.MATCH_COINS["draw"] >= U.MATCH_COINS["loss"])


def test_match_call_classification():
    """CREATEMATCH and PLAYGAME share a path; only the latter has a matchId."""
    cases = (
        ("/ut/game/fifa17/match", "POST", {}, "create"),
        ("/ut/game/fifa17/match", "POST", {"matchId": 1234}, "play"),
        ("/ut/game/fifa17/match/ready", "POST", {"matchId": 1234}, "ready"),
        ("/ut/game/fifa17/match/end", "POST", {"matchId": 1234}, "end"),
        ("/ut/game/fifa17/match/reset", "PUT", {"matchId": 1234}, "reset"),
        ("/ut/game/fifa17/match/keepalive", "POST", {"matchId": 1234}, "keepalive"),
    )
    for path, method, body, want in cases:
        got = U._match_call(path, method, body)
        check("%s %s -> %s" % (method, path, want), got == want, "got %s" % got)


def test_match_ready_body():
    """FutMatchReadyServerResponse parses these two scalar identifiers."""
    body = U.match_ready_body(1234, 33068179)
    check("ready echoes matchId", body.get("matchId") == 1234, repr(body))
    check("ready has opponentPersonaId", body.get("opponentPersonaId") == 33068179,
          repr(body))
    check("ready IDs are scalar ints",
          all(isinstance(v, int) and not isinstance(v, bool) for v in body.values()),
          repr(body))
    check("ready omits unproven nested items", "items" not in body, repr(body))


def main():
    for t in (test_result_detection, test_reward_body,
              test_end_reason_is_authoritative, test_payout_table,
              test_match_call_classification, test_match_ready_body):
        try:
            t()
        except Exception as e:
            _fail.append("%s raised %s: %s" % (t.__name__, type(e).__name__, e))
    print("\n%d checks passed, %d failed" % (_pass, len(_fail)))
    for f in _fail:
        print("  FAIL:", f)
    return 0 if not _fail else 1


if __name__ == "__main__":
    sys.exit(main())