test_bot/test_external_moves.py

"""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))