diff --git a/test_bot/conftest.py b/test_bot/conftest.py index 5563bb29a..0df8d6496 100644 --- a/test_bot/conftest.py +++ b/test_bot/conftest.py @@ -8,5 +8,3 @@ def pytest_sessionfinish(session: Any, exitstatus: Any) -> None: """Remove files created when testing lichess-bot.""" if os.path.exists("TEMP") and not os.getenv("GITHUB_ACTIONS"): shutil.rmtree("TEMP") - if os.path.exists("logs"): - shutil.rmtree("logs") diff --git a/test_bot/lichess.py b/test_bot/lichess.py index e2555b8b1..204148187 100644 --- a/test_bot/lichess.py +++ b/test_bot/lichess.py @@ -5,8 +5,10 @@ import json import logging import traceback -from lib.timer import seconds, to_msec +import datetime +from queue import Queue from typing import Union, Any, Optional, Generator +from lib.timer import to_msec JSON_REPLY_TYPE = dict[str, Any] REQUESTS_PAYLOAD_TYPE = dict[str, Any] @@ -29,9 +31,18 @@ def is_final(error: Any) -> bool: class GameStream: """Imitate lichess.org's GameStream. Used in tests.""" - def __init__(self) -> None: - """Initialize `self.moves_sent` to an empty string. It stores the moves that we have already sent.""" - self.moves_sent = "" + def __init__(self, + board_queue: Queue[chess.Board], + clock_queue: Queue[tuple[datetime.timedelta, datetime.timedelta, datetime.timedelta]]) -> None: + """ + Capture the interprocess queues that will feed the gameStream with game information. + + :param board_queue: Updated board positions from the lichess_org_simulator() function. + :param clock_queue: Updated game clock timings (white time, black time, and increment) from the + lichess_org_simulator() function. + """ + self.board_queue = board_queue + self.clock_queue = clock_queue def iter_lines(self) -> Generator[bytes, None, None]: """Send the game events to lichess-bot.""" @@ -64,40 +75,27 @@ def iter_lines(self) -> Generator[bytes, None, None]: "winc": 100, "binc": 100, "status": "started"}}).encode("utf-8") - time.sleep(1) while True: - time.sleep(0.001) - with open("./logs/events.txt") as events: - event = events.read() - while True: - try: - with open("./logs/states.txt") as states: - state = states.read().split("\n") - moves = state[0] - board = chess.Board() - for move in moves.split(): - board.push_uci(move) - wtime, btime = [seconds(float(n)) for n in state[1].split(",")] - if len(moves) <= len(self.moves_sent) and not event: - time.sleep(0.001) - continue - self.moves_sent = moves - break - except (IndexError, ValueError): - pass - time.sleep(0.1) + board = self.board_queue.get() + self.board_queue.task_done() + + wtime, btime, increment = self.clock_queue.get() + self.clock_queue.task_done() + new_game_state = {"type": "gameState", - "moves": moves, + "moves": " ".join(move.uci() for move in board.move_stack), "wtime": int(to_msec(wtime)), "btime": int(to_msec(btime)), - "winc": 100, - "binc": 100} - if event == "end": + "winc": int(to_msec(increment)), + "binc": int(to_msec(increment))} + + if board.is_game_over(): new_game_state["status"] = "outoftime" new_game_state["winner"] = "black" yield json.dumps(new_game_state).encode("utf-8") break - if moves: + + if board.move_stack: new_game_state["status"] = "started" yield json.dumps(new_game_state).encode("utf-8") @@ -106,7 +104,11 @@ class EventStream: """Imitate lichess.org's EventStream. Used in tests.""" def __init__(self, sent_game: bool = False) -> None: - """:param sent_game: If we have already sent the `gameStart` event, so we don't send it again.""" + """ + Start the event stream for the lichess_bot_main() loop. + + :param sent_game: If we have already sent the `gameStart` event, so we don't send it again. + """ self.sent_game = sent_game def iter_lines(self) -> Generator[bytes, None, None]: @@ -127,26 +129,31 @@ def iter_lines(self) -> Generator[bytes, None, None]: class Lichess: """Imitate communication with lichess.org.""" - def __init__(self, token: str, url: str, version: str, logging_level: int, max_retries: int) -> None: - """Has the same parameters as `lichess.Lichess` to be able to be used in its placed without any modification.""" - self.baseUrl = url - self.game_accepted = False - self.moves: list[chess.engine.PlayResult] = [] + def __init__(self, + move_queue: Queue[Optional[chess.Move]], + board_queue: Queue[chess.Board], + clock_queue: Queue[tuple[datetime.timedelta, datetime.timedelta, datetime.timedelta]]) -> None: + """ + Capture the interprocess queues to distribute them to the eventStream and gameStream instances. + + :param move_queue: An interprocess queue to send moves chosen by the bot under test to the mock lichess function. + :param board_queue: An interprocess queue to send board positions to the mock game stream. + :param clock_queue: An interprocess queue to send game clock information to the mock game stream. + """ + self.baseUrl = "testing" + self.move_queue = move_queue + self.board_queue = board_queue + self.clock_queue = clock_queue self.sent_game = False + self.started_game_stream = False def upgrade_to_bot_account(self) -> JSON_REPLY_TYPE: """Isn't used in tests.""" return {} def make_move(self, game_id: str, move: chess.engine.PlayResult) -> JSON_REPLY_TYPE: - """Write a move to `./logs/states.txt`, to be read by the opponent.""" - self.moves.append(move) - uci_move = move.move.uci() if move.move else "error" - with open("./logs/states.txt") as file: - contents = file.read().split("\n") - contents[0] += f" {uci_move}" - with open("./logs/states.txt", "w") as file: - file.write("\n".join(contents)) + """Send a move to the opponent engine thread.""" + self.move_queue.put(move.move) return {} def chat(self, game_id: str, room: str, text: str) -> JSON_REPLY_TYPE: @@ -165,11 +172,13 @@ def get_event_stream(self) -> EventStream: def get_game_stream(self, game_id: str) -> GameStream: """Send the `GameStream`.""" - return GameStream() + if self.started_game_stream: + self.move_queue.put(None) + self.started_game_stream = True + return GameStream(self.board_queue, self.clock_queue) def accept_challenge(self, challenge_id: str) -> JSON_REPLY_TYPE: - """Set `self.game_accepted` to true.""" - self.game_accepted = True + """Isn't used in tests.""" return {} def decline_challenge(self, challenge_id: str, reason: str = "generic") -> JSON_REPLY_TYPE: diff --git a/test_bot/test_bot.py b/test_bot/test_bot.py index 9b2bc25b3..4a9104d6f 100644 --- a/test_bot/test_bot.py +++ b/test_bot/test_bot.py @@ -1,337 +1,312 @@ -"""Test lichess-bot.""" -import pytest -import zipfile -import requests -import time -import yaml -import chess -import chess.engine -import threading -import os -import sys -import stat -import shutil -import importlib -import tarfile -import test_bot.lichess -from lib import config -from lib.timer import Timer, to_seconds, seconds -from typing import Any -from lib.engine_wrapper import test_suffix -if "pytest" not in sys.modules: - sys.exit(f"The script {os.path.basename(__file__)} should only be run by pytest.") -lichess_bot = importlib.import_module("lichess-bot") - -platform = sys.platform -file_extension = ".exe" if platform == "win32" else "" -stockfish_path = f"./TEMP/sf{file_extension}" - - -def download_sf() -> None: - """Download Stockfish 15.""" - if os.path.exists(stockfish_path): - return - - windows_or_linux = "windows" if platform == "win32" else "ubuntu" - sf_base = f"stockfish-{windows_or_linux}-x86-64-modern" - archive_ext = "zip" if platform == "win32" else "tar" - archive_link = f"https://github.com/official-stockfish/Stockfish/releases/download/sf_16/{sf_base}.{archive_ext}" - - response = requests.get(archive_link, allow_redirects=True) - archive_name = f"./TEMP/sf_zip.{archive_ext}" - 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/") - - exe_ext = ".exe" if platform == "win32" else "" - shutil.copyfile(f"./TEMP/stockfish/{sf_base}{exe_ext}", stockfish_path) - - if windows_or_linux == "ubuntu": - st = os.stat(stockfish_path) - os.chmod(stockfish_path, st.st_mode | stat.S_IEXEC) - - -def download_lc0() -> None: - """Download Leela Chess Zero 0.29.0.""" - if os.path.exists("./TEMP/lc0.exe"): - return - response = requests.get("https://github.com/LeelaChessZero/lc0/releases/download/v0.29.0/lc0-v0.29.0-windows-cpu-dnnl.zip", - allow_redirects=True) - 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/") - - -def download_sjeng() -> None: - """Download Sjeng.""" - if os.path.exists("./TEMP/sjeng.exe"): - return - response = requests.get("https://sjeng.org/ftp/Sjeng112.zip", allow_redirects=True) - with open("./TEMP/sjeng_zip.zip", "wb") as file: - file.write(response.content) - with zipfile.ZipFile("./TEMP/sjeng_zip.zip", "r") as zip_ref: - zip_ref.extractall("./TEMP/") - shutil.copyfile("./TEMP/Release/Sjeng112.exe", "./TEMP/sjeng.exe") - - -if not os.path.exists("TEMP"): - os.mkdir("TEMP") -download_sf() -if platform == "win32": - download_lc0() - download_sjeng() -logging_level = lichess_bot.logging.DEBUG -testing_log_file_name = None -lichess_bot.logging_configurer(logging_level, testing_log_file_name, None, False) -lichess_bot.logger.info("Downloaded engines") - - -def thread_for_test(opponent_path: str) -> None: - """Play the moves for the opponent of lichess-bot.""" - open("./logs/events.txt", "w").close() - open("./logs/states.txt", "w").close() - open("./logs/result.txt", "w").close() - - start_time = seconds(10) - increment = seconds(0.1) - - board = chess.Board() - wtime = start_time - btime = start_time - - with open("./logs/states.txt", "w") as file: - file.write(f"\n{to_seconds(wtime)},{to_seconds(btime)}") - - engine = chess.engine.SimpleEngine.popen_uci(opponent_path) - engine.configure({"Skill Level": 0, "Move Overhead": 1000, "Use NNUE": False} - if opponent_path == stockfish_path else {}) - - while not board.is_game_over(): - if len(board.move_stack) % 2 == 0: - if not board.move_stack: - move = engine.play(board, - chess.engine.Limit(time=1), - ponder=False) - else: - move_timer = Timer() - move = engine.play(board, - chess.engine.Limit(white_clock=to_seconds(wtime) - 2, - white_inc=to_seconds(increment)), - ponder=False) - wtime -= move_timer.time_since_reset() - wtime += increment - engine_move = move.move - if engine_move is None: - raise RuntimeError("Engine attempted to make null move.") - board.push(engine_move) - - uci_move = engine_move.uci() - with open("./logs/states.txt") as states: - state_str = states.read() - state = state_str.split("\n") - state[0] += f" {uci_move}" - state_str = "\n".join(state) - with open("./logs/states.txt", "w") as file: - file.write(state_str) - - else: # lichess-bot move. - move_timer = Timer() - state2 = state_str - moves_are_correct = False - while state2 == state_str or not moves_are_correct: - with open("./logs/states.txt") as states: - state2 = states.read() - time.sleep(0.001) - moves = state2.split("\n")[0] - temp_board = chess.Board() - moves_are_correct = True - for move_str in moves.split(): - try: - temp_board.push_uci(move_str) - except ValueError: - moves_are_correct = False - with open("./logs/states.txt") as states: - state2 = states.read() - if len(board.move_stack) > 1: - btime -= move_timer.time_since_reset() - btime += increment - move_str = state2.split("\n")[0].split(" ")[-1] - board.push_uci(move_str) - - time.sleep(0.001) - with open("./logs/states.txt") as states: - state_str = states.read() - state = state_str.split("\n") - state[1] = f"{to_seconds(wtime)},{to_seconds(btime)}" - state_str = "\n".join(state) - with open("./logs/states.txt", "w") as file: - file.write(state_str) - - with open("./logs/events.txt", "w") as file: - file.write("end") - engine.quit() - outcome = board.outcome() - win = outcome.winner == chess.BLACK if outcome else False - with open("./logs/result.txt", "w") as file: - file.write("1" if win else "0") - - -def run_bot(raw_config: dict[str, Any], logging_level: int, opponent_path: str = stockfish_path) -> str: - """Start lichess-bot.""" - config.insert_default_values(raw_config) - CONFIG = config.Configuration(raw_config) - lichess_bot.logger.info(lichess_bot.intro()) - li = test_bot.lichess.Lichess(CONFIG.token, CONFIG.url, lichess_bot.__version__, logging_level, 1) - - user_profile = li.get_profile() - username = user_profile["username"] - if user_profile.get("title") != "BOT": - return "0" - lichess_bot.logger.info(f"Welcome {username}!") - lichess_bot.disable_restart() - - thr = threading.Thread(target=thread_for_test, args=[opponent_path]) - thr.start() - lichess_bot.start(li, user_profile, CONFIG, logging_level, testing_log_file_name, None, one_game=True) - thr.join() - - with open("./logs/result.txt") as file: - data = file.read() - return data - - -@pytest.mark.timeout(150, method="thread") -def test_sf() -> None: - """Test lichess-bot with Stockfish (UCI).""" - if platform != "linux" and platform != "win32": - assert True - return - if os.path.exists("logs"): - shutil.rmtree("logs") - os.mkdir("logs") - with open("./config.yml.default") as file: - CONFIG = yaml.safe_load(file) - CONFIG["token"] = "" - CONFIG["engine"]["dir"] = "./TEMP/" - CONFIG["engine"]["name"] = f"sf{file_extension}" - CONFIG["engine"]["uci_options"]["Threads"] = 1 - CONFIG["pgn_directory"] = "TEMP/sf_game_record" - win = run_bot(CONFIG, logging_level) - shutil.rmtree("logs") - lichess_bot.logger.info("Finished Testing SF") - assert win == "1" - assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], - "bo vs b - zzzzzzzz.pgn")) - - -@pytest.mark.timeout(150, method="thread") -def test_lc0() -> None: - """Test lichess-bot with Leela Chess Zero (UCI).""" - if platform != "win32": - assert True - return - if os.path.exists("logs"): - shutil.rmtree("logs") - os.mkdir("logs") - with open("./config.yml.default") as file: - CONFIG = yaml.safe_load(file) - CONFIG["token"] = "" - CONFIG["engine"]["dir"] = "./TEMP/" - CONFIG["engine"]["working_dir"] = "./TEMP/" - CONFIG["engine"]["name"] = "lc0.exe" - CONFIG["engine"]["uci_options"]["Threads"] = 1 - CONFIG["engine"]["uci_options"].pop("Hash", None) - CONFIG["engine"]["uci_options"].pop("Move Overhead", None) - CONFIG["pgn_directory"] = "TEMP/lc0_game_record" - win = run_bot(CONFIG, logging_level) - shutil.rmtree("logs") - lichess_bot.logger.info("Finished Testing LC0") - assert win == "1" - assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], - "bo vs b - zzzzzzzz.pgn")) - - -@pytest.mark.timeout(150, method="thread") -def test_sjeng() -> None: - """Test lichess-bot with Sjeng (XBoard).""" - if platform != "win32": - assert True - return - if os.path.exists("logs"): - shutil.rmtree("logs") - os.mkdir("logs") - with open("./config.yml.default") as file: - CONFIG = yaml.safe_load(file) - CONFIG["token"] = "" - CONFIG["engine"]["dir"] = "./TEMP/" - CONFIG["engine"]["working_dir"] = "./TEMP/" - CONFIG["engine"]["protocol"] = "xboard" - CONFIG["engine"]["name"] = "sjeng.exe" - CONFIG["engine"]["ponder"] = False - CONFIG["pgn_directory"] = "TEMP/sjeng_game_record" - win = run_bot(CONFIG, logging_level) - shutil.rmtree("logs") - lichess_bot.logger.info("Finished Testing Sjeng") - assert win == "1" - assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], - "bo vs b - zzzzzzzz.pgn")) - - -@pytest.mark.timeout(150, method="thread") -def test_homemade() -> None: - """Test lichess-bot with a homemade engine running Stockfish (Homemade).""" - if platform != "linux" and platform != "win32": - assert True - return - if os.path.exists("logs"): - shutil.rmtree("logs") - os.mkdir("logs") - with open("./config.yml.default") as file: - CONFIG = yaml.safe_load(file) - CONFIG["token"] = "" - CONFIG["engine"]["name"] = f"Stockfish{test_suffix}" - CONFIG["engine"]["protocol"] = "homemade" - CONFIG["pgn_directory"] = "TEMP/homemade_game_record" - win = run_bot(CONFIG, logging_level) - shutil.rmtree("logs") - lichess_bot.logger.info("Finished Testing Homemade") - assert win == "1" - assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], - "bo vs b - zzzzzzzz.pgn")) - - -@pytest.mark.timeout(30, method="thread") -def test_buggy_engine() -> None: - """Test lichess-bot with an engine that causes a timeout error within python-chess.""" - if os.path.exists("logs"): - shutil.rmtree("logs") - os.mkdir("logs") - with open("./config.yml.default") as file: - CONFIG = yaml.safe_load(file) - CONFIG["token"] = "" - CONFIG["engine"]["dir"] = "test_bot" - - def engine_path(CONFIG: dict[str, Any]) -> str: - dir: str = CONFIG["engine"]["dir"] - name: str = CONFIG["engine"]["name"] - return os.path.join(dir, name) - - if platform == "win32": - CONFIG["engine"]["name"] = "buggy_engine.bat" - else: - CONFIG["engine"]["name"] = "buggy_engine" - st = os.stat(engine_path(CONFIG)) - os.chmod(engine_path(CONFIG), st.st_mode | stat.S_IEXEC) - CONFIG["engine"]["uci_options"] = {"go_commands": {"movetime": 100}} - CONFIG["pgn_directory"] = "TEMP/bug_game_record" - - win = run_bot(CONFIG, logging_level, engine_path(CONFIG)) - shutil.rmtree("logs") - lichess_bot.logger.info("Finished Testing buggy engine") - assert win == "1" - assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], - "bo vs b - zzzzzzzz.pgn")) +"""Test lichess-bot.""" +import pytest +import zipfile +import requests +import yaml +import chess +import chess.engine +import threading +import os +import sys +import stat +import shutil +import importlib +import tarfile +import datetime +from multiprocessing import Manager +from queue import Queue +import test_bot.lichess +from lib import config +from lib.timer import Timer, to_seconds, seconds +from typing import Any, Optional +from lib.engine_wrapper import test_suffix +if "pytest" not in sys.modules: + sys.exit(f"The script {os.path.basename(__file__)} should only be run by pytest.") +lichess_bot = importlib.import_module("lichess-bot") + +platform = sys.platform +file_extension = ".exe" if platform == "win32" else "" +stockfish_path = f"./TEMP/sf{file_extension}" + + +def download_sf() -> None: + """Download Stockfish 15.""" + if os.path.exists(stockfish_path): + return + + windows_or_linux = "windows" if platform == "win32" else "ubuntu" + sf_base = f"stockfish-{windows_or_linux}-x86-64-modern" + archive_ext = "zip" if platform == "win32" else "tar" + archive_link = f"https://github.com/official-stockfish/Stockfish/releases/download/sf_16/{sf_base}.{archive_ext}" + + response = requests.get(archive_link, allow_redirects=True) + archive_name = f"./TEMP/sf_zip.{archive_ext}" + 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/") + + exe_ext = ".exe" if platform == "win32" else "" + shutil.copyfile(f"./TEMP/stockfish/{sf_base}{exe_ext}", stockfish_path) + + if windows_or_linux == "ubuntu": + st = os.stat(stockfish_path) + os.chmod(stockfish_path, st.st_mode | stat.S_IEXEC) + + +def download_lc0() -> None: + """Download Leela Chess Zero 0.29.0.""" + if os.path.exists("./TEMP/lc0.exe"): + return + response = requests.get("https://github.com/LeelaChessZero/lc0/releases/download/v0.29.0/lc0-v0.29.0-windows-cpu-dnnl.zip", + allow_redirects=True) + 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/") + + +def download_sjeng() -> None: + """Download Sjeng.""" + if os.path.exists("./TEMP/sjeng.exe"): + return + response = requests.get("https://sjeng.org/ftp/Sjeng112.zip", allow_redirects=True) + with open("./TEMP/sjeng_zip.zip", "wb") as file: + file.write(response.content) + with zipfile.ZipFile("./TEMP/sjeng_zip.zip", "r") as zip_ref: + zip_ref.extractall("./TEMP/") + shutil.copyfile("./TEMP/Release/Sjeng112.exe", "./TEMP/sjeng.exe") + + +os.makedirs("TEMP", exist_ok=True) +download_sf() +if platform == "win32": + download_lc0() + download_sjeng() +logging_level = lichess_bot.logging.DEBUG +testing_log_file_name = None +lichess_bot.logging_configurer(logging_level, testing_log_file_name, None, False) +lichess_bot.logger.info("Downloaded engines") + + +def lichess_org_simulator(opponent_path: str, + move_queue: Queue[Optional[chess.Move]], + board_queue: Queue[chess.Board], + clock_queue: Queue[tuple[datetime.timedelta, datetime.timedelta, datetime.timedelta]], + results: Queue[bool]) -> None: + """ + Run a mocked version of the lichess.org server to provide an opponent for a test. This opponent always plays white. + + :param opponent_path: The path to the executable of the opponent. Usually Stockfish. + :param move_queue: An interprocess queue that supplies the moves chosen by the bot being tested. + :param board_queue: An interprocess queue where this function sends the updated board after choosing a move. + :param clock_queue: An interprocess queue where this function sends the updated game clock after choosing a move. + :param results: An interprocess queue where this function sends the result of the game to the testing function. + """ + start_time = seconds(10) + increment = seconds(0.1) + + board = chess.Board() + wtime = start_time + btime = start_time + + engine = chess.engine.SimpleEngine.popen_uci(opponent_path) + engine.configure({"Skill Level": 0, "Move Overhead": 1000, "Use NNUE": False} + if opponent_path == stockfish_path else {}) + + while not board.is_game_over(): + if board.turn == chess.WHITE: + if not board.move_stack: + move = engine.play(board, + chess.engine.Limit(time=1), + ponder=False) + else: + move_timer = Timer() + move = engine.play(board, + chess.engine.Limit(white_clock=to_seconds(wtime - seconds(2.0)), + white_inc=to_seconds(increment), + black_clock=to_seconds(btime), + black_inc=to_seconds(increment)), + ponder=False) + wtime -= move_timer.time_since_reset() + wtime += increment + engine_move = move.move + if engine_move is None: + raise RuntimeError("Engine attempted to make null move.") + board.push(engine_move) + board_queue.put(board) + clock_queue.put((wtime, btime, increment)) + else: + move_timer = Timer() + while (bot_move := move_queue.get()) is None: + board_queue.put(board) + clock_queue.put((wtime, btime, increment)) + move_queue.task_done() + board.push(bot_move) + move_queue.task_done() + if len(board.move_stack) > 2: + btime -= move_timer.time_since_reset() + btime += increment + + board_queue.put(board) + clock_queue.put((wtime, btime, increment)) + engine.quit() + outcome = board.outcome() + results.put(outcome is not None and outcome.winner == chess.BLACK) + + +def run_bot(raw_config: dict[str, Any], logging_level: int, opponent_path: str = stockfish_path) -> bool: + """ + Start lichess-bot test with a mocked version of the lichess.org site. + + :param raw_config: A dictionary of values to specify the engine to test. This engine will play as white. + :param logging_level: The level of logging to use during the test. Usually logging.DEBUG. + :param opponent_path: The path to the executable that will play the opponent. The opponent plays as black. + """ + config.insert_default_values(raw_config) + CONFIG = config.Configuration(raw_config) + lichess_bot.logger.info(lichess_bot.intro()) + manager = Manager() + board_queue: Queue[chess.Board] = manager.Queue() + clock_queue: Queue[tuple[datetime.timedelta, datetime.timedelta, datetime.timedelta]] = manager.Queue() + move_queue: Queue[Optional[chess.Move]] = manager.Queue() + li = test_bot.lichess.Lichess(move_queue, board_queue, clock_queue) + + user_profile = li.get_profile() + username = user_profile["username"] + if user_profile.get("title") != "BOT": + return False + lichess_bot.logger.info(f"Welcome {username}!") + lichess_bot.disable_restart() + + results: Queue[bool] = manager.Queue() + thr = threading.Thread(target=lichess_org_simulator, args=[opponent_path, move_queue, board_queue, clock_queue, results]) + thr.start() + lichess_bot.start(li, user_profile, CONFIG, logging_level, testing_log_file_name, None, one_game=True) + + result = results.get() + results.task_done() + + results.join() + board_queue.join() + clock_queue.join() + move_queue.join() + + thr.join() + + return result + + +@pytest.mark.timeout(150, method="thread") +def test_sf() -> None: + """Test lichess-bot with Stockfish (UCI).""" + if platform != "linux" and platform != "win32": + assert True + return + with open("./config.yml.default") as file: + CONFIG = yaml.safe_load(file) + CONFIG["token"] = "" + CONFIG["engine"]["dir"] = "./TEMP/" + CONFIG["engine"]["name"] = f"sf{file_extension}" + CONFIG["engine"]["uci_options"]["Threads"] = 1 + CONFIG["pgn_directory"] = "TEMP/sf_game_record" + win = run_bot(CONFIG, logging_level) + lichess_bot.logger.info("Finished Testing SF") + assert win + assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], + "bo vs b - zzzzzzzz.pgn")) + + +@pytest.mark.timeout(150, method="thread") +def test_lc0() -> None: + """Test lichess-bot with Leela Chess Zero (UCI).""" + if platform != "win32": + assert True + return + with open("./config.yml.default") as file: + CONFIG = yaml.safe_load(file) + CONFIG["token"] = "" + CONFIG["engine"]["dir"] = "./TEMP/" + CONFIG["engine"]["working_dir"] = "./TEMP/" + CONFIG["engine"]["name"] = "lc0.exe" + CONFIG["engine"]["uci_options"]["Threads"] = 1 + CONFIG["engine"]["uci_options"].pop("Hash", None) + CONFIG["engine"]["uci_options"].pop("Move Overhead", None) + CONFIG["pgn_directory"] = "TEMP/lc0_game_record" + win = run_bot(CONFIG, logging_level) + lichess_bot.logger.info("Finished Testing LC0") + assert win + assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], + "bo vs b - zzzzzzzz.pgn")) + + +@pytest.mark.timeout(150, method="thread") +def test_sjeng() -> None: + """Test lichess-bot with Sjeng (XBoard).""" + if platform != "win32": + assert True + return + with open("./config.yml.default") as file: + CONFIG = yaml.safe_load(file) + CONFIG["token"] = "" + CONFIG["engine"]["dir"] = "./TEMP/" + CONFIG["engine"]["working_dir"] = "./TEMP/" + CONFIG["engine"]["protocol"] = "xboard" + CONFIG["engine"]["name"] = "sjeng.exe" + CONFIG["engine"]["ponder"] = False + CONFIG["pgn_directory"] = "TEMP/sjeng_game_record" + win = run_bot(CONFIG, logging_level) + lichess_bot.logger.info("Finished Testing Sjeng") + assert win + assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], + "bo vs b - zzzzzzzz.pgn")) + + +@pytest.mark.timeout(150, method="thread") +def test_homemade() -> None: + """Test lichess-bot with a homemade engine running Stockfish (Homemade).""" + if platform != "linux" and platform != "win32": + assert True + return + with open("./config.yml.default") as file: + CONFIG = yaml.safe_load(file) + CONFIG["token"] = "" + CONFIG["engine"]["name"] = f"Stockfish{test_suffix}" + CONFIG["engine"]["protocol"] = "homemade" + CONFIG["pgn_directory"] = "TEMP/homemade_game_record" + win = run_bot(CONFIG, logging_level) + lichess_bot.logger.info("Finished Testing Homemade") + assert win + assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], + "bo vs b - zzzzzzzz.pgn")) + + +@pytest.mark.timeout(30, method="thread") +def test_buggy_engine() -> None: + """Test lichess-bot with an engine that causes a timeout error within python-chess.""" + with open("./config.yml.default") as file: + CONFIG = yaml.safe_load(file) + CONFIG["token"] = "" + CONFIG["engine"]["dir"] = "test_bot" + + def engine_path(CONFIG: dict[str, Any]) -> str: + dir: str = CONFIG["engine"]["dir"] + name: str = CONFIG["engine"]["name"] + return os.path.join(dir, name) + + if platform == "win32": + CONFIG["engine"]["name"] = "buggy_engine.bat" + else: + CONFIG["engine"]["name"] = "buggy_engine" + st = os.stat(engine_path(CONFIG)) + os.chmod(engine_path(CONFIG), st.st_mode | stat.S_IEXEC) + CONFIG["engine"]["uci_options"] = {"go_commands": {"movetime": 100}} + CONFIG["pgn_directory"] = "TEMP/bug_game_record" + + win = run_bot(CONFIG, logging_level, engine_path(CONFIG)) + lichess_bot.logger.info("Finished Testing buggy engine") + assert win + assert os.path.isfile(os.path.join(CONFIG["pgn_directory"], + "bo vs b - zzzzzzzz.pgn"))