"""Test the functions that get the external moves.""" import pytest import backoff import requests import yaml import os import chess import logging import chess.engine from datetime import timedelta from copy import deepcopy from requests.exceptions import ConnectionError as RequestsConnectionError, HTTPError, ReadTimeout, RequestException from http.client import RemoteDisconnected from lib.lichess_types import OnlineType, GameEventType from typing import cast from lib.lichess import is_final, backoff_handler, Lichess from lib.config import Configuration, insert_default_values from lib.model import Game from lib.engine_wrapper import get_online_move, get_book_move, is_op1_position class MockLichess(Lichess): """A modified Lichess class for communication with external move sources.""" def __init__(self) -> None: """Initialize only self.other_session and not self.session.""" self.max_retries = 3 self.other_session = requests.Session() def online_book_get(self, path: str, params: dict[str, str | int] | None = None, *, stream: bool = False, authenticated: bool = False) -> OnlineType: """ Get an external move from online sources (chessdb or lichess.org). Ignore authentication for tests. """ del authenticated @backoff.on_exception(backoff.constant, (RemoteDisconnected, RequestsConnectionError, HTTPError, ReadTimeout), max_time=60, max_tries=self.max_retries, interval=0.1, giveup=is_final, on_backoff=backoff_handler, backoff_log_level=logging.DEBUG, giveup_log_level=logging.DEBUG) def online_book_get() -> OnlineType: json_response: OnlineType = self.other_session.get(path, timeout=2, params=params, stream=stream).json() return json_response return online_book_get() def is_website_up(self, url: str) -> bool: """Check if a website is up.""" try: self.other_session.get(url, timeout=2) return True except RequestException: return False def get_configs() -> tuple[Configuration, Configuration, Configuration, Configuration]: """Create the configs used for the tests.""" with open("./config.yml.default") as file: CONFIG = yaml.safe_load(file) insert_default_values(CONFIG) CONFIG["engine"]["online_moves"]["lichess_cloud_analysis"]["enabled"] = True CONFIG["engine"]["online_moves"]["online_egtb"]["enabled"] = True CONFIG["engine"]["draw_or_resign"]["resign_enabled"] = True CONFIG["engine"]["polyglot"]["enabled"] = True CONFIG["engine"]["polyglot"]["book"]["standard"] = ["TEMP/gm2001.bin"] engine_cfg = Configuration(CONFIG).engine CONFIG_2 = deepcopy(CONFIG) CONFIG_2["engine"]["online_moves"]["chessdb_book"]["enabled"] = True CONFIG_2["engine"]["online_moves"]["online_egtb"]["source"] = "chessdb" engine_cfg_2 = Configuration(CONFIG_2).engine return engine_cfg.online_moves, engine_cfg_2.online_moves, engine_cfg.draw_or_resign, engine_cfg.polyglot def get_game() -> Game: """Create a model.Game to be used in the tests.""" game_event: GameEventType = {"id": "zzzzzzzz", "variant": {"key": "standard", "name": "Standard", "short": "Std"}, "clock": {"initial": 60000, "increment": 2000}, "speed": "bullet", "perf": {"name": "Bullet"}, "rated": True, "createdAt": 1600000000000, "white": {"id": "bo", "name": "bo", "title": "BOT", "rating": 3000}, "black": {"id": "b", "name": "b", "title": "BOT", "rating": 3000, "provisional": True}, "initialFen": "startpos", "type": "gameFull", "state": {"type": "gameState", "moves": "", "wtime": 1000000, "btime": 1000000, "winc": 2000, "binc": 2000, "status": "started"}} return Game(game_event, "b", "https://lichess.org", timedelta(seconds=60)) def download_opening_book() -> None: """Download gm2001.bin.""" if os.path.exists("./TEMP/gm2001.bin"): return os.makedirs("TEMP", exist_ok=True) response = requests.get("https://github.com/gmcheems-org/free-opening-books/raw/main/books/bin/gm2001.bin", allow_redirects=True, timeout=60) if response.status_code != 200: pytest.xfail("Could not download opening book.") with open("./TEMP/gm2001.bin", "wb") as file: file.write(response.content) def get_online_move_wrapper(li: Lichess, board: chess.Board, game: Game, online_moves_cfg: Configuration, draw_or_resign_cfg: Configuration, *, expect_none: bool = False) -> chess.engine.PlayResult: """Wrap `lib.engine_wrapper.get_online_move` so that it only returns a PlayResult type.""" online_move = get_online_move(li, board, game, online_moves_cfg, draw_or_resign_cfg) online_move = cast(chess.engine.PlayResult, online_move) if not expect_none and online_move.move is None: pytest.xfail("Could not contact external move source.") return online_move class TestExternalMoves: """Test that the code for external moves works properly.""" li = MockLichess() game = get_game() online_cfg, online_cfg_2, draw_or_resign_cfg, polyglot_cfg = get_configs() starting_fen = "rnbqkbnr/pppppppp/8/8/8/8/PPPPPPPP/RNBQKBNR w KQkq - 0 1" opening_fen = "rn1q1rk1/pbp1bpp1/1p2pn1p/3p4/2PP3B/2N1PN2/PP2BPPP/R2QK2R w KQ - 2 9" middlegame_fen = "8/5p2/1n1p1nk1/1p1Pp1p1/1Pp1P1Pp/r1P2B1P/2RNKP2/8 w - - 0 31" endgame_wdl2_fen = "2k5/4n2Q/5N2/8/8/8/1r6/2K5 b - - 0 123" endgame_wdl1_fen = "6N1/3n4/3k1b2/8/8/7Q/1r6/5K2 b - - 6 9" endgame_wdl0_fen = "6N1/3n4/3k1b2/8/8/7Q/5K2/1r6 b - - 8 10" endgame_op1_fen = "r7/p1b1k3/8/8/P7/8/8/K2R1R2 w - - 1 2" def test_lichess_cloud_analysis(self) -> None: """Test lichess_cloud_analysis.""" if not self.li.is_website_up("https://lichess.org/api/cloud-eval"): pytest.xfail("Lichess cloud eval is down.") assert get_online_move_wrapper(self.li, chess.Board(self.starting_fen), self.game, self.online_cfg, self.draw_or_resign_cfg).move is not None assert get_online_move_wrapper(self.li, chess.Board(self.opening_fen), self.game, self.online_cfg, self.draw_or_resign_cfg).move is not None assert get_online_move_wrapper(self.li, chess.Board(self.middlegame_fen), self.game, self.online_cfg, self.draw_or_resign_cfg, expect_none=True).move is None def test_chessdb_book(self) -> None: """Test chessdb_book.""" if not self.li.is_website_up("https://www.chessdb.cn/cdb.php"): pytest.xfail("ChessDB is down.") assert get_online_move_wrapper(self.li, chess.Board(self.starting_fen), self.game, self.online_cfg_2, self.draw_or_resign_cfg).move is not None assert get_online_move_wrapper(self.li, chess.Board(self.opening_fen), self.game, self.online_cfg_2, self.draw_or_resign_cfg).move is not None assert get_online_move_wrapper(self.li, chess.Board(self.middlegame_fen), self.game, self.online_cfg_2, self.draw_or_resign_cfg, expect_none=True).move is None def test_online_egtb_with_lichess(self) -> None: """Test online_egtb with lichess.""" if not self.li.is_website_up("https://tablebase.lichess.ovh/standard"): pytest.xfail("Lichess tablebase is down.") assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl2_fen), self.game, self.online_cfg, self.draw_or_resign_cfg).resigned assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl0_fen), self.game, self.online_cfg, self.draw_or_resign_cfg).draw_offered wdl1_move = get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl1_fen), self.game, self.online_cfg, self.draw_or_resign_cfg) assert not wdl1_move.resigned and not wdl1_move.draw_offered op1_move = get_online_move_wrapper(self.li, chess.Board(self.endgame_op1_fen), self.game, self.online_cfg, self.draw_or_resign_cfg) assert op1_move.move == chess.Move.from_uci("f1e1") # Test with reversed colors. assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl2_fen).mirror(), self.game, self.online_cfg, self.draw_or_resign_cfg).resigned assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl0_fen).mirror(), self.game, self.online_cfg, self.draw_or_resign_cfg).draw_offered wdl1_move = get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl1_fen).mirror(), self.game, self.online_cfg, self.draw_or_resign_cfg) assert not wdl1_move.resigned and not wdl1_move.draw_offered def test_online_egtb_with_chessdb(self) -> None: """Test online_egtb with chessdb.""" if not self.li.is_website_up("https://www.chessdb.cn/cdb.php"): pytest.xfail("ChessDB is down.") assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl2_fen), self.game, self.online_cfg_2, self.draw_or_resign_cfg).resigned assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl0_fen), self.game, self.online_cfg_2, self.draw_or_resign_cfg).draw_offered wdl1_move = get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl1_fen), self.game, self.online_cfg_2, self.draw_or_resign_cfg) assert not wdl1_move.resigned and not wdl1_move.draw_offered # Test with reversed colors. assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl2_fen).mirror(), self.game, self.online_cfg_2, self.draw_or_resign_cfg).resigned assert get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl0_fen).mirror(), self.game, self.online_cfg_2, self.draw_or_resign_cfg).draw_offered wdl1_move = get_online_move_wrapper(self.li, chess.Board(self.endgame_wdl1_fen).mirror(), self.game, self.online_cfg_2, self.draw_or_resign_cfg) assert not wdl1_move.resigned and not wdl1_move.draw_offered def test_opening_book(self) -> None: """Test opening book.""" download_opening_book() assert get_book_move(chess.Board(self.opening_fen), self.game, self.polyglot_cfg).move == chess.Move.from_uci("h4f6") class TestExternalMoveHelpers: """Test the helper functions for external moves.""" def test_op1_positions(self) -> None: """Test that a position is op1.""" fens_op1 = ["r7/p1b1k3/8/8/P7/8/8/K2R1R2 w - - 1 2", "3qk3/1n1pp3/8/8/8/8/4P2P/4K3 w - - 0 1"] fens_not_op1 = ["r3k3/2b5/8/p7/P7/8/8/K2R1R2 w q - 0 2", "3qk3/1n1pp1n1/8/8/8/8/4P3/4K3 w - - 0 1"] for fen in fens_op1: assert is_op1_position(chess.Board(fen)) for fen in fens_not_op1: assert not is_op1_position(chess.Board(fen))