diff --git a/blinkpy/blinkpy.py b/blinkpy/blinkpy.py index f9aafa7..653074f 100644 --- a/blinkpy/blinkpy.py +++ b/blinkpy/blinkpy.py @@ -76,6 +76,7 @@ class Blink(): self.get_ids() self.sync = BlinkSyncModule(self) + self.sync.start() def login(self): """Prompt user for username and password.""" diff --git a/blinkpy/sync_module.py b/blinkpy/sync_module.py index 93e4c28..f909b8f 100644 --- a/blinkpy/sync_module.py +++ b/blinkpy/sync_module.py @@ -37,8 +37,6 @@ class BlinkSyncModule(): self.cameras = CaseInsensitiveDict({}) self.all_clips = {} - self.start() - @property def attributes(self): """Return sync attributes.""" @@ -98,7 +96,8 @@ class BlinkSyncModule(): self.videos = self.get_videos() for camera_config in camera_info: name = camera_config['name'] - self.cameras[name].update(camera_config, force_cache=True) + if name in self.cameras: + self.cameras[name].update(camera_config, force_cache=True) def get_events(self): """Retrieve events from server.""" @@ -130,8 +129,6 @@ class BlinkSyncModule(): """ videos = list() all_dates = dict() - for camera in self.cameras: - self.all_clips[camera] = {} for page_num in range(start_page, end_page + 1): this_page = api.request_videos(self.blink, page=page_num) @@ -148,7 +145,11 @@ class BlinkSyncModule(): clip_date = clip_addr.split('_')[-6:] clip_date = '_'.join(clip_date) clip_date = clip_date.split('.')[0] - self.all_clips[camera_name][clip_date] = clip_addr + try: + self.all_clips[camera_name][clip_date] = clip_addr + except KeyError: + self.all_clips[camera_name] = {clip_date: clip_addr} + if camera_name not in all_dates: all_dates[camera_name] = list() all_dates[camera_name].append(clip_date) diff --git a/tests/test_blink_functions.py b/tests/test_blink_functions.py index 5f84ab7..dcfdc35 100644 --- a/tests/test_blink_functions.py +++ b/tests/test_blink_functions.py @@ -6,9 +6,7 @@ from requests import Request from blinkpy import blinkpy from blinkpy.sync_module import BlinkSyncModule -from blinkpy.camera import BlinkCamera from blinkpy.helpers.util import create_session -from blinkpy.helpers.constants import BLINK_URL import tests.mock_responses as mresp USERNAME = 'foobar' @@ -52,79 +50,13 @@ class TestBlinkFunctions(unittest.TestCase): 'TOKEN_AUTH': 'foobar123' } self.blink.urls = blinkpy.BlinkURLHandler('test') - self.config = { - 'device_id': 1111, - 'name': 'foobar', - 'armed': False, - 'active': 'disabled', - 'thumbnail': '/test', - 'video': '/test.mp4', - 'temp': 80, - 'battery': 3, - 'notifications': 2, - 'region_id': 'test', - 'device_type': 'camera' - } - self.blink.sync = MockSyncModule( - self.blink, self.blink._auth_header) self.blink.session = create_session() - self.camera = BlinkCamera(self.config, self.blink.sync) def tearDown(self): """Clean up after test.""" self.blink = None - self.config = {} - self.camera = None - def test_image_refresh(self, mock_sess): - """Test image refresh function.""" - self.blink.sync.return_value = {'devices': [self.config]} - image = self.camera.image_refresh() - self.assertEqual(image, - 'https://rest.test.{}/test.jpg'.format(BLINK_URL)) - - @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request') - @mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request') - def test_refresh(self, vid_req, req, mock_sess): - """Test blinkpy refresh function.""" - req.return_value = {'foo': 'bar'} - self.blink.sync.cameras = {'foobar': self.camera} - self.blink.sync.return_value = {'devices': [{'foo': 'bar'}]} - # pylint: disable=protected-access - summary = {'devices': [self.config]} - # pylint: disable=protected-access - events = {'foo': 'bar'} - vid_req.return_value = [ - { - 'camera_name': 'foobar', - 'address': '/new.mp4', - 'thumbnail': '/new', - } - ] - with mock.patch('blinkpy.blinkpy.Blink.summary_request', - return_value=summary): - with mock.patch('blinkpy.blinkpy.Blink.events_request', - return_value=events): - self.blink.refresh_rate = 0 - self.blink.refresh() - test_camera = self.blink.sync.cameras['foobar'] - self.assertEqual(test_camera.clip, - 'https://rest.test.{}/new.mp4'.format(BLINK_URL)) - self.assertEqual(test_camera.thumbnail, - 'https://rest.test.{}/new.jpg'.format(BLINK_URL)) - - def test_set_links(self, mock_sess): - """Test the link set method.""" - self.blink.sync.cameras = {'foobar': self.camera} - self.blink.network_id = 9999 - self.blink.sync.set_links() - net_url = "{}/{}".format(self.blink.urls.network_url, 9999) - self.assertEqual(self.camera.image_link, - "{}/camera/1111/thumbnail".format(net_url)) - self.assertEqual(self.camera.arm_link, - "{}/camera/1111/".format(net_url)) - - @mock.patch('blinkpy.blinkpy.http_req') + @mock.patch('blinkpy.blinkpy.api.http_req') def test_backup_url(self, req, mock_sess): """Test backup login method.""" fake_req = Request('POST', 'http://wrong.url').prepare() diff --git a/tests/test_blink_setup.py b/tests/test_blinkpy.py similarity index 78% rename from tests/test_blink_setup.py rename to tests/test_blinkpy.py index 898ca5f..62c9a6a 100644 --- a/tests/test_blink_setup.py +++ b/tests/test_blinkpy.py @@ -8,7 +8,8 @@ any communication related errors at startup. import unittest from unittest import mock -from blinkpy import blinkpy +from blinkpy import api +from blinkpy.blinkpy import Blink from blinkpy.sync_module import BlinkSyncModule from blinkpy.helpers.util import ( http_req, create_session, BlinkAuthenticationException, @@ -26,10 +27,10 @@ class TestBlinkSetup(unittest.TestCase): def setUp(self): """Set up Blink module.""" - self.blink_no_cred = blinkpy.Blink() - self.blink = blinkpy.Blink(username=USERNAME, - password=PASSWORD) - self.blink.sync = BlinkSyncModule(self.blink, dict(), self.blink.urls) + self.blink_no_cred = Blink() + self.blink = Blink(username=USERNAME, + password=PASSWORD) + self.blink.sync = BlinkSyncModule(self.blink) def tearDown(self): """Clean up after test.""" @@ -59,8 +60,6 @@ class TestBlinkSetup(unittest.TestCase): self.blink.urls = BlinkURLHandler(region_id) with self.assertRaises(BlinkException): self.blink.get_ids() - with self.assertRaises(BlinkException): - self.blink.summary_request() @mock.patch('blinkpy.blinkpy.getpass.getpass') def test_manual_login(self, getpwd, mock_sess): @@ -95,8 +94,18 @@ class TestBlinkSetup(unittest.TestCase): bad_header = {'Host': self.blink._host, 'TOKEN_AUTH': 'BADTOKEN'} # pylint: disable=protected-access self.blink._auth_header = bad_header - # pylint: disable=protected-access - self.assertEqual(self.blink._auth_header, bad_header) - self.blink.summary_request() - # pylint: disable=protected-access - self.assertEqual(self.blink._auth_header, original_header) + self.assertEqual(self.blink.auth_header, bad_header) + api.request_homescreen(self.blink, '1234') + self.assertEqual(self.blink.auth_header, original_header) + + @mock.patch('blinkpy.blinkpy.time.time') + def test_throttle(self, mock_time, mock_sess): + """Check throttling functionality.""" + now = self.blink.refresh_rate + 1 + mock_time.return_value = now + self.assertEqual(self.blink.last_refresh, None) + result = self.blink.check_if_ok_to_update() + self.assertEqual(self.blink.last_refresh, now) + self.assertEqual(result, True) + self.assertEqual(self.blink.check_if_ok_to_update(), False) + self.assertEqual(self.blink.last_refresh, now) diff --git a/tests/test_cameras.py b/tests/test_cameras.py index 923270f..9d6ff90 100644 --- a/tests/test_cameras.py +++ b/tests/test_cameras.py @@ -11,8 +11,7 @@ from unittest import mock from blinkpy import blinkpy from blinkpy.helpers.util import create_session, BlinkURLHandler from blinkpy.sync_module import BlinkSyncModule -from blinkpy.camera import BlinkCamera -from blinkpy.helpers.constants import BLINK_URL +from blinkpy.camera import BlinkCamera, MAX_CLIPS import tests.mock_responses as mresp USERNAME = 'foobar' @@ -39,177 +38,82 @@ class TestBlinkCameraSetup(unittest.TestCase): """Set up Blink module.""" self.blink = blinkpy.Blink(username=USERNAME, password=PASSWORD) - self.camera_config = { - 'device_id': 1111, - 'name': 'foobar', - 'armed': False, - 'active': 'disarmed', - 'thumbnail': '/test/image', - 'video': '/test/clip/clip.mp4', - 'temp': 70, - 'battery': 3, - 'notifications': 2, - 'region_id': 'test' - } - header = { 'Host': 'abc.zxc', 'TOKEN_AUTH': mresp.LOGIN_RESPONSE['authtoken']['authtoken'] } + # pylint: disable=protected-access + self.blink._auth_header = header self.blink.session = create_session() self.blink.urls = BlinkURLHandler('test') - self.blink.network_id = '0000' - self.sync = BlinkSyncModule(self.blink, header, self.blink.urls) + self.blink.sync = BlinkSyncModule(self.blink) + self.camera = BlinkCamera(self.blink.sync) + self.camera.name = 'foobar' + self.blink.sync.cameras['foobar'] = self.camera def tearDown(self): """Clean up after test.""" self.blink = None - @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request', - return_value=CAMERA_CFG) - def test_camera_properties(self, mock_cfg, mock_sess): - """Tests all property set/recall.""" - self.blink.urls = BlinkURLHandler('test') + def test_check_for_motion(self, mock_sess): + """Test check for motion function.""" + self.assertEqual(self.camera.last_record, []) + self.assertEqual(self.camera.motion_detected, None) + self.camera.sync.record_dates = {'foobar': [1, 3, 2, 4]} + self.camera.check_for_motion() + self.assertEqual(self.camera.last_record, [4]) + self.assertEqual(self.camera.motion_detected, False) + self.camera.sync.record_dates = {'foobar': [7, 1, 3, 4]} + self.camera.check_for_motion() + self.assertEqual(self.camera.last_record, [7, 4]) + self.assertEqual(self.camera.motion_detected, True) + self.camera.check_for_motion() + self.assertEqual(self.camera.last_record, [7, 4]) + self.assertEqual(self.camera.motion_detected, False) - self.sync.cameras = { - 'foobar': BlinkCamera(self.camera_config, self.sync) + def test_max_motion_clips(self, mock_sess): + """Test that we only maintain certain number of records.""" + for i in range(0, MAX_CLIPS): + self.camera.last_record.append(i) + self.camera.sync.record_dates['foobar'] = [MAX_CLIPS+2] + self.assertEqual(len(self.camera.last_record), MAX_CLIPS) + self.camera.check_for_motion() + self.assertEqual(self.camera.motion_detected, True) + self.assertEqual(len(self.camera.last_record), MAX_CLIPS) + + def test_camera_update(self, mock_sess): + """Test that we can properly update camera properties.""" + config = { + 'name': 'new', + 'camera_id': 1234, + 'network_id': 5678, + 'serial': '12345678', + 'enabled': False, + 'battery_voltage': 90, + 'battery_state': 'ok', + 'temperature': 68, + 'wifi_strength': 4, + 'thumbnail': '/thumb', } - - for name in self.sync.cameras: - camera = self.sync.cameras[name] - camera.update(self.camera_config, skip_cache=True) - self.assertEqual(camera.id, '1111') - self.assertEqual(camera.name, 'foobar') - self.assertEqual(camera.armed, False) - self.assertEqual( - camera.thumbnail, - "https://rest.test.{}/test/image.jpg".format(BLINK_URL) - ) - self.assertEqual( - camera.clip, - "https://rest.test.{}/test/clip/clip.mp4".format(BLINK_URL) - ) - self.assertEqual(camera.temperature, 68) - self.assertEqual(camera.temperature_c, 20.0) - self.assertEqual(camera.battery, 50) - self.assertEqual(camera.battery_string, "OK") - self.assertEqual(camera.notifications, 2) - self.assertEqual(camera.region_id, 'test') - self.assertEqual(camera.motion_enabled, True) - self.assertEqual(camera.wifi_strength, -30) - - camera_config = self.camera_config - camera_config['active'] = 'armed' - camera_config['thumbnail'] = '/test2/image' - camera_config['video'] = '/test2/clip.mp4' - camera_config['temp'] = 60 - camera_config['battery'] = 0 - camera_config['notifications'] = 4 - for name in self.sync.cameras: - camera = self.sync.cameras[name] - camera.update(camera_config, skip_cache=True) - self.assertEqual(camera.armed, True) - self.assertEqual( - camera.thumbnail, - "https://rest.test.{}/test2/image.jpg".format(BLINK_URL) - ) - self.assertEqual( - camera.clip, - "https://rest.test.{}/test2/clip.mp4".format(BLINK_URL) - ) - self.assertEqual(camera.temperature, 68) - self.assertEqual(camera.battery, 50) - self.assertEqual(camera.battery_string, "Low") - self.assertEqual(camera.notifications, 4) - camera_config['battery'] = -10 - camera.update(camera_config, skip_cache=True) - self.assertEqual(camera.battery_string, "Unknown") - - def test_camera_case(self, mock_sess): - """Tests camera case sensitivity.""" - camera_object = BlinkCamera(self.camera_config, self.sync) - self.sync.cameras['foobar'] = camera_object - self.assertEqual(camera_object, self.sync.cameras['fOoBaR']) - - @mock.patch('blinkpy.sync_module.BlinkSyncModule.camera_config_request', - return_value=CAMERA_CFG) - def test_camera_attributes(self, mock_cfg, mock_sess): - """Tests camera attributes.""" - self.blink.urls = BlinkURLHandler('test') - - self.sync.cameras = { - 'foobar': BlinkCamera(self.camera_config, self.sync) - } - - for name in self.sync.cameras: - camera = self.sync.cameras[name] - camera.update(self.camera_config, skip_cache=True) - camera_attr = camera.attributes - self.assertEqual(camera_attr['device_id'], '1111') - self.assertEqual(camera_attr['name'], 'foobar') - self.assertEqual(camera_attr['armed'], False) - self.assertEqual( - camera_attr['thumbnail'], - "https://rest.test.{}/test/image.jpg".format(BLINK_URL) - ) - self.assertEqual( - camera_attr['video'], - "https://rest.test.{}/test/clip/clip.mp4".format(BLINK_URL) - ) - self.assertEqual(camera_attr['temperature'], 68) - self.assertEqual(camera_attr['temperature_c'], 20.0) - self.assertEqual(camera_attr['battery'], 50) - self.assertEqual(camera_attr['notifications'], 2) - self.assertEqual(camera_attr['network_id'], '0000') - self.assertEqual(camera_attr['motion_enabled'], True) - self.assertEqual(camera_attr['wifi_strength'], -30) - - @mock.patch('blinkpy.camera.BlinkCamera.image_refresh', - return_value='https://fake.url') - def test_camera_cache(self, img_refresh, mock_sess): - """Tests camera cache.""" - update_vals = { - 'name': 'foobar', - 'active': 'disabled', - 'video': '/clip.mp4', - 'thumbnail': '/image', - 'battery': 3, - 'notifications': 1, - } - self.sync.cameras = { - 'foobar': BlinkCamera(self.camera_config, self.sync) - } - - test_image = 'https://rest.test.immedia-semi.com/image.jpg' - test_clip = 'https://rest.test.immedia-semi.com/clip.mp4' - - for name, camera in self.sync.cameras.items(): - # Check that no cache returns None - self.assertEqual(camera.name, name) - self.assertEqual(camera.image_from_cache, None) - self.assertEqual(camera.video_from_cache, None) - - # Now, call an update with a new thumbnail to see if we update - self.sync.records = [] - # pylint: disable=protected-access - camera.update(update_vals) - self.assertEqual(camera.thumbnail, test_image) - self.assertEqual(camera.image_from_cache.status_code, 200) - - # Now update the clip - self.sync.record_dates = {camera.name: ['7', '1', '4', '3']} - self.assertEqual(camera.last_record, list()) - camera.update(update_vals) - self.assertEqual(camera.clip, test_clip) - self.assertEqual(camera.last_record, list('7')) - # First update should be false - self.assertEqual(camera.motion_detected, False) - self.sync.record_dates[camera.name].append('88') - camera.update(update_vals) - self.assertEqual(camera.last_record, ['88', '7']) - self.assertEqual(camera.motion_detected, True) - self.assertEqual(camera.video_from_cache.status_code, 200) - # Next update shouldn't change records, and motion_dected=False - camera.update(update_vals) - self.assertEqual(camera.motion_detected, False) - self.assertEqual(camera.video_from_cache.status_code, 200) + self.camera.last_record = ['1'] + self.camera.sync.all_clips = {'new': {'1': '/test.mp4'}} + mock_sess.side_effect = [ + 'test', + 'foobar' + ] + self.camera.update(config) + self.assertEqual(self.camera.name, 'new') + self.assertEqual(self.camera.camera_id, '1234') + self.assertEqual(self.camera.network_id, '5678') + self.assertEqual(self.camera.serial, '12345678') + self.assertEqual(self.camera.motion_enabled, False) + self.assertEqual(self.camera.battery, 50) + self.assertEqual(self.camera.temperature, 68) + self.assertEqual(self.camera.temperature_c, 20) + self.assertEqual(self.camera.wifi_strength, 4) + self.assertEqual(self.camera.thumbnail, + 'https://rest.test.immedia-semi.com/thumb.jpg') + self.assertEqual(self.camera.clip, + 'https://rest.test.immedia-semi.com/test.mp4') + self.assertEqual(self.camera.image_from_cache, 'test') + self.assertEqual(self.camera.video_from_cache, 'foobar') diff --git a/tests/test_sync_module.py b/tests/test_sync_module.py index a3d2738..bda50fa 100644 --- a/tests/test_sync_module.py +++ b/tests/test_sync_module.py @@ -10,28 +10,7 @@ USERNAME = 'foobar' PASSWORD = 'deadbeef' -class MockSyncModule(BlinkSyncModule): - """Mock http requests from sync module.""" - - def __init__(self, blink, header): - """Create mock sync module instance.""" - super().__init__(blink, header) - self.blink = blink - self.header = header - self.return_value = None - self.return_value2 = None - - def http_get(self, url, stream=False, json=True): - """Mock get request.""" - if stream and self.return_value2 is not None: - return self.return_value2 - return self.return_value - - def http_post(self, url): - """Mock post request.""" - return self.return_value - - +@mock.patch('blinkpy.api.http_req') class TestBlinkSyncModule(unittest.TestCase): """Test BlinkSyncModule functions in blinkpy.""" @@ -45,63 +24,77 @@ class TestBlinkSyncModule(unittest.TestCase): 'TOKEN_AUTH': 'foobar123' } self.blink.urls = blinkpy.BlinkURLHandler('test') - self.config = { - 'device_id': 1111, - 'name': 'foobar', - 'armed': False, - 'active': 'disabled', - 'thumbnail': '/test', - 'video': '/test.mp4', - 'temp': 80, - 'battery': 3, - 'notifications': 2, - 'region_id': 'test', - 'device_type': 'camera' - } - self.blink.sync = MockSyncModule( - self.blink, self.blink._auth_header) - - self.camera = BlinkCamera(self.config, self.blink.sync) + self.blink.sync = BlinkSyncModule(self.blink) + self.camera = BlinkCamera(self.blink.sync) def tearDown(self): """Clean up after test.""" self.blink = None - self.config = {} self.camera = None - def test_get_videos(self): + def test_get_events(self, mock_resp): + """Test get events function.""" + mock_resp.return_value = {'event': True} + self.assertEqual(self.blink.sync.get_events(), True) + + def test_get_camera_info(self, mock_resp): + """Test get camera info function.""" + mock_resp.return_value = {'devicestatus': True} + self.assertEqual(self.blink.sync.get_camera_info(), True) + + def test_get_videos_one_page(self, mock_resp): """Test video access.""" - self.blink.sync.return_value = [ + mock_resp.return_value = [ { 'camera_name': 'foobar', - 'address': '/new/test.mp4', + 'address': '/test/clip_1900_01_01_12_00_00AM.mp4', 'thumbnail': '/test/thumb' } ] - self.blink.sync.get_videos() - self.assertEqual(self.blink.sync.videos['foobar'][0]['clip'], - '/new/test.mp4') - self.assertEqual(self.blink.sync.videos['foobar'][0]['thumb'], - '/test/thumb') + expected_videos = {'foobar': [ + {'clip': '/test/clip_1900_01_01_12_00_00AM.mp4', + 'thumb': '/test/thumb'}]} + expected_records = {'foobar': ['1900_01_01_12_00_00AM']} + expected_clips = {'foobar': { + '1900_01_01_12_00_00AM': '/test/clip_1900_01_01_12_00_00AM.mp4'}} + self.blink.sync.get_videos(start_page=0, end_page=0) + self.assertEqual(self.blink.sync.videos, expected_videos) + self.assertEqual(self.blink.sync.record_dates, expected_records) + self.assertEqual(self.blink.sync.all_clips, expected_clips) - @mock.patch('blinkpy.blinkpy.Blink.refresh') - @mock.patch('blinkpy.sync_module.BlinkSyncModule._summary_request') - @mock.patch('blinkpy.sync_module.BlinkSyncModule._video_request') - def test_get_cameras(self, vid_req, req, refresh): - """Test camera extraction.""" - refresh.return_value = True - req.return_value = {'devices': [self.config]} - vid_req.return_value = [ + def test_get_videos_multi_page(self, mock_resp): + """Test video access with multiple pages.""" + mock_resp.return_value = [ { - 'camera_name': 'foobar', - 'address': '/new.mp4', - 'thumbnail': '/new' + 'camera_name': 'test', + 'address': '/foo/bar_1900_01_01_12_00_00AM.mp4', + 'thumbnail': '/foobar' } ] - self.blink.sync.get_cameras() - self.assertTrue('foobar' in self.blink.sync.cameras) + self.blink.sync.get_videos() + self.assertEqual(mock_resp.call_count, 2) + mock_resp.reset_mock() + self.blink.sync.get_videos(start_page=0, end_page=9) + self.assertEqual(mock_resp.call_count, 10) - def test_video_count(self): - """Test video count function.""" - self.blink.sync.return_value = {'count': 1} - self.assertEqual(self.blink.sync.video_count, 1) + def test_sync_start(self, mock_resp): + """Test sync start function.""" + mock_resp.side_effect = [ + {'syncmodule': { + 'name': 'test', + 'id': 1234, + 'network_id': 5678, + 'serial': '12345678', + 'status': 'foobar'}}, + {'event': True}, + {}, + {'devicestatus': {}}, + None, + None + ] + self.blink.sync.start() + self.assertEqual(self.blink.sync.name, 'test') + self.assertEqual(self.blink.sync.sync_id, 1234) + self.assertEqual(self.blink.sync.network_id, 5678) + self.assertEqual(self.blink.sync.serial, '12345678') + self.assertEqual(self.blink.sync.status, 'foobar')