1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
#!/usr/bin/env run.sh
"""Telegram notification module for Kidcam."""
# : out kidcam-notifier
# : dep python-telegram-bot
# : dep pytest
# : dep pytest-asyncio
import asyncio
import datetime as dt
import logging
import Omni.Test as Test
import os
import pytest
import typing
import unittest.mock as mock
class TelegramNotifier:
def __init__(self, bot_token: str, chat_id: str) -> None:
self.bot_token = bot_token
self.chat_id = chat_id
self.logger = logging.getLogger(__name__)
async def send_notification(
self, stream_url: str, message: str = "Kids are playing!"
) -> bool:
try:
import telegram as tg # type: ignore[import-not-found,unused-ignore]
bot = tg.Bot(token=self.bot_token)
timestamp = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
full_message = (
f"{message}\n\n🎥 Watch: {stream_url}\n📅 {timestamp}"
)
await bot.send_message(chat_id=self.chat_id, text=full_message)
self.logger.info("Notification sent: %s", message)
return True
except Exception as e:
self.logger.exception("Failed to send notification: %s", e)
return False
async def send_stream_ended(self) -> bool:
timestamp = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
message = f"Stream ended\n📅 {timestamp}"
try:
import telegram as tg # type: ignore[import-not-found,unused-ignore]
bot = tg.Bot(token=self.bot_token)
await bot.send_message(chat_id=self.chat_id, text=message)
self.logger.info("Stream ended notification sent")
return True
except Exception as e:
self.logger.exception("Failed to send stream ended notification: %s", e)
return False
def from_env() -> TelegramNotifier:
bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
chat_id = os.getenv("TELEGRAM_CHAT_ID")
if not bot_token:
msg = "TELEGRAM_BOT_TOKEN environment variable not set"
raise ValueError(msg)
if not chat_id:
msg = "TELEGRAM_CHAT_ID environment variable not set"
raise ValueError(msg)
return TelegramNotifier(bot_token=bot_token, chat_id=chat_id)
async def main() -> None:
notifier = from_env()
stream_url = "https://example.com/stream/123"
await notifier.send_notification(stream_url)
class TestTelegramNotifier(Test.TestCase):
@mock.patch.dict(
os.environ,
{"TELEGRAM_BOT_TOKEN": "test_token", "TELEGRAM_CHAT_ID": "test_chat"},
)
def test_from_env(self) -> None:
notifier = from_env()
self.assertEqual(notifier.bot_token, "test_token")
self.assertEqual(notifier.chat_id, "test_chat")
def test_from_env_missing_token(self) -> None:
with pytest.raises(ValueError):
from_env()
@mock.patch("telegram.Bot")
async def test_send_notification(self, mock_bot_class: typing.Any) -> None:
mock_bot = mock.AsyncMock()
mock_bot_class.return_value = mock_bot
notifier = TelegramNotifier(bot_token="test_token", chat_id="test_chat")
result = await notifier.send_notification("https://example.com/stream")
self.assertTrue(result)
mock_bot.send_message.assert_called_once()
call_args = mock_bot.send_message.call_args
self.assertEqual(call_args.kwargs["chat_id"], "test_chat")
self.assertIn("Kids are playing!", call_args.kwargs["text"])
self.assertIn("https://example.com/stream", call_args.kwargs["text"])
@mock.patch("telegram.Bot")
async def test_send_stream_ended(self, mock_bot_class: typing.Any) -> None:
mock_bot = mock.AsyncMock()
mock_bot_class.return_value = mock_bot
notifier = TelegramNotifier(bot_token="test_token", chat_id="test_chat")
result = await notifier.send_stream_ended()
self.assertTrue(result)
mock_bot.send_message.assert_called_once()
call_args = mock_bot.send_message.call_args
self.assertIn("Stream ended", call_args.kwargs["text"])
@mock.patch("telegram.Bot")
async def test_send_notification_error(
self, mock_bot_class: typing.Any
) -> None:
mock_bot = mock.AsyncMock()
mock_bot.send_message.side_effect = Exception("Network error")
mock_bot_class.return_value = mock_bot
notifier = TelegramNotifier(bot_token="test_token", chat_id="test_chat")
result = await notifier.send_notification("https://example.com/stream")
self.assertFalse(result)
def test() -> None:
import Omni.App as App
Test.run(App.Area.Test, [TestTelegramNotifier])
if __name__ == "__main__":
asyncio.run(main())
|
