Add more tests

lib/engine_wrapper.py

@@ -1068,15 +1068,15 @@ def get_lichess_egtb_move(li: LICHESS_TYPE, game: model.Game, board: chess.Board
                 dtm *= -1
             logger.info(f"Got move {move} from tablebase.lichess.ovh (wdl: {wdl}, dtz: {dtz}, dtm: {dtm}) for game {game.id}")
         else:  # quality == "suggest":
-            best_wdl = name_to_wld[data["moves"][0]["category"]]
+            best_wdl = name_to_wld[data["moves"][0]["category"]] * -1
 
             def good_enough(possible_move: LichessEGTBMoveType) -> bool:
-                return name_to_wld[possible_move["category"]] == best_wdl
+                return name_to_wld[possible_move["category"]] * -1 == best_wdl
 
             possible_moves = list(filter(good_enough, data["moves"]))
             if len(possible_moves) > 1:
                 move_list = [move["uci"] for move in possible_moves]
-                wdl = best_wdl * -1
+                wdl = best_wdl
                 logger.info(f"Suggesting moves from tablebase.lichess.ovh (wdl: {wdl}) for game {game.id}")
                 return move_list, wdl, {"string": "lichess-bot-source:Lichess EGTB"}
             else:
@@ -1102,7 +1102,7 @@ def get_chessdb_egtb_move(li: LICHESS_TYPE, game: model.Game, board: chess.Board
     If `move_quality` is `suggest`, then it will return a list of moves for the engine to choose from.
     """
     def score_to_wdl(score: int) -> int:
-        return piecewise_function([(-20000, 'e', 2),
+        return piecewise_function([(-20000, 'e', -2),
                                    (0, 'e', -1),
                                    (0, 'i', 0),
                                    (20000, 'i', 1)], 2, score)

test_bot/conftest.py

@@ -7,6 +7,10 @@
 
 
 def pytest_sessionfinish(session: Session, exitstatus: Union[int, ExitCode]) -> None:
-    """Remove files created when testing lichess-bot."""
+    """
+    Remove files created when testing lichess-bot.
+
+    The only exception is if running in a GitHub action, in which case we save the engines to the cache.
+    """
     if os.path.exists("TEMP") and not os.getenv("GITHUB_ACTIONS"):
         shutil.rmtree("TEMP")

test_bot/test_external_moves.py

@@ -0,0 +1,171 @@
+"""Test the functions that get the external moves."""
+import backoff
+import requests
+import yaml
+import os
+import chess
+import logging
+import test_bot.lichess
+import chess.engine
+from datetime import timedelta
+from copy import deepcopy
+from requests.exceptions import ConnectionError, HTTPError, ReadTimeout
+from http.client import RemoteDisconnected
+from lib.types import OnlineType, GameEventType
+from typing import Optional, Union, 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
+LICHESS_TYPE = Union[Lichess, test_bot.lichess.Lichess]
+
+
+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: Optional[dict[str, Union[str, int]]] = None,
+                        stream: bool = False) -> OnlineType:
+        """Get an external move from online sources (chessdb or lichess.org)."""
+
+        @backoff.on_exception(backoff.constant,
+                              (RemoteDisconnected, ConnectionError, 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 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"}}
+    game = Game(game_event, "b", "https://lichess.org", timedelta(seconds=60))
+    return game
+
+
+def download_opening_book() -> None:
+    """Download gm2001.bin."""
+    if os.path.exists("./TEMP/gm2001.bin"):
+        return
+    response = requests.get("https://github.com/gmcheems-org/free-opening-books/raw/main/books/bin/gm2001.bin",
+                            allow_redirects=True)
+    with open("./TEMP/gm2001.bin", "wb") as file:
+        file.write(response.content)
+
+
+os.makedirs("TEMP", exist_ok=True)
+download_opening_book()
+
+
+def get_online_move_wrapper(li: LICHESS_TYPE, board: chess.Board, game: Game, online_moves_cfg: Configuration,
+                            draw_or_resign_cfg: Configuration) -> chess.engine.PlayResult:
+    """Wrap `lib.engine_wrapper.get_online_move` so that it only returns a PlayResult type."""
+    return cast(chess.engine.PlayResult, get_online_move(li, board, game, online_moves_cfg, draw_or_resign_cfg))
+
+
+def test_external_moves() -> None:
+    """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"
+
+    # Test lichess_cloud_analysis.
+    assert get_online_move_wrapper(li, chess.Board(starting_fen), game, online_cfg, draw_or_resign_cfg).move is not None
+    assert get_online_move_wrapper(li, chess.Board(opening_fen), game, online_cfg, draw_or_resign_cfg).move is not None
+    assert get_online_move_wrapper(li, chess.Board(middlegame_fen), game, online_cfg, draw_or_resign_cfg).move is None
+
+    # Test chessdb_book.
+    assert get_online_move_wrapper(li, chess.Board(starting_fen), game, online_cfg_2, draw_or_resign_cfg).move is not None
+    assert get_online_move_wrapper(li, chess.Board(opening_fen), game, online_cfg_2, draw_or_resign_cfg).move is not None
+    assert get_online_move_wrapper(li, chess.Board(middlegame_fen), game, online_cfg_2, draw_or_resign_cfg).move is None
+
+    # Test online_egtb with lichess.
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl2_fen), game, online_cfg, draw_or_resign_cfg).resigned
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl0_fen), game, online_cfg, draw_or_resign_cfg).draw_offered
+    wdl1_move = get_online_move_wrapper(li, chess.Board(endgame_wdl1_fen), game, online_cfg, draw_or_resign_cfg)
+    assert not wdl1_move.resigned and not wdl1_move.draw_offered
+    # Test with reversed colors.
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl2_fen).mirror(), game, online_cfg, draw_or_resign_cfg).resigned
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl0_fen).mirror(), game, online_cfg,
+                                   draw_or_resign_cfg).draw_offered
+    wdl1_move = get_online_move_wrapper(li, chess.Board(endgame_wdl1_fen).mirror(), game, online_cfg, draw_or_resign_cfg)
+    assert not wdl1_move.resigned and not wdl1_move.draw_offered
+
+    # Test online_egtb with chessdb.
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl2_fen), game, online_cfg_2, draw_or_resign_cfg).resigned
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl0_fen), game, online_cfg_2, draw_or_resign_cfg).draw_offered
+    wdl1_move = get_online_move_wrapper(li, chess.Board(endgame_wdl1_fen), game, online_cfg_2, draw_or_resign_cfg)
+    assert not wdl1_move.resigned and not wdl1_move.draw_offered
+    # Test with reversed colors.
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl2_fen).mirror(), game, online_cfg_2, draw_or_resign_cfg).resigned
+    assert get_online_move_wrapper(li, chess.Board(endgame_wdl0_fen).mirror(), game, online_cfg_2,
+                                   draw_or_resign_cfg).draw_offered
+    wdl1_move = get_online_move_wrapper(li, chess.Board(endgame_wdl1_fen).mirror(), game, online_cfg_2, draw_or_resign_cfg)
+    assert not wdl1_move.resigned and not wdl1_move.draw_offered
+
+    # Test opening book.
+    assert get_book_move(chess.Board(opening_fen), game, polyglot_cfg).move == chess.Move.from_uci("h4f6")