diff --git a/tests/test_custom_bot_manager.py b/tests/test_custom_bot_manager.py new file mode 100644 index 0000000..67a6ad5 --- /dev/null +++ b/tests/test_custom_bot_manager.py @@ -0,0 +1,50 @@ +import os +import sys +import time +import asyncio +from unittest.mock import patch + +# Ensure the project root is on sys.path so we can import modules +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import custom_bot_manager as cbm + + +async def fake_start(self, token): + cbm.custom_bot_status[self.user_id] = cbm.STATUS_RUNNING + while not getattr(self, "_stop_flag", False): + await asyncio.sleep(0.01) + + +async def fake_close(self): + self._stop_flag = True + self._closed = True + + +def test_custom_bot_lifecycle(): + user_id = "test_user" + token = "fake_token" + + with patch("custom_bot_manager.commands.Bot.start", new=fake_start), patch( + "custom_bot_manager.CustomBot.close", new=fake_close + ): + success, _ = cbm.create_custom_bot(user_id, token) + assert success + assert cbm.custom_bot_status[user_id] == cbm.STATUS_STOPPED + + success, _ = cbm.run_custom_bot_in_thread(user_id, token) + assert success + assert user_id in cbm.custom_bot_threads + + # Allow the start coroutine to run + time.sleep(0.05) + assert cbm.custom_bot_status[user_id] == cbm.STATUS_RUNNING + + thread = cbm.custom_bot_threads[user_id] + success, _ = cbm.stop_custom_bot(user_id) + assert success + + # Wait for the bot thread to exit + thread.join(timeout=1.0) + assert cbm.custom_bot_status[user_id] == cbm.STATUS_STOPPED + assert user_id not in cbm.custom_bot_threads