OpenFUT/fifa17-recon/tools/test_fut_contract.py

#!/usr/bin/env python3
"""Contract / freeze-safety regression tests for the FIFA 17 FUT backend.

Hits the LIVE utas_server (default http://127.0.0.1:8099) and asserts each
response matches the shape reversed from CardsDLL (see docs/ENDPOINT_MAP.md).
The point is freeze-safety: FIFA's SAX deserializers hard-freeze (busy-loop at
0x1801c7f1a) if a field that must be an array/object arrives as a scalar. These
tests encode "must be array" / "must be object" / "must be number" per the
reversed schemas so a future edit that reintroduces that class of bug fails here
instead of freezing the game.

Read-only: only GET endpoints are exercised (no pack buys / squad writes), so it
never mutates the profile. Run:  python3 tools/test_fut_contract.py
Exit 0 = all pass. No pytest dependency (stdlib only).
"""
import json, sys, urllib.request

BASE = "http://127.0.0.1:8099"
G = "/ut/game/fifa17"
V2 = "/ut/v2/game/fifa17"

_fail = []
_pass = 0


def _get(path):
    with urllib.request.urlopen(BASE + path, timeout=5) as r:
        raw = r.read()
    return json.loads(raw) if raw else {}


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


def is_arr(x):  return isinstance(x, list)
def is_obj(x):  return isinstance(x, dict)
def is_num(x):  return isinstance(x, (int, float)) and not isinstance(x, bool)
def is_str(x):  return isinstance(x, str)


def test_credits():
    d = _get(G + "/user/credits")
    # coin counter binds to currencies[name==coins].funds (deser 0x180122c50), NOT "credits"
    check("credits.currencies is array", is_arr(d.get("currencies")), repr(d.get("currencies")))
    coins = next((c for c in d.get("currencies", []) if c.get("name") == "coins"), None)
    check("credits has coins currency", coins is not None)
    if coins:
        check("coins.funds is number", is_num(coins.get("funds")), repr(coins.get("funds")))


def test_v2_store_gate():
    d = _get(V2 + "/store")
    # FutStorePackQuantities eligibility gate (deser 0x1801758c0): result must be SUCCESS
    check("v2/store result == SUCCESS", d.get("result") == "SUCCESS", repr(d))


def test_store_catalog():
    d = _get(G + "/store/purchasegroup/all")
    # FutStoreGetPackTypes (deser 0x1801234e0): root key "purchase" MUST be an array
    check("catalog.purchase is array", is_arr(d.get("purchase")), repr(type(d.get("purchase"))))
    for p in d.get("purchase", []):
        check("pack has assetId (real identity)", "assetId" in p, repr(p.get("assetId")))
        check("pack.currencies is array (coin price)", is_arr(p.get("currencies")))
        check("pack.packContentInfo is object", is_obj(p.get("packContentInfo")))
        check("pack.extPrice is object", is_obj(p.get("extPrice")))
        ep = p.get("extPrice", {})
        check("extPrice.finalPrice is object", is_obj(ep.get("finalPrice")))


def test_market_bodies():
    # Shared IS-list body (deser 0x18013e7f0): auctionInfo MUST be array, credits number
    for path in [G + "/auctionhouse?type=player&start=0&num=21",
                 G + "/tradePile", G + "/watchList", G + "/trade/123"]:
        d = _get(path)
        check(f"{path} auctionInfo is array", is_arr(d.get("auctionInfo")), repr(d.get("auctionInfo")))
        check(f"{path} credits is number", is_num(d.get("credits")), repr(d.get("credits")))
    dup = _get(G + "/auctionhouse?type=player").get("duplicateItemIdList")
    check("auctionhouse duplicateItemIdList is array", is_arr(dup), repr(dup))
    md = _get(G + "/marketdata?defId=1")
    check("marketdata minPrice is number", is_num(md.get("minPrice")))
    check("marketdata maxPrice is number", is_num(md.get("maxPrice")))


def test_auction_record_shape():
    # Every populated auction record MUST match the reversed schema (deser
    # 0x18013e410) field-for-field, or the market screen freezes. This proves the
    # sample listings are freeze-safe OFFLINE, before the game ever parses them.
    d = _get(G + "/auctionhouse?type=player&start=0&num=21")
    recs = d.get("auctionInfo", [])
    check("auctionhouse returns >=1 listing (or FUT_MARKET=empty)", is_arr(recs))
    numeric = ["tradeId", "buyNowPrice", "startingBid", "currentBid", "expires",
               "sellerEstablished", "coinsProcessed"]
    strings = ["tradeState", "bidState", "sellerName"]
    for r in recs:
        check("record.itemData is object", is_obj(r.get("itemData")), repr(type(r.get("itemData"))))
        check("record.watched is bool", isinstance(r.get("watched"), bool), repr(r.get("watched")))
        for k in numeric:
            check(f"record.{k} is number", is_num(r.get(k)), repr(r.get(k)))
        for k in strings:
            check(f"record.{k} is string", is_str(r.get(k)), repr(r.get(k)))
        # itemData must itself be a valid card object (reuses club/squad parser)
        it = r.get("itemData", {})
        check("record.itemData.attributeList is array", is_arr(it.get("attributeList")))
        check("record.itemData.resourceId is number", is_num(it.get("resourceId")))


def test_squad_boot():
    # LoadActiveSquad (deser 0x18013d1f0): players MUST be array; empty body would reset
    # the 23 slots. formation is a string. This is the boot-critical path.
    d = _get(G + "/squad/0")
    check("squad.players is array", is_arr(d.get("players")), repr(type(d.get("players"))))
    check("squad.formation is string", is_str(d.get("formation")), repr(d.get("formation")))
    for pl in d.get("players", []):
        # Empty bench/reserve slots legitimately carry itemData=null (proven-safe,
        # matches the working in-game squad). Only a PRESENT itemData must be an
        # object -- a scalar there would desync the reader.
        it = pl.get("itemData")
        check("squad slot itemData is object-or-null", it is None or is_obj(it), repr(it))


def test_massinfo_empty():
    # GetUserMassInfo (deser 0x180174630) MUST stay {} -- any populated userInfo/squad
    # desyncs the parser -> freeze (CARD_SYSTEM.md). Guard against accidental population.
    d = _get(G + "/userMassInfo")
    check("userMassInfo is empty {}", d == {}, repr(d))


def test_club_items():
    # /club serves {"itemData":[...]} (SKIP'd by GetClubInfo, but must stay array-safe)
    d = _get(G + "/club?type=player&count=5")
    check("club.itemData is array", is_arr(d.get("itemData")), repr(type(d.get("itemData"))))


def main():
    tests = [test_credits, test_v2_store_gate, test_store_catalog, test_market_bodies,
             test_auction_record_shape, test_squad_boot, test_massinfo_empty, test_club_items]
    try:
        _get(G + "/user/credits")
    except Exception as e:
        print(f"SERVER NOT REACHABLE at {BASE}: {e}\nStart it: python3 tools/utas_server.py")
        return 2
    for t in tests:
        try:
            t()
        except Exception as e:
            _fail.append(f"{t.__name__} raised {type(e).__name__}: {e}")
    print(f"\n{_pass} checks passed, {len(_fail)} failed")
    for f in _fail:
        print("  FAIL:", f)
    return 0 if not _fail else 1


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