diff options
Diffstat (limited to 'shared/tests')
| -rw-r--r-- | shared/tests/test_notifier.py | 177 |
1 files changed, 177 insertions, 0 deletions
diff --git a/shared/tests/test_notifier.py b/shared/tests/test_notifier.py new file mode 100644 index 0000000..09e731a --- /dev/null +++ b/shared/tests/test_notifier.py @@ -0,0 +1,177 @@ +"""Tests for Telegram notification service.""" +import os +import uuid +from decimal import Decimal +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +import pytest_asyncio + +from shared.models import Signal, Order, OrderSide, OrderType, OrderStatus, Position +from shared.notifier import TelegramNotifier + + +class TestTelegramNotifierEnabled: + """Test the enabled property.""" + + def test_disabled_when_no_token(self): + notifier = TelegramNotifier(bot_token="", chat_id="123") + assert notifier.enabled is False + + def test_enabled_with_token(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="123") + assert notifier.enabled is True + + def test_disabled_when_token_is_empty_string(self): + notifier = TelegramNotifier(bot_token="", chat_id="") + assert notifier.enabled is False + + +class TestTelegramNotifierSend: + """Test send method.""" + + @pytest.mark.asyncio + async def test_send_does_nothing_when_disabled(self): + notifier = TelegramNotifier(bot_token="", chat_id="123") + # Should not raise or do anything + await notifier.send("test message") + + @pytest.mark.asyncio + async def test_send_posts_to_api(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="12345") + + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.json = AsyncMock(return_value={"ok": True}) + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=False) + + mock_session = AsyncMock() + mock_session.post = MagicMock(return_value=mock_response) + + with patch.object(notifier, "_session", mock_session): + await notifier.send("Hello, world!") + + mock_session.post.assert_called_once() + call_args = mock_session.post.call_args + assert "fake-token" in call_args[0][0] + assert call_args[1]["json"]["chat_id"] == "12345" + assert call_args[1]["json"]["text"] == "Hello, world!" + assert call_args[1]["json"]["parse_mode"] == "HTML" + + @pytest.mark.asyncio + async def test_send_with_custom_parse_mode(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="12345") + + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.json = AsyncMock(return_value={"ok": True}) + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=False) + + mock_session = AsyncMock() + mock_session.post = MagicMock(return_value=mock_response) + + with patch.object(notifier, "_session", mock_session): + await notifier.send("test", parse_mode="Markdown") + + call_args = mock_session.post.call_args + assert call_args[1]["json"]["parse_mode"] == "Markdown" + + +class TestTelegramNotifierFormatters: + """Test message formatting methods.""" + + @pytest.mark.asyncio + async def test_send_signal_formats_message(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="123") + signal = Signal( + strategy="rsi_strategy", + symbol="BTCUSDT", + side=OrderSide.BUY, + price=Decimal("50000.00"), + quantity=Decimal("0.01"), + reason="RSI oversold", + ) + + with patch.object(notifier, "send", new_callable=AsyncMock) as mock_send: + await notifier.send_signal(signal) + mock_send.assert_called_once() + msg = mock_send.call_args[0][0] + assert "BUY" in msg + assert "rsi_strategy" in msg + assert "BTCUSDT" in msg + assert "50000.00" in msg + assert "0.01" in msg + assert "RSI oversold" in msg + + @pytest.mark.asyncio + async def test_send_order_formats_message(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="123") + order = Order( + signal_id=str(uuid.uuid4()), + symbol="ETHUSDT", + side=OrderSide.SELL, + type=OrderType.LIMIT, + price=Decimal("3000.50"), + quantity=Decimal("1.5"), + status=OrderStatus.FILLED, + ) + + with patch.object(notifier, "send", new_callable=AsyncMock) as mock_send: + await notifier.send_order(order) + mock_send.assert_called_once() + msg = mock_send.call_args[0][0] + assert "FILLED" in msg + assert "ETHUSDT" in msg + assert "SELL" in msg + assert "3000.50" in msg + assert "1.5" in msg + + @pytest.mark.asyncio + async def test_send_error_formats_message(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="123") + + with patch.object(notifier, "send", new_callable=AsyncMock) as mock_send: + await notifier.send_error("Connection failed", service="executor") + mock_send.assert_called_once() + msg = mock_send.call_args[0][0] + assert "Connection failed" in msg + assert "executor" in msg + + @pytest.mark.asyncio + async def test_send_daily_summary_formats_message(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="123") + positions = [ + Position( + symbol="BTCUSDT", + quantity=Decimal("0.1"), + avg_entry_price=Decimal("50000"), + current_price=Decimal("51000"), + ), + ] + + with patch.object(notifier, "send", new_callable=AsyncMock) as mock_send: + await notifier.send_daily_summary( + positions=positions, + total_value=Decimal("5100.00"), + daily_pnl=Decimal("100.00"), + ) + mock_send.assert_called_once() + msg = mock_send.call_args[0][0] + assert "BTCUSDT" in msg + assert "5100.00" in msg + assert "100.00" in msg + + +class TestTelegramNotifierClose: + """Test close method.""" + + @pytest.mark.asyncio + async def test_close_closes_session(self): + notifier = TelegramNotifier(bot_token="fake-token", chat_id="123") + mock_session = AsyncMock() + notifier._session = mock_session + await notifier.close() + mock_session.close.assert_called_once() |