Added disconnect to TwythonStreamer, more tests, update example

* Stream and Twython core tests
* Import TwythonStreamError from twython

See more in 2.10.1 section of HISTORY.rst

.travis.yml

@@ -26,7 +26,7 @@ env:
   - PROTECTED_TWITTER_2=TwythonSecure2
   - TEST_TWEET_ID=332992304010899457
   - TEST_LIST_ID=574
-script: nosetests -v test_twython:TwythonAPITestCase test_twython:TwythonAuthTestCase --logging-filter="twython" --cover-package="twython" --with-coverage
+script: nosetests -v test_twython:TwythonAPITestCase test_twython:TwythonAuthTestCase test_twython:TwythonStreamTestCase --logging-filter="twython" --cover-package="twython" --with-coverage
 install: pip install -r requirements.txt
 notifications:
   email: false

HISTORY.rst

@@ -9,6 +9,8 @@ History
 - Fix ``search_gen``
 - Fixed ``get_lastfunction_header`` to actually do what its docstring says, returns ``None`` if header is not found
 - Updated some internal API code, ``__init__`` didn't need to have ``self.auth`` and ``self.headers`` because they were never used anywhere else but the ``__init__``
+- Added ``disconnect`` method to ``TwythonStreamer``, allowing users to disconnect as they desire
+- Updated ``TwythonStreamError`` docstring, also allow importing it from ``twython``
 
 2.10.0 (2013-05-21)
 ++++++++++++++++++

examples/stream.py

@@ -4,6 +4,8 @@ from twython import TwythonStreamer
 class MyStreamer(TwythonStreamer):
     def on_success(self, data):
         print data
+        # Want to disconnect after the first result?
+        # self.disconnect()
 
     def on_error(self, status_code, data):
         print status_code, data

test_twython.py

@@ -1,7 +1,11 @@
-import unittest
-import os
+from twython import(
+    Twython, TwythonStreamer, TwythonError,
+    TwythonAuthError, TwythonStreamError
+)
 
-from twython import Twython, TwythonError, TwythonAuthError
+import os
+import time
+import unittest
 
 app_key = os.environ.get('APP_KEY')
 app_secret = os.environ.get('APP_SECRET')
@@ -94,10 +98,17 @@ class TwythonAPITestCase(unittest.TestCase):
 
     def test_search_gen(self):
         '''Test looping through the generator results works, at least once that is'''
-        search = self.api.search_gen('python')
-        for result in search:
-            if result:
-                break
+        search = self.api.search_gen('twitter', count=1)
+        counter = 0
+        while counter < 2:
+            counter += 1
+            result = search.next()
+            new_id_str = int(result['id_str'])
+            if counter == 1:
+                prev_id_str = new_id_str
+                time.sleep(1)  # Give time for another tweet to come into search
+            if counter == 2:
+                self.assertTrue(new_id_str > prev_id_str)
 
     def test_encode(self):
         '''Test encoding UTF-8 works'''
@@ -480,5 +491,47 @@ class TwythonAPITestCase(unittest.TestCase):
         self.api.get_closest_trends(lat='37', long='-122')
 
 
+class TwythonStreamTestCase(unittest.TestCase):
+    def setUp(self):
+        class MyStreamer(TwythonStreamer):
+            def on_success(self, data):
+                self.disconnect()
+
+            def on_error(self, status_code, data):
+                raise TwythonStreamError(data)
+
+            def on_delete(self, data):
+                return
+
+            def on_limit(self, data):
+                return
+
+            def on_disconnect(self, data):
+                return
+
+            def on_timeout(self, data):
+                return
+
+        self.api = MyStreamer(app_key, app_secret,
+                              oauth_token, oauth_token_secret)
+
+    def test_stream_status_filter(self):
+        self.api.statuses.filter(track='twitter')
+
+    def test_stream_status_sample(self):
+        self.api.statuses.sample()
+
+    def test_stream_status_firehose(self):
+        self.assertRaises(TwythonStreamError, self.api.statuses.firehose,
+                          track='twitter')
+
+    def test_stream_site(self):
+        self.assertRaises(TwythonStreamError, self.api.site,
+                          follow='twitter')
+
+    def test_stream_user(self):
+        self.api.user(track='twitter')
+
+
 if __name__ == '__main__':
     unittest.main()

twython/__init__.py

@@ -22,4 +22,7 @@ __version__ = '2.10.1'
 
 from .twython import Twython
 from .streaming import TwythonStreamer
-from .exceptions import TwythonError, TwythonRateLimitError, TwythonAuthError
+from .exceptions import (
+    TwythonError, TwythonRateLimitError, TwythonAuthError,
+    TwythonStreamError
+)

twython/exceptions.py

@@ -2,8 +2,7 @@ from .endpoints import twitter_http_status_codes
 
 
 class TwythonError(Exception):
-    """
-        Generic error class, catch-all for most Twython issues.
+    """Generic error class, catch-all for most Twython issues.
     Special cases are handled by TwythonAuthError & TwythonRateLimitError.
 
     Note: Syntax has changed as of Twython 1.3. To catch these,
@@ -11,8 +10,7 @@ class TwythonError(Exception):
 
     from twython import (
         TwythonError, TwythonRateLimitError, TwythonAuthError
-        )
-    """
+    )"""
     def __init__(self, msg, error_code=None, retry_after=None):
         self.error_code = error_code
 
@@ -31,8 +29,7 @@ class TwythonError(Exception):
 
 class TwythonAuthError(TwythonError):
     """Raised when you try to access a protected resource and it fails due to
-        some issue with your authentication.
-    """
+    some issue with your authentication."""
     pass
 
 
@@ -40,8 +37,7 @@ class TwythonRateLimitError(TwythonError):
     """Raised when you've hit a rate limit.
 
     The amount of seconds to retry your request in will be appended
-        to the message.
-    """
+    to the message."""
     def __init__(self, msg, error_code, retry_after=None):
         if isinstance(retry_after, int):
             msg = '%s (Retry after %d seconds)' % (msg, retry_after)
@@ -49,5 +45,5 @@ class TwythonRateLimitError(TwythonError):
 
 
 class TwythonStreamError(TwythonError):
-    """Test"""
+    """Raised when an invalid response from the Stream API is received"""
     pass

twython/streaming/api.py

@@ -55,14 +55,18 @@ class TwythonStreamer(object):
         self.user = StreamTypes.user
         self.site = StreamTypes.site
 
+        self.connected = False
+
     def _request(self, url, method='GET', params=None):
         """Internal stream request handling"""
+        self.connected = True
         retry_counter = 0
 
         method = method.lower()
         func = getattr(self.client, method)
 
         def _send(retry_counter):
+            while self.connected:
                 try:
                     if method == 'get':
                         response = func(url, params=params, timeout=self.timeout)
@@ -81,9 +85,12 @@ class TwythonStreamer(object):
 
                     return response
 
+        while self.connected:
             response = _send(retry_counter)
 
             for line in response.iter_lines():
+                if not self.connected:
+                    break
                 if line:
                     try:
                         self.on_success(json.loads(line))
@@ -161,3 +168,6 @@ class TwythonStreamer(object):
 
     def on_timeout(self):
         return
+
+    def disconnect(self):
+        self.connected = False

twython/streaming/types.py

@@ -61,7 +61,7 @@ class TwythonStreamerTypesStatuses(object):
         self.streamer._request(url, params=params)
 
     def firehose(self, **params):
-        """Stream statuses/filter
+        """Stream statuses/firehose
 
         Accepted params found at:
         https://dev.twitter.com/docs/api/1.1/get/statuses/firehose

twython/twython.py

@@ -388,11 +388,12 @@ class Twython(object):
             yield tweet
 
         try:
-            kwargs['page'] = 2 if not 'page' in kwargs else (int(kwargs['page']) + 1)
+            if not 'since_id' in kwargs:
+                kwargs['since_id'] = (int(content['statuses'][0]['id_str']) + 1)
         except (TypeError, ValueError):
             raise TwythonError('Unable to generate next page of search results, `page` is not a number.')
 
-        for tweet in self.searchGen(search_query, **kwargs):
+        for tweet in self.search_gen(search_query, **kwargs):
             yield tweet
 
     @staticmethod