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
|
import unittest
from unittest.mock import Mock, patch
import main
class SlopbotTests(unittest.TestCase):
def setUp(self):
self.old_client = main.client
self.old_executor = main.executor
def tearDown(self):
main.client = self.old_client
main.executor = self.old_executor
def stream_msg(self, content, **overrides):
msg = {
"id": 123,
"type": "stream",
"sender_email": "alice@example.com",
"sender_full_name": "Alice",
"display_recipient": "test stream",
"stream_id": 23,
"subject": "slop",
"content": content,
}
msg.update(overrides)
return msg
def test_ticker_regex_deduplicates_in_handle_message(self):
sent = []
main.client = Mock(email="slop-bot@example.com")
main.client.send_message.side_effect = sent.append
with patch.object(main, "fetch_price", side_effect=lambda ticker: f"${ticker}: $1.00") as fetch:
main.handle_message(self.stream_msg("$NVDA $AAPL $NVDA"))
self.assertEqual([call.args[0] for call in fetch.call_args_list], ["NVDA", "AAPL"])
self.assertEqual(len(sent), 1)
self.assertEqual(sent[0]["type"], "stream")
self.assertEqual(sent[0]["to"], 23)
self.assertEqual(sent[0]["topic"], "slop")
self.assertEqual(sent[0]["content"], "$NVDA: $1.00\n$AAPL: $1.00")
def test_mention_calls_agent_and_appends_prices(self):
sent = []
main.client = Mock(email="slop-bot@example.com")
main.client.send_message.side_effect = sent.append
with patch.object(main, "ask_agent", return_value="agent reply") as ask_agent, \
patch.object(main, "fetch_price", return_value="$NVDA: $1.00") as fetch_price:
main.handle_message(self.stream_msg("@**slop-bot** summarize $NVDA"))
ask_agent.assert_called_once()
self.assertEqual(ask_agent.call_args.args[1], "summarize $NVDA")
fetch_price.assert_called_once_with("NVDA")
self.assertEqual(sent[0]["content"], "agent reply\n\n$NVDA: $1.00")
def test_ignores_self_and_irrelevant_messages(self):
main.client = Mock(email="slop-bot@example.com")
main.handle_message(self.stream_msg("$NVDA", sender_email="slop-bot@example.com"))
main.handle_message(self.stream_msg("nothing to do here"))
main.client.send_message.assert_not_called()
def test_private_reply_goes_to_sender_email(self):
sent = []
main.client = Mock(email="slop-bot@example.com")
main.client.send_message.side_effect = sent.append
msg = self.stream_msg("$NVDA", type="private", sender_email="alice@example.com")
with patch.object(main, "fetch_price", return_value="$NVDA: $1.00"):
main.handle_message(msg)
self.assertEqual(sent[0]["type"], "private")
self.assertEqual(sent[0]["to"], "alice@example.com")
def test_handle_event_submits_message_work(self):
submitted = []
main.executor = Mock()
main.executor.submit.side_effect = lambda fn, msg: submitted.append((fn, msg))
msg = self.stream_msg("$NVDA")
main.handle_event({"type": "message", "message": msg})
main.handle_event({"type": "heartbeat"})
self.assertEqual(submitted, [(main.handle_message, msg)])
def test_agent_debug_streams_stderr_to_journal(self):
with patch.object(main.subprocess, "run") as run:
run.return_value = Mock(returncode=0, stdout="ok\n", stderr=None)
with patch.object(main, "AGENT_DEBUG", True):
rc, out, err = main._agent("hello")
self.assertEqual((rc, out, err), (0, "ok", ""))
kwargs = run.call_args.kwargs
command = run.call_args.args[0]
self.assertIn("--debug", command)
self.assertIsNone(kwargs["stderr"])
if __name__ == "__main__":
unittest.main()
|
