Repository navigation
Expand file tree
/
Copy pathsmoke_test_commander.py
More file actions
104 lines (88 loc) · 3.76 KB
/
Copy pathsmoke_test_commander.py
File metadata and controls
104 lines (88 loc) · 3.76 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import asyncio
import json
import os
import sys
import websockets
DECK = """Deck
99 Mountain
Commander
1 Krenko, Mob Boss"""
def load_invite_code():
"""Lê BRIDGE_INVITE_CODE do .env na raiz do repo (mesmo arquivo que
start-xmage.sh gera/usa) -- o Bridge agora exige esse código no login."""
env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
try:
with open(env_path, encoding="utf-8") as f:
for line in f:
if line.startswith("BRIDGE_INVITE_CODE="):
return line.split("=", 1)[1].strip()
except FileNotFoundError:
pass
return os.environ.get("BRIDGE_INVITE_CODE", "")
async def main():
uri = "ws://localhost:8000"
async with websockets.connect(uri) as ws:
async def send(obj):
print(">>", json.dumps(obj)[:150])
await ws.send(json.dumps(obj))
async def recv_until(pred, timeout=25):
while True:
raw = await asyncio.wait_for(ws.recv(), timeout=timeout)
msg = json.loads(raw)
print("<<", (msg.get("type") or msg.get("method")), "|", json.dumps(msg)[:250])
if pred(msg):
return msg
await send({"action": "login", "username": "claudebot3", "password": "test12345",
"inviteCode": load_invite_code()})
login = await recv_until(lambda m: m.get("type") == "login_result")
if not login.get("success"):
print("LOGIN FALHOU:", login.get("error"))
sys.exit(1)
await send({
"action": "createTable",
"gameType": "Commander Free For All",
"deckType": "Variant Magic - Commander",
"deckList": DECK,
"opponents": ["COMPUTER_MAD", "COMPUTER_MAD", "COMPUTER_MAD"],
})
created = await recv_until(lambda m: m.get("type") == "create_table_result")
if not created.get("success"):
print("CREATE TABLE FALHOU:", created.get("error"))
sys.exit(1)
table_id = created["tableId"]
print("MESA CRIADA (4 jogadores):", table_id)
await send({"action": "startMatch", "tableId": table_id})
started = await recv_until(lambda m: m.get("type") == "start_match_result")
if not started.get("success"):
print("START MATCH FALHOU:", started.get("error"))
sys.exit(1)
print("PARTIDA INICIADA, ouvindo eventos por 25s...")
counts = {}
players_seen = set()
try:
end = asyncio.get_event_loop().time() + 25
while True:
remaining = end - asyncio.get_event_loop().time()
if remaining <= 0:
break
raw = await asyncio.wait_for(ws.recv(), timeout=remaining)
msg = json.loads(raw)
key = msg.get("type") or msg.get("method") or "?"
counts[key] = counts.get(key, 0) + 1
data = msg.get("data")
gv = None
if isinstance(data, dict):
gv = data.get("gameView") or (data if "players" in data else None)
if gv and "players" in gv:
for p in gv["players"]:
players_seen.add(p.get("name"))
if msg.get("method") in ("GAME_ASK", "GAME_SELECT"):
game_id = msg.get("objectId")
await send({"action": "sendPlayerBoolean", "gameId": game_id, "boolValue": True})
except asyncio.TimeoutError:
pass
print("\nJOGADORES VISTOS NO GAMEVIEW:", players_seen)
print("RESUMO DE EVENTOS RECEBIDOS:")
for k, v in sorted(counts.items(), key=lambda x: -x[1]):
print(f" {k}: {v}")
asyncio.run(main())