Don't copy files when testing

lib/conversation.py

@@ -1,20 +1,23 @@
 """Allows lichess-bot to send messages to the chat."""
 from __future__ import annotations
 import logging
+import test_bot.lichess
 from lib import model
 from lib.engine_wrapper import EngineWrapper
-from lib.lichess import Lichess
+from lib import lichess
 from collections.abc import Sequence
 from lib.timer import seconds
+from typing import Union
 MULTIPROCESSING_LIST_TYPE = Sequence[model.Challenge]
+LICHESS_TYPE = Union[lichess.Lichess, test_bot.lichess.Lichess]
 
 logger = logging.getLogger(__name__)
 
 
 class Conversation:
     """Enables the bot to communicate with its opponent and the spectators."""
 
-    def __init__(self, game: model.Game, engine: EngineWrapper, li: Lichess, version: str,
+    def __init__(self, game: model.Game, engine: EngineWrapper, li: LICHESS_TYPE, version: str,
                  challenge_queue: MULTIPROCESSING_LIST_TYPE) -> None:
         """
         Communication between lichess-bot and the game chats.

lib/engine_wrapper.py

@@ -12,6 +12,7 @@
 import time
 import random
 import math
+import test_bot.lichess
 from collections import Counter
 from collections.abc import Callable
 from lib import config, model, lichess
@@ -25,6 +26,7 @@
 LICHESS_EGTB_MOVE = dict[str, Any]
 CHESSDB_EGTB_MOVE = dict[str, Any]
 MOVE = Union[chess.engine.PlayResult, list[chess.Move]]
+LICHESS_TYPE = Union[lichess.Lichess, test_bot.lichess.Lichess]
 
 logger = logging.getLogger(__name__)
 
@@ -110,7 +112,7 @@ def __exit__(self, exc_type: Optional[Type[BaseException]],
     def play_move(self,
                   board: chess.Board,
                   game: model.Game,
-                  li: lichess.Lichess,
+                  li: LICHESS_TYPE,
                   setup_timer: Timer,
                   move_overhead: datetime.timedelta,
                   can_ponder: bool,
@@ -593,7 +595,12 @@ def getHomemadeEngine(name: str) -> type[MinimalEngine]:
     :return: The engine with this name.
     """
     from lib import strategies
-    engine: type[MinimalEngine] = getattr(strategies, name)
+    from test_bot import strategies as test_strategies
+    engine: type[MinimalEngine]
+    if __name__ == "main":
+        engine = getattr(strategies, name)
+    else:
+        engine = getattr(test_strategies, name)
     return engine
 
 
@@ -728,7 +735,7 @@ def get_book_move(board: chess.Board, game: model.Game,
     return no_book_move
 
 
-def get_online_move(li: lichess.Lichess, board: chess.Board, game: model.Game, online_moves_cfg: config.Configuration,
+def get_online_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, online_moves_cfg: config.Configuration,
                     draw_or_resign_cfg: config.Configuration) -> Union[chess.engine.PlayResult, list[chess.Move]]:
     """
     Get a move from an online source.
@@ -780,7 +787,7 @@ def get_online_move(li: lichess.Lichess, board: chess.Board, game: model.Game, o
     return chess.engine.PlayResult(None, None)
 
 
-def get_chessdb_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
+def get_chessdb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,
                      chessdb_cfg: config.Configuration) -> tuple[Optional[str], chess.engine.InfoDict]:
     """Get a move from chessdb.cn's opening book."""
     wb = "w" if board.turn == chess.WHITE else "b"
@@ -822,7 +829,7 @@ def get_chessdb_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
     return move, comment
 
 
-def get_lichess_cloud_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
+def get_lichess_cloud_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,
                            lichess_cloud_cfg: config.Configuration) -> tuple[Optional[str], chess.engine.InfoDict]:
     """Get a move from the lichess's cloud analysis."""
     wb = "w" if board.turn == chess.WHITE else "b"
@@ -876,7 +883,7 @@ def get_lichess_cloud_move(li: lichess.Lichess, board: chess.Board, game: model.
     return move, comment
 
 
-def get_opening_explorer_move(li: lichess.Lichess, board: chess.Board, game: model.Game,
+def get_opening_explorer_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game,
                               opening_explorer_cfg: config.Configuration
                               ) -> tuple[Optional[str], chess.engine.InfoDict]:
     """Get a move from lichess's opening explorer."""
@@ -926,7 +933,7 @@ def get_opening_explorer_move(li: lichess.Lichess, board: chess.Board, game: mod
     return move, comment
 
 
-def get_online_egtb_move(li: lichess.Lichess, board: chess.Board, game: model.Game, online_egtb_cfg: config.Configuration
+def get_online_egtb_move(li: LICHESS_TYPE, board: chess.Board, game: model.Game, online_egtb_cfg: config.Configuration
                          ) -> tuple[Union[str, list[str], None], int, chess.engine.InfoDict]:
     """
     Get a move from an online egtb (either by lichess or chessdb).
@@ -992,7 +999,7 @@ def get_egtb_move(board: chess.Board, game: model.Game, lichess_bot_tbs: config.
     return chess.engine.PlayResult(None, None)
 
 
-def get_lichess_egtb_move(li: lichess.Lichess, game: model.Game, board: chess.Board, quality: str,
+def get_lichess_egtb_move(li: LICHESS_TYPE, game: model.Game, board: chess.Board, quality: str,
                           variant: str) -> tuple[Union[str, list[str], None], int, chess.engine.InfoDict]:
     """
     Get a move from lichess's egtb.
@@ -1045,7 +1052,7 @@ def good_enough(possible_move: LICHESS_EGTB_MOVE) -> bool:
     return None, -3, {}
 
 
-def get_chessdb_egtb_move(li: lichess.Lichess, game: model.Game, board: chess.Board,
+def get_chessdb_egtb_move(li: LICHESS_TYPE, game: model.Game, board: chess.Board,
                           quality: str) -> tuple[Union[str, list[str], None], int, chess.engine.InfoDict]:
     """
     Get a move from chessdb's egtb.

lib/matchmaking.py

@@ -1,18 +1,20 @@
 """Challenge other bots."""
 import random
 import logging
+import datetime
+import test_bot.lichess
 from lib import model
 from lib.timer import Timer, seconds, minutes, days
 from collections import defaultdict
 from collections.abc import Sequence
 from lib import lichess
-import datetime
 from lib.config import Configuration, FilterType
-from typing import Any, Optional
+from typing import Any, Optional, Union
 USER_PROFILE_TYPE = dict[str, Any]
 EVENT_TYPE = dict[str, Any]
 MULTIPROCESSING_LIST_TYPE = Sequence[model.Challenge]
 DAILY_TIMERS_TYPE = list[Timer]
+LICHESS_TYPE = Union[lichess.Lichess, test_bot.lichess.Lichess]
 
 logger = logging.getLogger(__name__)
 
@@ -43,7 +45,7 @@ def write_daily_challenges(daily_challenges: DAILY_TIMERS_TYPE) -> None:
 class Matchmaking:
     """Challenge other bots."""
 
-    def __init__(self, li: lichess.Lichess, config: Configuration, user_profile: USER_PROFILE_TYPE) -> None:
+    def __init__(self, li: LICHESS_TYPE, config: Configuration, user_profile: USER_PROFILE_TYPE) -> None:
         """Initialize values needed for matchmaking."""
         self.li = li
         self.variants = list(filter(lambda variant: variant != "fromPosition", config.challenge.variants))

lichess-bot.py

@@ -3,7 +3,7 @@
 import chess
 import chess.pgn
 from chess.variant import find_variant
-from lib import engine_wrapper, model, lichess, matchmaking
+from lib import engine_wrapper, model, matchmaking, lichess
 import json
 import logging
 import logging.handlers
@@ -19,6 +19,7 @@
 import sys
 import yaml
 import traceback
+import test_bot.lichess
 from lib.config import load_config, Configuration
 from lib.conversation import Conversation, ChatLine
 from lib.timer import Timer, seconds, msec, hours, to_seconds
@@ -40,9 +41,12 @@
 LOGGING_QUEUE_TYPE = Queue[logging.LogRecord]
 MULTIPROCESSING_LIST_TYPE = MutableSequence[model.Challenge]
 POOL_TYPE = Pool
+LICHESS_TYPE = Union[lichess.Lichess, test_bot.lichess.Lichess]
 
 logger = logging.getLogger(__name__)
 
+lichess_communicator: type[LICHESS_TYPE] = lichess.Lichess
+
 with open("lib/versioning.yml") as version_file:
     versioning_info = yaml.safe_load(version_file)
 
@@ -68,7 +72,7 @@ def signal_handler(signal: int, frame: Any) -> None:
 signal.signal(signal.SIGINT, signal_handler)
 
 
-def upgrade_account(li: lichess.Lichess) -> bool:
+def upgrade_account(li: LICHESS_TYPE) -> bool:
     """Upgrade the account to a BOT account."""
     if li.upgrade_to_bot_account() is None:
         return False
@@ -77,7 +81,7 @@ def upgrade_account(li: lichess.Lichess) -> bool:
     return True
 
 
-def watch_control_stream(control_queue: CONTROL_QUEUE_TYPE, li: lichess.Lichess) -> None:
+def watch_control_stream(control_queue: CONTROL_QUEUE_TYPE, li: LICHESS_TYPE) -> None:
     """Put the events in a queue."""
     error = None
     while not terminated:
@@ -189,7 +193,7 @@ def thread_logging_configurer(queue: Union[CONTROL_QUEUE_TYPE, LOGGING_QUEUE_TYP
     root.setLevel(logging.DEBUG)
 
 
-def start(li: lichess.Lichess, user_profile: USER_PROFILE_TYPE, config: Configuration, logging_level: int,
+def start(li: LICHESS_TYPE, user_profile: USER_PROFILE_TYPE, config: Configuration, logging_level: int,
           log_filename: Optional[str], auto_log_filename: Optional[str], one_game: bool = False) -> None:
     """
     Start lichess-bot.
@@ -253,7 +257,7 @@ def log_proc_count(change: str, active_games: set[str]) -> None:
     logger.info(f"{symbol} Process {change}. Count: {len(active_games)}. IDs: {active_games or None}")
 
 
-def lichess_bot_main(li: lichess.Lichess,
+def lichess_bot_main(li: LICHESS_TYPE,
                      user_profile: USER_PROFILE_TYPE,
                      config: Configuration,
                      challenge_queue: MULTIPROCESSING_LIST_TYPE,
@@ -407,7 +411,7 @@ def start_low_time_games(low_time_games: list[EVENT_GETATTR_GAME_TYPE], active_g
         start_game_thread(active_games, game_id, play_game_args, pool)
 
 
-def accept_challenges(li: lichess.Lichess, challenge_queue: MULTIPROCESSING_LIST_TYPE, active_games: set[str],
+def accept_challenges(li: LICHESS_TYPE, challenge_queue: MULTIPROCESSING_LIST_TYPE, active_games: set[str],
                       max_games: int) -> None:
     """Accept a challenge."""
     while len(active_games) < max_games and challenge_queue:
@@ -425,7 +429,7 @@ def accept_challenges(li: lichess.Lichess, challenge_queue: MULTIPROCESSING_LIST
                 logger.info(f"Skip missing {chlng}")
 
 
-def check_online_status(li: lichess.Lichess, user_profile: USER_PROFILE_TYPE, last_check_online_time: Timer) -> None:
+def check_online_status(li: LICHESS_TYPE, user_profile: USER_PROFILE_TYPE, last_check_online_time: Timer) -> None:
     """Check if lichess.org thinks the bot is online or not. If it isn't, we restart it."""
     global restart
 
@@ -452,7 +456,7 @@ def sort_challenges(challenge_queue: MULTIPROCESSING_LIST_TYPE, challenge_config
         challenge_queue[:] = list_c
 
 
-def game_is_active(li: lichess.Lichess, game_id: str) -> bool:
+def game_is_active(li: LICHESS_TYPE, game_id: str) -> bool:
     """Determine if a game is still being played."""
     return game_id in (ongoing_game["gameId"] for ongoing_game in li.get_ongoing_games())
 
@@ -517,7 +521,7 @@ def enough_time_to_queue(event: EVENT_TYPE, config: Configuration) -> bool:
     return not game["isMyTurn"] or game.get("secondsLeft", math.inf) > minimum_time
 
 
-def handle_challenge(event: EVENT_TYPE, li: lichess.Lichess, challenge_queue: MULTIPROCESSING_LIST_TYPE,
+def handle_challenge(event: EVENT_TYPE, li: LICHESS_TYPE, challenge_queue: MULTIPROCESSING_LIST_TYPE,
                      challenge_config: Configuration, user_profile: USER_PROFILE_TYPE,
                      recent_bot_challenges: defaultdict[str, list[Timer]]) -> None:
     """Handle incoming challenges. It either accepts, declines, or queues them to accept later."""
@@ -538,7 +542,7 @@ def handle_challenge(event: EVENT_TYPE, li: lichess.Lichess, challenge_queue: MU
 
 @backoff.on_exception(backoff.expo, BaseException, max_time=600, giveup=lichess.is_final,  # type: ignore[arg-type]
                       on_backoff=lichess.backoff_handler)
-def play_game(li: lichess.Lichess,
+def play_game(li: LICHESS_TYPE,
               game_id: str,
               control_queue: CONTROL_QUEUE_TYPE,
               user_profile: USER_PROFILE_TYPE,
@@ -719,7 +723,7 @@ def is_game_over(game: model.Game) -> bool:
     return status != "started"
 
 
-def should_exit_game(board: chess.Board, game: model.Game, prior_game: Optional[model.Game], li: lichess.Lichess,
+def should_exit_game(board: chess.Board, game: model.Game, prior_game: Optional[model.Game], li: LICHESS_TYPE,
                      is_correspondence: bool) -> bool:
     """Whether we should exit a game."""
     if (is_correspondence
@@ -807,7 +811,7 @@ def tell_user_game_result(game: model.Game, board: chess.Board) -> None:
         logger.info(f"Game ended by {termination}")
 
 
-def try_get_pgn_game_record(li: lichess.Lichess, config: Configuration, game: model.Game, board: chess.Board,
+def try_get_pgn_game_record(li: LICHESS_TYPE, config: Configuration, game: model.Game, board: chess.Board,
                             engine: engine_wrapper.EngineWrapper) -> str:
     """
     Call `print_pgn_game_record` to write the game to a PGN file and handle errors raised by it.
@@ -825,7 +829,7 @@ def try_get_pgn_game_record(li: lichess.Lichess, config: Configuration, game: mo
         return ""
 
 
-def pgn_game_record(li: lichess.Lichess, config: Configuration, game: model.Game, board: chess.Board,
+def pgn_game_record(li: LICHESS_TYPE, config: Configuration, game: model.Game, board: chess.Board,
                     engine: engine_wrapper.EngineWrapper) -> str:
     """
     Return the text of the game's PGN.
@@ -1021,7 +1025,7 @@ def start_lichess_bot() -> None:
 
     max_retries = CONFIG.engine.online_moves.max_retries
     check_python_version()
-    li = lichess.Lichess(CONFIG.token, CONFIG.url, __version__, logging_level, max_retries)
+    li = lichess_communicator(CONFIG.token, CONFIG.url, __version__, logging_level, max_retries)
 
     user_profile = li.get_profile()
     username = user_profile["username"]

test_bot/conftest.py

@@ -6,8 +6,6 @@
 
 def pytest_sessionfinish(session: Any, exitstatus: Any) -> None:
     """Remove files created when testing lichess-bot."""
-    shutil.copyfile("lib/correct_lichess.py", "lib/lichess.py")
-    os.remove("lib/correct_lichess.py")
     if os.path.exists("TEMP") and not os.getenv("GITHUB_ACTIONS"):
         shutil.rmtree("TEMP")
     if os.path.exists("logs"):