From 4db978b67b05bccbca706d5a1710798a29440fcd Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Tue, 2 Dec 2025 21:53:15 +0100 Subject: [PATCH] Add comprehensive tests for lichess_bot main.py and lichess_api.py - 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) --- python_pkg/lichess_bot/tests/test_main.py | 1093 +++++++++++++++++++++ python_pkg/screen_locker/workout_log.json | 6 + 2 files changed, 1099 insertions(+) create mode 100644 python_pkg/lichess_bot/tests/test_main.py diff --git a/python_pkg/lichess_bot/tests/test_main.py b/python_pkg/lichess_bot/tests/test_main.py new file mode 100644 index 0000000..99b173f --- /dev/null +++ b/python_pkg/lichess_bot/tests/test_main.py @@ -0,0 +1,1093 @@ +"""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) diff --git a/python_pkg/screen_locker/workout_log.json b/python_pkg/screen_locker/workout_log.json index dae29cc..8a19852 100644 --- a/python_pkg/screen_locker/workout_log.json +++ b/python_pkg/screen_locker/workout_log.json @@ -22,5 +22,11 @@ "workout_data": { "type": "strength" } + }, + "2025-12-02": { + "timestamp": "2025-12-02T19:52:07.899613+00:00", + "workout_data": { + "type": "strength" + } } }