diff --git a/.github/workflows/python-build.yml b/.github/workflows/python-build.yml index 10ca50d02..9921602af 100644 --- a/.github/workflows/python-build.yml +++ b/.github/workflows/python-build.yml @@ -28,17 +28,7 @@ jobs: python -m pip install --upgrade pip pip install -r requirements.txt pip install -r test_bot/test-requirements.txt - - name: Lint with flake8 + - name: Lint with ruff run: | - # stop the build if there are Python syntax errors or undefined names - flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics - # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide. - # W503 and W504 are mutually exclusive. W504 is considered the best practice now. - flake8 . --count --max-complexity=10 --max-line-length=127 --statistics --ignore=D,W503 - - name: Lint with flake8-markdown - run: | - flake8-markdown "*.md" - flake8-markdown "wiki/*.md" - - name: Lint with flake8-docstrings - run: | - flake8 . --count --max-line-length=127 --statistics --select=D + # Check for python syntax errors and inconsistent code style. + ruff check --config test_bot/ruff.toml diff --git a/.github/workflows/update_version.py b/.github/workflows/update_version.py index 56645ebcd..92b1c6e59 100644 --- a/.github/workflows/update_version.py +++ b/.github/workflows/update_version.py @@ -3,6 +3,9 @@ import datetime import os +# File is part of an implicit namespace package. Add an `__init__.py`. +# ruff: noqa: INP001 + with open("lib/versioning.yml") as version_file: versioning_info = yaml.safe_load(version_file) @@ -22,5 +25,5 @@ with open("lib/versioning.yml", "w") as version_file: yaml.dump(versioning_info, version_file, sort_keys=False) -with open(os.environ['GITHUB_OUTPUT'], 'a') as fh: +with open(os.environ["GITHUB_OUTPUT"], "a") as fh: print(f"new_version={new_version}", file=fh) diff --git a/extra_game_handlers.py b/extra_game_handlers.py index 89a42b5e2..7d3f97bae 100644 --- a/extra_game_handlers.py +++ b/extra_game_handlers.py @@ -3,7 +3,7 @@ from lib.types import OPTIONS_TYPE -def game_specific_options(game: model.Game) -> OPTIONS_TYPE: +def game_specific_options(game: model.Game) -> OPTIONS_TYPE: # noqa: ARG001 """ Return a dictionary of engine options based on game aspects. @@ -12,7 +12,7 @@ def game_specific_options(game: model.Game) -> OPTIONS_TYPE: return {} -def is_supported_extra(challenge: model.Challenge) -> bool: +def is_supported_extra(challenge: model.Challenge) -> bool: # noqa: ARG001 """ Determine whether to accept a challenge. diff --git a/homemade.py b/homemade.py index a16a72dec..19633d311 100644 --- a/homemade.py +++ b/homemade.py @@ -20,15 +20,13 @@ class ExampleEngine(MinimalEngine): """An example engine that all homemade engines inherit.""" - pass - # Bot names and ideas from tom7's excellent eloWorld video class RandomMove(ExampleEngine): """Get a random move.""" - def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: + def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: # noqa: ARG002 """Choose a random move.""" return PlayResult(random.choice(list(board.legal_moves)), None) @@ -36,7 +34,7 @@ def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: class Alphabetical(ExampleEngine): """Get the first move when sorted by san representation.""" - def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: + def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: # noqa: ARG002 """Choose the first move alphabetically.""" moves = list(board.legal_moves) moves.sort(key=board.san) @@ -46,7 +44,7 @@ def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: class FirstMove(ExampleEngine): """Get the first move when sorted by uci representation.""" - def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: + def search(self, board: chess.Board, *args: HOMEMADE_ARGS_TYPE) -> PlayResult: # noqa: ARG002 """Choose the first move alphabetically in uci representation.""" moves = list(board.legal_moves) moves.sort(key=str) @@ -60,7 +58,12 @@ class ComboEngine(ExampleEngine): This engine demonstrates how one can use `time_limit`, `draw_offered`, and `root_moves`. """ - def search(self, board: chess.Board, time_limit: Limit, ponder: bool, draw_offered: bool, root_moves: MOVE) -> PlayResult: + def search(self, + board: chess.Board, + time_limit: Limit, + ponder: bool, # noqa: ARG002 + draw_offered: bool, + root_moves: MOVE) -> PlayResult: """ Choose a move using multiple different methods. diff --git a/lib/config.py b/lib/config.py index 0f79af8ff..dffd8985b 100644 --- a/lib/config.py +++ b/lib/config.py @@ -244,9 +244,9 @@ def insert_default_values(CONFIG: CONFIG_DICT_TYPE) -> None: for ponder in ["ponder", "uci_ponder"]: set_config_default(CONFIG, section, key=ponder, default=False) - for type in ["hello", "goodbye"]: + for greeting in ["hello", "goodbye"]: for target in ["", "_spectators"]: - set_config_default(CONFIG, "greeting", key=type + target, default="", force_empty_values=True) + set_config_default(CONFIG, "greeting", key=greeting + target, default="", force_empty_values=True) if CONFIG["matchmaking"]["include_challenge_block_list"]: CONFIG["matchmaking"]["block_list"].extend(CONFIG["challenge"]["block_list"]) @@ -259,7 +259,7 @@ def log_config(CONFIG: CONFIG_DICT_TYPE, alternate_log_function: Callable[[str], :param CONFIG: The bot's config. """ logger_config = CONFIG.copy() - logger_config["token"] = "logger" + logger_config["token"] = "logger" # noqa: S105 (Possible hardcoded password) destination = alternate_log_function or logger.debug destination(f"Config:\n{yaml.dump(logger_config, sort_keys=False)}") destination("====================") @@ -321,10 +321,11 @@ def validate_config(CONFIG: CONFIG_DICT_TYPE) -> None: pgn_directory = CONFIG["pgn_directory"] in_docker = os.environ.get("LICHESS_BOT_DOCKER") - config_warn(not pgn_directory or not in_docker, "Games will be saved to '{}', please ensure this folder is in a mounted " - "volume; Using the Docker's container internal file system will prevent " - "you accessing the saved files and can lead to disk " - "saturation.".format(pgn_directory)) + config_warn(not pgn_directory or not in_docker, + f"Games will be saved to '{pgn_directory}', please ensure this folder is in a mounted " + "volume; Using the Docker's container internal file system will prevent " + "you accessing the saved files and can lead to disk " + "saturation.") valid_pgn_grouping_options = ["game", "opponent", "all"] config_pgn_choice = CONFIG["pgn_file_grouping"] diff --git a/lib/conversation.py b/lib/conversation.py index 5cddb14f6..84351b5ba 100644 --- a/lib/conversation.py +++ b/lib/conversation.py @@ -55,7 +55,7 @@ def react(self, line: ChatLine) -> None: :param line: Information about the message. """ - logger.info(f'*** {self.game.url()} [{line.room}] {line.username}: {line.text}') + logger.info(f"*** {self.game.url()} [{line.room}] {line.username}: {line.text}") if line.text[0] == self.command_prefix: self.command(line, line.text[1:].lower()) @@ -68,7 +68,7 @@ def command(self, line: ChatLine, cmd: str) -> None: """ from_self = line.username == self.game.username is_eval = cmd.startswith("eval") - if cmd == "commands" or cmd == "help": + if cmd in ("commands", "help"): self.send_reply(line, "Supported commands: !wait (wait a minute for my first move), !name, " "!eval (or any text starting with !eval), !queue") @@ -97,7 +97,7 @@ def send_reply(self, line: ChatLine, reply: str) -> None: :param line: Information about the original message that we reply to. :param reply: The reply to send. """ - logger.info(f'*** {self.game.url()} [{line.room}] {self.game.username}: {reply}') + logger.info(f"*** {self.game.url()} [{line.room}] {self.game.username}: {reply}") self.li.chat(self.game.id, line.room, reply) def send_message(self, room: str, message: str) -> None: diff --git a/lib/engine_wrapper.py b/lib/engine_wrapper.py index 35a86103f..5554a2471 100644 --- a/lib/engine_wrapper.py +++ b/lib/engine_wrapper.py @@ -12,6 +12,7 @@ import time import random import math +import contextlib import test_bot.lichess from collections import Counter from collections.abc import Callable @@ -22,7 +23,7 @@ COMMANDS_TYPE, MOVE, InfoStrDict, InfoDictKeys, InfoDictValue, GO_COMMANDS_TYPE, EGTPATH_TYPE, ENGINE_INPUT_ARGS_TYPE, ENGINE_INPUT_KWARGS_TYPE) from extra_game_handlers import game_specific_options -from typing import Any, Optional, Union, Literal, Type, cast +from typing import Any, Optional, Union, Literal, cast from types import TracebackType LICHESS_TYPE = Union[lichess.Lichess, test_bot.lichess.Lichess] @@ -55,13 +56,13 @@ def create_engine(engine_config: Configuration, game: Optional[model.Game] = Non stderr = None if cfg.silence_stderr else subprocess.DEVNULL - Engine: Union[type[UCIEngine], type[XBoardEngine], type[MinimalEngine]] + Engine: type[Union[UCIEngine, XBoardEngine, MinimalEngine]] if engine_type == "xboard": Engine = XBoardEngine elif engine_type == "uci": Engine = UCIEngine elif engine_type == "homemade": - Engine = getHomemadeEngine(cfg.name) + Engine = get_homemade_engine(cfg.name) else: raise ValueError( f" Invalid engine type: {engine_type}. Expected xboard, uci, or homemade.") @@ -113,12 +114,12 @@ def configure(self, options: OPTIONS_GO_EGTB_TYPE, game: Optional[model.Game]) - self.engine.close() raise - def __enter__(self) -> EngineWrapper: + def __enter__(self) -> EngineWrapper: # noqa: PYI034 (return Self not available until 3.11) """Enter context so engine communication will be properly shutdown.""" self.engine.__enter__() return self - def __exit__(self, exc_type: Optional[Type[BaseException]], + def __exit__(self, exc_type: Optional[type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[TracebackType]) -> None: """Exit context and allow engine to shutdown nicely if there was no exception.""" @@ -270,8 +271,7 @@ def search(self, board: chess.Board, time_limit: chess.engine.Limit, ponder: boo # Use null_score to have no effect on draw/resign decisions null_score = chess.engine.PovScore(chess.engine.Mate(1), board.turn) self.scores.append(result.info.get("score", null_score)) - result = self.offer_draw_or_resign(result, board) - return result + return self.offer_draw_or_resign(result, board) def comment_index(self, move_stack_index: int) -> int: """ @@ -326,10 +326,8 @@ def discard_last_move_commentary(self) -> None: Used after allowing an opponent to take back a move. """ - try: + with contextlib.suppress(IndexError): self.move_commentary.pop() - except IndexError: - pass def print_stats(self) -> None: """Print the engine stats.""" @@ -432,9 +430,7 @@ def get_opponent_info(self, game: model.Game) -> None: def name(self) -> str: """Get the name of the engine.""" - engine_info: dict[str, str] = dict(self.engine.id) - name = engine_info["name"] - return name + return self.engine.id["name"] def get_pid(self) -> str: """Get the pid of the engine.""" @@ -545,9 +541,9 @@ class MinimalEngine(EngineWrapper): `notify`, etc. """ - def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_GO_EGTB_TYPE, stderr: Optional[int], - draw_or_resign: Configuration, game: Optional[model.Game] = None, name: Optional[str] = None, - **popen_args: str) -> None: + def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_GO_EGTB_TYPE, stderr: Optional[int], # noqa: ARG002 + draw_or_resign: Configuration, game: Optional[model.Game] = None, name: Optional[str] = None, # noqa: ARG002 + **popen_args: str) -> None: # noqa: ARG002 Unused argument popen_args """ Initialize the values of the engine that all homemade engines inherit. @@ -590,7 +586,6 @@ def notify(self, method_name: str, *args: ENGINE_INPUT_ARGS_TYPE, **kwargs: ENGI self.engine.(<*args>, <**kwargs>) self.notify(, <*args>, <**kwargs>) """ - pass class FillerEngine: @@ -623,7 +618,7 @@ def method(*args: ENGINE_INPUT_ARGS_TYPE, **kwargs: ENGINE_INPUT_KWARGS_TYPE) -> test_suffix = "-for-lichess-bot-testing-only" -def getHomemadeEngine(name: str) -> type[MinimalEngine]: +def get_homemade_engine(name: str) -> type[MinimalEngine]: """ Get the homemade engine with name `name`. e.g. If `name` is `RandomMove` then we will return `homemade.RandomMove`. @@ -848,7 +843,7 @@ def get_chessdb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, action = {"best": "querypv", "good": "querybest", "all": "query"} - try: + with contextlib.suppress(Exception): params: dict[str, Union[str, int]] = {"action": action[quality], "board": board.fen(), "json": 1} data = li.online_book_get(site, params=params) if data["status"] == "ok": @@ -865,8 +860,6 @@ def get_chessdb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, else: move = data["move"] logger.info(f"Got move {move} from chessdb.cn for game {game.id}") - except Exception: - pass return move, comment @@ -888,7 +881,7 @@ def get_lichess_cloud_move(li: LICHESS_TYPE, board: chess.Board, game: model.Gam multipv = 1 if quality == "best" else 5 variant = "standard" if board.uci_variant == "chess" else str(board.uci_variant) # `str` is there only for mypy. - try: + with contextlib.suppress(Exception): data = li.online_book_get("https://lichess.org/api/cloud-eval", params={"fen": board.fen(), "multiPv": multipv, @@ -919,8 +912,6 @@ def get_lichess_cloud_move(li: LICHESS_TYPE, board: chess.Board, game: model.Gam comment["string"] = "lichess-bot-source:Lichess Cloud Analysis" logger.info(f"Got move {move} from lichess cloud analysis (depth: {depth}, score: {score}, knodes: {knodes})" f" for game {game.id}") - except Exception: - pass return move, comment @@ -939,7 +930,7 @@ def get_opening_explorer_move(li: LICHESS_TYPE, board: chess.Board, game: model. move = None comment: chess.engine.InfoDict = {} variant = "standard" if board.uci_variant == "chess" else str(board.uci_variant) # `str` is there only for mypy - try: + with contextlib.suppress(Exception): params: dict[str, Union[str, int]] if source == "masters": params = {"fen": board.fen(), "moves": 100} @@ -972,8 +963,6 @@ def get_opening_explorer_move(li: LICHESS_TYPE, board: chess.Board, game: model. move = moves[0][2] logger.info(f"Got move {move} from lichess opening explorer ({opening_explorer_cfg.sort}: {moves[0][0]})" f" for game {game.id}") - except Exception: - pass return move, comment @@ -1004,13 +993,11 @@ def get_online_egtb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, quality = online_egtb_cfg.move_quality variant = "standard" if board.uci_variant == "chess" else str(board.uci_variant) - try: + with contextlib.suppress(Exception): if source == "lichess": return get_lichess_egtb_move(li, game, board, quality, variant) elif source == "chessdb": return get_chessdb_egtb_move(li, game, board, quality) - except Exception: - pass return None, -3, {} @@ -1106,16 +1093,16 @@ 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), - (0, 'e', -1), - (0, 'i', 0), - (20000, 'i', 1)], 2, score) + return piecewise_function([(-20000, "e", -2), + (0, "e", -1), + (0, "i", 0), + (20000, "i", 1)], 2, score) def score_to_dtz(score: int) -> int: - return piecewise_function([(-20000, 'e', -30000 - score), - (0, 'e', -20000 - score), - (0, 'i', 0), - (20000, 'i', 20000 - score)], 30000 - score, score) + return piecewise_function([(-20000, "e", -30000 - score), + (0, "e", -20000 - score), + (0, "i", 0), + (20000, "i", 20000 - score)], 30000 - score, score) action = "querypv" if quality == "best" else "queryall" data = li.online_book_get("https://www.chessdb.cn/cdb.php", @@ -1214,13 +1201,14 @@ def dtz_scorer(tablebase: chess.syzygy.Tablebase, board: chess.Board) -> Union[i return dtz + (math.copysign(board.halfmove_clock, dtz) if dtz else 0) -def dtz_to_wdl(dtz: Union[int, float]) -> int: - """Convert DTZ scores to syzygy WDL scores. +def dtz_to_wdl(dtz: float) -> int: + """ + Convert DTZ scores to syzygy WDL scores. A DTZ of +/-100 returns a draw score of +/-1 instead of a win/loss score of +/-2 because a 50-move draw can be forced before checkmate can be forced. """ - return piecewise_function([(-100, 'i', -1), (0, 'e', -2), (0, 'i', 0), (100, 'e', 2)], 1, dtz) + return piecewise_function([(-100, "i", -1), (0, "e", -2), (0, "i", 0), (100, "e", 2)], 1, dtz) def get_gaviota(board: chess.Board, game: model.Game, @@ -1282,14 +1270,14 @@ def dtm_scorer(tablebase: Union[chess.gaviota.NativeTablebase, chess.gaviota.Pyt def dtm_to_gaviota_wdl(dtm: int) -> int: """Convert DTM scores to gaviota WDL scores.""" - return piecewise_function([(-1, 'i', -1), (0, 'i', 0)], 1, dtm) + return piecewise_function([(-1, "i", -1), (0, "i", 0)], 1, dtm) def dtm_to_wdl(dtm: int, min_dtm_to_consider_as_wdl_1: int) -> int: """Convert DTM scores to syzygy WDL scores.""" # We use 100 and not min_dtm_to_consider_as_wdl_1, because we want to play it safe and not resign in a # position where dtz=-102 (only if resign_for_egtb_minus_two is enabled). - return piecewise_function([(-100, 'i', -1), (-1, 'i', -2), (0, 'i', 0), (min_dtm_to_consider_as_wdl_1, 'e', 2)], 1, dtm) + return piecewise_function([(-100, "i", -1), (-1, "i", -2), (0, "i", 0), (min_dtm_to_consider_as_wdl_1, "e", 2)], 1, dtm) def good_enough_gaviota_moves(good_moves: list[tuple[chess.Move, int]], best_dtm: int, @@ -1324,15 +1312,15 @@ def good_enough_gaviota_moves(good_moves: list[tuple[chess.Move, int]], best_dtm return good_moves -def piecewise_function(range_definitions: list[tuple[Union[int, float], Literal['e', 'i'], int]], last_value: int, - position: Union[int, float]) -> int: +def piecewise_function(range_definitions: list[tuple[float, Literal["e", "i"], int]], last_value: int, + position: float) -> int: """ Return a value according to a position argument. This function is meant to replace if-elif-else blocks that turn ranges into discrete values. Each tuple in the list has three parts: an upper limit, and inclusive/exclusive indicator, and a value. For example, - `piecewise_function([(-20000, 'e', 2), (0, 'e' -1), (0, 'i', 0), (20000, 'i', 1)], 2, score)` is equivalent to: + `piecewise_function([(-20000, "e", 2), (0, "e" -1), (0, "i", 0), (20000, "i", 1)], 2, score)` is equivalent to: if score < -20000: return -2 @@ -1348,16 +1336,16 @@ def piecewise_function(range_definitions: list[tuple[Union[int, float], Literal[ Arguments: range_definitions: A list of tuples with the first element being the inclusive right border of region and the second - element being the associated value. An element of this list (a, 'i', b) corresponds to an + element being the associated value. An element of this list (a, "i", b) corresponds to an inclusive limit and is equivalent to if x <= a: return b - where x is the value of the position argument. An element of the form (a, 'e', b) corresponds to + where x is the value of the position argument. An element of the form (a, "e", b) corresponds to an exclusive limit and is equivalent to if x < a: return b For correct operation, this argument should be sorted by the first element. If two ranges have the - same border, one with 'e' and the other with 'i', the 'e' element should be first. + same border, one with "e" and the other with "i", the "e" element should be first. last_value: If the position argument does not fall in any of the ranges in the range_definition argument, return this value. @@ -1366,7 +1354,7 @@ def piecewise_function(range_definitions: list[tuple[Union[int, float], Literal[ """ for border, inc_exc, value in range_definitions: - if position < border or (inc_exc == 'i' and position == border): + if position < border or (inc_exc == "i" and position == border): return value return last_value diff --git a/lib/lichess.py b/lib/lichess.py index d4687478b..f878b428c 100644 --- a/lib/lichess.py +++ b/lib/lichess.py @@ -9,6 +9,7 @@ import traceback from collections import defaultdict import datetime +import contextlib from lib.timer import Timer, seconds, sec_str from typing import Optional, Union, cast import chess.engine @@ -47,8 +48,6 @@ class RateLimited(RuntimeError): """Exception raised when we are rate limited (status code 429).""" - pass - def is_new_rate_limit(response: requests.models.Response) -> bool: """Check if the status code is 429, which means that we are rate limited.""" @@ -317,13 +316,11 @@ def accept_challenge(self, challenge_id: str) -> None: def decline_challenge(self, challenge_id: str, reason: str = "generic") -> None: """Decline a challenge.""" - try: + with contextlib.suppress(Exception): self.api_post("decline", challenge_id, data=f"reason={reason}", headers={"Content-Type": "application/x-www-form-urlencoded"}, raise_for_status=False) - except Exception: - pass def get_profile(self) -> UserProfileType: """Get the bot's profile (e.g. username).""" @@ -334,11 +331,9 @@ def get_profile(self) -> UserProfileType: def get_ongoing_games(self) -> list[GameType]: """Get the bot's ongoing games.""" ongoing_games: list[GameType] = [] - try: + with contextlib.suppress(Exception): response = cast(dict[str, list[GameType]], self.api_get_json("playing")) ongoing_games = response["nowPlaying"] - except Exception: - pass return ongoing_games def resign(self, game_id: str) -> None: diff --git a/lib/lichess_bot.py b/lib/lichess_bot.py index 746d7613f..1de52ac09 100644 --- a/lib/lichess_bot.py +++ b/lib/lichess_bot.py @@ -23,6 +23,7 @@ import glob import platform import importlib.metadata +import contextlib import test_bot.lichess from lib.config import load_config, Configuration, log_config from lib.conversation import Conversation, ChatLine @@ -90,7 +91,7 @@ def disable_restart() -> None: restart = False -def signal_handler(signal: int, frame: Optional[FrameType]) -> None: +def signal_handler(signal: int, frame: Optional[FrameType]) -> None: # noqa: ARG001 """Terminate lichess-bot.""" global terminated global force_quit @@ -198,7 +199,7 @@ def logging_configurer(level: int, filename: Optional[str], disable_auto_logs: b auto_file_handler = logging.handlers.TimedRotatingFileHandler(auto_log_filename, delay=True, encoding="utf-8", - when='midnight', + when="midnight", backupCount=7) auto_file_handler.setLevel(logging.DEBUG) @@ -230,7 +231,7 @@ def logging_listener_proc(queue: LOGGING_QUEUE_TYPE, level: int, log_filename: O time.sleep(0.1) except InterruptedError: pass - except Exception: + except Exception: # noqa: S110 pass if task is None: @@ -357,9 +358,9 @@ def lichess_bot_main(li: LICHESS_TYPE, startup_correspondence_games = [game["gameId"] for game in all_games if game["speed"] == "correspondence"] - active_games = set(game["gameId"] - for game in all_games - if game["gameId"] not in startup_correspondence_games) + active_games = {game["gameId"] + for game in all_games + if game["gameId"] not in startup_correspondence_games} low_time_games: list[GameType] = [] last_check_online_time = Timer(hours(1)) @@ -589,10 +590,10 @@ def start_game(event: EventType, game_id = event["game"]["id"] if game_id in startup_correspondence_games: if enough_time_to_queue(event, config): - logger.info(f'--- Enqueue {config.url + game_id}') + logger.info(f"--- Enqueue {config.url + game_id}") correspondence_queue.put_nowait(game_id) else: - logger.info(f'--- Will start {config.url + game_id} as soon as possible') + logger.info(f"--- Will start {config.url + game_id} as soon as possible") low_time_games.append(event["game"]) startup_correspondence_games.remove(game_id) else: @@ -771,24 +772,20 @@ def record_takeback(game: model.Game, accepted_count: int) -> None: def delete_takeback_record(game: model.Game) -> None: """Delete the takeback record from a game if it has finished.""" if is_game_over(game): - try: + with contextlib.suppress(Exception): os.remove(takeback_record_file_name(game.id)) - except Exception: - pass def prune_takeback_records(all_games: list[GameType]) -> None: """Delete takeback records from games that have ended.""" - active_game_ids = set(game["gameId"] for game in all_games) + active_game_ids = {game["gameId"] for game in all_games} takeback_file_template = takeback_record_file_name("*") prefix, suffix = takeback_file_template.split("*") for takeback_file_name in glob.glob(takeback_file_template): game_id = takeback_file_name.removeprefix(prefix).removesuffix(suffix) if game_id not in active_game_ids: - try: + with contextlib.suppress(Exception): os.remove(takeback_file_name) - except Exception: - pass def takeback_record_file_name(game_id: str) -> str: @@ -1248,7 +1245,7 @@ def version_str(version: list[int]) -> str: def start_program() -> None: """Start lichess-bot and restart when needed.""" - multiprocessing.set_start_method('spawn') + multiprocessing.set_start_method("spawn") try: while should_restart(): disable_restart() diff --git a/lib/matchmaking.py b/lib/matchmaking.py index ca499be2c..a2eea9bec 100644 --- a/lib/matchmaking.py +++ b/lib/matchmaking.py @@ -2,6 +2,7 @@ import random import logging import datetime +import contextlib import test_bot.lichess from lib import model from lib.timer import Timer, seconds, minutes, days, years @@ -145,10 +146,8 @@ def update_user_profile(self) -> None: """Update our user profile data, to get our latest rating.""" if self.last_user_profile_update_time.is_expired(): self.last_user_profile_update_time.reset() - try: + with contextlib.suppress(Exception): self.user_profile = self.li.get_profile() - except Exception: - pass def get_weights(self, online_bots: list[UserProfileType], rating_preference: str, min_rating: int, max_rating: int, game_type: str) -> list[int]: diff --git a/lib/timer.py b/lib/timer.py index 4b02b2a7e..1f220052e 100644 --- a/lib/timer.py +++ b/lib/timer.py @@ -98,6 +98,6 @@ def time_until_expiration(self) -> datetime.timedelta: """How much time is left until it expires.""" return max(seconds(0), self.duration - self.time_since_reset()) - def starting_timestamp(self, format: str) -> str: + def starting_timestamp(self, timestamp_format: str) -> str: """When the timer started.""" - return (datetime.datetime.now() - self.time_since_reset()).strftime(format) + return (datetime.datetime.now() - self.time_since_reset()).strftime(timestamp_format) diff --git a/lib/types.py b/lib/types.py index 758752295..38410db3b 100644 --- a/lib/types.py +++ b/lib/types.py @@ -1,5 +1,5 @@ """Some type hints that can be accessed by all other python files.""" -from typing import Any, Callable, Optional, Union, TypedDict, Literal, Type +from typing import Any, Callable, Optional, Union, TypedDict, Literal from chess.engine import PovWdl, PovScore, PlayResult, Limit, Opponent from chess import Move, Board from queue import Queue @@ -455,5 +455,5 @@ class BackoffDetails(_BackoffDetails, total=False): value: Any # present in the on_predicate decorator case -ENGINE_INPUT_ARGS_TYPE = Union[None, OPTIONS_TYPE, Type[BaseException], BaseException, TracebackType, Board, Limit, str, bool] +ENGINE_INPUT_ARGS_TYPE = Union[None, OPTIONS_TYPE, type[BaseException], BaseException, TracebackType, Board, Limit, str, bool] ENGINE_INPUT_KWARGS_TYPE = Union[None, int, bool, list[Move], Opponent] diff --git a/lib/versioning.yml b/lib/versioning.yml index bb37fa674..e5cf8e897 100644 --- a/lib/versioning.yml +++ b/lib/versioning.yml @@ -1,4 +1,4 @@ -lichess_bot_version: 2025.1.3.1 +lichess_bot_version: 2025.1.4.1 minimum_python_version: '3.9' deprecated_python_version: '3.8' deprecation_date: 2023-05-01 diff --git a/test_bot/buggy_engine.py b/test_bot/buggy_engine.py index 81908c211..f397fcb09 100644 --- a/test_bot/buggy_engine.py +++ b/test_bot/buggy_engine.py @@ -8,7 +8,7 @@ def send_command(command: str) -> None: """Send UCI commands to lichess-bot without output buffering.""" - print(command, flush=True) + print(command, flush=True) # noqa: T201 (print() found) send_command("id name Procrastinator") diff --git a/test_bot/conftest.py b/test_bot/conftest.py index d15a6c60f..22ad7bb49 100644 --- a/test_bot/conftest.py +++ b/test_bot/conftest.py @@ -6,7 +6,7 @@ from typing import Union -def pytest_sessionfinish(session: Session, exitstatus: Union[int, ExitCode]) -> None: +def pytest_sessionfinish(session: Session, exitstatus: Union[int, ExitCode]) -> None: # noqa: ARG001 """ Remove files created when testing lichess-bot. diff --git a/test_bot/homemade.py b/test_bot/homemade.py index caa40ea07..ab96c5bc6 100644 --- a/test_bot/homemade.py +++ b/test_bot/homemade.py @@ -1,27 +1,29 @@ -"""Homemade engine using Stockfish (used in testing).""" -from homemade import ExampleEngine -import chess -import chess.engine -import sys -from lib.config import Configuration -from lib import model -from typing import Optional -from lib.types import OPTIONS_GO_EGTB_TYPE, COMMANDS_TYPE, MOVE - -platform = sys.platform -file_extension = ".exe" if platform == "win32" else "" - - -class Stockfish(ExampleEngine): - """A homemade engine that uses Stockfish.""" - - def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_GO_EGTB_TYPE, stderr: Optional[int], - draw_or_resign: Configuration, game: Optional[model.Game], **popen_args: str): - """Start Stockfish.""" - super().__init__(commands, options, stderr, draw_or_resign, game, **popen_args) - self.engine = chess.engine.SimpleEngine.popen_uci(f"./TEMP/sf{file_extension}") - - def search(self, board: chess.Board, time_limit: chess.engine.Limit, ponder: bool, draw_offered: bool, - root_moves: MOVE) -> chess.engine.PlayResult: - """Get a move using Stockfish.""" - return self.engine.play(board, time_limit) +"""Homemade engine using Stockfish (used in testing).""" +from homemade import ExampleEngine +import chess +import chess.engine +import sys +from lib.config import Configuration +from lib import model +from typing import Optional +from lib.types import OPTIONS_GO_EGTB_TYPE, COMMANDS_TYPE, MOVE + +# ruff: noqa: ARG002 + +platform = sys.platform +file_extension = ".exe" if platform == "win32" else "" + + +class Stockfish(ExampleEngine): + """A homemade engine that uses Stockfish.""" + + def __init__(self, commands: COMMANDS_TYPE, options: OPTIONS_GO_EGTB_TYPE, stderr: Optional[int], + draw_or_resign: Configuration, game: Optional[model.Game], **popen_args: str) -> None: + """Start Stockfish.""" + super().__init__(commands, options, stderr, draw_or_resign, game, **popen_args) + self.engine = chess.engine.SimpleEngine.popen_uci(f"./TEMP/sf{file_extension}") + + def search(self, board: chess.Board, time_limit: chess.engine.Limit, ponder: bool, draw_offered: bool, + root_moves: MOVE) -> chess.engine.PlayResult: + """Get a move using Stockfish.""" + return self.engine.play(board, time_limit) diff --git a/test_bot/lichess.py b/test_bot/lichess.py index db31edb1e..40d1fb504 100644 --- a/test_bot/lichess.py +++ b/test_bot/lichess.py @@ -12,6 +12,8 @@ from lib.types import (UserProfileType, ChallengeType, REQUESTS_PAYLOAD_TYPE, GameType, OnlineType, PublicDataType, BackoffDetails) +# Unused method argument +# ruff: noqa: ARG002 logger = logging.getLogger(__name__) @@ -115,7 +117,7 @@ def __init__(self, sent_game: bool = False) -> None: def iter_lines(self) -> Generator[bytes, None, None]: """Send the events to lichess-bot.""" if self.sent_game: - yield b'' + yield b"" time.sleep(1) else: yield json.dumps( @@ -150,7 +152,6 @@ def __init__(self, def upgrade_to_bot_account(self) -> None: """Isn't used in tests.""" - pass def make_move(self, game_id: str, move: chess.engine.PlayResult) -> None: """Send a move to the opponent engine thread.""" @@ -158,15 +159,12 @@ def make_move(self, game_id: str, move: chess.engine.PlayResult) -> None: def accept_takeback(self, game_id: str, accept: bool) -> None: """Isn't used in tests.""" - pass def chat(self, game_id: str, room: str, text: str) -> None: """Isn't used in tests.""" - pass def abort(self, game_id: str) -> None: """Isn't used in tests.""" - pass def get_event_stream(self) -> EventStream: """Send the `EventStream`.""" @@ -183,11 +181,9 @@ def get_game_stream(self, game_id: str) -> GameStream: def accept_challenge(self, challenge_id: str) -> None: """Isn't used in tests.""" - pass def decline_challenge(self, challenge_id: str, reason: str = "generic") -> None: """Isn't used in tests.""" - pass def get_profile(self) -> UserProfileType: """Return a simple profile for the bot that lichess-bot uses when testing.""" @@ -208,7 +204,6 @@ def get_ongoing_games(self) -> list[GameType]: def resign(self, game_id: str) -> None: """Isn't used in tests.""" - pass def get_game_pgn(self, game_id: str) -> str: """Return a simple PGN.""" @@ -234,7 +229,6 @@ def challenge(self, username: str, payload: REQUESTS_PAYLOAD_TYPE) -> ChallengeT def cancel(self, challenge_id: str) -> None: """Isn't used in tests.""" - pass def online_book_get(self, path: str, params: Optional[dict[str, Union[str, int]]] = None, stream: bool = False) -> OnlineType: diff --git a/test_bot/ruff.toml b/test_bot/ruff.toml new file mode 100644 index 000000000..7a1f45564 --- /dev/null +++ b/test_bot/ruff.toml @@ -0,0 +1,52 @@ +target-version = "py39" + +# The GitHub editor is 127 chars wide. +line-length = 127 + +[lint] +select = ["ALL"] + +ignore = [ + "A004", # Import is shadowing a Python builtin + "ANN401", # Dynamically typed expressions (typing.Any) are disallowed + "B008", # Do not perform function call in argument defaults + "BLE001", # Do not catch blind exception: Exception + "COM812", # Trailing comma missing + "D203", # Require blank line after class declaration before docstring + "D212", # Start multiline docstring on same line as triple-quote + "D404", # Docstring should not start with the word "This" + "DTZ", # datetime without timezone + "EM101", # Exception must not use a string literal, assign to variable first + "EM102", # Exception must not use an f-string literal, assign to variable first + "ERA001", # Found commented-out code + "FA100", # Add from __future__ import annotations to simplify typing + "FBT", # Boolean argument in function definition + "G", # Logging + "I001", # Import block is un-sorted or un-formatted + "N803", # Argument name should be lowercase + "N806", # Variable in function should be lowercase + "N818", # Exception name should be named with an Error suffix + "PERF203", # try-except within a loop incurs performance overhead + "PLR0913", #Too many arguments in function definition + "PLR0915", # Too many statements + "PLR2004", # Magic value used in comparison, consider replacing `20` with a constant variable + "PLW0603", # Using the global statement to update variable is discouraged + "PT018", # Assertion should be broken down into multiple parts + "PTH", # Replace builtin functions with Path methods + "RET505", # Unnecessary else after return statement + "RET508", # Unnecessary elif after break statement + "RUF005", # Consider [*list1, None] instead of concatenation (list1 + [None]) + "RUF021", # Parenthesize `a and b` expressions when chaining `and` and `or` together, to make the precedence clear + "S101", # Use of assert detected + "S113", # Probable use of `requests` call without timeout + "S311", # Standard pseudo-random generators are not suitable for cryptographic purposes + "SIM108", #Use ternary operator instead of `if`-`else`-block + "TC001", # Move application import into a type-checking block + "TC003", # Move standard library import into a type-checking block + "TRY", # Try-except suggestions + "UP035", # Import from collections.abc instead of typing + "UP007", # Use `X | Y` for type annotations +] + +[lint.mccabe] +max-complexity = 10 diff --git a/test_bot/test-requirements.txt b/test_bot/test-requirements.txt index c99503473..04621c311 100644 --- a/test_bot/test-requirements.txt +++ b/test_bot/test-requirements.txt @@ -1,8 +1,6 @@ pytest==8.3.4 pytest-timeout==2.3.1 -flake8==7.1.1 -flake8-markdown==0.6.0 -flake8-docstrings==1.7.0 -mypy==1.14.0 +ruff==0.8.5 +mypy==1.14.1 types-requests==2.32.0.20241016 types-PyYAML==6.0.12.20241230 diff --git a/test_bot/test_bot.py b/test_bot/test_bot.py index ee7a67050..3c5e0d937 100644 --- a/test_bot/test_bot.py +++ b/test_bot/test_bot.py @@ -46,9 +46,12 @@ def download_sf() -> None: with open(archive_name, "wb") as file: file.write(response.content) - archive_open = zipfile.ZipFile if archive_ext == "zip" else tarfile.TarFile - with archive_open(archive_name, "r") as archive_ref: - archive_ref.extractall("./TEMP/") + if archive_ext == "zip": + with zipfile.ZipFile(archive_name, "r") as archive_ref: + archive_ref.extractall("./TEMP/") # noqa: S202 + else: + with tarfile.TarFile(archive_name, "r") as archive_ref: + archive_ref.extractall("./TEMP/", filter="data") exe_ext = ".exe" if platform == "win32" else "" shutil.copyfile(f"./TEMP/stockfish/{sf_base}{exe_ext}", stockfish_path) @@ -69,7 +72,7 @@ def download_lc0() -> None: with open("./TEMP/lc0_zip.zip", "wb") as file: file.write(response.content) with zipfile.ZipFile("./TEMP/lc0_zip.zip", "r") as zip_ref: - zip_ref.extractall("./TEMP/") + zip_ref.extractall("./TEMP/") # noqa: S202 def download_arasan() -> None: @@ -83,9 +86,12 @@ def download_arasan() -> None: response.raise_for_status() with open(f"./TEMP/arasan.{archive_ext}", "wb") as file: file.write(response.content) - archive_open = zipfile.ZipFile if archive_ext == "zip" else tarfile.TarFile - with archive_open(f"./TEMP/arasan.{archive_ext}", "r") as archive_ref: - archive_ref.extractall("./TEMP/") + if archive_ext == "zip": + with zipfile.ZipFile(f"./TEMP/arasan.{archive_ext}", "r") as archive_ref: + archive_ref.extractall("./TEMP/") # noqa: S202 + else: + with tarfile.TarFile(f"./TEMP/arasan.{archive_ext}", "r") as archive_ref: + archive_ref.extractall("./TEMP/", filter="data") shutil.copyfile(f"./TEMP/arasanx-64{file_extension}", f"./TEMP/arasan{file_extension}") if platform != "win32": st = os.stat(f"./TEMP/arasan{file_extension}") @@ -270,7 +276,7 @@ def test_lc0() -> None: @pytest.mark.timeout(150, method="thread") def test_arasan() -> None: """Test lichess-bot with Arasan (XBoard).""" - if platform != "linux" and platform != "win32": + if platform not in ("linux", "win32"): pytest.skip("Platform must be Windows or Linux.") with open("./config.yml.default") as file: CONFIG = yaml.safe_load(file) @@ -325,9 +331,9 @@ def test_buggy_engine() -> None: CONFIG["engine"]["dir"] = "test_bot" def engine_path(CONFIG: CONFIG_DICT_TYPE) -> str: - dir: str = CONFIG["engine"]["dir"] + directory: str = CONFIG["engine"]["dir"] name: str = CONFIG["engine"]["name"].removesuffix(".py") - path = os.path.join(dir, name) + path = os.path.join(directory, name) if platform == "win32": path += ".bat" else: diff --git a/test_bot/test_external_moves.py b/test_bot/test_external_moves.py index e9a4e3dd4..20983893b 100644 --- a/test_bot/test_external_moves.py +++ b/test_bot/test_external_moves.py @@ -1,171 +1,170 @@ -"""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) - - -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() - download_opening_book() - 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") +"""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"}} + 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 + 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) + + +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() + download_opening_book() + 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") diff --git a/test_bot/test_lichess.py b/test_bot/test_lichess.py index 491144273..a63d5ede2 100644 --- a/test_bot/test_lichess.py +++ b/test_bot/test_lichess.py @@ -14,32 +14,32 @@ def test_lichess() -> None: li = lichess.Lichess(token, "https://lichess.org/", "0.0.0", logging.DEBUG, 3) assert len(li.get_online_bots()) > 20 profile = li.get_profile() - profile['seenAt'] = 1700000000000 - assert profile == {'blocking': False, - 'count': {'ai': 3, 'all': 12, 'bookmark': 0, 'draw': 1, 'drawH': 1, 'import': 0, - 'loss': 8, 'lossH': 5, 'me': 0, 'playing': 0, 'rated': 0, 'win': 3, 'winH': 3}, - 'createdAt': 1627834995597, 'followable': True, 'following': False, 'id': 'badsunfish', - 'perfs': {'blitz': {'games': 0, 'prog': 0, 'prov': True, 'rating': 1500, 'rd': 500}, - 'bullet': {'games': 0, 'prog': 0, 'prov': True, 'rating': 1500, 'rd': 500}, - 'classical': {'games': 0, 'prog': 0, 'prov': True, 'rating': 1500, 'rd': 500}, - 'correspondence': {'games': 0, 'prog': 0, 'prov': True, 'rating': 1500, 'rd': 500}, - 'rapid': {'games': 0, 'prog': 0, 'prov': True, 'rating': 1500, 'rd': 500}}, - 'playTime': {'total': 1873, 'tv': 0}, 'seenAt': 1700000000000, 'title': 'BOT', - 'url': 'https://lichess.org/@/BadSunfish', 'username': 'BadSunfish'} + profile["seenAt"] = 1700000000000 + assert profile == {"blocking": False, + "count": {"ai": 3, "all": 12, "bookmark": 0, "draw": 1, "drawH": 1, "import": 0, + "loss": 8, "lossH": 5, "me": 0, "playing": 0, "rated": 0, "win": 3, "winH": 3}, + "createdAt": 1627834995597, "followable": True, "following": False, "id": "badsunfish", + "perfs": {"blitz": {"games": 0, "prog": 0, "prov": True, "rating": 1500, "rd": 500}, + "bullet": {"games": 0, "prog": 0, "prov": True, "rating": 1500, "rd": 500}, + "classical": {"games": 0, "prog": 0, "prov": True, "rating": 1500, "rd": 500}, + "correspondence": {"games": 0, "prog": 0, "prov": True, "rating": 1500, "rd": 500}, + "rapid": {"games": 0, "prog": 0, "prov": True, "rating": 1500, "rd": 500}}, + "playTime": {"total": 1873, "tv": 0}, "seenAt": 1700000000000, "title": "BOT", + "url": "https://lichess.org/@/BadSunfish", "username": "BadSunfish"} assert li.get_ongoing_games() == [] assert li.is_online("NNWithSF") is False public_data = li.get_public_data("lichapibot") for key in public_data["perfs"]: public_data["perfs"][key]["rd"] = 0 - assert public_data == {'blocking': False, 'count': {'ai': 1, 'all': 15774, 'bookmark': 0, 'draw': 3009, 'drawH': 3009, - 'import': 0, 'loss': 6423, 'lossH': 6423, - 'me': 0, 'playing': 0, 'rated': 15121, 'win': 6342, 'winH': 6341}, - 'createdAt': 1524037267522, 'followable': True, 'following': False, 'id': 'lichapibot', - 'perfs': {'blitz': {'games': 2430, 'prog': 3, 'prov': True, 'rating': 2388, 'rd': 0}, - 'bullet': {'games': 7293, 'prog': 9, 'prov': True, 'rating': 2298, 'rd': 0}, - 'classical': {'games': 0, 'prog': 0, 'prov': True, 'rating': 1500, 'rd': 0}, - 'correspondence': {'games': 0, 'prog': 0, 'prov': True, 'rating': 1500, 'rd': 0}, - 'rapid': {'games': 993, 'prog': -80, 'prov': True, 'rating': 2363, 'rd': 0}}, - 'playTime': {'total': 4111502, 'tv': 1582068}, 'profile': {}, - 'seenAt': 1669272254317, 'title': 'BOT', 'tosViolation': True, - 'url': 'https://lichess.org/@/lichapibot', 'username': 'lichapibot'} + assert public_data == {"blocking": False, "count": {"ai": 1, "all": 15774, "bookmark": 0, "draw": 3009, "drawH": 3009, + "import": 0, "loss": 6423, "lossH": 6423, + "me": 0, "playing": 0, "rated": 15121, "win": 6342, "winH": 6341}, + "createdAt": 1524037267522, "followable": True, "following": False, "id": "lichapibot", + "perfs": {"blitz": {"games": 2430, "prog": 3, "prov": True, "rating": 2388, "rd": 0}, + "bullet": {"games": 7293, "prog": 9, "prov": True, "rating": 2298, "rd": 0}, + "classical": {"games": 0, "prog": 0, "prov": True, "rating": 1500, "rd": 0}, + "correspondence": {"games": 0, "prog": 0, "prov": True, "rating": 1500, "rd": 0}, + "rapid": {"games": 993, "prog": -80, "prov": True, "rating": 2363, "rd": 0}}, + "playTime": {"total": 4111502, "tv": 1582068}, "profile": {}, + "seenAt": 1669272254317, "title": "BOT", "tosViolation": True, + "url": "https://lichess.org/@/lichapibot", "username": "lichapibot"}