From 3ef256973fbbe6af6188f1786b9b918d2f5b6fa5 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 1 Dec 2025 20:41:51 +0100 Subject: [PATCH] feat(tests): add comprehensive tests for lichess_bot engine and API - Add test_engine.py with 100% coverage for engine.py - Tests for Engine class initialization - Tests for choose_move with various scenarios - Tests for best move parsing and validation - Tests for checkmate and stalemate detection - Add test_lichess_api.py with 98% coverage for lichess_api.py - Tests for API initialization and request handling - Tests for stream_events with proper infinite loop handling - Tests for challenge accept/decline - Tests for game streaming and move submission - Tests for rate limit handling and retry logic - Remove unreachable return statement in lichess_api.make_move Coverage: engine.py 17% -> 100%, lichess_api.py 0% -> 98% --- python_pkg/lichess_bot/lichess_api.py | 1 - python_pkg/lichess_bot/tests/test_engine.py | 300 +++++++++ .../lichess_bot/tests/test_lichess_api.py | 624 ++++++++++++++++++ 3 files changed, 924 insertions(+), 1 deletion(-) create mode 100644 python_pkg/lichess_bot/tests/test_engine.py create mode 100644 python_pkg/lichess_bot/tests/test_lichess_api.py diff --git a/python_pkg/lichess_bot/lichess_api.py b/python_pkg/lichess_bot/lichess_api.py index 0e7f9ee..d27e00c 100644 --- a/python_pkg/lichess_bot/lichess_api.py +++ b/python_pkg/lichess_bot/lichess_api.py @@ -184,7 +184,6 @@ class LichessAPI: if r.status_code in (HTTPStatus.BAD_REQUEST, HTTPStatus.CONFLICT): # Likely not our turn or move already played; do not retry to avoid spam r.raise_for_status() - return if r.status_code == HTTPStatus.TOO_MANY_REQUESTS: _logger.warning("HTTP POST %s -> 429; retrying once after 0.5s", url) time.sleep(0.5) diff --git a/python_pkg/lichess_bot/tests/test_engine.py b/python_pkg/lichess_bot/tests/test_engine.py new file mode 100644 index 0000000..3204497 --- /dev/null +++ b/python_pkg/lichess_bot/tests/test_engine.py @@ -0,0 +1,300 @@ +"""Unit tests for lichess_bot engine module.""" + +# ruff: noqa: SLF001 +# Tests need to access private members to verify internal logic + +import json +from pathlib import Path +import subprocess +from unittest.mock import MagicMock, patch + +import chess +import pytest + +from python_pkg.lichess_bot.engine import RandomEngine + + +class TestRandomEngineInit: + """Tests for RandomEngine initialization.""" + + def test_init_with_missing_engine_raises(self) -> None: + """Test that missing engine raises FileNotFoundError.""" + with pytest.raises(FileNotFoundError, match="C engine not found"): + RandomEngine(engine_path="/nonexistent/path/to/engine") + + def test_init_with_non_executable_raises(self, tmp_path: Path) -> None: + """Test that non-executable engine raises FileNotFoundError.""" + fake_engine = tmp_path / "fake_engine" + fake_engine.write_text("not executable") + with pytest.raises(FileNotFoundError, match="not executable"): + RandomEngine(engine_path=str(fake_engine)) + + def test_init_with_valid_engine(self, tmp_path: Path) -> None: + """Test successful initialization with valid engine.""" + fake_engine = tmp_path / "fake_engine" + fake_engine.write_text("#!/bin/bash\necho test") + fake_engine.chmod(0o755) + + engine = RandomEngine(engine_path=str(fake_engine), max_time_sec=1.0, depth=5) + + assert engine.engine_path == fake_engine + assert engine.max_time_sec == 1.0 + assert engine.depth == 5 + + +class TestCallEngine: + """Tests for _call_engine method.""" + + @pytest.fixture + def mock_engine(self, tmp_path: Path) -> RandomEngine: + """Create an engine with a mock executable.""" + fake_engine = tmp_path / "fake_engine" + fake_engine.write_text("#!/bin/bash\necho test") + fake_engine.chmod(0o755) + return RandomEngine(engine_path=str(fake_engine)) + + def test_call_engine_success(self, mock_engine: RandomEngine) -> None: + """Test successful engine call.""" + mock_result = MagicMock() + mock_result.stdout = "e2e4\n" + + with patch("subprocess.run", return_value=mock_result): + result = mock_engine._call_engine(["--test"], timeout=1.0) + + assert result == "e2e4" + + def test_call_engine_called_process_error(self, mock_engine: RandomEngine) -> None: + """Test engine call with CalledProcessError.""" + error = subprocess.CalledProcessError(1, "cmd", stderr="engine error") + + with ( + patch("subprocess.run", side_effect=error), + pytest.raises(RuntimeError, match="C engine failed"), + ): + mock_engine._call_engine(["--test"], timeout=1.0) + + def test_call_engine_timeout(self, mock_engine: RandomEngine) -> None: + """Test engine call timeout.""" + error = subprocess.TimeoutExpired("cmd", 1.0) + + with ( + patch("subprocess.run", side_effect=error), + pytest.raises(TimeoutError, match="C engine timed out"), + ): + mock_engine._call_engine(["--test"], timeout=1.0) + + +class TestChooseMove: + """Tests for choose_move methods.""" + + @pytest.fixture + def mock_engine(self, tmp_path: Path) -> RandomEngine: + """Create an engine with a mock executable.""" + fake_engine = tmp_path / "fake_engine" + fake_engine.write_text("#!/bin/bash\necho test") + fake_engine.chmod(0o755) + return RandomEngine(engine_path=str(fake_engine)) + + def test_choose_move_returns_valid_move(self, mock_engine: RandomEngine) -> None: + """Test choose_move returns a valid move.""" + board = chess.Board() + + with patch.object(mock_engine, "_call_engine", return_value="e2e4"): + move = mock_engine.choose_move(board) + + assert move == chess.Move.from_uci("e2e4") + assert move in board.legal_moves + + def test_choose_move_with_explanation_no_legal_moves( + self, mock_engine: RandomEngine + ) -> None: + """Test choose_move_with_explanation when no legal moves.""" + # Create a checkmate position - black king checkmated by rook + board = chess.Board("k7/2K5/8/8/8/8/8/R7 b - - 0 1") + + move, reason = mock_engine.choose_move_with_explanation( + board, time_budget_sec=1.0 + ) + + assert move is None + assert reason == "no_legal_moves" + + def test_choose_move_with_explanation_invalid_move( + self, mock_engine: RandomEngine + ) -> None: + """Test choose_move_with_explanation with invalid move from engine.""" + board = chess.Board() + + with ( + patch.object(mock_engine, "_call_engine", return_value="invalid"), + pytest.raises(RuntimeError, match="Engine returned invalid move"), + ): + mock_engine.choose_move_with_explanation(board, time_budget_sec=1.0) + + def test_choose_move_with_explanation_illegal_move( + self, mock_engine: RandomEngine + ) -> None: + """Test choose_move_with_explanation with illegal move from engine.""" + board = chess.Board() + + # e2e5 is a valid UCI format but illegal from starting position + with ( + patch.object(mock_engine, "_call_engine", return_value="e2e5"), + pytest.raises(RuntimeError, match="Engine returned illegal move"), + ): + mock_engine.choose_move_with_explanation(board, time_budget_sec=1.0) + + +class TestParseEngineAnalysis: + """Tests for _parse_engine_analysis method.""" + + @pytest.fixture + def mock_engine(self, tmp_path: Path) -> RandomEngine: + """Create an engine with a mock executable.""" + fake_engine = tmp_path / "fake_engine" + fake_engine.write_text("#!/bin/bash\necho test") + fake_engine.chmod(0o755) + return RandomEngine(engine_path=str(fake_engine)) + + def test_parse_valid_json(self, mock_engine: RandomEngine) -> None: + """Test parsing valid JSON output.""" + board = chess.Board() + legal_moves = list(board.legal_moves) + output = json.dumps( + { + "analyze": {"candidate_score": 0.5}, + "chosen_move": "e2e4", + "chosen_index": 0, + } + ) + + score, cand_expl, best_move, best_expl = mock_engine._parse_engine_analysis( + output, legal_moves + ) + + assert score == 0.5 + assert best_move == chess.Move.from_uci("e2e4") + assert "candidate_score" in cand_expl + assert "chosen_move" in best_expl + + def test_parse_invalid_json(self, mock_engine: RandomEngine) -> None: + """Test parsing invalid JSON output.""" + board = chess.Board() + legal_moves = list(board.legal_moves) + + score, cand_expl, best_move, _best_expl = mock_engine._parse_engine_analysis( + "not json", legal_moves + ) + + assert score == 0.0 + assert best_move is None + assert cand_expl == "not json" + + def test_parse_json_with_illegal_move(self, mock_engine: RandomEngine) -> None: + """Test parsing JSON with illegal move.""" + legal_moves = [chess.Move.from_uci("e2e4")] + output = json.dumps( + { + "analyze": {"candidate_score": 1.0}, + "chosen_move": "a1a8", # Not in legal moves + "chosen_index": 0, + } + ) + + score, _cand_expl, best_move, _best_expl = mock_engine._parse_engine_analysis( + output, legal_moves + ) + + assert score == 1.0 + assert best_move is None # Move not in legal moves + + def test_parse_json_without_chosen_move(self, mock_engine: RandomEngine) -> None: + """Test parsing JSON without chosen_move field.""" + legal_moves = [chess.Move.from_uci("e2e4")] + output = json.dumps( + { + "analyze": {"candidate_score": 0.7}, + "chosen_index": 0, + # No chosen_move field + } + ) + + score, _cand_expl, best_move, _best_expl = mock_engine._parse_engine_analysis( + output, legal_moves + ) + + assert score == 0.7 + assert best_move is None + + def test_parse_json_without_score(self, mock_engine: RandomEngine) -> None: + """Test parsing JSON without candidate_score field.""" + board = chess.Board() + legal_moves = list(board.legal_moves) + output = json.dumps( + { + "analyze": {}, # No candidate_score + "chosen_move": "e2e4", + "chosen_index": 0, + } + ) + + score, _cand_expl, best_move, _best_expl = mock_engine._parse_engine_analysis( + output, legal_moves + ) + + assert score == 0.0 # Default score + assert best_move == chess.Move.from_uci("e2e4") + + +class TestEvaluateProposedMove: + """Tests for evaluate_proposed_move_with_suggestion method.""" + + @pytest.fixture + def mock_engine(self, tmp_path: Path) -> RandomEngine: + """Create an engine with a mock executable.""" + fake_engine = tmp_path / "fake_engine" + fake_engine.write_text("#!/bin/bash\necho test") + fake_engine.chmod(0o755) + return RandomEngine(engine_path=str(fake_engine)) + + def test_evaluate_no_legal_moves(self, mock_engine: RandomEngine) -> None: + """Test evaluate when no legal moves available.""" + # Create a checkmate position - black king checkmated by rook + board = chess.Board("k7/2K5/8/8/8/8/8/R7 b - - 0 1") + + score, cand_expl, best_move, best_expl = ( + mock_engine.evaluate_proposed_move_with_suggestion( + board, "e1e2", time_budget_sec=1.0 + ) + ) + + assert score == 0.0 + assert cand_expl == "no_legal_moves" + assert best_move is None + assert best_expl == "no_best_move" + + assert score == 0.0 + assert cand_expl == "no_legal_moves" + assert best_move is None + assert best_expl == "no_best_move" + + def test_evaluate_with_valid_position(self, mock_engine: RandomEngine) -> None: + """Test evaluate with a valid position.""" + board = chess.Board() + output = json.dumps( + { + "analyze": {"candidate_score": 0.3}, + "chosen_move": "e2e4", + "chosen_index": 0, + } + ) + + with patch.object(mock_engine, "_call_engine", return_value=output): + score, _cand_expl, best_move, _best_expl = ( + mock_engine.evaluate_proposed_move_with_suggestion( + board, "d2d4", time_budget_sec=1.0 + ) + ) + + assert score == 0.3 + assert best_move == chess.Move.from_uci("e2e4") diff --git a/python_pkg/lichess_bot/tests/test_lichess_api.py b/python_pkg/lichess_bot/tests/test_lichess_api.py new file mode 100644 index 0000000..85b2acc --- /dev/null +++ b/python_pkg/lichess_bot/tests/test_lichess_api.py @@ -0,0 +1,624 @@ +"""Unit tests for lichess_bot lichess_api module.""" + +# ruff: noqa: SLF001, S105, ARG001, PT012, TRY003, EM101, PERF402 +# SLF001: Tests need to access private members to verify internal logic +# S105: Test tokens are not real passwords +# ARG001: Mock functions need *args, **kwargs signature +# PT012: pytest.raises can contain multiple statements for generator testing +# TRY003, EM101: Exception messages in tests are fine +# PERF402: We need loop append for generator consumption with exception break + +from http import HTTPStatus +import json +from unittest.mock import MagicMock, patch + +import chess +import pytest +import requests + +from python_pkg.lichess_bot.lichess_api import LichessAPI + + +class _TestTerminationError(Exception): + """Custom exception to break out of infinite loops in tests.""" + + +class TestLichessAPIInit: + """Tests for LichessAPI initialization.""" + + def test_init_creates_session_with_headers(self) -> None: + """Test initialization creates session with proper headers.""" + api = LichessAPI("test_token") + + assert api.token == "test_token" + assert "Bearer test_token" in api.session.headers["Authorization"] + assert "application/json" in api.session.headers["Accept"] + + def test_init_with_custom_session(self) -> None: + """Test initialization with custom session.""" + custom_session = requests.Session() + api = LichessAPI("test_token", session=custom_session) + + assert api.session is custom_session + + +class TestRequest: + """Tests for _request method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance with mocked session.""" + return LichessAPI("test_token") + + def test_request_success(self, api: LichessAPI) -> None: + """Test successful request logs appropriately.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + + with patch.object(api.session, "request", return_value=mock_response): + result = api._request("GET", "http://test.com") + + assert result == mock_response + + def test_request_error_logs_body(self, api: LichessAPI) -> None: + """Test error response logs body snippet.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.BAD_REQUEST + mock_response.text = "Error message here" + + with patch.object(api.session, "request", return_value=mock_response): + result = api._request("GET", "http://test.com") + + assert result == mock_response + + def test_request_error_no_body(self, api: LichessAPI) -> None: + """Test error response without body.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR + mock_response.text = None + + with patch.object(api.session, "request", return_value=mock_response): + result = api._request("GET", "http://test.com") + + assert result == mock_response + + def test_request_raises_for_status(self, api: LichessAPI) -> None: + """Test request raises for status when flag is set.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.NOT_FOUND + mock_response.text = "" + mock_response.raise_for_status.side_effect = requests.HTTPError() + + with ( + patch.object(api.session, "request", return_value=mock_response), + pytest.raises(requests.HTTPError), + ): + api._request("GET", "http://test.com", raise_for_status=True) + + def test_request_exception_handling(self, api: LichessAPI) -> None: + """Test request handles exceptions.""" + with ( + patch.object( + api.session, "request", side_effect=requests.ConnectionError() + ), + pytest.raises(requests.ConnectionError), + ): + api._request("GET", "http://test.com") + + +class TestStreamEvents: + """Tests for stream_events method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_stream_events_yields_json_lines(self, api: LichessAPI) -> None: + """Test stream_events yields parsed JSON lines.""" + # stream_events has a while True loop, so we need to break out of it + # by raising an exception after yielding our test data + + def iter_lines_with_stop(*args: object, **kwargs: object) -> list[str]: + """Return lines then signal generator to stop.""" + return ['{"type": "challenge"}', "", '{"type": "gameStart"}'] + + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + mock_response.iter_lines = iter_lines_with_stop + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + call_count = 0 + + def mock_request(*args: object, **kwargs: object) -> MagicMock: + nonlocal call_count + call_count += 1 + if call_count > 1: + # Break out of while True on second iteration + raise _TestTerminationError("Test termination") + return mock_response + + events_collected: list[dict] = [] + with ( + patch.object(api, "_request", side_effect=mock_request), + pytest.raises(_TestTerminationError), + ): + for event in api.stream_events(): + events_collected.append(event) + + assert len(events_collected) == 2 + assert events_collected[0]["type"] == "challenge" + assert events_collected[1]["type"] == "gameStart" + + def test_stream_events_skips_invalid_json(self, api: LichessAPI) -> None: + """Test stream_events skips non-JSON lines.""" + + def iter_lines_with_invalid(*args: object, **kwargs: object) -> list[str]: + return ['{"type": "challenge"}', "not json", '{"type": "gameStart"}'] + + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + mock_response.iter_lines = iter_lines_with_invalid + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + call_count = 0 + + def mock_request(*args: object, **kwargs: object) -> MagicMock: + nonlocal call_count + call_count += 1 + if call_count > 1: + raise _TestTerminationError("Test termination") + return mock_response + + events_collected: list[dict] = [] + with ( + patch.object(api, "_request", side_effect=mock_request), + pytest.raises(_TestTerminationError), + ): + for event in api.stream_events(): + events_collected.append(event) + + assert len(events_collected) == 2 + + def test_stream_events_handles_rate_limit(self, api: LichessAPI) -> None: + """Test stream_events backs off on rate limit.""" + mock_response_429 = MagicMock() + mock_response_429.status_code = HTTPStatus.TOO_MANY_REQUESTS + mock_response_429.raise_for_status.side_effect = requests.HTTPError( + response=MagicMock(status_code=HTTPStatus.TOO_MANY_REQUESTS) + ) + mock_response_429.__enter__ = MagicMock(return_value=mock_response_429) + mock_response_429.__exit__ = MagicMock(return_value=False) + + def iter_lines_ok(*args: object, **kwargs: object) -> list[str]: + return ['{"type": "test"}'] + + mock_response_ok = MagicMock() + mock_response_ok.status_code = HTTPStatus.OK + mock_response_ok.iter_lines = iter_lines_ok + mock_response_ok.__enter__ = MagicMock(return_value=mock_response_ok) + mock_response_ok.__exit__ = MagicMock(return_value=False) + + call_count = 0 + + def mock_request(*args: object, **kwargs: object) -> MagicMock: + nonlocal call_count + call_count += 1 + if call_count == 1: + return mock_response_429 + if call_count == 2: + return mock_response_ok + raise _TestTerminationError("Test termination") + + events_collected: list[dict] = [] + with ( + patch.object(api, "_request", side_effect=mock_request), + patch("python_pkg.lichess_bot.lichess_api.time.sleep"), + pytest.raises(_TestTerminationError), + ): + for event in api.stream_events(): + events_collected.append(event) + + assert len(events_collected) == 1 + assert call_count == 3 # 429 + OK + termination + + +class TestChallenges: + """Tests for challenge-related methods.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_accept_challenge(self, api: LichessAPI) -> None: + """Test accepting a challenge.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + + with patch.object(api, "_request", return_value=mock_response) as mock_req: + api.accept_challenge("test_challenge_id") + + mock_req.assert_called_once() + call_args = mock_req.call_args + assert "test_challenge_id" in call_args[0][1] + assert call_args[1]["raise_for_status"] is True + + def test_decline_challenge(self, api: LichessAPI) -> None: + """Test declining a challenge.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + + with patch.object(api, "_request", return_value=mock_response) as mock_req: + api.decline_challenge("test_challenge_id", reason="tooSlow") + + mock_req.assert_called_once() + call_args = mock_req.call_args + assert "decline" in call_args[0][1] + assert call_args[1]["data"]["reason"] == "tooSlow" + + +class TestParseGameFullEvent: + """Tests for _parse_game_full_event method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_parse_game_full_as_white(self, api: LichessAPI) -> None: + """Test parsing gameFull event when playing as white.""" + event = { + "white": {"id": "my_user"}, + "black": {"id": "opponent"}, + "state": {"moves": "e2e4 e7e5"}, + } + board = chess.Board() + + with patch.object(api, "get_my_user_id", return_value="my_user"): + color = api._parse_game_full_event(event, board, "unknown") + + assert color == "white" + assert len(board.move_stack) == 2 + + def test_parse_game_full_as_black(self, api: LichessAPI) -> None: + """Test parsing gameFull event when playing as black.""" + event = { + "white": {"id": "opponent"}, + "black": {"id": "my_user"}, + "state": {"moves": ""}, + } + board = chess.Board() + + with patch.object(api, "get_my_user_id", return_value="my_user"): + color = api._parse_game_full_event(event, board, "unknown") + + assert color == "black" + + +class TestJoinGameStream: + """Tests for join_game_stream method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_join_game_stream(self, api: LichessAPI) -> None: + """Test joining a game stream.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + event = json.dumps( + { + "type": "gameFull", + "white": {"id": "my_user"}, + "black": {"id": "opponent"}, + "state": {"moves": ""}, + } + ) + mock_response.iter_lines.return_value = iter([event]) + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + with ( + patch.object(api, "_request", return_value=mock_response), + patch.object(api, "get_my_user_id", return_value="my_user"), + ): + board, color = api.join_game_stream("game123", None) + + assert color == "white" + assert board.fen() == chess.STARTING_FEN + + +class TestStreamGameEvents: + """Tests for stream_game_events method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_stream_game_events_yields_events(self, api: LichessAPI) -> None: + """Test stream_game_events yields parsed events.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + mock_response.iter_lines.return_value = iter( + ['{"type": "gameFull"}', '{"type": "gameState"}'] + ) + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + with patch.object(api, "_request", return_value=mock_response): + events = list(api.stream_game_events("game123")) + + assert len(events) == 2 + + +class TestMakeMove: + """Tests for make_move method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_make_move_success(self, api: LichessAPI) -> None: + """Test successful move submission.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + + with patch.object(api, "_request", return_value=mock_response): + api.make_move("game123", chess.Move.from_uci("e2e4")) + + def test_make_move_conflict_raises(self, api: LichessAPI) -> None: + """Test move submission with conflict.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.CONFLICT + mock_response.raise_for_status.side_effect = requests.HTTPError() + + with ( + patch.object(api, "_request", return_value=mock_response), + pytest.raises(requests.HTTPError), + ): + api.make_move("game123", chess.Move.from_uci("e2e4")) + + def test_make_move_rate_limit_retries(self, api: LichessAPI) -> None: + """Test move submission retries on rate limit.""" + mock_response_429 = MagicMock() + mock_response_429.status_code = HTTPStatus.TOO_MANY_REQUESTS + + mock_response_ok = MagicMock() + mock_response_ok.status_code = HTTPStatus.OK + + call_count = 0 + + def mock_request(*args: object, **kwargs: object) -> MagicMock: + nonlocal call_count + call_count += 1 + if call_count == 1: + return mock_response_429 + return mock_response_ok + + with ( + patch.object(api, "_request", side_effect=mock_request), + patch("python_pkg.lichess_bot.lichess_api.time.sleep"), + ): + api.make_move("game123", chess.Move.from_uci("e2e4")) + + assert call_count == 2 + + def test_make_move_bad_request_raises(self, api: LichessAPI) -> None: + """Test move submission with bad request raises but returns.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.BAD_REQUEST + mock_response.raise_for_status.side_effect = requests.HTTPError() + + with ( + patch.object(api, "_request", return_value=mock_response), + pytest.raises(requests.HTTPError), + ): + api.make_move("game123", chess.Move.from_uci("e2e4")) + + +class TestGetGameState: + """Tests for get_game_state method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_get_game_state_returns_none(self, api: LichessAPI) -> None: + """Test deprecated get_game_state returns None.""" + result = api.get_game_state("game123") + assert result is None + + +class TestGetMyUserId: + """Tests for get_my_user_id method.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_get_my_user_id_success(self, api: LichessAPI) -> None: + """Test getting user ID successfully.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + mock_response.json.return_value = {"id": "my_username"} + + with patch.object(api, "_request", return_value=mock_response): + user_id = api.get_my_user_id() + + assert user_id == "my_username" + + def test_get_my_user_id_failure(self, api: LichessAPI) -> None: + """Test getting user ID when request fails.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.UNAUTHORIZED + + with patch.object(api, "_request", return_value=mock_response): + user_id = api.get_my_user_id() + + assert user_id is None + + +class TestRequestEdgeCases: + """Additional tests for _request edge cases.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_request_error_with_attribute_error_on_text(self, api: LichessAPI) -> None: + """Test error response when text property raises AttributeError.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.BAD_REQUEST + # Make text property raise AttributeError when accessed + del mock_response.text # Remove the default mock + type(mock_response).text = property( + fget=lambda _self: (_ for _ in ()).throw(AttributeError("no text")) + ) + + with patch.object(api.session, "request", return_value=mock_response): + result = api._request("GET", "http://test.com") + + assert result == mock_response + + def test_request_error_with_type_error_on_text(self, api: LichessAPI) -> None: + """Test error response when text causes TypeError.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.BAD_REQUEST + # Make text return something that causes TypeError when sliced + mock_response.text = 12345 # integer can't be sliced with [:200] + + with patch.object(api.session, "request", return_value=mock_response): + result = api._request("GET", "http://test.com") + + assert result == mock_response + + +class TestStreamEventsNon429Error: + """Test stream_events with non-429 HTTP errors.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_stream_events_raises_non_429_error(self, api: LichessAPI) -> None: + """Test stream_events raises non-429 HTTP errors.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.INTERNAL_SERVER_ERROR + mock_response.raise_for_status.side_effect = requests.HTTPError( + response=MagicMock(status_code=HTTPStatus.INTERNAL_SERVER_ERROR) + ) + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + with ( + patch.object(api, "_request", return_value=mock_response), + pytest.raises(requests.HTTPError), + ): + # Try to get the first event - should raise + next(api.stream_events()) + + +class TestJoinGameStreamEdgeCases: + """Additional tests for join_game_stream edge cases.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_join_game_stream_skips_empty_lines(self, api: LichessAPI) -> None: + """Test join_game_stream skips empty lines.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + event = json.dumps( + { + "type": "gameFull", + "white": {"id": "my_user"}, + "black": {"id": "opponent"}, + "state": {"moves": ""}, + } + ) + mock_response.iter_lines.return_value = iter(["", "", event]) + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + with ( + patch.object(api, "_request", return_value=mock_response), + patch.object(api, "get_my_user_id", return_value="my_user"), + ): + __board, color = api.join_game_stream("game123", None) + + assert color == "white" + + def test_join_game_stream_skips_invalid_json(self, api: LichessAPI) -> None: + """Test join_game_stream skips invalid JSON lines.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + event = json.dumps( + { + "type": "gameFull", + "white": {"id": "my_user"}, + "black": {"id": "opponent"}, + "state": {"moves": ""}, + } + ) + mock_response.iter_lines.return_value = iter(["not json", event]) + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + with ( + patch.object(api, "_request", return_value=mock_response), + patch.object(api, "get_my_user_id", return_value="my_user"), + ): + __board, color = api.join_game_stream("game123", None) + + assert color == "white" + + +class TestStreamGameEventsEdgeCases: + """Additional tests for stream_game_events edge cases.""" + + @pytest.fixture + def api(self) -> LichessAPI: + """Create API instance.""" + return LichessAPI("test_token") + + def test_stream_game_events_skips_empty_lines(self, api: LichessAPI) -> None: + """Test stream_game_events skips empty lines.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + mock_response.iter_lines.return_value = iter( + ["", '{"type": "gameFull"}', "", '{"type": "gameState"}'] + ) + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + with patch.object(api, "_request", return_value=mock_response): + events = list(api.stream_game_events("game123")) + + assert len(events) == 2 + + def test_stream_game_events_skips_invalid_json(self, api: LichessAPI) -> None: + """Test stream_game_events skips invalid JSON lines.""" + mock_response = MagicMock() + mock_response.status_code = HTTPStatus.OK + mock_response.iter_lines.return_value = iter( + ['{"type": "gameFull"}', "invalid json", '{"type": "gameState"}'] + ) + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=False) + + with patch.object(api, "_request", return_value=mock_response): + events = list(api.stream_game_events("game123")) + + assert len(events) == 2