Merge pull request openfusion-dev#37 from dmtucker/full-coverage

Get to 100% code coverage

.travis.yml

@@ -4,6 +4,7 @@ python:
   - "3.3"
   - "3.4"
   - "3.5"
+  - "3.6"
 branches:
   only:
   - master

ogre/Twitter.py

@@ -9,16 +9,16 @@
 import logging
 import sys
 import time
-try:
-    from urllib.request import urlopen
-except ImportError:
-    from urllib2 import urlopen  # pylint: disable=import-error
 from datetime import datetime
 from twython import Twython
 from ogre.validation import sanitize
 from ogre.exceptions import OGReError, OGReLimitError
 from snowflake2time.snowflake import snowflake2utc, utc2snowflake
 
+from future.standard_library import hooks
+with hooks():
+    from urllib.request import urlopen  # pylint: disable=import-error
+
 
 def sanitize_twitter(
         keys,
@@ -97,7 +97,7 @@ def sanitize_twitter(
         for clean_medium in clean_media:
             if clean_medium in ("image", "text"):
                 kinds.append(clean_medium)
-    kinds = tuple(kinds)  # pylint: disable=redefined-variable-type
+    kinds = tuple(kinds)
 
     keywords = clean_keyword
     if kinds == ("image",):
@@ -312,6 +312,7 @@ def twitter(
             modifiers["query_limit"] = limit
     except KeyError:
         log.warning(qid+" Unobtainable Rate Limit")
+        raise
     total = remaining
 
     collection = []
@@ -325,7 +326,7 @@ def twitter(
                 since_id=since_id,
                 max_id=max_id
             )
-        except:
+        except Exception:
             log.info(
                 qid+" Failure: " +
                 str(query+1)+" queries produced " +
@@ -401,7 +402,7 @@ def twitter(
             )
             break
         if results.get("search_metadata", {}).get("next_results") is None:
-            outcome = "Success" if len(collection) > 0 else "Failure"
+            outcome = "Success" if collection else "Failure"
             log.info(
                 qid+" "+outcome+": " +
                 str(query+1)+" queries produced " +
@@ -415,7 +416,7 @@ def twitter(
             .split("&")[0]
         )
         if query+1 >= modifiers["query_limit"]:
-            outcome = "Success" if len(collection) > 0 else "Failure"
+            outcome = "Success" if collection else "Failure"
             log.info(
                 qid+" "+outcome+": " +
                 str(query+1)+" queries produced " +

ogre/test/test_api.py

@@ -16,10 +16,7 @@
 import logging
 import os
 import unittest
-try:
-    from io import StringIO
-except ImportError:
-    from StringIO import StringIO  # pylint: disable=import-error
+from io import StringIO
 from mock import MagicMock
 from ogre import OGRe
 from ogre.Twitter import twitter
@@ -219,7 +216,3 @@ def test_fetch(self):
                 network=self.network
             )
         )
-
-
-if __name__ == "__main__":
-    unittest.main()

ogre/test/test_cli.py

@@ -4,11 +4,21 @@
 
 from __future__ import absolute_import
 
+import random
+
 import pytest
 
 import ogre.cli
 
 
+@pytest.fixture
+def source():
+    """Return a valid source."""
+    return random.choice(['twitter'])
+
+# pylint: disable=redefined-outer-name
+
+
 def test___main__():
     """Test python -m functionality."""
     with pytest.raises(SystemExit) as excinfo:
@@ -21,3 +31,45 @@ def test_empty():
     with pytest.raises(SystemExit) as excinfo:
         ogre.cli.main()
     assert excinfo.value != 0
+
+
+def test_no_credentials(source):
+    """Test an invocation without API keys."""
+    with pytest.raises(ValueError) as excinfo:
+        ogre.cli.main(['-s', source])
+    assert excinfo.value != 0
+
+
+def test_invalid_keys(source):
+    """Test an invocation with invalid API keys."""
+    with pytest.raises(ValueError) as excinfo:
+        ogre.cli.main(['-s', source, '--keys', 'invalid'])
+    assert excinfo.value != 0
+
+
+def test_invalid_location(source):
+    """Test an invocation with an invalid location."""
+    with pytest.raises(ValueError) as excinfo:
+        ogre.cli.main(['-s', source, '-l', '0', '0', 'invalid', 'km'])
+    assert excinfo.value != 0
+
+
+def test_invalid_interval(source):
+    """Test an invocation with an invalid interval."""
+    with pytest.raises(ValueError) as excinfo:
+        ogre.cli.main(['-s', source, '-i', '0', 'invalid'])
+    assert excinfo.value != 0
+
+
+def test_invalid_limit(source):
+    """Test an invocation with an invalid limit."""
+    with pytest.raises(ValueError) as excinfo:
+        ogre.cli.main(['-s', source, '--limit', 'invalid'])
+    assert excinfo.value != 0
+
+
+def test_invalid_log(source):
+    """Test an invocation with an invalid log."""
+    with pytest.raises(AttributeError) as excinfo:
+        ogre.cli.main(['-s', source, '--log', 'invalid'])
+    assert excinfo.value != 0

ogre/test/test_exceptions.py

@@ -0,0 +1,12 @@
+# coding: utf-8
+
+"""Tests for ogre.exceptions"""
+
+from __future__ import absolute_import
+
+import ogre.exceptions
+
+
+def test_ogreerror__str__():
+    """Test the __str__ method of OGReError."""
+    assert isinstance(str(ogre.exceptions.OGReError()), str)

ogre/test/test_twitter.py

@@ -17,10 +17,7 @@
 import os
 import unittest
 from datetime import datetime
-try:
-    from StringIO import StringIO
-except ImportError:
-    from io import StringIO
+from io import StringIO
 from mock import MagicMock
 from twython import TwythonError
 from snowflake2time import snowflake
@@ -109,7 +106,29 @@ def setUp(self):
                 },
                 "network": {
                     "return": None,
-                    "effect": lambda _: StringIO("test_image")
+                    "effect": lambda _: StringIO(u"test_image")
+                }
+            },
+            "malformed_limits": {
+                "api": {
+                    "limit": {},
+                    "return": copy.deepcopy(self.tweets),
+                    "effect": None
+                },
+                "network": {
+                    "return": None,
+                    "effect": lambda _: StringIO(u"test_image")
+                }
+            },
+            "low_limits": {
+                "api": {
+                    "limit": twitter_limits(1, 1234567890),
+                    "return": copy.deepcopy(self.tweets),
+                    "effect": None
+                },
+                "network": {
+                    "return": None,
+                    "effect": lambda _: StringIO(u"test_image")
                 }
             },
             "limited": {
@@ -162,7 +181,7 @@ def setUp(self):
                     "effect": None
                 },
                 "network": {
-                    "return": StringIO("test_image"),
+                    "return": StringIO(u"test_image"),
                     "effect": None
                 }
             }
@@ -484,6 +503,43 @@ def test_rate_limit_obedience(self):
         self.assertEqual(0, api().search.call_count)
         self.assertEqual(0, network.call_count)
 
+    def test_rate_limit_failure(self):
+        """Failure to retrieve a rate limit is fatal."""
+        self.log.debug("Testing rate limit failure...")
+        api = self.injectors["api"]["malformed_limits"]
+        network = self.injectors["network"]["malformed_limits"]
+        with self.assertRaises(KeyError):
+            twitter(
+                keys=self.retriever.keychain[
+                    self.retriever.keyring["twitter"]
+                ],
+                keyword="test",
+                api=api,
+                network=network,
+            )
+        self.assertEqual(1, api.call_count)
+        self.assertEqual(1, api().get_application_rate_limit_status.call_count)
+        self.assertEqual(0, api().search.call_count)
+        self.assertEqual(0, network.call_count)
+
+    def test_rate_limit_low(self):
+        """The rate limit should be obeyed during paging."""
+        self.log.debug("Testing rate low limit obedience...")
+        api = self.injectors["api"]["low_limits"]
+        network = self.injectors["network"]["low_limits"]
+        twitter(
+            keys=self.retriever.keychain[
+                self.retriever.keyring["twitter"]
+            ],
+            keyword="test",
+            api=api,
+            network=network,
+        )
+        self.assertEqual(1, api.call_count)
+        self.assertEqual(1, api().get_application_rate_limit_status.call_count)
+        self.assertEqual(1, api().search.call_count)
+        self.assertEqual(1, network.call_count)
+
     def test_hard_failure(self):
         """Failing hard raises exceptions instead of returning empty."""
         self.log.debug("Testing hard failure...")
@@ -864,7 +920,3 @@ def test_return_format(self):
                 }
             ]
         )
-
-
-if __name__ == "__main__":
-    unittest.main()

ogre/test/test_validation.py

@@ -38,6 +38,10 @@ def test_validate(self):
 
         self.log.debug("Testing the OGRe validator...")
 
+        class _StrTrap(Exception):
+            def __str__(self):
+                raise self
+
         with self.assertRaises(AttributeError):
             validate(media=(0,))
         with self.assertRaises(AttributeError):
@@ -46,6 +50,8 @@ def test_validate(self):
             validate(media=("invalid",))
         with self.assertRaises(ValueError):
             validate(media=("text", "invalid"))
+        with self.assertRaises(ValueError):
+            validate(keyword=_StrTrap())
         with self.assertRaises(ValueError):
             validate(quantity=-1)
         with self.assertRaises(ValueError):
@@ -123,7 +129,3 @@ def test_sanitize(self):
             sanitize(interval=(1, 0)),
             (("image", "sound", "text", "video"), "", 15, None, (0, 1))
         )
-
-
-if __name__ == "__main__":
-    unittest.main()

ogre/validation.py

@@ -56,7 +56,7 @@ def validate(
 
     try:
         str(keyword)
-    except:
+    except Exception:
         raise ValueError("Keyword must be a string.")
 
     if int(quantity) < 0:
@@ -142,7 +142,7 @@ def sanitize(
             if medium not in clean_media:
                 clean_media.append(medium)
 
-    clean_media = tuple(clean_media)  # pylint: disable=redefined-variable-type
+    clean_media = tuple(clean_media)
     clean_keyword = str(keyword)
     clean_quantity = int(quantity)
 

setup.cfg

@@ -1,2 +1,2 @@
-[pep8]
+[pycodestyle]
 max-line-length = 100

setup.py

@@ -26,6 +26,7 @@
     include_package_data=True,
     test_suite="ogre.test",
     install_requires=[
+        'future ~= 0.16.0',
         'mock ~= 1.0.1',
         'twython ~= 3.4',
     ],
@@ -39,6 +40,7 @@
         'Programming Language :: Python :: 3.3',
         'Programming Language :: Python :: 3.4',
         'Programming Language :: Python :: 3.5',
-        'Development Status :: 4 - Beta',
+        'Programming Language :: Python :: 3.6',
+        'Development Status :: 5 - Production/Stable',
     ],
 )

tox.ini

@@ -3,15 +3,16 @@ envlist = py27
           py33
           py34
           py35
+          py36
 [testenv]
 deps = coverage
-       pep8
+       pycodestyle
        pylint
        pytest
 setenv = TWITTER_ACCESS_TOKEN=fake
          TWITTER_CONSUMER_KEY=fake
 commands = coverage run --source=ogre -m pytest
            coverage report -m
            ogre --help
-           pep8 ogre setup.py
+           pycodestyle ogre setup.py
            pylint ogre setup.py