1 files changed, 141 insertions, 4 deletions

diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index ac8e691..33d1381 100644
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -1,11 +1,51 @@
 import six
 import pytest
 import itertools
-from mastodon.streaming import StreamListener
+from mastodon.streaming import StreamListener, CallbackStreamListener
 from mastodon.Mastodon import MastodonMalformedEventError
-
-
-
+from mastodon import Mastodon
+
+import threading
+import time
+
+# For monkeypatching so we can make vcrpy better
+import vcr.stubs
+
+streamingIsPatched = False
+realConnections = []
+
+def patchStreaming():
+    global streamingIsPatched
+    if streamingIsPatched == True:
+        return
+    streamingIsPatched = True
+    
+    realGetResponse = vcr.stubs.VCRConnection.getresponse
+    def fakeGetResponse(*args, **kwargs):
+        if args[0]._vcr_request.path.startswith("/api/v1/streaming/"):
+            realConnections.append(args[0].real_connection)
+            realConnectionRealGetresponse = args[0].real_connection.getresponse
+            def fakeRealConnectionGetresponse(*args, **kwargs):
+                response = realConnectionRealGetresponse(*args, **kwargs)
+                real_body = b""
+                try:
+                    while True:
+                        chunk = response.read(1)
+                        real_body += chunk
+                except AttributeError: 
+                    pass # Connection closed
+                response.read = (lambda: real_body)
+                return response
+            args[0].real_connection.getresponse = fakeRealConnectionGetresponse
+        return realGetResponse(*args, **kwargs)
+    vcr.stubs.VCRConnection.getresponse = fakeGetResponse
+
+def streamingClose():
+    global realConnections
+    for connection in realConnections:
+        connection.close()
+    realConnections = []
+    
 class Listener(StreamListener):
     def __init__(self):
         self.updates = []
@@ -114,6 +154,25 @@ def test_unknown_event():
     assert listener.deletes == []
     assert listener.heartbeats == 0
 
+def test_invalid_event():
+    """But not too tolerant"""
+    listener = Listener()
+    with pytest.raises(MastodonMalformedEventError):
+        listener.handle_stream_([
+            'event: whatup',
+            'data: {}',
+            '',
+        ])
+
+def test_invalid_json():
+    """But not too tolerant"""
+    listener = Listener()
+    with pytest.raises(MastodonMalformedEventError):
+        listener.handle_stream_([
+            'event: blahblah',
+            'data: {kjaslkdjalskdjasd asdkjhak ajdasldasd}',
+            '',
+        ])
 
 def test_missing_event_name():
     listener = Listener()
@@ -210,3 +269,81 @@ def test_multiline_payload():
         '',
     ])
     assert listener.updates == [{"foo": "bar"}]
+
+@pytest.mark.vcr(match_on=['path'])
+def test_stream_user(api, api2):
+    patchStreaming()
+    
+    updates = []
+    notifications = []
+    deletes = []
+    listener = CallbackStreamListener(
+        update_handler = lambda x: updates.append(x),
+        notification_handler = lambda x: notifications.append(x),
+        delete_handler = lambda x: deletes.append(x)
+    )
+    
+    posted = []
+    def do_activities():
+        time.sleep(5)
+        posted.append(api.status_post("only real cars respond."))
+        posted.append(api2.status_post("@mastodonpy_test beep beep I'm a jeep"))
+        posted.append(api2.status_post("on the internet, nobody knows you're a plane"))
+        time.sleep(1)
+        api.status_delete(posted[0])
+        time.sleep(2)
+        streamingClose()
+        
+    t = threading.Thread(args=(), target=do_activities)
+    t.start()
+    
+    stream = None
+    try:
+        stream = api.stream_user(listener, run_async=True)
+        time.sleep(13)
+    finally:
+        if stream != None:
+            stream.close()
+        
+    assert len(updates) == 1
+    assert len(notifications) == 1
+    assert len(deletes) == 1
+    
+    assert updates[0].id == posted[0].id
+    assert deletes[0] == posted[0].id
+    assert notifications[0].status.id == posted[1].id
+    
+    t.join()
+    
+@pytest.mark.vcr(match_on=['path'])
+def test_stream_user_local(api, api2):
+    patchStreaming()
+    
+    updates = []
+    notifications = []
+    listener = CallbackStreamListener(
+        local_update_handler = lambda x: updates.append(x),
+    )
+    
+    posted = []
+    def do_activities():
+        time.sleep(5)
+        posted.append(api.status_post("it's cool guy"))
+        time.sleep(3)
+        streamingClose()
+        
+    t = threading.Thread(args=(), target=do_activities)
+    t.start()
+    
+    stream = None
+    try:
+        stream = api.stream_user(listener, run_async=True)
+        time.sleep(13)
+    finally:
+        if stream != None:
+            stream.close()
+        
+    assert len(updates) == 1
+    assert updates[0].id == posted[0].id
+    
+    t.join()
\ No newline at end of file