mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-07-04 16:23:04 +02:00
Krzysztof kuhy Rudnicki
4db978b67b
- Fix test_process_game_event_game_end to properly mock engine - Fix test_handle_game tests to mock _run_analysis_subprocess to prevent hanging - Fix test_process_game_start_event to mock analysis subprocess - Add test_process_game_event_game_end_after_move to cover game end status path - Add test_process_game_event_color_unknown_on_gamefull to cover gameFull with spectator - Add tests for lichess_api.py covering spectator case and join_game_stream edge cases - Fix mypy type errors with Event and GameThreads type aliases Coverage: - lichess_api.py: 100% (was 98%) - main.py: 99% (branch partials for loop exits only)
1094 lines
40 KiB
Python
1094 lines
40 KiB
Python
"""Tests for lichess_bot main module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import threading
|
|
from typing import TYPE_CHECKING, Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import chess
|
|
import pytest
|
|
import requests
|
|
|
|
from python_pkg.lichess_bot.main import (
|
|
BotContext,
|
|
GameMeta,
|
|
GameState,
|
|
_apply_move_to_board,
|
|
_attempt_move,
|
|
_calculate_time_budget,
|
|
_extract_game_full_data,
|
|
_extract_game_state_data,
|
|
_extract_player_info,
|
|
_finalize_game,
|
|
_handle_challenge,
|
|
_handle_game,
|
|
_handle_move_if_needed,
|
|
_init_game_log,
|
|
_insert_analysis_into_log,
|
|
_is_my_turn,
|
|
_log_analysis_progress,
|
|
_log_move_to_file,
|
|
_process_analysis_output,
|
|
_process_bot_event,
|
|
_process_game_event,
|
|
_rebuild_board_from_moves,
|
|
_run_analysis_subprocess,
|
|
_run_event_loop_iteration,
|
|
_safe_event_loop_iteration,
|
|
_stream_bot_events,
|
|
_update_clocks_from_state,
|
|
_write_pgn_to_log,
|
|
main,
|
|
run_bot,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
# Type alias to make mypy happy with test event dicts
|
|
Event = dict[str, Any]
|
|
GameThreads = dict[str, threading.Thread]
|
|
|
|
|
|
class TestApplyMoveToBoard:
|
|
"""Tests for _apply_move_to_board."""
|
|
|
|
def test_apply_valid_move(self) -> None:
|
|
"""Test applying a valid move."""
|
|
board = chess.Board()
|
|
_apply_move_to_board(board, "e2e4", "game1")
|
|
assert board.fen() != chess.STARTING_FEN
|
|
|
|
def test_apply_invalid_move(self) -> None:
|
|
"""Test applying an invalid move logs debug."""
|
|
board = chess.Board()
|
|
with patch("python_pkg.lichess_bot.main._logger") as mock_logger:
|
|
_apply_move_to_board(board, "invalid", "game1")
|
|
mock_logger.debug.assert_called_once()
|
|
|
|
|
|
class TestInitGameLog:
|
|
"""Tests for _init_game_log."""
|
|
|
|
def test_init_game_log_success(self, tmp_path: Path) -> None:
|
|
"""Test successful log initialization."""
|
|
with patch("python_pkg.lichess_bot.main.Path.cwd", return_value=tmp_path):
|
|
result = _init_game_log("game123", 42)
|
|
assert result is not None
|
|
assert result.exists()
|
|
content = result.read_text()
|
|
assert "game game123 started" in content
|
|
assert "bot_version v42" in content
|
|
|
|
def test_init_game_log_oserror(self) -> None:
|
|
"""Test log initialization with OSError."""
|
|
with patch("python_pkg.lichess_bot.main.Path.cwd") as mock_cwd:
|
|
mock_path = MagicMock()
|
|
mock_path.__truediv__ = MagicMock(return_value=mock_path)
|
|
mock_path.open.side_effect = OSError("Permission denied")
|
|
mock_cwd.return_value = mock_path
|
|
result = _init_game_log("game123", 42)
|
|
assert result is None
|
|
|
|
|
|
class TestUpdateClocksFromState:
|
|
"""Tests for _update_clocks_from_state."""
|
|
|
|
def test_update_clocks_white(self) -> None:
|
|
"""Test clock update when playing as white."""
|
|
state = GameState(color="white")
|
|
state_data: Event = {"wtime": 60000, "btime": 55000, "winc": 1000}
|
|
_update_clocks_from_state(state_data, state)
|
|
assert state.my_ms == 60000
|
|
assert state.opp_ms == 55000
|
|
assert state.inc_ms == 1000
|
|
|
|
def test_update_clocks_black(self) -> None:
|
|
"""Test clock update when playing as black."""
|
|
state = GameState(color="black")
|
|
state_data: Event = {"wtime": 60000, "btime": 55000, "binc": 2000}
|
|
_update_clocks_from_state(state_data, state)
|
|
assert state.my_ms == 55000
|
|
assert state.opp_ms == 60000
|
|
assert state.inc_ms == 2000
|
|
|
|
def test_update_clocks_float_values(self) -> None:
|
|
"""Test clock update with float values."""
|
|
state = GameState(color="white")
|
|
state_data: Event = {"wtime": 60000.5, "btime": 55000.5}
|
|
_update_clocks_from_state(state_data, state)
|
|
assert state.my_ms == 60000
|
|
assert state.opp_ms == 55000
|
|
|
|
def test_update_clocks_none_values(self) -> None:
|
|
"""Test clock update with None values."""
|
|
state = GameState(color="white")
|
|
state_data: Event = {"wtime": None, "btime": None}
|
|
_update_clocks_from_state(state_data, state)
|
|
assert state.my_ms is None
|
|
assert state.opp_ms is None
|
|
|
|
|
|
class TestExtractPlayerInfo:
|
|
"""Tests for _extract_player_info."""
|
|
|
|
def test_extract_player_info_white(self) -> None:
|
|
"""Test extracting player info when bot is white."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"white": {"id": "mybot", "name": "MyBot"},
|
|
"black": {"id": "opp", "name": "Opponent"},
|
|
}
|
|
_extract_player_info(event, state, meta, api)
|
|
assert state.color == "white"
|
|
assert meta.white_name == "MyBot"
|
|
assert meta.black_name == "Opponent"
|
|
|
|
def test_extract_player_info_black(self) -> None:
|
|
"""Test extracting player info when bot is black."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"white": {"id": "opp", "name": "Opponent"},
|
|
"black": {"id": "mybot", "name": "MyBot"},
|
|
}
|
|
_extract_player_info(event, state, meta, api)
|
|
assert state.color == "black"
|
|
|
|
def test_extract_player_info_invalid_data(self) -> None:
|
|
"""Test extracting player info with invalid data."""
|
|
api = MagicMock()
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {"white": "invalid", "black": "invalid"}
|
|
_extract_player_info(event, state, meta, api)
|
|
assert state.color is None
|
|
|
|
def test_extract_player_info_missing_name(self) -> None:
|
|
"""Test extracting player info with missing name uses id."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"white": {"id": "mybot"},
|
|
"black": {"id": "opponent"},
|
|
}
|
|
_extract_player_info(event, state, meta, api)
|
|
assert meta.white_name == "mybot"
|
|
assert meta.black_name == "opponent"
|
|
|
|
|
|
class TestExtractGameFullData:
|
|
"""Tests for _extract_game_full_data."""
|
|
|
|
def test_extract_game_full_data(self) -> None:
|
|
"""Test extracting gameFull data."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"state": {"moves": "e2e4 e7e5", "status": "started", "wtime": 60000},
|
|
"white": {"id": "mybot"},
|
|
"black": {"id": "opp"},
|
|
"createdAt": 1609459200000, # 2021-01-01
|
|
}
|
|
moves, status = _extract_game_full_data(event, state, meta, api)
|
|
assert moves == "e2e4 e7e5"
|
|
assert status == "started"
|
|
assert meta.site_url == "https://lichess.org/game1"
|
|
assert meta.date_iso == "2021.01.01"
|
|
|
|
def test_extract_game_full_data_invalid_state(self) -> None:
|
|
"""Test extracting gameFull data with invalid state."""
|
|
api = MagicMock()
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {"state": "invalid"}
|
|
moves, status = _extract_game_full_data(event, state, meta, api)
|
|
assert moves == ""
|
|
assert status is None
|
|
|
|
|
|
class TestExtractGameStateData:
|
|
"""Tests for _extract_game_state_data."""
|
|
|
|
def test_extract_game_state_as_white(self) -> None:
|
|
"""Test extracting gameState data as white."""
|
|
state = GameState(color="white", my_ms=60000)
|
|
event: Event = {
|
|
"moves": "e2e4",
|
|
"status": "started",
|
|
"wtime": 59000,
|
|
"btime": 60000,
|
|
}
|
|
moves, status = _extract_game_state_data(event, state)
|
|
assert moves == "e2e4"
|
|
assert status == "started"
|
|
assert state.my_ms == 59000
|
|
assert state.opp_ms == 60000
|
|
|
|
def test_extract_game_state_as_black(self) -> None:
|
|
"""Test extracting gameState data as black."""
|
|
state = GameState(color="black")
|
|
event: Event = {
|
|
"moves": "e2e4 e7e5",
|
|
"wtime": 60000,
|
|
"btime": 59000,
|
|
"binc": 1000,
|
|
}
|
|
moves, __status = _extract_game_state_data(event, state)
|
|
assert moves == "e2e4 e7e5"
|
|
assert state.my_ms == 59000
|
|
assert state.opp_ms == 60000
|
|
assert state.inc_ms == 1000
|
|
|
|
|
|
class TestCalculateTimeBudget:
|
|
"""Tests for _calculate_time_budget."""
|
|
|
|
def test_calculate_time_budget_normal(self) -> None:
|
|
"""Test time budget calculation."""
|
|
state = GameState(my_ms=60000, inc_ms=1000)
|
|
board = chess.Board()
|
|
budget = _calculate_time_budget(state, board, 10.0)
|
|
assert 0.05 <= budget <= 10.0
|
|
|
|
def test_calculate_time_budget_low_time(self) -> None:
|
|
"""Test time budget with low time."""
|
|
state = GameState(my_ms=1000, inc_ms=0)
|
|
board = chess.Board()
|
|
budget = _calculate_time_budget(state, board, 10.0)
|
|
assert budget >= 0.05
|
|
|
|
|
|
class TestLogMoveToFile:
|
|
"""Tests for _log_move_to_file."""
|
|
|
|
def test_log_move_to_file(self, tmp_path: Path) -> None:
|
|
"""Test logging a move to file."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("header\n")
|
|
move = chess.Move.from_uci("e2e4")
|
|
_log_move_to_file(log_path, 1, move, "best move")
|
|
content = log_path.read_text()
|
|
assert "ply 1: e2e4" in content
|
|
assert "best move" in content
|
|
|
|
def test_log_move_to_file_none_path(self) -> None:
|
|
"""Test logging with None path does nothing."""
|
|
move = chess.Move.from_uci("e2e4")
|
|
_log_move_to_file(None, 1, move, "reason") # Should not raise
|
|
|
|
|
|
class TestAttemptMove:
|
|
"""Tests for _attempt_move."""
|
|
|
|
def test_attempt_move_success(self) -> None:
|
|
"""Test successful move attempt."""
|
|
api = MagicMock()
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (
|
|
chess.Move.from_uci("e2e4"),
|
|
"opening",
|
|
)
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState(my_ms=60000)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
board = chess.Board()
|
|
|
|
result = _attempt_move(ctx, state, meta, board)
|
|
assert result is True
|
|
api.make_move.assert_called_once()
|
|
|
|
def test_attempt_move_no_moves(self) -> None:
|
|
"""Test move attempt with no legal moves."""
|
|
api = MagicMock()
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (None, "no moves")
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
board = chess.Board()
|
|
|
|
result = _attempt_move(ctx, state, meta, board)
|
|
assert result is False
|
|
|
|
def test_attempt_move_illegal(self) -> None:
|
|
"""Test move attempt with illegal move."""
|
|
api = MagicMock()
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
# Return a move that's not legal (e.g., random square move)
|
|
engine.choose_move_with_explanation.return_value = (
|
|
chess.Move.from_uci("a1a8"),
|
|
"bad",
|
|
)
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState(my_ms=60000)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
board = chess.Board()
|
|
|
|
result = _attempt_move(ctx, state, meta, board)
|
|
assert result is True
|
|
api.make_move.assert_not_called()
|
|
|
|
def test_attempt_move_request_error(self) -> None:
|
|
"""Test move attempt with request error."""
|
|
api = MagicMock()
|
|
api.make_move.side_effect = requests.RequestException("Network error")
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (
|
|
chess.Move.from_uci("e2e4"),
|
|
"opening",
|
|
)
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState(my_ms=60000)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
board = chess.Board()
|
|
|
|
result = _attempt_move(ctx, state, meta, board)
|
|
assert result is True # Still returns True
|
|
|
|
|
|
class TestIsMyTurn:
|
|
"""Tests for _is_my_turn."""
|
|
|
|
def test_is_my_turn_white_to_move(self) -> None:
|
|
"""Test checking turn when white to move."""
|
|
board = chess.Board() # White to move
|
|
assert _is_my_turn(board, "white") is True
|
|
assert _is_my_turn(board, "black") is False
|
|
|
|
def test_is_my_turn_black_to_move(self) -> None:
|
|
"""Test checking turn when black to move."""
|
|
board = chess.Board()
|
|
board.push_uci("e2e4") # Black to move
|
|
assert _is_my_turn(board, "white") is False
|
|
assert _is_my_turn(board, "black") is True
|
|
|
|
|
|
class TestRebuildBoardFromMoves:
|
|
"""Tests for _rebuild_board_from_moves."""
|
|
|
|
def test_rebuild_board_from_moves(self) -> None:
|
|
"""Test rebuilding board from moves list."""
|
|
moves_list = ["e2e4", "e7e5", "g1f3"]
|
|
board = _rebuild_board_from_moves(moves_list, "game1")
|
|
assert len(board.move_stack) == 3
|
|
|
|
|
|
class TestHandleMoveIfNeeded:
|
|
"""Tests for _handle_move_if_needed."""
|
|
|
|
def test_handle_move_game_state_my_turn(self) -> None:
|
|
"""Test handling move on gameState when my turn."""
|
|
api = MagicMock()
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (
|
|
chess.Move.from_uci("e2e4"),
|
|
"opening",
|
|
)
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState(color="white", my_ms=60000, board=chess.Board())
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
|
|
result = _handle_move_if_needed(ctx, state, meta, "gameState", 0)
|
|
assert result is True
|
|
|
|
def test_handle_move_game_full_with_moves(self) -> None:
|
|
"""Test handling move on gameFull with existing moves (opponent's turn)."""
|
|
api = MagicMock()
|
|
engine = MagicMock()
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState(color="white", my_ms=60000, board=chess.Board())
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
|
|
# gameFull with moves - don't move
|
|
result = _handle_move_if_needed(ctx, state, meta, "gameFull", 1)
|
|
assert result is True
|
|
engine.choose_move_with_explanation.assert_not_called()
|
|
|
|
|
|
class TestProcessGameEvent:
|
|
"""Tests for _process_game_event."""
|
|
|
|
def test_process_game_event_unhandled_type(self) -> None:
|
|
"""Test processing unhandled event type."""
|
|
ctx = MagicMock()
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {"type": "chatLine", "text": "hello"}
|
|
result = _process_game_event(event, ctx, state, meta)
|
|
assert result is True
|
|
|
|
def test_process_game_event_game_full(self) -> None:
|
|
"""Test processing gameFull event."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (
|
|
chess.Move.from_uci("e2e4"),
|
|
"opening",
|
|
)
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState()
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"type": "gameFull",
|
|
"state": {"moves": "", "status": "started"},
|
|
"white": {"id": "mybot"},
|
|
"black": {"id": "opp"},
|
|
}
|
|
result = _process_game_event(event, ctx, state, meta)
|
|
assert result is True
|
|
|
|
def test_process_game_event_game_end(self) -> None:
|
|
"""Test processing game end event."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (None, "no moves")
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
state = GameState(color="white", last_handled_len=-1)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"type": "gameState",
|
|
"moves": "e2e4 e7e5",
|
|
"status": "mate",
|
|
}
|
|
result = _process_game_event(event, ctx, state, meta)
|
|
assert result is False
|
|
|
|
def test_process_game_event_game_end_after_move(self) -> None:
|
|
"""Test game ends with status after handling move.
|
|
|
|
This covers the case where _handle_move_if_needed returns True
|
|
but status indicates game end.
|
|
"""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (
|
|
chess.Move.from_uci("d2d4"),
|
|
"response",
|
|
)
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
# Black's turn - it's opponent's move, so we don't need to move
|
|
state = GameState(color="black", last_handled_len=-1)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"type": "gameState",
|
|
"moves": "e2e4", # One move - now it's black's turn
|
|
"status": "resign", # Game ended with resign
|
|
}
|
|
result = _process_game_event(event, ctx, state, meta)
|
|
assert result is False # Game should end
|
|
|
|
def test_process_game_event_unchanged_position(self) -> None:
|
|
"""Test processing event with unchanged position."""
|
|
api = MagicMock()
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
state = GameState(last_handled_len=2, color="white")
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {"type": "gameState", "moves": "e2e4 e7e5"}
|
|
result = _process_game_event(event, ctx, state, meta)
|
|
assert result is True
|
|
|
|
def test_process_game_event_color_unknown(self) -> None:
|
|
"""Test processing event with unknown color."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
state = GameState(last_handled_len=-1)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {"type": "gameState", "moves": "e2e4"}
|
|
result = _process_game_event(event, ctx, state, meta)
|
|
assert result is True
|
|
assert state.last_handled_len == 1
|
|
|
|
def test_process_game_event_color_unknown_on_gamefull(self) -> None:
|
|
"""Test processing gameFull event with still unknown color.
|
|
|
|
This covers the branch where event_type is gameFull but color
|
|
is not determined (e.g., spectator watching game).
|
|
"""
|
|
api = MagicMock()
|
|
# Return a user id that doesn't match either player
|
|
api.get_my_user_id.return_value = "spectator"
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
state = GameState(last_handled_len=-1)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
event: Event = {
|
|
"type": "gameFull",
|
|
"state": {"moves": "e2e4", "status": "started"},
|
|
"white": {"id": "player1"},
|
|
"black": {"id": "player2"},
|
|
}
|
|
result = _process_game_event(event, ctx, state, meta)
|
|
assert result is True
|
|
# last_handled_len should NOT be updated for gameFull with unknown color
|
|
assert state.last_handled_len == -1
|
|
|
|
|
|
class TestWritePgnToLog:
|
|
"""Tests for _write_pgn_to_log."""
|
|
|
|
def test_write_pgn_to_log(self, tmp_path: Path) -> None:
|
|
"""Test writing PGN to log file."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("header\n")
|
|
board = chess.Board()
|
|
board.push_uci("e2e4")
|
|
meta = GameMeta(
|
|
game_id="game1",
|
|
bot_version=1,
|
|
site_url="https://lichess.org/game1",
|
|
date_iso="2021.01.01",
|
|
white_name="White",
|
|
black_name="Black",
|
|
)
|
|
_write_pgn_to_log(log_path, board, meta)
|
|
content = log_path.read_text()
|
|
assert "PGN:" in content
|
|
assert "e4" in content
|
|
|
|
|
|
class TestRunAnalysisSubprocess:
|
|
"""Tests for _run_analysis_subprocess."""
|
|
|
|
def test_run_analysis_subprocess_script_not_found(self, tmp_path: Path) -> None:
|
|
"""Test analysis when script not found."""
|
|
log_path = tmp_path / "game.log"
|
|
with patch("python_pkg.lichess_bot.main.Path") as mock_path:
|
|
mock_script = MagicMock()
|
|
mock_script.is_file.return_value = False
|
|
resolve = mock_path.return_value.resolve.return_value
|
|
resolve.parent.parent.__truediv__.return_value.__truediv__.return_value = (
|
|
mock_script
|
|
)
|
|
result = _run_analysis_subprocess("game1", log_path, 10)
|
|
assert result is None
|
|
|
|
def test_run_analysis_subprocess_success(self, tmp_path: Path) -> None:
|
|
"""Test successful analysis subprocess."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("test")
|
|
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = iter([" 1 e4\n", " 2 e5\n"])
|
|
mock_proc.stderr.read.return_value = ""
|
|
mock_proc.wait.return_value = 0
|
|
mock_proc.__enter__ = MagicMock(return_value=mock_proc)
|
|
mock_proc.__exit__ = MagicMock(return_value=False)
|
|
|
|
with (
|
|
patch("python_pkg.lichess_bot.main.Path") as mock_path,
|
|
patch("subprocess.Popen", return_value=mock_proc),
|
|
):
|
|
mock_script = MagicMock()
|
|
mock_script.is_file.return_value = True
|
|
resolve = mock_path.return_value.resolve.return_value
|
|
resolve.parent.parent.__truediv__.return_value.__truediv__.return_value = (
|
|
mock_script
|
|
)
|
|
result = _run_analysis_subprocess("game1", log_path, 2)
|
|
|
|
assert result is not None
|
|
|
|
|
|
class TestProcessAnalysisOutput:
|
|
"""Tests for _process_analysis_output."""
|
|
|
|
def test_process_analysis_output_success(self) -> None:
|
|
"""Test processing analysis output successfully."""
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = iter([" 1 e4\n", " 2 e5\n"])
|
|
mock_proc.stderr.read.return_value = ""
|
|
mock_proc.wait.return_value = 0
|
|
|
|
result = _process_analysis_output(mock_proc, "game1", 2)
|
|
assert result is not None
|
|
assert "e4" in result
|
|
|
|
def test_process_analysis_output_error_exit(self) -> None:
|
|
"""Test processing analysis output with error exit."""
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = iter(["output\n"])
|
|
mock_proc.stderr.read.return_value = "error message"
|
|
mock_proc.wait.return_value = 1
|
|
|
|
result = _process_analysis_output(mock_proc, "game1", 1)
|
|
assert result is not None
|
|
assert "stderr" in result
|
|
|
|
def test_process_analysis_output_none_pipes(self) -> None:
|
|
"""Test processing analysis output with None pipes."""
|
|
mock_proc = MagicMock()
|
|
mock_proc.stdout = None
|
|
mock_proc.stderr = None
|
|
|
|
with pytest.raises(RuntimeError, match="pipes unexpectedly None"):
|
|
_process_analysis_output(mock_proc, "game1", 1)
|
|
|
|
|
|
class TestLogAnalysisProgress:
|
|
"""Tests for _log_analysis_progress."""
|
|
|
|
def test_log_analysis_progress_with_total(self) -> None:
|
|
"""Test logging progress with known total."""
|
|
with patch("python_pkg.lichess_bot.main._logger") as mock_logger:
|
|
_log_analysis_progress("game1", 5, 10)
|
|
mock_logger.info.assert_called_once()
|
|
call_args = mock_logger.info.call_args[0]
|
|
assert "50%" in call_args[0] % call_args[1:]
|
|
|
|
def test_log_analysis_progress_zero_total(self) -> None:
|
|
"""Test logging progress with zero total."""
|
|
with patch("python_pkg.lichess_bot.main._logger") as mock_logger:
|
|
_log_analysis_progress("game1", 5, 0)
|
|
mock_logger.info.assert_called_once()
|
|
call_args = mock_logger.info.call_args[0]
|
|
assert "unknown" in call_args[0]
|
|
|
|
|
|
class TestInsertAnalysisIntoLog:
|
|
"""Tests for _insert_analysis_into_log."""
|
|
|
|
def test_insert_analysis_before_pgn(self, tmp_path: Path) -> None:
|
|
"""Test inserting analysis before PGN section."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("header\n\nPGN:\n1. e4\n")
|
|
meta = GameMeta(
|
|
game_id="game1",
|
|
bot_version=1,
|
|
date_iso="2021.01.01",
|
|
white_name="White",
|
|
black_name="Black",
|
|
)
|
|
_insert_analysis_into_log(log_path, "Analysis here", meta)
|
|
content = log_path.read_text()
|
|
assert "ANALYSIS:" in content
|
|
assert content.index("ANALYSIS:") < content.index("PGN:")
|
|
|
|
def test_insert_analysis_at_start(self, tmp_path: Path) -> None:
|
|
"""Test inserting analysis when PGN at start."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("PGN:\n1. e4\n")
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
_insert_analysis_into_log(log_path, "Analysis here", meta)
|
|
content = log_path.read_text()
|
|
assert "ANALYSIS:" in content
|
|
|
|
def test_insert_analysis_no_pgn(self, tmp_path: Path) -> None:
|
|
"""Test inserting analysis when no PGN section."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("header\n")
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
_insert_analysis_into_log(log_path, "Analysis here", meta)
|
|
content = log_path.read_text()
|
|
assert "ANALYSIS:" in content
|
|
|
|
def test_insert_analysis_oserror(self, tmp_path: Path) -> None:
|
|
"""Test inserting analysis with OSError."""
|
|
log_path = tmp_path / "nonexistent" / "game.log"
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
# Should not raise, just log debug
|
|
_insert_analysis_into_log(log_path, "Analysis", meta)
|
|
|
|
|
|
class TestFinalizeGame:
|
|
"""Tests for _finalize_game."""
|
|
|
|
def test_finalize_game_no_log_path(self) -> None:
|
|
"""Test finalize game with no log path."""
|
|
state = GameState(log_path=None)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
_finalize_game(state, meta) # Should not raise
|
|
|
|
def test_finalize_game_write_error(self, tmp_path: Path) -> None:
|
|
"""Test finalize game with write error."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("header")
|
|
state = GameState(log_path=log_path)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
|
|
with patch(
|
|
"python_pkg.lichess_bot.main._write_pgn_to_log",
|
|
side_effect=OSError("error"),
|
|
):
|
|
_finalize_game(state, meta) # Should not raise
|
|
|
|
def test_finalize_game_type_error_on_move_stack(self, tmp_path: Path) -> None:
|
|
"""Test finalize game with TypeError on move_stack."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("header\n")
|
|
state = GameState(log_path=log_path)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
|
|
mock_board = MagicMock()
|
|
# Create a property that raises TypeError when accessed
|
|
type(mock_board).move_stack = property(
|
|
lambda _self: (_ for _ in ()).throw(TypeError())
|
|
)
|
|
state.board = mock_board
|
|
|
|
with patch("python_pkg.lichess_bot.main._write_pgn_to_log"):
|
|
_finalize_game(state, meta) # Should not raise
|
|
|
|
def test_finalize_game_analysis_error(self, tmp_path: Path) -> None:
|
|
"""Test finalize game with analysis error."""
|
|
log_path = tmp_path / "game.log"
|
|
log_path.write_text("header\n")
|
|
state = GameState(log_path=log_path)
|
|
meta = GameMeta(game_id="game1", bot_version=1)
|
|
|
|
with (
|
|
patch("python_pkg.lichess_bot.main._write_pgn_to_log"),
|
|
patch(
|
|
"python_pkg.lichess_bot.main._run_analysis_subprocess",
|
|
side_effect=OSError("error"),
|
|
),
|
|
):
|
|
_finalize_game(state, meta) # Should not raise
|
|
|
|
|
|
class TestHandleGame:
|
|
"""Tests for _handle_game."""
|
|
|
|
def test_handle_game_success(self, tmp_path: Path) -> None:
|
|
"""Test handling a game successfully."""
|
|
api = MagicMock()
|
|
api.get_my_user_id.return_value = "mybot"
|
|
api.stream_game_events.return_value = iter(
|
|
[
|
|
{
|
|
"type": "gameFull",
|
|
"state": {"moves": "", "status": "started"},
|
|
"white": {"id": "mybot"},
|
|
"black": {"id": "opp"},
|
|
},
|
|
{"type": "gameState", "moves": "e2e4", "status": "mate"},
|
|
]
|
|
)
|
|
engine = MagicMock()
|
|
engine.max_time_sec = 5.0
|
|
engine.choose_move_with_explanation.return_value = (None, "no moves")
|
|
ctx = BotContext(api=api, engine=engine, bot_version=1)
|
|
|
|
with (
|
|
patch("python_pkg.lichess_bot.main.Path.cwd", return_value=tmp_path),
|
|
patch(
|
|
"python_pkg.lichess_bot.main._run_analysis_subprocess",
|
|
return_value=None,
|
|
),
|
|
):
|
|
_handle_game("game1", ctx, None)
|
|
|
|
def test_handle_game_request_error(self, tmp_path: Path) -> None:
|
|
"""Test handling a game with request error."""
|
|
api = MagicMock()
|
|
api.stream_game_events.side_effect = requests.RequestException("error")
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
|
|
with (
|
|
patch("python_pkg.lichess_bot.main.Path.cwd", return_value=tmp_path),
|
|
patch(
|
|
"python_pkg.lichess_bot.main._run_analysis_subprocess",
|
|
return_value=None,
|
|
),
|
|
):
|
|
_handle_game("game1", ctx, None) # Should not raise
|
|
|
|
def test_handle_game_skips_chat_events(self, tmp_path: Path) -> None:
|
|
"""Test handling a game skips chat events."""
|
|
api = MagicMock()
|
|
api.stream_game_events.return_value = iter(
|
|
[
|
|
{"type": "chatLine", "text": "hello"},
|
|
{"type": "opponentGone", "gone": True},
|
|
]
|
|
)
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
|
|
with (
|
|
patch("python_pkg.lichess_bot.main.Path.cwd", return_value=tmp_path),
|
|
patch(
|
|
"python_pkg.lichess_bot.main._run_analysis_subprocess",
|
|
return_value=None,
|
|
),
|
|
):
|
|
_handle_game("game1", ctx, None)
|
|
|
|
|
|
class TestHandleChallenge:
|
|
"""Tests for _handle_challenge."""
|
|
|
|
def test_accept_standard_blitz(self) -> None:
|
|
"""Test accepting standard blitz challenge."""
|
|
api = MagicMock()
|
|
challenge: Event = {
|
|
"id": "ch1",
|
|
"variant": {"key": "standard"},
|
|
"speed": "blitz",
|
|
}
|
|
_handle_challenge(challenge, api, decline_correspondence=False)
|
|
api.accept_challenge.assert_called_once_with("ch1")
|
|
|
|
def test_decline_variant(self) -> None:
|
|
"""Test declining non-standard variant."""
|
|
api = MagicMock()
|
|
challenge: Event = {
|
|
"id": "ch1",
|
|
"variant": {"key": "chess960"},
|
|
"speed": "blitz",
|
|
}
|
|
_handle_challenge(challenge, api, decline_correspondence=False)
|
|
api.decline_challenge.assert_called_once()
|
|
|
|
def test_decline_correspondence(self) -> None:
|
|
"""Test declining correspondence when flag set."""
|
|
api = MagicMock()
|
|
challenge: Event = {
|
|
"id": "ch1",
|
|
"variant": {"key": "standard"},
|
|
"speed": "correspondence",
|
|
}
|
|
_handle_challenge(challenge, api, decline_correspondence=True)
|
|
api.decline_challenge.assert_called_once()
|
|
|
|
def test_accept_correspondence_when_allowed(self) -> None:
|
|
"""Test accepting correspondence when flag not set."""
|
|
api = MagicMock()
|
|
challenge: Event = {
|
|
"id": "ch1",
|
|
"variant": {"key": "standard"},
|
|
"speed": "correspondence",
|
|
}
|
|
_handle_challenge(challenge, api, decline_correspondence=False)
|
|
api.decline_challenge.assert_called_once() # Still declined due to perf_ok
|
|
|
|
def test_invalid_variant_data(self) -> None:
|
|
"""Test handling invalid variant data."""
|
|
api = MagicMock()
|
|
challenge: Event = {
|
|
"id": "ch1",
|
|
"variant": "invalid",
|
|
"speed": "blitz",
|
|
}
|
|
_handle_challenge(challenge, api, decline_correspondence=False)
|
|
api.accept_challenge.assert_called_once()
|
|
|
|
|
|
class TestProcessBotEvent:
|
|
"""Tests for _process_bot_event."""
|
|
|
|
def test_process_challenge_event(self) -> None:
|
|
"""Test processing challenge event."""
|
|
api = MagicMock()
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
event: Event = {
|
|
"type": "challenge",
|
|
"challenge": {
|
|
"id": "ch1",
|
|
"variant": {"key": "standard"},
|
|
"speed": "blitz",
|
|
},
|
|
}
|
|
_process_bot_event(event, ctx, game_threads)
|
|
api.accept_challenge.assert_called_once()
|
|
|
|
def test_process_game_start_event(self, tmp_path: Path) -> None:
|
|
"""Test processing gameStart event."""
|
|
api = MagicMock()
|
|
api.stream_game_events.return_value = iter([])
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
event: Event = {"type": "gameStart", "game": {"id": "game1"}}
|
|
|
|
with (
|
|
patch("python_pkg.lichess_bot.main.Path.cwd", return_value=tmp_path),
|
|
patch(
|
|
"python_pkg.lichess_bot.main._run_analysis_subprocess",
|
|
return_value=None,
|
|
),
|
|
):
|
|
_process_bot_event(event, ctx, game_threads)
|
|
|
|
assert "game1" in game_threads
|
|
game_threads["game1"].join(timeout=1)
|
|
|
|
def test_process_game_start_existing_thread(self) -> None:
|
|
"""Test processing gameStart with existing alive thread."""
|
|
api = MagicMock()
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
mock_thread = MagicMock(spec=threading.Thread)
|
|
mock_thread.is_alive.return_value = True
|
|
game_threads: GameThreads = {"game1": mock_thread}
|
|
event: Event = {"type": "gameStart", "game": {"id": "game1"}}
|
|
_process_bot_event(event, ctx, game_threads)
|
|
# Should not create new thread
|
|
assert game_threads["game1"] is mock_thread
|
|
|
|
def test_process_game_finish_event(self) -> None:
|
|
"""Test processing gameFinish event."""
|
|
api = MagicMock()
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
event: Event = {"type": "gameFinish", "game": {"id": "game1"}}
|
|
with patch("python_pkg.lichess_bot.main._logger") as mock_logger:
|
|
_process_bot_event(event, ctx, game_threads)
|
|
mock_logger.info.assert_called()
|
|
|
|
def test_process_unknown_event(self) -> None:
|
|
"""Test processing unknown event."""
|
|
api = MagicMock()
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
event: Event = {"type": "unknown", "data": "test"}
|
|
with patch("python_pkg.lichess_bot.main._logger") as mock_logger:
|
|
_process_bot_event(event, ctx, game_threads)
|
|
mock_logger.debug.assert_called()
|
|
|
|
def test_process_challenge_invalid_data(self) -> None:
|
|
"""Test processing challenge with invalid data."""
|
|
api = MagicMock()
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
event: Event = {"type": "challenge", "challenge": "invalid"}
|
|
_process_bot_event(event, ctx, game_threads)
|
|
api.accept_challenge.assert_not_called()
|
|
|
|
def test_process_game_start_invalid_data(self) -> None:
|
|
"""Test processing gameStart with invalid data."""
|
|
api = MagicMock()
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
event: Event = {"type": "gameStart", "game": "invalid"}
|
|
_process_bot_event(event, ctx, game_threads)
|
|
assert len(game_threads) == 0
|
|
|
|
|
|
class TestStreamBotEvents:
|
|
"""Tests for _stream_bot_events."""
|
|
|
|
def test_stream_bot_events(self) -> None:
|
|
"""Test streaming bot events."""
|
|
api = MagicMock()
|
|
api.stream_events.return_value = iter([{"type": "test"}])
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
events = list(_stream_bot_events(ctx))
|
|
assert len(events) == 1
|
|
|
|
|
|
class TestRunEventLoopIteration:
|
|
"""Tests for _run_event_loop_iteration."""
|
|
|
|
def test_run_event_loop_iteration(self) -> None:
|
|
"""Test running event loop iteration."""
|
|
api = MagicMock()
|
|
api.stream_events.return_value = iter([{"type": "unknown"}])
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
result = _run_event_loop_iteration(ctx, game_threads)
|
|
assert result == 0
|
|
|
|
|
|
class TestSafeEventLoopIteration:
|
|
"""Tests for _safe_event_loop_iteration."""
|
|
|
|
def test_safe_event_loop_iteration_success(self) -> None:
|
|
"""Test safe event loop iteration success."""
|
|
api = MagicMock()
|
|
api.stream_events.return_value = iter([])
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
result = _safe_event_loop_iteration(ctx, game_threads, 0)
|
|
assert result == 0
|
|
|
|
def test_safe_event_loop_iteration_error(self) -> None:
|
|
"""Test safe event loop iteration with error."""
|
|
api = MagicMock()
|
|
api.stream_events.side_effect = requests.RequestException("error")
|
|
ctx = BotContext(api=api, engine=MagicMock(), bot_version=1)
|
|
game_threads: GameThreads = {}
|
|
with patch("python_pkg.lichess_bot.main.backoff_sleep", return_value=5):
|
|
result = _safe_event_loop_iteration(ctx, game_threads, 2)
|
|
assert result == 5
|
|
|
|
|
|
class TestRunBot:
|
|
"""Tests for run_bot."""
|
|
|
|
def test_run_bot_no_token(self) -> None:
|
|
"""Test run_bot without token raises error."""
|
|
with (
|
|
patch.dict(os.environ, {}, clear=True),
|
|
pytest.raises(RuntimeError, match="LICHESS_TOKEN"),
|
|
):
|
|
run_bot()
|
|
|
|
def test_run_bot_with_token(self) -> None:
|
|
"""Test run_bot with token starts event loop."""
|
|
|
|
class _StopLoopError(Exception):
|
|
"""Custom exception to stop the loop."""
|
|
|
|
def stop_loop(*_args: object, **_kwargs: object) -> None:
|
|
raise _StopLoopError
|
|
|
|
with (
|
|
patch.dict(os.environ, {"LICHESS_TOKEN": "test_token"}),
|
|
patch(
|
|
"python_pkg.lichess_bot.main.get_and_increment_version",
|
|
return_value=1,
|
|
),
|
|
patch("python_pkg.lichess_bot.main.LichessAPI"),
|
|
patch("python_pkg.lichess_bot.main.RandomEngine"),
|
|
patch(
|
|
"python_pkg.lichess_bot.main._safe_event_loop_iteration",
|
|
side_effect=stop_loop,
|
|
),
|
|
pytest.raises(_StopLoopError),
|
|
):
|
|
run_bot("DEBUG", decline_correspondence=True)
|
|
|
|
|
|
class TestMain:
|
|
"""Tests for main function."""
|
|
|
|
def test_main_parses_args(self) -> None:
|
|
"""Test main parses command line arguments."""
|
|
|
|
class _StopExecutionError(Exception):
|
|
"""Custom exception to stop execution."""
|
|
|
|
with (
|
|
patch(
|
|
"sys.argv",
|
|
["main.py", "--log-level", "DEBUG", "--decline-correspondence"],
|
|
),
|
|
patch(
|
|
"python_pkg.lichess_bot.main.run_bot", side_effect=_StopExecutionError
|
|
) as mock_run_bot,
|
|
pytest.raises(_StopExecutionError),
|
|
):
|
|
main()
|
|
mock_run_bot.assert_called_once_with("DEBUG", decline_correspondence=True)
|