aboutsummaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/test_streaming.py195
1 files changed, 195 insertions, 0 deletions
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
new file mode 100644
index 0000000..c79a8e9
--- /dev/null
+++ b/tests/test_streaming.py
@@ -0,0 +1,195 @@
1import six
2import pytest
3import itertools
4from mastodon.streaming import StreamListener, MalformedEventError
5
6
7class Listener(StreamListener):
8 def __init__(self):
9 self.updates = []
10 self.notifications = []
11 self.deletes = []
12 self.heartbeats = 0
13
14 def on_update(self, status):
15 self.updates.append(status)
16
17 def on_notification(self, notification):
18 self.notifications.append(notification)
19
20 def on_delete(self, status_id):
21 self.deletes.append(status_id)
22
23 def handle_heartbeat(self):
24 self.heartbeats += 1
25
26 def handle_stream_(self, lines):
27 '''Test helper to avoid littering all tests with six.b().'''
28 return self.handle_stream(map(six.b, lines))
29
30def test_heartbeat():
31 listener = Listener()
32 listener.handle_stream_([':one', ':two'])
33 assert listener.heartbeats == 2
34
35
36def test_status():
37 listener = Listener()
38 listener.handle_stream_([
39 'event: update',
40 'data: {"foo": "bar"}',
41 '',
42 ])
43 assert listener.updates == [{"foo": "bar"}]
44
45
46def test_notification():
47 listener = Listener()
48 listener.handle_stream_([
49 'event: notification',
50 'data: {"foo": "bar"}',
51 '',
52 ])
53 assert listener.notifications == [{"foo": "bar"}]
54
55
56def test_delete():
57 listener = Listener()
58 listener.handle_stream_([
59 'event: delete',
60 'data: 123',
61 '',
62 ])
63 assert listener.deletes == [123]
64
65
66@pytest.mark.parametrize('events', itertools.permutations([
67 ['event: update', 'data: {"foo": "bar"}', ''],
68 ['event: notification', 'data: {"foo": "bar"}', ''],
69 ['event: delete', 'data: 123', ''],
70 [':toot toot'],
71 [':beep beep'],
72]))
73def test_many(events):
74 listener = Listener()
75 stream = [
76 line
77 for event in events
78 for line in event
79 ]
80 listener.handle_stream_(stream)
81 assert listener.updates == [{"foo": "bar"}]
82 assert listener.notifications == [{"foo": "bar"}]
83 assert listener.deletes == [123]
84 assert listener.heartbeats == 2
85
86
87def test_unknown_event():
88 '''Be tolerant of new event types'''
89 listener = Listener()
90 listener.handle_stream_([
91 'event: blahblah',
92 'data: {}',
93 '',
94 ])
95 assert listener.updates == []
96 assert listener.notifications == []
97 assert listener.deletes == []
98 assert listener.heartbeats == 0
99
100
101def test_missing_event_name():
102 listener = Listener()
103 with pytest.raises(MalformedEventError):
104 listener.handle_stream_([
105 'data: {}',
106 '',
107 ])
108
109 assert listener.updates == []
110 assert listener.notifications == []
111 assert listener.deletes == []
112 assert listener.heartbeats == 0
113
114
115def test_missing_data():
116 listener = Listener()
117 with pytest.raises(MalformedEventError):
118 listener.handle_stream_([
119 'event: update',
120 '',
121 ])
122
123 assert listener.updates == []
124 assert listener.notifications == []
125 assert listener.deletes == []
126 assert listener.heartbeats == 0
127
128
129def test_sse_order_doesnt_matter():
130 listener = Listener()
131 listener.handle_stream_([
132 'data: {"foo": "bar"}',
133 'event: update',
134 '',
135 ])
136 assert listener.updates == [{"foo": "bar"}]
137
138
139def test_extra_keys_ignored():
140 '''
141 https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Event_stream_format
142 defines 'id' and 'retry' keys which the Mastodon streaming API doesn't use,
143 and alleges that "All other field names are ignored".
144 '''
145 listener = Listener()
146 listener.handle_stream_([
147 'event: update',
148 'data: {"foo": "bar"}',
149 'id: 123',
150 'retry: 456',
151 'ignoreme: blah blah blah',
152 '',
153 ])
154 assert listener.updates == [{"foo": "bar"}]
155
156
157def test_valid_utf8():
158 '''Snowman Cat Face With Tears Of Joy'''
159 listener = Listener()
160 listener.handle_stream_([
161 'event: update',
162 'data: {"foo": "\xE2\x98\x83\xF0\x9F\x98\xB9"}',
163 '',
164 ])
165 assert listener.updates == [{"foo": u"\u2603\U0001F639"}]
166
167
168def test_invalid_utf8():
169 '''Cat Face With Tears O'''
170 listener = Listener()
171 with pytest.raises(MalformedEventError):
172 listener.handle_stream_([
173 'event: update',
174 'data: {"foo": "\xF0\x9F\x98"}',
175 '',
176 ])
177
178
179def test_multiline_payload():
180 '''
181 https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Data-only_messages
182 says that newlines in the 'data' field can be encoded by sending the field
183 twice! This would be really pathological for Mastodon because the payload
184 is JSON, but technically literal newlines are permissible (outside strings)
185 so let's handle this case.
186 '''
187 listener = Listener()
188 listener.handle_stream_([
189 'event: update',
190 'data: {"foo":',
191 'data: "bar"',
192 'data: }',
193 '',
194 ])
195 assert listener.updates == [{"foo": "bar"}]
Powered by cgit v1.2.3 (git 2.41.0)