add stream tests

Author: boogheta

tests/test_sanity.py

@@ -51,6 +51,10 @@
                        auth=noauth,
                        api_version='1.1')
 
+twitter11_stream = TwitterStream(domain='stream.twitter.com',
+                    auth=oauth,
+                    api_version='1.1')
+
 twitter2 = Twitter(domain='api.twitter.com',
                     auth=oauth,
                     api_version='2',
@@ -61,6 +65,11 @@
                     api_version='2',
                     format='')
 
+twitter2_stream = TwitterStream2(domain='api.twitter.com',
+                    auth=oauth2,
+                    api_version='2',
+                    format='')
+
 AZaz = "abcdefghijklmnopqrstuvwxyz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 
 b64_image_data = b"iVBORw0KGgoAAAANSUhEUgAAABAAAAAQCAIAAACQkWg2AAAAAXNSR0IArs4c6QAAAAlwSFlzAAALEwAACxMBAJqcGAAAAAd0SU1FB94JFhMBAJv5kaUAAAAZdEVYdENvbW1lbnQAQ3JlYXRlZCB3aXRoIEdJTVBXgQ4XAAAA4UlEQVQoz7WSIZLGIAxG6c5OFZjianBcIOfgPkju1DsEBWfAUEcNGGpY8Xe7dDoVFRvHfO8NJGRorZE39UVe1nd/WNfVObcsi3OOEAIASikAmOf5D2q/FWPUWgshKKWfiFIqhNBaxxhPjPQ05/z+Bs557xw9hBC89ymlu5BS8t6HEC5NW2sR8alRRLTWXoRSSinlSejT12M9BAAAgCeoTw9BSimlfBIu6WdYtVZEVErdaaUUItZaL/9wOsaY83YAMMb0dGtt6Jdv3/ec87ZtOWdCCGNsmibG2DiOJzP8+7b+AAOmsiPxyHWCAAAAAElFTkSuQmCC"
@@ -185,6 +194,54 @@ def test_search():
     assert results
 
 
+def test_stream():
+    # Sometimes Twitter matches tweets with keywords from urls or users
+    # which are more complex to check so we allow a few mismatches
+    mismatch = 0
+    for tweet in twitter11_stream.statuses.filter(track="gaga,bieber", filter_level='none', stall_warnings='true'):
+        if "timeout" in tweet:
+            continue
+        assert "text" in tweet
+        lowtext = tweet["text"].lower()
+        if "gaga" in lowtext or "bieber" in lowtext:
+            break
+        mismatch += 1
+        if mismatch > 5:
+            assert "gaga" in lowtext or "bieber" in lowtext
+
+
+def test_stream_v2():
+    rules = twitter2_app.tweets.search.stream.rules()
+    assert "data" in rules and "meta" in rules and len(rules["data"]) == rules["meta"].get("result_count", 0)
+
+    if len(rules["data"]):
+        remove = twitter2_app.tweets.search.stream.rules(
+            _json={"delete": {"ids": [rule["id"] for rule in rules["data"]]}}
+        )
+        assert "meta" in remove and "summary" in remove["meta"] and remove["meta"]["summary"].get("deleted", 0) == len(rules["data"])
+
+    add = twitter2_app.tweets.search.stream.rules(
+        _json={"add": [{"value": '"gaga"'}, {"value": '"bieber"'}]}
+    )
+    assert "meta" in add and "summary" in add["meta"] and add["meta"]["summary"].get("created", 0) == 2
+    assert "data" in add and len(add["data"]) == 2
+
+    # Sometimes Twitter matches tweets with keywords from urls or users
+    # which are more complex to check so we allow a few mismatches
+    mismatch = 0
+    for tweet in twitter2_stream.tweets.search.stream(
+        expansions="referenced_tweets.id",
+        params={"tweet.fields": "referenced_tweets,text"}
+    ):
+        assert "data" in tweet and "text" in tweet["data"]
+        lowtext = tweet["data"]["text"].lower()
+        if "gaga" in lowtext or "bieber" in lowtext:
+            break
+        mismatch += 1
+        if mismatch > 5:
+            assert "gaga" in lowtext or "bieber" in lowtext
+
+
 def test_get_trends():
     # This is one method of inserting parameters, using named
     # underscore params.