API: in-process boti ako hraci

Boti su sedadla bez socketu: ucty bot:<kind>-<n> v tabulke players
(nehijacknutelne, recyklovane medzi hrami), handlery add_bot/remove_bot
(len hostitel, pred startom) a tahova slucka _run_bot_turns s pauzou
BOT_MOVE_DELAY_SECONDS a MC/inferenciou v executori. Druhy: heuristic,
random, neural (pure-Python siet; bez suboru vah jasna chyba a pri
restore fallback na heuristiku). Botie sedadla preziju restart servera.

Oprava po ceste: emit kariet hracovi so sid=None (offline sedadlo po
restore) broadcastoval jeho karty vsetkym klientom -- preskakuje sa.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
tim
2026-07-07 18:51:14 +02:00
co-authored by Claude Fable 5
parent 9a750756c5
commit 23de3ae3ac
5 changed files with 490 additions and 5 deletions
+3
View File
@@ -10,3 +10,6 @@ frontend
.env .env
.env.* .env.*
docker-compose*.yaml docker-compose*.yaml
# trenovacie artefakty do image nepatria (produkcia cita len rl/weights/)
rl/checkpoints
rl/runs
+1
View File
@@ -11,6 +11,7 @@ RUN pip install --no-cache-dir -r requirements.txt
COPY bridzik.py app.py ./ COPY bridzik.py app.py ./
COPY api ./api COPY api ./api
COPY db ./db COPY db ./db
COPY rl ./rl
COPY tests ./tests COPY tests ./tests
RUN useradd --create-home --uid 1000 appuser \ RUN useradd --create-home --uid 1000 appuser \
+148 -3
View File
@@ -13,8 +13,10 @@ import socketio
from bridzik import Bridzik, BridzikException, Card from bridzik import Bridzik, BridzikException, Card
from db.db import init_db from db.db import init_db
from api import auth as auth_module, history from api import auth as auth_module, history
from api import bots as bots_module
from api import stats as stats_module from api import stats as stats_module
from api.auth import AuthError, RegistrationIncomplete from api.auth import AuthError, RegistrationIncomplete
from rl.encoding import index_card
def _env_bool(name: str, default: bool) -> bool: def _env_bool(name: str, default: bool) -> bool:
@@ -179,6 +181,9 @@ class Game:
self.players: list["Player"] = [] self.players: list["Player"] = []
self.started = False self.started = False
self.bridzik_core: Bridzik | None = None self.bridzik_core: Bridzik | None = None
# Serializuje tahovu slucku botov -- dva sucasne _run_bot_turns tasky
# by inak mohli tahat za to iste sedadlo.
self.bot_lock = asyncio.Lock()
def start(self): def start(self):
self.bridzik_core = Bridzik() self.bridzik_core = Bridzik()
@@ -202,6 +207,10 @@ class Player:
self.player_id = player_id # persistent account id (db.models.Player.id) self.player_id = player_id # persistent account id (db.models.Player.id)
self.token = str(uuid.uuid4()) # secret token used for secure reconnect self.token = str(uuid.uuid4()) # secret token used for secure reconnect
self.connected = True self.connected = True
# Bot = sedadlo bez socketu; `brain` je rozhodovaci objekt s rozhranim
# guess(rnd, seat) / play(rnd, seat) z rl/players.py.
self.is_bot = False
self.brain = None
class CardStatusEncoder(JSONEncoder): class CardStatusEncoder(JSONEncoder):
@@ -232,6 +241,7 @@ def public_games() -> list:
"name": p.name, "name": p.name,
"connected": p.connected, "connected": p.connected,
"player_id": p.player_id, "player_id": p.player_id,
"is_bot": p.is_bot,
} }
for p in g.players for p in g.players
], ],
@@ -262,7 +272,8 @@ async def send_game_status(gid: str):
"completed": core.is_completed(), "completed": core.is_completed(),
# Self-contained roster so the game view doesn't depend on the lobby snapshot. # Self-contained roster so the game view doesn't depend on the lobby snapshot.
"players": [ "players": [
{"order": p.order, "name": p.name, "connected": p.connected} {"order": p.order, "name": p.name, "connected": p.connected,
"is_bot": p.is_bot}
for p in sorted(game.players, key=lambda p: p.order) for p in sorted(game.players, key=lambda p: p.order)
], ],
"series_number": core.series[-1].series_number, "series_number": core.series[-1].series_number,
@@ -299,11 +310,16 @@ async def _cleanup_abandoned_lobby(gid: str):
task neskodny no-op (netreba nic rusit).""" task neskodny no-op (netreba nic rusit)."""
await asyncio.sleep(LOBBY_ABANDON_GRACE_SECONDS) await asyncio.sleep(LOBBY_ABANDON_GRACE_SECONDS)
game = games.get(gid) game = games.get(gid)
if game is not None and not game.started and not any(p.connected for p in game.players): if game is not None and not game.started and not _any_human_connected(game):
del games[gid] del games[gid]
await broadcast_lobby() await broadcast_lobby()
def _any_human_connected(game: "Game") -> bool:
"""Boti su 'pripojeni' stale, pre opustenost lobby sa pocitaju len ludia."""
return any(p.connected for p in game.players if not p.is_bot)
async def _mark_player_offline(game: "Game", player: "Player"): async def _mark_player_offline(game: "Game", player: "Player"):
"""Mark player disconnected. An unstarted game with nobody left gets a """Mark player disconnected. An unstarted game with nobody left gets a
delayed cleanup (mobile sockets drop on screen lock, so an immediate delayed cleanup (mobile sockets drop on screen lock, so an immediate
@@ -311,7 +327,7 @@ async def _mark_player_offline(game: "Game", player: "Player"):
is kept in memory so it stays in the lobby and can be resumed (it's torn is kept in memory so it stays in the lobby and can be resumed (it's torn
down only by end_game).""" down only by end_game)."""
player.connected = False player.connected = False
if not any(p.connected for p in game.players) and not game.started: if not _any_human_connected(game) and not game.started:
asyncio.create_task(_cleanup_abandoned_lobby(game.gid)) asyncio.create_task(_cleanup_abandoned_lobby(game.gid))
await sio.emit( await sio.emit(
"player_connection", "player_connection",
@@ -352,11 +368,75 @@ def _load_game_into_memory(info: dict) -> "Game":
for seat, (pid, uname) in enumerate(info["seats"]): for seat, (pid, uname) in enumerate(info["seats"]):
player = Player(None, uname, seat, pid) player = Player(None, uname, seat, pid)
player.connected = False player.connected = False
# Boti sa rozpoznaju konvenciou mena a dostanu novy mozog -- ozivi ich
# prvy _kick_bots (napr. ked sa clovek vrati cez rejoin_game).
if bots_module.is_bot_username(uname):
player.is_bot = True
player.brain = bots_module.make_brain(uname)
player.connected = True
game.players.append(player) game.players.append(player)
games[info["gid"]] = game games[info["gid"]] = game
return game return game
# --- bot turns --------------------------------------------------------------
# Pauza medzi tahmi bota, nech ludia stihaju sledovat hru (0 = okamzite).
BOT_MOVE_DELAY_SECONDS = float(os.environ.get("BOT_MOVE_DELAY_SECONDS", "0.8"))
def _kick_bots(gid: str) -> None:
"""Ak je v hre bot, spusti (na pozadi) dohratie vsetkych botich tahov.
Vola sa po kazdej akcii, ktora mohla posunut tah na botie sedadlo."""
game = games.get(gid)
if game is not None and game.started and any(p.is_bot for p in game.players):
asyncio.create_task(_run_bot_turns(gid))
async def _run_bot_turns(gid: str):
"""Kym je na tahu botie sedadlo, vykonavaj jeho tahy tym istym internym
postupom ako handlery add_guess/play_card (engine validuje, historia sa
zapisuje, room dostava game_status). MC vypocet bezi v executori, aby
nedrzal event loop ostatnych hier."""
game = games.get(gid)
if game is None or not game.started or game.bridzik_core is None:
return
async with game.bot_lock:
core = game.bridzik_core
loop = asyncio.get_running_loop()
while not core.is_completed():
rnd = core.series[-1].get_last_round()
seat = rnd.get_active_player()
bot = game.player_by_order(seat)
if bot is None or not bot.is_bot:
return # na tahu je clovek
if BOT_MOVE_DELAY_SECONDS > 0:
await asyncio.sleep(BOT_MOVE_DELAY_SECONDS)
if games.get(gid) is not game:
return # hru medzitym niekto ukoncil (end_game)
try:
if not rnd.is_guessing_completed():
guess = await loop.run_in_executor(None, bot.brain.guess, rnd, seat)
core.add_player_guess(seat, guess)
else:
action = await loop.run_in_executor(None, bot.brain.play, rnd, seat)
core.play_card(seat, index_card(action))
await history.record_completed_rounds(gid, core)
for player in game.players:
if player.sid:
await send_player_cards(gid, player.order, player.sid)
except BridzikException as exc:
# Nemalo by nastat (bot hra len legalne tahy) -- nezacykli sa,
# slucku znovu spusti dalsia akcia cloveka.
await send_error_room(gid, str(exc))
return
await send_game_status(gid)
async def send_error_room(gid: str, message: str):
await sio.emit("error", {"error": message}, room=gid)
# --- connection lifecycle ------------------------------------------------- # --- connection lifecycle -------------------------------------------------
@sio.event @sio.event
@@ -488,6 +568,62 @@ async def register_player(sid, gid):
await broadcast_lobby() await broadcast_lobby()
@sio.on("add_bot")
async def add_bot(sid, gid, kind=None):
"""Hostitel prida bota na najnizsie volne sedadlo nezacatej hry."""
sess = sessions.get(sid)
if sess is None or sess["gid"] != gid:
return await send_error(sid, "Nie ste v tejto hre.")
if sess["order"] != 0:
return await send_error(sid, "Iba hostitel moze pridavat botov.")
game = games.get(gid)
if game is None:
return await send_error(sid, "Hra neexistuje.")
if game.started:
return await send_error(sid, "Hra uz zacala.")
if len(game.players) >= 4:
return await send_error(sid, "Prekroceny pocet hracov.")
if kind == "neural" and not bots_module.neural_available():
return await send_error(sid, "AI bot nie je na tomto serveri dostupny.")
if kind not in bots_module.BOT_KINDS:
kind = bots_module.DEFAULT_KIND
account = await bots_module.ensure_bot_account(
kind, {p.player_id for p in game.players}
)
used = {p.order for p in game.players}
order = next(o for o in range(4) if o not in used)
player = Player(None, account["username"], order, account["player_id"])
player.is_bot = True
player.brain = bots_module.make_brain(account["username"])
game.players.append(player)
await broadcast_lobby()
@sio.on("remove_bot")
async def remove_bot(sid, gid, order):
"""Hostitel odoberie bota z nezacatej hry (sedadlo sa uvolni)."""
sess = sessions.get(sid)
if sess is None or sess["gid"] != gid:
return await send_error(sid, "Nie ste v tejto hre.")
if sess["order"] != 0:
return await send_error(sid, "Iba hostitel moze odoberat botov.")
game = games.get(gid)
if game is None:
return await send_error(sid, "Hra neexistuje.")
if game.started:
return await send_error(sid, "Hra uz zacala.")
try:
seat = int(order)
except (TypeError, ValueError):
return await send_error(sid, "Neplatne sedadlo.")
player = game.player_by_order(seat)
if player is None or not player.is_bot:
return await send_error(sid, "Na tomto sedadle nie je bot.")
game.players.remove(player)
await broadcast_lobby()
@sio.on("leave_game") @sio.on("leave_game")
async def leave_game(sid): async def leave_game(sid):
"""Explicit exit (e.g. a 'Back to lobby' button). The socket stays """Explicit exit (e.g. a 'Back to lobby' button). The socket stays
@@ -535,7 +671,11 @@ async def start_game(sid, gid):
await broadcast_lobby() await broadcast_lobby()
await send_game_status(gid) await send_game_status(gid)
for player in game.players: for player in game.players:
# sid None = bot alebo offline sedadlo -- emit s to=None by karty
# broadcastol VSETKYM klientom, preto sa preskakuje.
if player.sid:
await send_player_cards(gid, player.order, player.sid) await send_player_cards(gid, player.order, player.sid)
_kick_bots(gid)
@sio.on("end_game") @sio.on("end_game")
@@ -590,6 +730,7 @@ async def reconnect_to_game(sid, gid, token):
await send_player_cards(gid, player.order, sid) await send_player_cards(gid, player.order, sid)
await sio.emit("player_connection", {"order": player.order, "connected": True}, room=gid) await sio.emit("player_connection", {"order": player.order, "connected": True}, room=gid)
await broadcast_lobby() await broadcast_lobby()
_kick_bots(gid)
@sio.on("rejoin_game") @sio.on("rejoin_game")
@@ -627,6 +768,7 @@ async def rejoin_game(sid, gid):
await send_player_cards(gid, player.order, sid) await send_player_cards(gid, player.order, sid)
await sio.emit("player_connection", {"order": player.order, "connected": True}, room=gid) await sio.emit("player_connection", {"order": player.order, "connected": True}, room=gid)
await broadcast_lobby() await broadcast_lobby()
_kick_bots(gid)
@sio.on("restore_game") @sio.on("restore_game")
@@ -685,6 +827,7 @@ async def add_guess(sid, guess):
except BridzikException as exc: except BridzikException as exc:
return await send_error(sid, str(exc)) return await send_error(sid, str(exc))
await send_game_status(game.gid) await send_game_status(game.gid)
_kick_bots(game.gid)
@sio.on("play_card") @sio.on("play_card")
@@ -710,12 +853,14 @@ async def play_card(sid, card_key):
await history.record_completed_rounds(game.gid, core) await history.record_completed_rounds(game.gid, core)
await send_game_status(game.gid) await send_game_status(game.gid)
for player in game.players: for player in game.players:
if player.sid: # None (bot/offline) by broadcastoval karty vsetkym
await send_player_cards(game.gid, player.order, player.sid) await send_player_cards(game.gid, player.order, player.sid)
# A naturally-finished game (all 4 series played out) has nothing left to # A naturally-finished game (all 4 series played out) has nothing left to
# offer the lobby -- refresh the list so it drops out immediately instead # offer the lobby -- refresh the list so it drops out immediately instead
# of lingering as "started"/resumable until someone happens to leave it. # of lingering as "started"/resumable until someone happens to leave it.
if core.is_completed(): if core.is_completed():
await broadcast_lobby() await broadcast_lobby()
_kick_bots(game.gid)
# --- history (read-only) -------------------------------------------------- # --- history (read-only) --------------------------------------------------
+120
View File
@@ -0,0 +1,120 @@
"""In-process boti: DB ucty botov a ich rozhodovacie "mozgy" z rl/players.py.
Bot je normalny hrac na sedadle -- ma riadok v tabulke `players` (aby
historia, standings a restore fungovali bez zmeny), ale ziadny socket.
Tahovu slucku botov ma api/__init__.py (_run_bot_turns); tu je len to,
co potrebuje DB a rl vrstvu.
Bezpecnost botich uctov: totp_secret je nahodny a nikde sa neuklada v
citatelnej podobe, totp_last_step sa nastavi na aktualny krok -- ucet tym
padom NIE JE "nedokoncena registracia" (viz auth._is_unconfirmed), takze
register_account ho odmietne prepisat a login bez secretu neprejde.
"""
import os
from random import Random
import pyotp
from sqlalchemy import select
from api import auth
from db import crypto
from db.db import async_session
from db.models import Player
from rl.players import HeuristicPlayer, RandomPlayer
from rl.pure_net import DEFAULT_WEIGHTS_PATH, PureNet, PureNeuralPlayer
# Prefix je konvencia na rozpoznanie bota (aj po restarte servera, kedy sa
# sedadla obnovuju z DB len ako (player_id, username)).
BOT_PREFIX = "bot:"
DEFAULT_KIND = "heuristic"
# Natrenovana siet -- vahy (rl/weights/, export z rl/export.py) sa nacitaju
# raz a zdielaju medzi botmi (PureNet je bezstavovy, len cita).
_pure_net: PureNet | None = None
def _neural_brain() -> PureNeuralPlayer:
global _pure_net
if _pure_net is None:
_pure_net = PureNet.load()
return PureNeuralPlayer(_pure_net)
def neural_available() -> bool:
return os.path.exists(DEFAULT_WEIGHTS_PATH)
_BRAINS = {
"heuristic": lambda: HeuristicPlayer(Random()),
"random": lambda: RandomPlayer(Random()),
"neural": _neural_brain,
}
BOT_KINDS = tuple(_BRAINS)
def available_kinds() -> tuple:
"""Druhy botov ponuknutelne na tomto serveri (neural len s vahami)."""
return tuple(k for k in _BRAINS if k != "neural" or neural_available())
def is_bot_username(username: str) -> bool:
return bool(username) and username.startswith(BOT_PREFIX)
def kind_of(username: str) -> str:
"""'bot:heuristic-2' -> 'heuristic'; neznamy druh padne na DEFAULT_KIND."""
body = username[len(BOT_PREFIX):]
kind = body.rsplit("-", 1)[0]
return kind if kind in _BRAINS else DEFAULT_KIND
def make_brain(username: str):
"""Rozhodovaci objekt (guess/play rozhranie z rl/players.py) pre bota.
Neural bez suboru vah (napr. restore hry na serveri bez exportu) padne
na heuristiku -- sedadlo hra dalej, len inym mozgom.
"""
kind = kind_of(username)
if kind == "neural" and not neural_available():
kind = DEFAULT_KIND
return _BRAINS[kind]()
def _suffix_number(username: str) -> int:
try:
return int(username.rsplit("-", 1)[1])
except (IndexError, ValueError):
return 0
async def ensure_bot_account(kind: str, exclude_ids: set) -> dict:
"""Najde alebo zalozi boti ucet daneho druhu; vrati {player_id, username}.
`exclude_ids` su ucty uz obsadene v danej hre -- kazde sedadlo potrebuje
INY ucet (Game.playerN_id aj unikat v Guess predpokladaju 4 rozne ID).
Ucty sa cisluju bot:<kind>-1, -2, ... a recykluju sa medzi hrami.
"""
prefix = f"{BOT_PREFIX}{kind}-"
async with async_session() as session:
rows = (
await session.scalars(
select(Player).where(Player.username.like(prefix + "%"))
)
).all()
for player in sorted(rows, key=lambda p: _suffix_number(p.username)):
if player.id not in exclude_ids:
return {"player_id": player.id, "username": player.username}
number = 1 + max((_suffix_number(p.username) for p in rows), default=0)
username = f"{prefix}{number}"
player = Player(
username=username,
# nahodny secret, ktory sa zahodi -- nikto sa zan neprihlasi
totp_secret=crypto.encrypt(pyotp.random_base32()),
# nenulovy last_step = ucet sa netvari ako nedokoncena registracia
totp_last_step=auth._current_step(),
)
session.add(player)
await session.commit()
return {"player_id": player.id, "username": player.username}
+216
View File
@@ -0,0 +1,216 @@
"""Testy in-process botov (api/bots.py + tahova slucka v api/__init__.py).
Rovnaky setup ako tests/test_history.py: docasny SQLite subor, env pred
importom. Socket.IO emity idu do prazdnych roomov (ziadny klient), takze
handlery a slucka sa daju volat priamo bez klienta.
"""
import asyncio
import os
import tempfile
import unittest
import uuid
# Nastav DB/ENCRYPTION_KEY PRED importom db/api modulov.
_DB_FILE = os.path.join(tempfile.gettempdir(), f"bridzik_test_{uuid.uuid4().hex}.db")
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///" + _DB_FILE.replace("\\", "/")
from cryptography.fernet import Fernet # noqa: E402
os.environ.setdefault("ENCRYPTION_KEY", Fernet.generate_key().decode())
from random import Random # noqa: E402
import api # noqa: E402
from api import auth, bots, history # noqa: E402
from db.db import init_db # noqa: E402
from rl.players import HeuristicPlayer, RandomPlayer # noqa: E402
def run(coro):
return asyncio.run(coro)
def _make_bot_accounts(n):
exclude = set()
accounts = []
for _ in range(n):
acc = run(bots.ensure_bot_account("random", exclude))
exclude.add(acc["player_id"])
accounts.append(acc)
return accounts
def _make_game(seat_accounts, brains):
"""Postavi zacatu in-memory hru + Game riadok v DB."""
gid = str(uuid.uuid4())
game = api.Game(gid, "test")
for seat, acc in enumerate(seat_accounts):
player = api.Player(None, acc["username"], seat, acc["player_id"])
if brains[seat] is not None:
player.is_bot = True
player.brain = brains[seat]
player.connected = True
else:
player.connected = False
game.players.append(player)
game.start()
api.games[gid] = game
run(history.record_game_started(
gid, "test", [acc["player_id"] for acc in seat_accounts]
))
return game
class BotAccountCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
run(init_db())
api.BOT_MOVE_DELAY_SECONDS = 0
def test_username_conventions(self):
self.assertTrue(bots.is_bot_username("bot:heuristic-1"))
self.assertFalse(bots.is_bot_username("alice"))
self.assertEqual(bots.kind_of("bot:heuristic-2"), "heuristic")
self.assertEqual(bots.kind_of("bot:random-1"), "random")
self.assertEqual(bots.kind_of("bot:neural-1"), "neural")
self.assertEqual(bots.kind_of("bot:nezmysel-9"), bots.DEFAULT_KIND)
self.assertIsInstance(bots.make_brain("bot:heuristic-1"), HeuristicPlayer)
self.assertIsInstance(bots.make_brain("bot:random-3"), RandomPlayer)
@unittest.skipUnless(bots.neural_available(),
'chyba export vah (py -m rl.export)')
def test_neural_kind(self):
from rl.pure_net import PureNeuralPlayer
self.assertIn("neural", bots.available_kinds())
brain = bots.make_brain("bot:neural-1")
self.assertIsInstance(brain, PureNeuralPlayer)
# zdielana instancia PureNet (vahy sa nacitavaju len raz)
self.assertIs(brain.net, bots.make_brain("bot:neural-2").net)
def test_ensure_bot_account_reuse_and_exclude(self):
first = run(bots.ensure_bot_account("heuristic", set()))
self.assertTrue(first["username"].startswith("bot:heuristic-"))
# bez vylucenia sa ucet recykluje
again = run(bots.ensure_bot_account("heuristic", set()))
self.assertEqual(first["player_id"], again["player_id"])
# s vylucenim vznikne dalsi ucet s inym ID
second = run(bots.ensure_bot_account("heuristic", {first["player_id"]}))
self.assertNotEqual(first["player_id"], second["player_id"])
self.assertNotEqual(first["username"], second["username"])
def test_bot_account_cannot_be_hijacked(self):
acc = run(bots.ensure_bot_account("heuristic", set()))
# registracia mena zlyha -- ucet sa netvari ako nedokoncena registracia
with self.assertRaises(auth.AuthError):
run(auth.register_account(acc["username"]))
# login zlyha na kode (secret nikto nepozna), NIE na RegistrationIncomplete
with self.assertRaises(auth.AuthError) as ctx:
run(auth.login(acc["username"], "000000"))
self.assertNotIsInstance(ctx.exception, auth.RegistrationIncomplete)
class BotTurnLoopCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
run(init_db())
api.BOT_MOVE_DELAY_SECONDS = 0
def setUp(self):
api.games.clear()
api.sessions.clear()
api.accounts.clear()
def test_four_bots_play_whole_game(self):
accounts = _make_bot_accounts(4)
brains = [RandomPlayer(Random(seat)) for seat in range(4)]
game = _make_game(accounts, brains)
run(api._run_bot_turns(game.gid))
self.assertTrue(game.bridzik_core.is_completed())
# cela hra je zapisana: 4 serie x 8 kol, ended_at nastaveny
detail = run(history.get_game_detail(game.gid))
self.assertIsNotNone(detail["ended_at"])
self.assertEqual(len(detail["rounds"]), history.FULL_GAME_ROUNDS * 4)
def test_bots_stop_at_human_turn(self):
accounts = _make_bot_accounts(3)
# sedadlo 0 = clovek; identitu v DB mu robi dalsi (nepouzity) boti
# ucet -- pre historiu je to len player_id, wrapper bez mozgu = clovek
human_acc = run(bots.ensure_bot_account(
"random", {a["player_id"] for a in accounts}
))
game = _make_game(
[human_acc] + accounts,
[None, HeuristicPlayer(Random(1), n_samples=20),
RandomPlayer(Random(2)), RandomPlayer(Random(3))],
)
core = game.bridzik_core
rnd = core.series[-1].get_last_round()
async def scenario():
# na tahu je clovek (first_player serie 0 je sedadlo 0) -> boti nic
await api._run_bot_turns(game.gid)
self.assertEqual(len(rnd.guesses), 0)
# clovek tipne -> boti dotipuju a hraju az po dalsi tah cloveka
core.add_player_guess(0, 1)
await api._run_bot_turns(game.gid)
run(scenario())
self.assertTrue(rnd.is_guessing_completed())
self.assertEqual(rnd.get_active_player(), 0)
def test_add_and_remove_bot_handlers(self):
async def scenario():
gid = str(uuid.uuid4())
api.games[gid] = api.Game(gid, "lobby-test")
host = api.Player("sid-host", "hostiteľ", 0, 999_100)
api.games[gid].players.append(host)
api.sessions["sid-host"] = {"gid": gid, "order": 0}
api.sessions["sid-guest"] = {"gid": gid, "order": 1}
# nehostitel nesmie pridat bota
await api.add_bot("sid-guest", gid)
self.assertEqual(len(api.games[gid].players), 1)
# hostitel prida dvoch botov -> rozne ucty, najnizsie volne sedadla
await api.add_bot("sid-host", gid)
await api.add_bot("sid-host", gid, "random")
players = api.games[gid].players
self.assertEqual(len(players), 3)
bots_added = [p for p in players if p.is_bot]
self.assertEqual(len(bots_added), 2)
self.assertEqual({p.order for p in bots_added}, {1, 2})
self.assertNotEqual(bots_added[0].player_id, bots_added[1].player_id)
self.assertIsNotNone(bots_added[0].brain)
# remove_bot: odmietne cloveka, odoberie bota
await api.remove_bot("sid-host", gid, 0)
self.assertEqual(len(api.games[gid].players), 3)
await api.remove_bot("sid-host", gid, 1)
self.assertEqual(len(api.games[gid].players), 2)
self.assertIsNone(api.games[gid].player_by_order(1))
run(scenario())
def test_restore_marks_bots(self):
from bridzik import Bridzik
info = {
"gid": str(uuid.uuid4()),
"name": "obnova",
"seats": [(1, "alice"), (2, "bot:heuristic-1"),
(3, "bot:random-1"), (4, "bob")],
"core": Bridzik(),
}
game = api._load_game_into_memory(info)
self.assertFalse(game.players[0].is_bot)
self.assertFalse(game.players[0].connected)
self.assertTrue(game.players[1].is_bot)
self.assertTrue(game.players[1].connected)
self.assertIsInstance(game.players[1].brain, HeuristicPlayer)
self.assertIsInstance(game.players[2].brain, RandomPlayer)
if __name__ == '__main__':
unittest.main(verbosity=2)