#!/usr/bin/env python3 """Acceptance harness for FIFA17 player-contract consumable APPLY (staging). Exercises the real route end to end and asserts the full observable contract: * the target's `contract` goes from B to min(99, B + grant), where `grant` is read from the shipped EA table `fcc_contractcards` INDEPENDENTLY of the Rust implementation (this is a cross-check, not a mirror); * exactly one source copy is consumed; * coins do NOT move (an apply is not an economy credit); * replaying the exhausted resource fails closed rather than granting again; * a manager contract and a non-contract consumable both fail closed. Read-only against production by construction: every URL is the staging port. """ import argparse import json import sqlite3 import sys import urllib.error import urllib.request HOST = "http://127.0.0.1:8299" CORE_DB = "/home/alex/openfut-sold-staging/staging-core.db" HDRS = {"X-OpenFUT-Game": "fifa17"} TABLE = "/home/alex/OpenFUT/fifa17-recon/data/tables/fcc_contractcards.json" PLAYER_CONTRACT_SUBTYPE = 201 MANAGER_CONTRACT_SUBTYPE = 202 CAP = 99 FAILURES = [] def check(label, got, want): ok = got == want print(f" [{'OK ' if ok else 'FAIL'}] {label}: got {got!r} want {want!r}") if not ok: FAILURES.append(label) return ok def req(method, path, body=None): data = json.dumps(body).encode() if body is not None else None r = urllib.request.Request(HOST + path, data=data, headers=HDRS, method=method) if data: r.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(r, timeout=30) as resp: raw = resp.read() return resp.status, (json.loads(raw) if raw else None) except urllib.error.HTTPError as e: raw = e.read() try: return e.code, json.loads(raw) except Exception: return e.code, raw.decode(errors="replace") def grant_from_table(resource_id, target_rating): """The authoritative grant, read straight from EA's shipped table. Column is selected by the TARGET's tier (bronze <65, silver 65..74, gold >=75) — NOT by the card's own tier. Verified 36/36 against the published FIFA 17 matrix. """ d = json.load(open(TABLE)) rows = d if isinstance(d, list) else (d.get("rows") or list(d.values())[0]) row = next((r for r in rows if r["carddbid"] == resource_id), None) if row is None: return None col = "bronze" if target_rating < 65 else ("silver" if target_rating < 75 else "gold") return row[col] def core_snapshot(): con = sqlite3.connect(f"file:{CORE_DB}?mode=ro", uri=True) try: snap = { "coins": con.execute("SELECT coins FROM clubs LIMIT 1").fetchone()[0], "owned": con.execute("SELECT COUNT(*) FROM owned_cards").fetchone()[0], "by_kind": dict( con.execute("SELECT content_kind, COUNT(*) FROM owned_cards GROUP BY 1") ), } cols = [r[1] for r in con.execute("PRAGMA table_info(owned_cards)")] snap["has_contract_column"] = "contract_matches" in cols if snap["has_contract_column"]: snap["contracts_set"] = con.execute( "SELECT COUNT(*) FROM owned_cards WHERE contract_matches IS NOT NULL" ).fetchone()[0] return snap finally: con.close() def club_players(): """Every owned player on the wire, with its contract, keyed by wire id.""" out = {} for pile in ("/ut/game/fifa17/club?type=player&start=0&count=200",): _, body = req("GET", pile) for it in (body or {}).get("itemData") or []: if it.get("itemType") == "player": out[it["id"]] = it return out def consumable_stacks(): _, body = req("GET", "/ut/game/fifa17/club/consumables/development") return {s["resourceId"]: s for s in (body or {}).get("itemData") or []} def pick_source(stacks, subtype_range): for rid, s in sorted(stacks.items()): if rid in subtype_range: return rid, s return None, None def main(): ap = argparse.ArgumentParser() ap.add_argument("--source", type=int, default=None, help="contract resource id to apply (default: first owned player contract)") ap.add_argument("--target", type=int, default=None, help="target player wire id (default: lowest-rated owned player)") args = ap.parse_args() print("== BEFORE ==") before = core_snapshot() print(json.dumps(before, indent=2, sort_keys=True)) if not before["has_contract_column"]: print("FATAL: migration 0028 not applied — owned_cards has no contract_matches column") return 2 stacks = consumable_stacks() players = club_players() if not players: print("FATAL: no owned players on the wire") return 2 player_contracts = {r: s for r, s in stacks.items() if 5001001 <= r <= 5001006 or r == 5001013} src = args.source or next(iter(sorted(player_contracts)), None) if src is None: print("FATAL: no owned PLAYER contract consumable to apply") return 2 # Lowest-rated target maximises the observable delta (a bronze target draws # the largest column) and exercises the tier selector rather than assuming gold. tgt_id = args.target or min(players, key=lambda i: players[i].get("rating", 0)) tgt = players[tgt_id] rating = tgt["rating"] grant = grant_from_table(src, rating) c_before = tgt.get("contract") expect_after = min(CAP, c_before + grant) print(f"\n== APPLY ==\n source resource {src} (stack count {stacks[src].get('count')})") print(f" target wire {tgt_id} rating {rating} -> tier " f"{'bronze' if rating < 65 else 'silver' if rating < 75 else 'gold'}") print(f" table grant {grant}; contract {c_before} -> expect {expect_after}") status, body = req("POST", f"/ut/game/fifa17/item/resource/{src}", {"apply": [{"id": tgt_id}]}) print(f" HTTP {status} {json.dumps(body)[:200] if body is not None else ''}") check("apply status", status, 200) check("apply body", body, {"itemData": []}) print("\n== AFTER ==") after = core_snapshot() players2 = club_players() stacks2 = consumable_stacks() print(json.dumps(after, indent=2, sort_keys=True)) check("coins unchanged", after["coins"], before["coins"]) check("one owned row consumed", after["owned"], before["owned"] - 1) check("one consumable consumed", after["by_kind"].get("consumable", 0), before["by_kind"].get("consumable", 0) - 1) check("target contract granted", players2.get(tgt_id, {}).get("contract"), expect_after) check("source stack decremented", (stacks2.get(src) or {}).get("count", 0), (stacks[src].get("count") or 1) - 1) # Untouched players must not have drifted. drifted = [i for i, p in players2.items() if i != tgt_id and p.get("contract") != players.get(i, {}).get("contract")] check("no collateral contract changes", drifted, []) print("\n== FAIL-CLOSED CASES ==") exhausted = (stacks2.get(src) or {}).get("count", 0) == 0 if exhausted: s, _ = req("POST", f"/ut/game/fifa17/item/resource/{src}", {"apply": [{"id": tgt_id}]}) check("replay of exhausted resource refused", s, 404) mgr = next((r for r in stacks2 if 5001007 <= r <= 5001012), None) if mgr: s, _ = req("POST", f"/ut/game/fifa17/item/resource/{mgr}", {"apply": [{"id": tgt_id}]}) check("manager contract fails closed (staff ratings unimported)", s, 409) other = next((r for r in stacks2 if not (5001001 <= r <= 5001013)), None) if other: s, _ = req("POST", f"/ut/game/fifa17/item/resource/{other}", {"apply": [{"id": tgt_id}]}) check("unproven family fails closed", s, 409) s, _ = req("POST", f"/ut/game/fifa17/item/resource/{src}", {"apply": [{"id": tgt_id}, {"id": tgt_id}]}) check("batch apply refused", s in (400, 404), True) print("\n== RESULT ==") if FAILURES: print("FAILED: " + ", ".join(FAILURES)) return 1 print(f"PASS — contract {c_before} -> {expect_after} on wire {tgt_id}, " f"one copy of {src} consumed, coins flat") return 0 if __name__ == "__main__": sys.exit(main())