"""Imitate `lichess.py`. Used in tests.""" import time import chess import chess.engine import json import logging import traceback from lib.timer import seconds, to_msec from typing import Union, Any, Optional, Generator logger = logging.getLogger(__name__) def backoff_handler(details: Any) -> None: """Log exceptions inside functions with the backoff decorator.""" logger.debug("Backing off {wait:0.1f} seconds after {tries} tries " "calling function {target} with args {args} and kwargs {kwargs}".format(**details)) logger.debug(f"Exception: {traceback.format_exc()}") def is_final(error: Any) -> bool: """Mock error handler for tests when a function has a backup decorator.""" logger.debug(error) return False 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 iter_lines(self) -> Generator[bytes, None, None]: """Send the game events to lichess_bot.""" yield json.dumps( {"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": 10000, "btime": 10000, "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) new_game_state = {"type": "gameState", "moves": moves, "wtime": int(to_msec(wtime)), "btime": int(to_msec(btime)), "winc": 100, "binc": 100} if event == "end": new_game_state["status"] = "outoftime" new_game_state["winner"] = "black" yield json.dumps(new_game_state).encode("utf-8") break if moves: new_game_state["status"] = "started" yield json.dumps(new_game_state).encode("utf-8") 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.""" self.sent_game = sent_game def iter_lines(self) -> Generator[bytes, None, None]: """Send the events to lichess_bot.""" if self.sent_game: yield b'' time.sleep(1) else: yield json.dumps( {"type": "gameStart", "game": {"id": "zzzzzzzz", "source": "friend", "compat": {"bot": True, "board": True}}}).encode("utf-8") # Docs: https://lichess.org/api. class Lichess: """Imitate communication with lichess.org.""" def __init__(self, token: str, url: str, version: str) -> 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] = [] self.sent_game = False def upgrade_to_bot_account(self) -> None: """Isn't used in tests.""" return def make_move(self, game_id: str, move: chess.engine.PlayResult) -> None: """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)) def chat(self, game_id: str, room: str, text: str) -> None: """Isn't used in tests.""" return def abort(self, game_id: str) -> None: """Isn't used in tests.""" return def get_event_stream(self) -> EventStream: """Send the `EventStream`.""" events = EventStream(self.sent_game) self.sent_game = True return events def get_game_stream(self, game_id: str) -> GameStream: """Send the `GameStream`.""" return GameStream() def accept_challenge(self, challenge_id: str) -> None: """Set `self.game_accepted` to true.""" self.game_accepted = True def decline_challenge(self, challenge_id: str, reason: str = "generic") -> None: """Isn't used in tests.""" return def get_profile(self) -> dict[str, Union[str, bool, dict[str, str]]]: """Return a simple profile for the bot that lichess_bot uses when testing.""" return {"id": "b", "username": "b", "online": True, "title": "BOT", "url": "https://lichess.org/@/b", "followable": True, "following": False, "blocking": False, "followsYou": False, "perfs": {}} def get_ongoing_games(self) -> list[str]: """Return that the bot isn't playing a game.""" return [] def resign(self, game_id: str) -> None: """Isn't used in tests.""" return def get_game_pgn(self, game_id: str) -> str: """Return a simple PGN.""" return """ [Event "Test game"] [Site "pytest"] [Date "2022.03.11"] [Round "1"] [White "bo"] [Black "b"] [Result "0-1"] * """ def get_online_bots(self) -> list[dict[str, Union[str, bool]]]: """Return that the only bot online is us.""" return [{"username": "b", "online": True}] def challenge(self, username: str, params: dict[str, str]) -> None: """Isn't used in tests.""" return def cancel(self, challenge_id: str) -> None: """Isn't used in tests.""" return def online_book_get(self, path: str, params: Optional[dict[str, str]] = None) -> None: """Isn't used in tests.""" return def is_online(self, user_id: str) -> bool: """Return that a bot is online.""" return True