mirror of
https://github.com/kuhyx/testsAndMisc.git
synced 2026-07-04 15:03:01 +02:00
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%
This commit is contained in:
parent
8619dfb3c7
commit
23a516de12
@ -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)
|
||||
|
||||
300
python_pkg/lichess_bot/tests/test_engine.py
Normal file
300
python_pkg/lichess_bot/tests/test_engine.py
Normal file
@ -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")
|
||||
624
python_pkg/lichess_bot/tests/test_lichess_api.py
Normal file
624
python_pkg/lichess_bot/tests/test_lichess_api.py
Normal file
@ -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
|
||||
Loading…
Reference in New Issue
Block a user