diff --git a/blinkpy/helpers/constants.py b/blinkpy/helpers/constants.py index 574f099..09c1db4 100644 --- a/blinkpy/helpers/constants.py +++ b/blinkpy/helpers/constants.py @@ -60,7 +60,7 @@ ONLINE = {'online': True, 'offline': False} ''' OTHER ''' -TIMESTAMP_FORMAT = '%Y-%m-%dT%H:%M:%S%Z' +TIMESTAMP_FORMAT = '%Y-%m-%dT%H:%M:%S%z' DEFAULT_MOTION_INTERVAL = 1 DEFAULT_REFRESH = 30 diff --git a/blinkpy/helpers/util.py b/blinkpy/helpers/util.py index 66c9f6b..4780a85 100644 --- a/blinkpy/helpers/util.py +++ b/blinkpy/helpers/util.py @@ -2,6 +2,8 @@ import logging import time +from calendar import timegm +from datetime import datetime from functools import partial, wraps from requests import Request, Session, exceptions from blinkpy.helpers.constants import BLINK_URL, TIMESTAMP_FORMAT @@ -11,6 +13,17 @@ import blinkpy.helpers.errors as ERROR _LOGGER = logging.getLogger(__name__) +def time_to_seconds(timestamp): + """Convert TIMESTAMP_FORMAT time to seconds.""" + try: + dtime = datetime.strptime(timestamp, TIMESTAMP_FORMAT) + except ValueError: + _LOGGER.error("Incorrect timestamp format for conversion: %s.", + timestamp) + return False + return timegm(dtime.timetuple()) + + def get_time(time_to_convert=None): """Create blink-compatible timestamp.""" if time_to_convert is None: diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 6c6a6ce..8984ae6 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -5,6 +5,7 @@ import logging from requests.structures import CaseInsensitiveDict from blinkpy import api from blinkpy.camera import BlinkCamera +from blinkpy.helpers.util import time_to_seconds from blinkpy.helpers.constants import ONLINE _LOGGER = logging.getLogger(__name__) @@ -187,9 +188,13 @@ class BlinkSyncModule(): name = entry['device_name'] clip = entry['media'] timestamp = entry['created_at'] - self.motion[name] = True + self.motion[name] = self.check_new_video_time(timestamp) self.last_record[name] = {'clip': clip, 'time': timestamp} except KeyError: _LOGGER.debug("No new videos since last refresh.") return True + + def check_new_video_time(self, timestamp): + """Check if video has timestamp since last refresh.""" + return time_to_seconds(timestamp) > self.blink.last_refresh diff --git a/requirements_test.txt b/requirements_test.txt index 3af0828..c7c4e0b 100644 --- a/requirements_test.txt +++ b/requirements_test.txt @@ -2,7 +2,7 @@ flake8==3.5.0 flake8-docstrings==1.3.0 pylint==2.3.0 pydocstyle==2.1.1 -pytest==3.7.1 +pytest==5.3.5 pytest-cov>=2.3.1 pytest-sugar>=0.9.0 pytest-timeout>=1.0.0 diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index 191e83c..348e351 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -73,23 +73,25 @@ class TestBlinkSyncModule(unittest.TestCase): 'media': [{ 'device_name': 'foo', 'media': '/foo/bar.mp4', - 'created_at': '1970-01-01T00:00:00+0:00' + 'created_at': '1990-01-01T00:00:00+0:00' }] } + sync_module = self.blink.sync['test'] sync_module.cameras = {'foo': None} + sync_module.blink.last_refresh = 0 self.assertEqual(sync_module.motion, {}) self.assertTrue(sync_module.check_new_videos()) self.assertEqual(sync_module.last_record['foo'], {'clip': '/foo/bar.mp4', - 'time': '1970-01-01T00:00:00+0:00'}) + 'time': '1990-01-01T00:00:00+0:00'}) self.assertEqual(sync_module.motion, {'foo': True}) mock_resp.return_value = {'media': []} self.assertTrue(sync_module.check_new_videos()) self.assertEqual(sync_module.motion, {'foo': False}) self.assertEqual(sync_module.last_record['foo'], {'clip': '/foo/bar.mp4', - 'time': '1970-01-01T00:00:00+0:00'}) + 'time': '1990-01-01T00:00:00+0:00'}) def test_check_new_videos_failed(self, mock_resp): """Test method when response is unexpected.""" diff --git a/tests/test_util.py b/tests/test_util.py index 21524bf..3f888b1 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -3,7 +3,7 @@ import unittest from unittest import mock import time -from blinkpy.helpers.util import Throttle, BlinkURLHandler +from blinkpy.helpers.util import Throttle, BlinkURLHandler, time_to_seconds class TestUtil(unittest.TestCase): @@ -105,3 +105,11 @@ class TestUtil(unittest.TestCase): self.assertEqual(urls.subdomain, 'rest-test') urls = BlinkURLHandler('test', legacy=True) self.assertEqual(urls.subdomain, 'rest.test') + + def test_time_to_seconds(self): + """Test time to seconds conversion.""" + correct_time = '1970-01-01T00:00:05+00:00' + wrong_time = '1/1/1970 00:00:03' + self.assertEqual(time_to_seconds(correct_time), 5) + self.assertFalse(time_to_seconds(wrong_time)) +