51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
import asyncio
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from aitrader.config import Settings
|
|
from aitrader.notify import discord
|
|
|
|
|
|
def test_discord_enabled_with_webhook():
|
|
s = Settings(discord=dict(enabled=True), discord_webhook_url="https://xyz")
|
|
assert discord._enabled(s)
|
|
|
|
|
|
def test_discord_enabled_with_bot():
|
|
s = Settings(discord=dict(enabled=True), discord_bot_token="xyz")
|
|
assert discord._enabled(s)
|
|
|
|
|
|
def test_discord_disabled():
|
|
s = Settings(
|
|
discord=dict(enabled=False),
|
|
discord_webhook_url="https://xyz",
|
|
discord_bot_token="xyz",
|
|
)
|
|
assert not discord._enabled(s)
|
|
|
|
|
|
@patch("aitrader.notify.bot.get_bot_instance")
|
|
def test_send_via_bot_success(mock_get_bot):
|
|
mock_bot = MagicMock()
|
|
mock_bot.is_ready.return_value = True
|
|
|
|
mock_channel = MagicMock()
|
|
mock_bot.get_channel.return_value = mock_channel
|
|
|
|
mock_get_bot.return_value = mock_bot
|
|
|
|
s = Settings(discord_channel_id="12345", discord_bot_token="token")
|
|
|
|
def run_coro_sync(coro, loop):
|
|
asyncio.run(coro)
|
|
fut = MagicMock()
|
|
fut.result.return_value = None
|
|
return fut
|
|
|
|
with patch("asyncio.run_coroutine_threadsafe", side_effect=run_coro_sync) as mock_run:
|
|
res = discord.send_via_bot(s, {"title": "Test"}, channel_type="trades")
|
|
assert res
|
|
mock_run.assert_called_once()
|
|
mock_bot.get_channel.assert_called_once_with(12345)
|