diff --git a/blinkpy/helpers/util.py b/blinkpy/helpers/util.py index 0023dd3..aabcbdb 100644 --- a/blinkpy/helpers/util.py +++ b/blinkpy/helpers/util.py @@ -2,6 +2,7 @@ import logging import time +from functools import wraps from requests import Request, Session, exceptions from blinkpy.helpers.constants import BLINK_URL, TIMESTAMP_FORMAT import blinkpy.helpers.errors as ERROR @@ -121,3 +122,33 @@ class BlinkURLHandler(): self.networks_url = "{}/networks".format(self.base_url) self.video_url = "{}/api/v2/videos".format(self.base_url) _LOGGER.debug("Setting base url to %s.", self.base_url) + + +class Throttle(): + """Class for throttling api calls.""" + + def __init__(self, seconds=10): + """Initialize throttle class.""" + self.throttle_time = seconds + self.last_call = 0 + + def __call__(self, method): + """Throttle caller method.""" + def throttle_method(): + """Call when method is throttled.""" + return None + + @wraps(method) + def wrapper(*args, **kwargs): + """Wrap that checks for throttling.""" + force = kwargs.pop('force', False) + now = int(time.time()) + last_call_delta = now - self.last_call + if force or last_call_delta > self.throttle_time: + result = method(*args, *kwargs) + self.last_call = now + return result + + return throttle_method() + + return wrapper diff --git a/pylintrc b/pylintrc index 95ef270..85dfdd6 100644 --- a/pylintrc +++ b/pylintrc @@ -8,6 +8,8 @@ reports=no # too-many-* - are not enforced for the sake of readability # too-few-* - same as too-many-* # no-else-return - I don't see any reason to enforce this. both forms are readable +# no-self-use - stupid and only annoying +# unexpected-keyword-arg - doesn't allow for use of **kwargs, which is dumb disable= locally-disabled, @@ -23,3 +25,5 @@ disable= too-many-lines, too-few-public-methods, no-else-return, + no-self-use, + unexpected-keyword-arg, diff --git a/tests/test_util.py b/tests/test_util.py new file mode 100644 index 0000000..f75cf76 --- /dev/null +++ b/tests/test_util.py @@ -0,0 +1,100 @@ +"""Test various api functions.""" + +import unittest +from unittest import mock +import time +from blinkpy.helpers.util import Throttle + + +class TestUtil(unittest.TestCase): + """Test the helpers/util module.""" + + def setUp(self): + """Initialize the blink module.""" + + def tearDown(self): + """Tear down blink module.""" + + def test_throttle(self): + """Test the throttle decorator.""" + calls = [] + + @Throttle(seconds=5) + def test_throttle(): + calls.append(1) + + now = int(time.time()) + now_plus_four = now + 4 + now_plus_six = now + 6 + + test_throttle() + self.assertEqual(1, len(calls)) + + # Call again, still shouldn't fire + test_throttle() + self.assertEqual(1, len(calls)) + + # Call with force + test_throttle(force=True) + self.assertEqual(2, len(calls)) + + # Call without throttle, shouldn't fire + test_throttle() + self.assertEqual(2, len(calls)) + + # Fake time as 4 seconds from now + with mock.patch('time.time', return_value=now_plus_four): + test_throttle() + self.assertEqual(2, len(calls)) + + # Fake time as 6 seconds from now + with mock.patch('time.time', return_value=now_plus_six): + test_throttle() + self.assertEqual(3, len(calls)) + + def test_throttle_per_instance(self): + """Test that throttle is done once per instance of class.""" + class Tester: + """A tester class for throttling.""" + + def test(self): + """Test the throttle.""" + return True + + tester = Tester() + throttled = Throttle(seconds=1)(tester.test) + self.assertEqual(throttled(), True) + self.assertEqual(throttled(), None) + + def test_throttle_on_two_methods(self): + """Test that throttle works for multiple methods.""" + class Tester: + """A tester class for throttling.""" + + @Throttle(seconds=3) + def test1(self): + """Test function for throttle.""" + return True + + @Throttle(seconds=5) + def test2(self): + """Test function for throttle.""" + return True + + tester = Tester() + now = time.time() + now_plus_4 = now + 4 + now_plus_6 = now + 6 + + self.assertEqual(tester.test1(), True) + self.assertEqual(tester.test2(), True) + self.assertEqual(tester.test1(), None) + self.assertEqual(tester.test2(), None) + + with mock.patch('time.time', return_value=now_plus_4): + self.assertEqual(tester.test1(), True) + self.assertEqual(tester.test2(), None) + + with mock.patch('time.time', return_value=now_plus_6): + self.assertEqual(tester.test1(), None) + self.assertEqual(tester.test2(), True)