51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
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
|