Add black formatting style
This commit is contained in:
Kevin Fronczak
+16
-18
@@ -4,11 +4,9 @@ from blinkpy.helpers.util import BlinkURLHandler
|
||||
import blinkpy.helpers.constants as const
|
||||
|
||||
LOGIN_RESPONSE = {
|
||||
'region': {'mock': 'Test'},
|
||||
'networks': {
|
||||
'1234': {'name': 'test', 'onboarded': True}
|
||||
},
|
||||
'authtoken': {'authtoken': 'foobar123', 'message': 'auth'}
|
||||
"region": {"mock": "Test"},
|
||||
"networks": {"1234": {"name": "test", "onboarded": True}},
|
||||
"authtoken": {"authtoken": "foobar123", "message": "auth"},
|
||||
}
|
||||
|
||||
|
||||
@@ -37,29 +35,29 @@ def mocked_session_send(*args, **kwargs):
|
||||
url = prepped.url
|
||||
header = prepped.headers
|
||||
method = prepped.method
|
||||
if method == 'GET':
|
||||
expected_token = LOGIN_RESPONSE['authtoken']['authtoken']
|
||||
if header['TOKEN_AUTH'] != expected_token:
|
||||
response = {'message': 'Not Authorized', 'code': 400}
|
||||
if method == "GET":
|
||||
expected_token = LOGIN_RESPONSE["authtoken"]["authtoken"]
|
||||
if header["TOKEN_AUTH"] != expected_token:
|
||||
response = {"message": "Not Authorized", "code": 400}
|
||||
status = 400
|
||||
elif url == 'use_bad_response':
|
||||
response = {'foo': 'bar'}
|
||||
elif url == "use_bad_response":
|
||||
response = {"foo": "bar"}
|
||||
status = 200
|
||||
elif url == 'reauth':
|
||||
response = {'message': 'REAUTH', 'code': 777}
|
||||
elif url == "reauth":
|
||||
response = {"message": "REAUTH", "code": 777}
|
||||
status = 777
|
||||
else:
|
||||
response = {'test': 'foo'}
|
||||
response = {"test": "foo"}
|
||||
status = 200
|
||||
elif method == 'POST':
|
||||
elif method == "POST":
|
||||
if url in (const.LOGIN_URL, const.LOGIN_BACKUP_URL):
|
||||
response = LOGIN_RESPONSE
|
||||
status = 200
|
||||
elif url == 'http://wrong.url/' or url is None:
|
||||
response = {'message': 'Error', 'code': 404}
|
||||
elif url == "http://wrong.url/" or url is None:
|
||||
response = {"message": "Error", "code": 404}
|
||||
status = 404
|
||||
else:
|
||||
response = {'message': 'foo', 'code': 200}
|
||||
response = {"message": "foo", "code": 200}
|
||||
status = 200
|
||||
|
||||
return MockResponse(response, status)
|
||||
|
||||
+16
-12
@@ -21,22 +21,26 @@ class TestBlinkAPI(unittest.TestCase):
|
||||
"""Tear down blink module."""
|
||||
self.blink = None
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.Blink.get_auth_token')
|
||||
@mock.patch("blinkpy.blinkpy.Blink.get_auth_token")
|
||||
def test_http_req_connect_error(self, mock_auth):
|
||||
"""Test http_get error condition."""
|
||||
mock_auth.return_value = {'foo': 'bar'}
|
||||
firstlog = ("INFO:blinkpy.helpers.util:"
|
||||
"Cannot connect to server with url {}.").format(
|
||||
'http://notreal.fake')
|
||||
nextlog = ("INFO:blinkpy.helpers.util:"
|
||||
"Auth token expired, attempting reauthorization.")
|
||||
lastlog = ("ERROR:blinkpy.helpers.util:"
|
||||
"Endpoint {} failed. Possible issue with "
|
||||
"Blink servers.").format('http://notreal.fake')
|
||||
mock_auth.return_value = {"foo": "bar"}
|
||||
firstlog = (
|
||||
"INFO:blinkpy.helpers.util:" "Cannot connect to server with url {}."
|
||||
).format("http://notreal.fake")
|
||||
nextlog = (
|
||||
"INFO:blinkpy.helpers.util:"
|
||||
"Auth token expired, attempting reauthorization."
|
||||
)
|
||||
lastlog = (
|
||||
"ERROR:blinkpy.helpers.util:"
|
||||
"Endpoint {} failed. Possible issue with "
|
||||
"Blink servers."
|
||||
).format("http://notreal.fake")
|
||||
expected = [firstlog, nextlog, firstlog, lastlog]
|
||||
with self.assertLogs() as getlog:
|
||||
api.http_get(self.blink, 'http://notreal.fake')
|
||||
api.http_get(self.blink, "http://notreal.fake")
|
||||
with self.assertLogs() as postlog:
|
||||
api.http_post(self.blink, 'http://notreal.fake')
|
||||
api.http_post(self.blink, "http://notreal.fake")
|
||||
self.assertEqual(getlog.output, expected)
|
||||
self.assertEqual(postlog.output, expected)
|
||||
|
||||
@@ -8,8 +8,8 @@ from blinkpy.sync_module import BlinkSyncModule
|
||||
from blinkpy.helpers.util import create_session, get_time
|
||||
import tests.mock_responses as mresp
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
USERNAME = "foobar"
|
||||
PASSWORD = "deadbeef"
|
||||
|
||||
|
||||
class MockSyncModule(BlinkSyncModule):
|
||||
@@ -34,69 +34,56 @@ class MockSyncModule(BlinkSyncModule):
|
||||
return self.return_value
|
||||
|
||||
|
||||
@mock.patch('blinkpy.helpers.util.Session.send',
|
||||
side_effect=mresp.mocked_session_send)
|
||||
@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send)
|
||||
class TestBlinkFunctions(unittest.TestCase):
|
||||
"""Test Blink and BlinkCamera functions in blinkpy."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up Blink module."""
|
||||
self.blink = blinkpy.Blink(username=USERNAME,
|
||||
password=PASSWORD)
|
||||
self.blink = blinkpy.Blink(username=USERNAME, password=PASSWORD)
|
||||
# pylint: disable=protected-access
|
||||
self.blink._auth_header = {
|
||||
'Host': 'test.url.tld',
|
||||
'TOKEN_AUTH': 'foobar123'
|
||||
}
|
||||
self.blink.urls = blinkpy.BlinkURLHandler('test')
|
||||
self.blink._auth_header = {"Host": "test.url.tld", "TOKEN_AUTH": "foobar123"}
|
||||
self.blink.urls = blinkpy.BlinkURLHandler("test")
|
||||
self.blink.session = create_session()
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
self.blink = None
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.api.request_login')
|
||||
@mock.patch("blinkpy.blinkpy.api.request_login")
|
||||
def test_backup_url(self, req, mock_sess):
|
||||
"""Test backup login method."""
|
||||
json_resp = {
|
||||
'authtoken': {'authtoken': 'foobar123'},
|
||||
'networks': {'1234': {'name': 'foobar', 'onboarded': True}}
|
||||
"authtoken": {"authtoken": "foobar123"},
|
||||
"networks": {"1234": {"name": "foobar", "onboarded": True}},
|
||||
}
|
||||
bad_req = mresp.MockResponse({}, 404)
|
||||
new_req = mresp.MockResponse(json_resp, 200)
|
||||
req.side_effect = [
|
||||
bad_req,
|
||||
bad_req,
|
||||
new_req
|
||||
]
|
||||
self.blink.login_urls = ['test1', 'test2', 'test3']
|
||||
req.side_effect = [bad_req, bad_req, new_req]
|
||||
self.blink.login_urls = ["test1", "test2", "test3"]
|
||||
self.blink.login_request()
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._login_url, 'test3')
|
||||
self.assertEqual(self.blink._login_url, "test3")
|
||||
|
||||
req.side_effect = [
|
||||
bad_req,
|
||||
new_req,
|
||||
bad_req
|
||||
]
|
||||
self.blink.login_urls = ['test1', 'test2', 'test3']
|
||||
req.side_effect = [bad_req, new_req, bad_req]
|
||||
self.blink.login_urls = ["test1", "test2", "test3"]
|
||||
self.blink.login_request()
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink._login_url, 'test2')
|
||||
self.assertEqual(self.blink._login_url, "test2")
|
||||
|
||||
def test_merge_cameras(self, mock_sess):
|
||||
"""Test merge camera functionality."""
|
||||
first_dict = {'foo': 'bar', 'test': 123}
|
||||
next_dict = {'foobar': 456, 'bar': 'foo'}
|
||||
self.blink.sync['foo'] = BlinkSyncModule(self.blink, 'foo', 1, [])
|
||||
self.blink.sync['bar'] = BlinkSyncModule(self.blink, 'bar', 2, [])
|
||||
self.blink.sync['foo'].cameras = first_dict
|
||||
self.blink.sync['bar'].cameras = next_dict
|
||||
first_dict = {"foo": "bar", "test": 123}
|
||||
next_dict = {"foobar": 456, "bar": "foo"}
|
||||
self.blink.sync["foo"] = BlinkSyncModule(self.blink, "foo", 1, [])
|
||||
self.blink.sync["bar"] = BlinkSyncModule(self.blink, "bar", 2, [])
|
||||
self.blink.sync["foo"].cameras = first_dict
|
||||
self.blink.sync["bar"].cameras = next_dict
|
||||
result = self.blink.merge_cameras()
|
||||
expected = {'foo': 'bar', 'test': 123, 'foobar': 456, 'bar': 'foo'}
|
||||
expected = {"foo": "bar", "test": 123, "foobar": 456, "bar": "foo"}
|
||||
self.assertEqual(expected, result)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.api.request_videos')
|
||||
@mock.patch("blinkpy.blinkpy.api.request_videos")
|
||||
def test_download_video_exit(self, mock_req, mock_sess):
|
||||
"""Test we exit method when provided bad response."""
|
||||
blink = blinkpy.Blink()
|
||||
@@ -106,63 +93,60 @@ class TestBlinkFunctions(unittest.TestCase):
|
||||
mock_req.return_value = {}
|
||||
formatted_date = get_time(blink.last_refresh)
|
||||
expected_log = [
|
||||
"INFO:blinkpy.blinkpy:Retrieving videos since {}".format(
|
||||
formatted_date),
|
||||
"INFO:blinkpy.blinkpy:Retrieving videos since {}".format(formatted_date),
|
||||
"DEBUG:blinkpy.blinkpy:Processing page 1",
|
||||
"INFO:blinkpy.blinkpy:No videos found on page 1. Exiting."
|
||||
"INFO:blinkpy.blinkpy:No videos found on page 1. Exiting.",
|
||||
]
|
||||
with self.assertLogs() as dl_log:
|
||||
blink.download_videos('/tmp')
|
||||
blink.download_videos("/tmp")
|
||||
self.assertEqual(dl_log.output, expected_log)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.api.request_videos')
|
||||
@mock.patch("blinkpy.blinkpy.api.request_videos")
|
||||
def test_parse_downloaded_items(self, mock_req, mock_sess):
|
||||
"""Test ability to parse downloaded items list."""
|
||||
blink = blinkpy.Blink()
|
||||
# pylint: disable=protected-access
|
||||
blinkpy._LOGGER.setLevel(logging.DEBUG)
|
||||
generic_entry = {
|
||||
'created_at': '1970',
|
||||
'device_name': 'foo',
|
||||
'deleted': True,
|
||||
'media': '/bar.mp4'
|
||||
"created_at": "1970",
|
||||
"device_name": "foo",
|
||||
"deleted": True,
|
||||
"media": "/bar.mp4",
|
||||
}
|
||||
result = [generic_entry]
|
||||
mock_req.return_value = {'media': result}
|
||||
mock_req.return_value = {"media": result}
|
||||
blink.last_refresh = 0
|
||||
formatted_date = get_time(blink.last_refresh)
|
||||
expected_log = [
|
||||
"INFO:blinkpy.blinkpy:Retrieving videos since {}".format(
|
||||
formatted_date),
|
||||
"INFO:blinkpy.blinkpy:Retrieving videos since {}".format(formatted_date),
|
||||
"DEBUG:blinkpy.blinkpy:Processing page 1",
|
||||
"DEBUG:blinkpy.blinkpy:foo: /bar.mp4 is marked as deleted."
|
||||
"DEBUG:blinkpy.blinkpy:foo: /bar.mp4 is marked as deleted.",
|
||||
]
|
||||
with self.assertLogs() as dl_log:
|
||||
blink.download_videos('/tmp', stop=2)
|
||||
blink.download_videos("/tmp", stop=2)
|
||||
self.assertEqual(dl_log.output, expected_log)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.api.request_videos')
|
||||
@mock.patch("blinkpy.blinkpy.api.request_videos")
|
||||
def test_parse_camera_not_in_list(self, mock_req, mock_sess):
|
||||
"""Test ability to parse downloaded items list."""
|
||||
blink = blinkpy.Blink()
|
||||
# pylint: disable=protected-access
|
||||
blinkpy._LOGGER.setLevel(logging.DEBUG)
|
||||
generic_entry = {
|
||||
'created_at': '1970',
|
||||
'device_name': 'foo',
|
||||
'deleted': True,
|
||||
'media': '/bar.mp4'
|
||||
"created_at": "1970",
|
||||
"device_name": "foo",
|
||||
"deleted": True,
|
||||
"media": "/bar.mp4",
|
||||
}
|
||||
result = [generic_entry]
|
||||
mock_req.return_value = {'media': result}
|
||||
mock_req.return_value = {"media": result}
|
||||
blink.last_refresh = 0
|
||||
formatted_date = get_time(blink.last_refresh)
|
||||
expected_log = [
|
||||
"INFO:blinkpy.blinkpy:Retrieving videos since {}".format(
|
||||
formatted_date),
|
||||
"INFO:blinkpy.blinkpy:Retrieving videos since {}".format(formatted_date),
|
||||
"DEBUG:blinkpy.blinkpy:Processing page 1",
|
||||
"DEBUG:blinkpy.blinkpy:Skipping videos for foo."
|
||||
"DEBUG:blinkpy.blinkpy:Skipping videos for foo.",
|
||||
]
|
||||
with self.assertLogs() as dl_log:
|
||||
blink.download_videos('/tmp', camera='bar', stop=2)
|
||||
blink.download_videos("/tmp", camera="bar", stop=2)
|
||||
self.assertEqual(dl_log.output, expected_log)
|
||||
|
||||
+82
-69
@@ -12,30 +12,29 @@ 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,
|
||||
BlinkException, BlinkURLHandler)
|
||||
http_req,
|
||||
create_session,
|
||||
BlinkAuthenticationException,
|
||||
BlinkException,
|
||||
BlinkURLHandler,
|
||||
)
|
||||
from blinkpy.helpers.constants import __version__
|
||||
import tests.mock_responses as mresp
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
USERNAME = "foobar"
|
||||
PASSWORD = "deadbeef"
|
||||
|
||||
|
||||
@mock.patch('blinkpy.helpers.util.Session.send',
|
||||
side_effect=mresp.mocked_session_send)
|
||||
@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send)
|
||||
class TestBlinkSetup(unittest.TestCase):
|
||||
"""Test the Blink class in blinkpy."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up Blink module."""
|
||||
self.blink_no_cred = Blink()
|
||||
self.blink = Blink(username=USERNAME,
|
||||
password=PASSWORD)
|
||||
self.blink.sync['test'] = BlinkSyncModule(self.blink,
|
||||
'test',
|
||||
'1234',
|
||||
[])
|
||||
self.blink.urls = BlinkURLHandler('test')
|
||||
self.blink = Blink(username=USERNAME, password=PASSWORD)
|
||||
self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", [])
|
||||
self.blink.urls = BlinkURLHandler("test")
|
||||
self.blink.session = create_session()
|
||||
|
||||
def tearDown(self):
|
||||
@@ -63,122 +62,129 @@ class TestBlinkSetup(unittest.TestCase):
|
||||
def test_no_auth_header(self, mock_sess):
|
||||
"""Check that we throw an exception when no auth header given."""
|
||||
# pylint: disable=unused-variable
|
||||
(region_id, region), = mresp.LOGIN_RESPONSE['region'].items()
|
||||
((region_id, region),) = mresp.LOGIN_RESPONSE["region"].items()
|
||||
self.blink.urls = BlinkURLHandler(region_id)
|
||||
with self.assertRaises(BlinkException):
|
||||
self.blink.get_ids()
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.getpass.getpass')
|
||||
@mock.patch("blinkpy.blinkpy.getpass.getpass")
|
||||
def test_manual_login(self, getpwd, mock_sess):
|
||||
"""Check that we can manually use the login() function."""
|
||||
getpwd.return_value = PASSWORD
|
||||
with mock.patch('builtins.input', return_value=USERNAME):
|
||||
with mock.patch("builtins.input", return_value=USERNAME):
|
||||
self.assertTrue(self.blink_no_cred.login())
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink_no_cred._username, USERNAME)
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink_no_cred._password, PASSWORD)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.getpass.getpass')
|
||||
@mock.patch('blinkpy.blinkpy.Blink.get_auth_token')
|
||||
@mock.patch("blinkpy.blinkpy.getpass.getpass")
|
||||
@mock.patch("blinkpy.blinkpy.Blink.get_auth_token")
|
||||
def test_no_cred_file(self, getpwd, getauth, mock_sess):
|
||||
"""Check that normal login occurs when cred file doesn't exist."""
|
||||
# pylint: disable=protected-access
|
||||
self.blink._cred_file = '/tmp/fake.file'
|
||||
self.blink._cred_file = "/tmp/fake.file"
|
||||
getpwd.return_value = PASSWORD
|
||||
getauth.return_value = True
|
||||
with mock.patch('builtins.input', return_value=USERNAME):
|
||||
with mock.patch("builtins.input", return_value=USERNAME):
|
||||
self.assertTrue(self.blink.login())
|
||||
|
||||
def test_exit_on_missing_json(self, mock_sess):
|
||||
"""Test that we fail on missing json data."""
|
||||
# pylint: disable=protected-access
|
||||
self.blink._cred_file = '/tmp/fake.file'
|
||||
with mock.patch('os.path.isfile', return_value=True):
|
||||
with mock.patch('builtins.open', mock.mock_open(read_data="{}")):
|
||||
self.blink._cred_file = "/tmp/fake.file"
|
||||
with mock.patch("os.path.isfile", return_value=True):
|
||||
with mock.patch("builtins.open", mock.mock_open(read_data="{}")):
|
||||
self.assertFalse(self.blink.login())
|
||||
|
||||
def test_exit_on_bad_json(self, mock_sess):
|
||||
"""Test that we fail on bad json format."""
|
||||
# pylint: disable=protected-access
|
||||
self.blink._cred_file = '/tmp/fake.file'
|
||||
with mock.patch('os.path.isfile', return_value=True):
|
||||
with mock.patch('builtins.open', mock.mock_open(read_data='{]')):
|
||||
self.blink._cred_file = "/tmp/fake.file"
|
||||
with mock.patch("os.path.isfile", return_value=True):
|
||||
with mock.patch("builtins.open", mock.mock_open(read_data="{]")):
|
||||
self.assertFalse(self.blink.login())
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.json.load')
|
||||
@mock.patch("blinkpy.blinkpy.json.load")
|
||||
def test_cred_file(self, mockjson, mock_sess):
|
||||
"""Test that loading credential file works."""
|
||||
# pylint: disable=protected-access
|
||||
self.blink_no_cred._cred_file = '/tmp/fake.file'
|
||||
mockjson.return_value = {'username': 'foo', 'password': 'bar'}
|
||||
with mock.patch('os.path.isfile', return_value=True):
|
||||
with mock.patch('builtins.open', mock.mock_open(read_data='')):
|
||||
self.blink_no_cred._cred_file = "/tmp/fake.file"
|
||||
mockjson.return_value = {"username": "foo", "password": "bar"}
|
||||
with mock.patch("os.path.isfile", return_value=True):
|
||||
with mock.patch("builtins.open", mock.mock_open(read_data="")):
|
||||
self.assertTrue(self.blink_no_cred.login())
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink_no_cred._username, 'foo')
|
||||
self.assertEqual(self.blink_no_cred._username, "foo")
|
||||
# pylint: disable=protected-access
|
||||
self.assertEqual(self.blink_no_cred._password, 'bar')
|
||||
self.assertEqual(self.blink_no_cred._password, "bar")
|
||||
|
||||
def test_bad_request(self, mock_sess):
|
||||
"""Check that we raise an Exception with a bad request."""
|
||||
self.blink.session = create_session()
|
||||
explog = ("WARNING:blinkpy.helpers.util:"
|
||||
"Response from server: 200 - foo")
|
||||
explog = "WARNING:blinkpy.helpers.util:" "Response from server: 200 - foo"
|
||||
with self.assertRaises(BlinkException):
|
||||
http_req(self.blink, reqtype='bad')
|
||||
http_req(self.blink, reqtype="bad")
|
||||
|
||||
with self.assertLogs() as logrecord:
|
||||
http_req(self.blink, reqtype='post', is_retry=True)
|
||||
http_req(self.blink, reqtype="post", is_retry=True)
|
||||
self.assertEqual(logrecord.output, [explog])
|
||||
|
||||
def test_authentication(self, mock_sess):
|
||||
"""Check that we can authenticate Blink up properly."""
|
||||
authtoken = self.blink.get_auth_token()['TOKEN_AUTH']
|
||||
expected = mresp.LOGIN_RESPONSE['authtoken']['authtoken']
|
||||
authtoken = self.blink.get_auth_token()["TOKEN_AUTH"]
|
||||
expected = mresp.LOGIN_RESPONSE["authtoken"]["authtoken"]
|
||||
self.assertEqual(authtoken, expected)
|
||||
|
||||
def test_reauthorization_attempt(self, mock_sess):
|
||||
"""Check that we can reauthorize after first unsuccessful attempt."""
|
||||
original_header = self.blink.get_auth_token()
|
||||
# pylint: disable=protected-access
|
||||
bad_header = {'Host': self.blink._host, 'TOKEN_AUTH': 'BADTOKEN'}
|
||||
bad_header = {"Host": self.blink._host, "TOKEN_AUTH": "BADTOKEN"}
|
||||
# pylint: disable=protected-access
|
||||
self.blink._auth_header = bad_header
|
||||
self.assertEqual(self.blink.auth_header, bad_header)
|
||||
api.request_homescreen(self.blink)
|
||||
self.assertEqual(self.blink.auth_header, original_header)
|
||||
|
||||
@mock.patch('blinkpy.api.request_networks')
|
||||
@mock.patch("blinkpy.api.request_networks")
|
||||
def test_multiple_networks(self, mock_net, mock_sess):
|
||||
"""Check that we handle multiple networks appropriately."""
|
||||
mock_net.return_value = {
|
||||
'networks': [{'id': 1234, 'account_id': 1111},
|
||||
{'id': 5678, 'account_id': 2222}]
|
||||
"networks": [
|
||||
{"id": 1234, "account_id": 1111},
|
||||
{"id": 5678, "account_id": 2222},
|
||||
]
|
||||
}
|
||||
self.blink.networks = {
|
||||
"0000": {"onboarded": False, "name": "foo"},
|
||||
"5678": {"onboarded": True, "name": "bar"},
|
||||
"1234": {"onboarded": False, "name": "test"},
|
||||
}
|
||||
self.blink.networks = {'0000': {'onboarded': False, 'name': 'foo'},
|
||||
'5678': {'onboarded': True, 'name': 'bar'},
|
||||
'1234': {'onboarded': False, 'name': 'test'}}
|
||||
self.blink.get_ids()
|
||||
self.assertTrue('5678' in self.blink.network_ids)
|
||||
self.assertTrue("5678" in self.blink.network_ids)
|
||||
self.assertEqual(self.blink.account_id, 2222)
|
||||
|
||||
@mock.patch('blinkpy.api.request_networks')
|
||||
@mock.patch("blinkpy.api.request_networks")
|
||||
def test_multiple_onboarded_networks(self, mock_net, mock_sess):
|
||||
"""Check that we handle multiple networks appropriately."""
|
||||
mock_net.return_value = {
|
||||
'networks': [{'id': 0000, 'account_id': 2222},
|
||||
{'id': 5678, 'account_id': 1111}]
|
||||
"networks": [
|
||||
{"id": 0000, "account_id": 2222},
|
||||
{"id": 5678, "account_id": 1111},
|
||||
]
|
||||
}
|
||||
self.blink.networks = {
|
||||
"0000": {"onboarded": False, "name": "foo"},
|
||||
"5678": {"onboarded": True, "name": "bar"},
|
||||
"1234": {"onboarded": True, "name": "test"},
|
||||
}
|
||||
self.blink.networks = {'0000': {'onboarded': False, 'name': 'foo'},
|
||||
'5678': {'onboarded': True, 'name': 'bar'},
|
||||
'1234': {'onboarded': True, 'name': 'test'}}
|
||||
self.blink.get_ids()
|
||||
self.assertTrue('5678' in self.blink.network_ids)
|
||||
self.assertTrue('1234' in self.blink.network_ids)
|
||||
self.assertTrue("5678" in self.blink.network_ids)
|
||||
self.assertTrue("1234" in self.blink.network_ids)
|
||||
self.assertEqual(self.blink.account_id, 1111)
|
||||
|
||||
@mock.patch('blinkpy.blinkpy.time.time')
|
||||
@mock.patch("blinkpy.blinkpy.time.time")
|
||||
def test_throttle(self, mock_time, mock_sess):
|
||||
"""Check throttling functionality."""
|
||||
now = self.blink.refresh_rate + 1
|
||||
@@ -186,8 +192,9 @@ class TestBlinkSetup(unittest.TestCase):
|
||||
self.assertEqual(self.blink.last_refresh, None)
|
||||
self.assertEqual(self.blink.check_if_ok_to_update(), True)
|
||||
self.assertEqual(self.blink.last_refresh, None)
|
||||
with mock.patch('blinkpy.sync_module.BlinkSyncModule.refresh',
|
||||
return_value=True):
|
||||
with mock.patch(
|
||||
"blinkpy.sync_module.BlinkSyncModule.refresh", return_value=True
|
||||
):
|
||||
self.blink.refresh()
|
||||
|
||||
self.assertEqual(self.blink.last_refresh, now)
|
||||
@@ -196,29 +203,35 @@ class TestBlinkSetup(unittest.TestCase):
|
||||
|
||||
def test_sync_case_insensitive_dict(self, mock_sess):
|
||||
"""Check that we can access sync modules ignoring case."""
|
||||
self.assertEqual(self.blink.sync['test'].name, 'test')
|
||||
self.assertEqual(self.blink.sync['TEST'].name, 'test')
|
||||
self.assertEqual(self.blink.sync["test"].name, "test")
|
||||
self.assertEqual(self.blink.sync["TEST"].name, "test")
|
||||
|
||||
@mock.patch('blinkpy.api.request_login')
|
||||
@mock.patch("blinkpy.api.request_login")
|
||||
def test_unexpected_login(self, mock_login, mock_sess):
|
||||
"""Check that we appropriately handle unexpected login info."""
|
||||
mock_login.return_value = None
|
||||
self.assertFalse(self.blink.get_auth_token())
|
||||
|
||||
@mock.patch('blinkpy.api.request_homescreen')
|
||||
@mock.patch("blinkpy.api.request_homescreen")
|
||||
def test_get_cameras(self, mock_home, mock_sess):
|
||||
"""Check retrieval of camera information."""
|
||||
mock_home.return_value = {
|
||||
'cameras': [{'name': 'foo', 'network_id': 1234, 'id': 5678},
|
||||
{'name': 'bar', 'network_id': 1234, 'id': 5679},
|
||||
{'name': 'test', 'network_id': 4321, 'id': 0000}]
|
||||
"cameras": [
|
||||
{"name": "foo", "network_id": 1234, "id": 5678},
|
||||
{"name": "bar", "network_id": 1234, "id": 5679},
|
||||
{"name": "test", "network_id": 4321, "id": 0000},
|
||||
]
|
||||
}
|
||||
result = self.blink.get_cameras()
|
||||
self.assertEqual(result, {'1234': [{'name': 'foo', 'id': 5678},
|
||||
{'name': 'bar', 'id': 5679}],
|
||||
'4321': [{'name': 'test', 'id': 0000}]})
|
||||
self.assertEqual(
|
||||
result,
|
||||
{
|
||||
"1234": [{"name": "foo", "id": 5678}, {"name": "bar", "id": 5679}],
|
||||
"4321": [{"name": "test", "id": 0000}],
|
||||
},
|
||||
)
|
||||
|
||||
@mock.patch('blinkpy.api.request_homescreen')
|
||||
@mock.patch("blinkpy.api.request_homescreen")
|
||||
def test_get_cameras_failure(self, mock_home, mock_sess):
|
||||
"""Check that on failure we initialize empty info and move on."""
|
||||
mock_home.return_value = {}
|
||||
|
||||
+80
-81
@@ -14,42 +14,40 @@ from blinkpy.sync_module import BlinkSyncModule
|
||||
from blinkpy.camera import BlinkCamera
|
||||
import tests.mock_responses as mresp
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
USERNAME = "foobar"
|
||||
PASSWORD = "deadbeef"
|
||||
|
||||
CAMERA_CFG = {
|
||||
'camera': [
|
||||
"camera": [
|
||||
{
|
||||
'battery_voltage': 90,
|
||||
'motion_alert': True,
|
||||
'wifi_strength': -30,
|
||||
'temperature': 68
|
||||
"battery_voltage": 90,
|
||||
"motion_alert": True,
|
||||
"wifi_strength": -30,
|
||||
"temperature": 68,
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@mock.patch('blinkpy.helpers.util.Session.send',
|
||||
side_effect=mresp.mocked_session_send)
|
||||
@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send)
|
||||
class TestBlinkCameraSetup(unittest.TestCase):
|
||||
"""Test the Blink class in blinkpy."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up Blink module."""
|
||||
self.blink = blinkpy.Blink(username=USERNAME,
|
||||
password=PASSWORD)
|
||||
self.blink = blinkpy.Blink(username=USERNAME, password=PASSWORD)
|
||||
header = {
|
||||
'Host': 'abc.zxc',
|
||||
'TOKEN_AUTH': mresp.LOGIN_RESPONSE['authtoken']['authtoken']
|
||||
"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.sync['test'] = BlinkSyncModule(self.blink, 'test', 1234, [])
|
||||
self.camera = BlinkCamera(self.blink.sync['test'])
|
||||
self.camera.name = 'foobar'
|
||||
self.blink.sync['test'].cameras['foobar'] = self.camera
|
||||
self.blink.urls = BlinkURLHandler("test")
|
||||
self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", 1234, [])
|
||||
self.camera = BlinkCamera(self.blink.sync["test"])
|
||||
self.camera.name = "foobar"
|
||||
self.blink.sync["test"].cameras["foobar"] = self.camera
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
@@ -58,98 +56,99 @@ class TestBlinkCameraSetup(unittest.TestCase):
|
||||
def test_camera_update(self, mock_sess):
|
||||
"""Test that we can properly update camera properties."""
|
||||
config = {
|
||||
'name': 'new',
|
||||
'id': 1234,
|
||||
'network_id': 5678,
|
||||
'serial': '12345678',
|
||||
'enabled': False,
|
||||
'battery_voltage': 90,
|
||||
'battery_state': 'ok',
|
||||
'temperature': 68,
|
||||
'wifi_strength': 4,
|
||||
'thumbnail': '/thumb',
|
||||
"name": "new",
|
||||
"id": 1234,
|
||||
"network_id": 5678,
|
||||
"serial": "12345678",
|
||||
"enabled": False,
|
||||
"battery_voltage": 90,
|
||||
"battery_state": "ok",
|
||||
"temperature": 68,
|
||||
"wifi_strength": 4,
|
||||
"thumbnail": "/thumb",
|
||||
}
|
||||
self.camera.last_record = ['1']
|
||||
self.camera.last_record = ["1"]
|
||||
self.camera.sync.last_record = {
|
||||
'new': {
|
||||
'clip': '/test.mp4',
|
||||
'time': '1970-01-01T00:00:00'
|
||||
}
|
||||
"new": {"clip": "/test.mp4", "time": "1970-01-01T00:00:00"}
|
||||
}
|
||||
mock_sess.side_effect = [
|
||||
mresp.MockResponse({'temp': 71}, 200),
|
||||
'test',
|
||||
'foobar'
|
||||
mresp.MockResponse({"temp": 71}, 200),
|
||||
"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.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, 'ok')
|
||||
self.assertEqual(self.camera.battery, "ok")
|
||||
self.assertEqual(self.camera.temperature, 68)
|
||||
self.assertEqual(self.camera.temperature_c, 20)
|
||||
self.assertEqual(self.camera.temperature_calibrated, 71)
|
||||
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')
|
||||
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")
|
||||
|
||||
def test_no_thumbnails(self, mock_sess):
|
||||
"""Tests that thumbnail is 'None' if none found."""
|
||||
mock_sess.return_value = 'foobar'
|
||||
self.camera.last_record = ['1']
|
||||
mock_sess.return_value = "foobar"
|
||||
self.camera.last_record = ["1"]
|
||||
config = {
|
||||
'name': 'new',
|
||||
'id': 1234,
|
||||
'network_id': 5678,
|
||||
'serial': '12345678',
|
||||
'enabled': False,
|
||||
'battery_voltage': 90,
|
||||
'battery_state': 'ok',
|
||||
'temperature': 68,
|
||||
'wifi_strength': 4,
|
||||
'thumbnail': '',
|
||||
}
|
||||
self.camera.sync.homescreen = {
|
||||
'devices': []
|
||||
"name": "new",
|
||||
"id": 1234,
|
||||
"network_id": 5678,
|
||||
"serial": "12345678",
|
||||
"enabled": False,
|
||||
"battery_voltage": 90,
|
||||
"battery_state": "ok",
|
||||
"temperature": 68,
|
||||
"wifi_strength": 4,
|
||||
"thumbnail": "",
|
||||
}
|
||||
self.camera.sync.homescreen = {"devices": []}
|
||||
self.assertEqual(self.camera.temperature_calibrated, None)
|
||||
with self.assertLogs() as logrecord:
|
||||
self.camera.update(config, force=True)
|
||||
self.assertEqual(self.camera.thumbnail, None)
|
||||
self.assertEqual(self.camera.last_record, ['1'])
|
||||
self.assertEqual(self.camera.last_record, ["1"])
|
||||
self.assertEqual(self.camera.temperature_calibrated, 68)
|
||||
self.assertEqual(
|
||||
logrecord.output,
|
||||
[("WARNING:blinkpy.camera:Could not retrieve calibrated "
|
||||
"temperature."),
|
||||
("WARNING:blinkpy.camera:Could not find thumbnail for camera new"
|
||||
"\nNoneType: None")]
|
||||
[
|
||||
(
|
||||
"WARNING:blinkpy.camera:Could not retrieve calibrated "
|
||||
"temperature."
|
||||
),
|
||||
(
|
||||
"WARNING:blinkpy.camera:Could not find thumbnail for camera new"
|
||||
"\nNoneType: None"
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
def test_no_video_clips(self, mock_sess):
|
||||
"""Tests that we still proceed with camera setup with no videos."""
|
||||
mock_sess.return_value = 'foobar'
|
||||
mock_sess.return_value = "foobar"
|
||||
config = {
|
||||
'name': 'new',
|
||||
'id': 1234,
|
||||
'network_id': 5678,
|
||||
'serial': '12345678',
|
||||
'enabled': False,
|
||||
'battery_voltage': 90,
|
||||
'battery_state': 'ok',
|
||||
'temperature': 68,
|
||||
'wifi_strength': 4,
|
||||
'thumbnail': '/foobar',
|
||||
}
|
||||
self.camera.sync.homescreen = {
|
||||
'devices': []
|
||||
"name": "new",
|
||||
"id": 1234,
|
||||
"network_id": 5678,
|
||||
"serial": "12345678",
|
||||
"enabled": False,
|
||||
"battery_voltage": 90,
|
||||
"battery_state": "ok",
|
||||
"temperature": 68,
|
||||
"wifi_strength": 4,
|
||||
"thumbnail": "/foobar",
|
||||
}
|
||||
self.camera.sync.homescreen = {"devices": []}
|
||||
self.camera.update(config, force_cache=True)
|
||||
self.assertEqual(self.camera.clip, None)
|
||||
self.assertEqual(self.camera.video_from_cache, None)
|
||||
|
||||
+109
-108
@@ -6,44 +6,41 @@ from blinkpy import blinkpy
|
||||
from blinkpy.sync_module import BlinkSyncModule
|
||||
from blinkpy.camera import BlinkCamera
|
||||
|
||||
USERNAME = 'foobar'
|
||||
PASSWORD = 'deadbeef'
|
||||
USERNAME = "foobar"
|
||||
PASSWORD = "deadbeef"
|
||||
|
||||
|
||||
@mock.patch('blinkpy.api.http_req')
|
||||
@mock.patch("blinkpy.api.http_req")
|
||||
class TestBlinkSyncModule(unittest.TestCase):
|
||||
"""Test BlinkSyncModule functions in blinkpy."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up Blink module."""
|
||||
self.blink = blinkpy.Blink(username=USERNAME,
|
||||
password=PASSWORD,
|
||||
motion_interval=0)
|
||||
self.blink = blinkpy.Blink(
|
||||
username=USERNAME, password=PASSWORD, motion_interval=0
|
||||
)
|
||||
# pylint: disable=protected-access
|
||||
self.blink._auth_header = {
|
||||
'Host': 'test.url.tld',
|
||||
'TOKEN_AUTH': 'foobar123'
|
||||
}
|
||||
self.blink._auth_header = {"Host": "test.url.tld", "TOKEN_AUTH": "foobar123"}
|
||||
self.blink.last_refresh = 0
|
||||
self.blink.urls = blinkpy.BlinkURLHandler('test')
|
||||
self.blink.sync['test'] = BlinkSyncModule(self.blink,
|
||||
'test',
|
||||
'1234',
|
||||
[])
|
||||
self.blink.urls = blinkpy.BlinkURLHandler("test")
|
||||
self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", [])
|
||||
self.camera = BlinkCamera(self.blink.sync)
|
||||
self.mock_start = [
|
||||
{'syncmodule': {
|
||||
'id': 1234,
|
||||
'network_id': 5678,
|
||||
'serial': '12345678',
|
||||
'status': 'foobar'}},
|
||||
{'event': True},
|
||||
{
|
||||
"syncmodule": {
|
||||
"id": 1234,
|
||||
"network_id": 5678,
|
||||
"serial": "12345678",
|
||||
"status": "foobar",
|
||||
}
|
||||
},
|
||||
{"event": True},
|
||||
{},
|
||||
{},
|
||||
None,
|
||||
{'devicestatus': {}},
|
||||
{"devicestatus": {}},
|
||||
]
|
||||
self.blink.sync['test'].network_info = {'network': {'armed': True}}
|
||||
self.blink.sync["test"].network_info = {"network": {"armed": True}}
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after test."""
|
||||
@@ -53,173 +50,177 @@ class TestBlinkSyncModule(unittest.TestCase):
|
||||
|
||||
def test_get_events(self, mock_resp):
|
||||
"""Test get events function."""
|
||||
mock_resp.return_value = {'event': True}
|
||||
self.assertEqual(self.blink.sync['test'].get_events(), True)
|
||||
mock_resp.return_value = {"event": True}
|
||||
self.assertEqual(self.blink.sync["test"].get_events(), True)
|
||||
|
||||
def test_get_camera_info(self, mock_resp):
|
||||
"""Test get camera info function."""
|
||||
mock_resp.return_value = {'camera': ['foobar']}
|
||||
self.assertEqual(self.blink.sync['test'].get_camera_info('1234'),
|
||||
'foobar')
|
||||
mock_resp.return_value = {"camera": ["foobar"]}
|
||||
self.assertEqual(self.blink.sync["test"].get_camera_info("1234"), "foobar")
|
||||
|
||||
def test_check_new_videos_startup(self, mock_resp):
|
||||
"""Test that check_new_videos does not block startup."""
|
||||
sync_module = self.blink.sync['test']
|
||||
sync_module = self.blink.sync["test"]
|
||||
self.blink.last_refresh = None
|
||||
self.assertFalse(sync_module.check_new_videos())
|
||||
|
||||
def test_check_new_videos(self, mock_resp):
|
||||
"""Test recent video response."""
|
||||
mock_resp.return_value = {
|
||||
'media': [{
|
||||
'device_name': 'foo',
|
||||
'media': '/foo/bar.mp4',
|
||||
'created_at': '1990-01-01T00:00:00+00:00'
|
||||
}]
|
||||
"media": [
|
||||
{
|
||||
"device_name": "foo",
|
||||
"media": "/foo/bar.mp4",
|
||||
"created_at": "1990-01-01T00:00:00+00:00",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
sync_module = self.blink.sync['test']
|
||||
sync_module.cameras = {'foo': None}
|
||||
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': '1990-01-01T00:00:00+00:00'})
|
||||
self.assertEqual(sync_module.motion, {'foo': True})
|
||||
mock_resp.return_value = {'media': []}
|
||||
self.assertEqual(
|
||||
sync_module.last_record["foo"],
|
||||
{"clip": "/foo/bar.mp4", "time": "1990-01-01T00:00:00+00: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': '1990-01-01T00:00:00+00:00'})
|
||||
self.assertEqual(sync_module.motion, {"foo": False})
|
||||
self.assertEqual(
|
||||
sync_module.last_record["foo"],
|
||||
{"clip": "/foo/bar.mp4", "time": "1990-01-01T00:00:00+00:00"},
|
||||
)
|
||||
|
||||
def test_check_new_videos_old_date(self, mock_resp):
|
||||
"""Test videos return response with old date."""
|
||||
mock_resp.return_value = {
|
||||
'media': [{
|
||||
'device_name': 'foo',
|
||||
'media': '/foo/bar.mp4',
|
||||
'created_at': '1970-01-01T00:00:00+00:00'
|
||||
}]
|
||||
"media": [
|
||||
{
|
||||
"device_name": "foo",
|
||||
"media": "/foo/bar.mp4",
|
||||
"created_at": "1970-01-01T00:00:00+00:00",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
sync_module = self.blink.sync['test']
|
||||
sync_module.cameras = {'foo': None}
|
||||
sync_module = self.blink.sync["test"]
|
||||
sync_module.cameras = {"foo": None}
|
||||
sync_module.blink.last_refresh = 1000
|
||||
self.assertTrue(sync_module.check_new_videos())
|
||||
self.assertEqual(sync_module.motion, {'foo': False})
|
||||
self.assertEqual(sync_module.motion, {"foo": False})
|
||||
|
||||
def test_check_no_motion_if_not_armed(self, mock_resp):
|
||||
"""Test that motion detection is not set if module unarmed."""
|
||||
mock_resp.return_value = {
|
||||
'media': [{
|
||||
'device_name': 'foo',
|
||||
'media': '/foo/bar.mp4',
|
||||
'created_at': '1990-01-01T00:00:00+00:00'
|
||||
}]
|
||||
"media": [
|
||||
{
|
||||
"device_name": "foo",
|
||||
"media": "/foo/bar.mp4",
|
||||
"created_at": "1990-01-01T00:00:00+00:00",
|
||||
}
|
||||
]
|
||||
}
|
||||
sync_module = self.blink.sync['test']
|
||||
sync_module.cameras = {'foo': None}
|
||||
sync_module = self.blink.sync["test"]
|
||||
sync_module.cameras = {"foo": None}
|
||||
sync_module.blink.last_refresh = 1000
|
||||
self.assertTrue(sync_module.check_new_videos())
|
||||
self.assertEqual(sync_module.motion, {'foo': True})
|
||||
sync_module.network_info = {'network': {'armed': False}}
|
||||
self.assertEqual(sync_module.motion, {"foo": True})
|
||||
sync_module.network_info = {"network": {"armed": False}}
|
||||
self.assertTrue(sync_module.check_new_videos())
|
||||
self.assertEqual(sync_module.motion, {'foo': False})
|
||||
self.assertEqual(sync_module.motion, {"foo": False})
|
||||
|
||||
def test_check_multiple_videos(self, mock_resp):
|
||||
"""Test motion found even with multiple videos."""
|
||||
mock_resp.return_value = {
|
||||
'media': [
|
||||
"media": [
|
||||
{
|
||||
'device_name': 'foo',
|
||||
'media': '/foo/bar.mp4',
|
||||
'created_at': '1970-01-01T00:00:00+00:00'
|
||||
"device_name": "foo",
|
||||
"media": "/foo/bar.mp4",
|
||||
"created_at": "1970-01-01T00:00:00+00:00",
|
||||
},
|
||||
{
|
||||
'device_name': 'foo',
|
||||
'media': '/bar/foo.mp4',
|
||||
'created_at': '1990-01-01T00:00:00+00:00'
|
||||
"device_name": "foo",
|
||||
"media": "/bar/foo.mp4",
|
||||
"created_at": "1990-01-01T00:00:00+00:00",
|
||||
},
|
||||
{
|
||||
'device_name': 'foo',
|
||||
'media': '/foobar.mp4',
|
||||
'created_at': '1970-01-01T00:00:01+00:00'
|
||||
}
|
||||
"device_name": "foo",
|
||||
"media": "/foobar.mp4",
|
||||
"created_at": "1970-01-01T00:00:01+00:00",
|
||||
},
|
||||
]
|
||||
}
|
||||
sync_module = self.blink.sync['test']
|
||||
sync_module.cameras = {'foo': None}
|
||||
sync_module = self.blink.sync["test"]
|
||||
sync_module.cameras = {"foo": None}
|
||||
sync_module.blink.last_refresh = 1000
|
||||
self.assertTrue(sync_module.check_new_videos())
|
||||
self.assertEqual(sync_module.motion, {'foo': True})
|
||||
self.assertEqual(sync_module.motion, {"foo": True})
|
||||
expected_result = {
|
||||
'foo': {
|
||||
'clip': '/bar/foo.mp4',
|
||||
'time': '1990-01-01T00:00:00+00:00'
|
||||
}
|
||||
"foo": {"clip": "/bar/foo.mp4", "time": "1990-01-01T00:00:00+00:00"}
|
||||
}
|
||||
self.assertEqual(sync_module.last_record, expected_result)
|
||||
|
||||
def test_check_new_videos_failed(self, mock_resp):
|
||||
"""Test method when response is unexpected."""
|
||||
mock_resp.side_effect = [None, 'just a string', {}]
|
||||
sync_module = self.blink.sync['test']
|
||||
sync_module.cameras = {'foo': None}
|
||||
mock_resp.side_effect = [None, "just a string", {}]
|
||||
sync_module = self.blink.sync["test"]
|
||||
sync_module.cameras = {"foo": None}
|
||||
|
||||
sync_module.motion['foo'] = True
|
||||
sync_module.motion["foo"] = True
|
||||
self.assertFalse(sync_module.check_new_videos())
|
||||
self.assertFalse(sync_module.motion['foo'])
|
||||
self.assertFalse(sync_module.motion["foo"])
|
||||
|
||||
sync_module.motion['foo'] = True
|
||||
sync_module.motion["foo"] = True
|
||||
self.assertFalse(sync_module.check_new_videos())
|
||||
self.assertFalse(sync_module.motion['foo'])
|
||||
self.assertFalse(sync_module.motion["foo"])
|
||||
|
||||
sync_module.motion['foo'] = True
|
||||
sync_module.motion["foo"] = True
|
||||
self.assertFalse(sync_module.check_new_videos())
|
||||
self.assertFalse(sync_module.motion['foo'])
|
||||
self.assertFalse(sync_module.motion["foo"])
|
||||
|
||||
def test_sync_start(self, mock_resp):
|
||||
"""Test sync start function."""
|
||||
mock_resp.side_effect = self.mock_start
|
||||
self.blink.sync['test'].start()
|
||||
self.assertEqual(self.blink.sync['test'].name, 'test')
|
||||
self.assertEqual(self.blink.sync['test'].sync_id, 1234)
|
||||
self.assertEqual(self.blink.sync['test'].network_id, 5678)
|
||||
self.assertEqual(self.blink.sync['test'].serial, '12345678')
|
||||
self.assertEqual(self.blink.sync['test'].status, 'foobar')
|
||||
self.blink.sync["test"].start()
|
||||
self.assertEqual(self.blink.sync["test"].name, "test")
|
||||
self.assertEqual(self.blink.sync["test"].sync_id, 1234)
|
||||
self.assertEqual(self.blink.sync["test"].network_id, 5678)
|
||||
self.assertEqual(self.blink.sync["test"].serial, "12345678")
|
||||
self.assertEqual(self.blink.sync["test"].status, "foobar")
|
||||
|
||||
def test_unexpected_summary(self, mock_resp):
|
||||
"""Test unexpected summary response."""
|
||||
self.mock_start[0] = None
|
||||
mock_resp.side_effect = self.mock_start
|
||||
self.assertFalse(self.blink.sync['test'].start())
|
||||
self.assertFalse(self.blink.sync["test"].start())
|
||||
|
||||
def test_summary_with_no_network_id(self, mock_resp):
|
||||
"""Test handling of bad summary."""
|
||||
self.mock_start[0]['syncmodule'] = None
|
||||
self.mock_start[0]["syncmodule"] = None
|
||||
mock_resp.side_effect = self.mock_start
|
||||
self.assertFalse(self.blink.sync['test'].start())
|
||||
self.assertFalse(self.blink.sync["test"].start())
|
||||
|
||||
def test_summary_with_only_network_id(self, mock_resp):
|
||||
"""Test handling of sparse summary."""
|
||||
self.mock_start[0]['syncmodule'] = {'network_id': 8675309}
|
||||
self.mock_start[0]["syncmodule"] = {"network_id": 8675309}
|
||||
mock_resp.side_effect = self.mock_start
|
||||
self.blink.sync['test'].start()
|
||||
self.assertEqual(self.blink.sync['test'].network_id, 8675309)
|
||||
self.blink.sync["test"].start()
|
||||
self.assertEqual(self.blink.sync["test"].network_id, 8675309)
|
||||
|
||||
def test_unexpected_camera_info(self, mock_resp):
|
||||
"""Test unexpected camera info response."""
|
||||
self.blink.sync['test'].cameras['foo'] = None
|
||||
self.blink.sync["test"].cameras["foo"] = None
|
||||
self.mock_start[5] = None
|
||||
mock_resp.side_effect = self.mock_start
|
||||
self.blink.sync['test'].start()
|
||||
self.assertEqual(self.blink.sync['test'].cameras, {'foo': None})
|
||||
self.blink.sync["test"].start()
|
||||
self.assertEqual(self.blink.sync["test"].cameras, {"foo": None})
|
||||
|
||||
def test_missing_camera_info(self, mock_resp):
|
||||
"""Test missing key from camera info response."""
|
||||
self.blink.sync['test'].cameras['foo'] = None
|
||||
self.blink.sync["test"].cameras["foo"] = None
|
||||
self.mock_start[5] = {}
|
||||
self.blink.sync['test'].start()
|
||||
self.assertEqual(self.blink.sync['test'].cameras, {'foo': None})
|
||||
self.blink.sync["test"].start()
|
||||
self.assertEqual(self.blink.sync["test"].cameras, {"foo": None})
|
||||
|
||||
+12
-10
@@ -43,17 +43,18 @@ class TestUtil(unittest.TestCase):
|
||||
self.assertEqual(2, len(calls))
|
||||
|
||||
# Fake time as 4 seconds from now
|
||||
with mock.patch('time.time', return_value=now_plus_four):
|
||||
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):
|
||||
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."""
|
||||
|
||||
@@ -68,6 +69,7 @@ class TestUtil(unittest.TestCase):
|
||||
|
||||
def test_throttle_on_two_methods(self):
|
||||
"""Test that throttle works for multiple methods."""
|
||||
|
||||
class Tester:
|
||||
"""A tester class for throttling."""
|
||||
|
||||
@@ -91,24 +93,24 @@ class TestUtil(unittest.TestCase):
|
||||
self.assertEqual(tester.test1(), None)
|
||||
self.assertEqual(tester.test2(), None)
|
||||
|
||||
with mock.patch('time.time', return_value=now_plus_4):
|
||||
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):
|
||||
with mock.patch("time.time", return_value=now_plus_6):
|
||||
self.assertEqual(tester.test1(), None)
|
||||
self.assertEqual(tester.test2(), True)
|
||||
|
||||
def test_legacy_subdomains(self):
|
||||
"""Test that subdomain can be set to legacy mode."""
|
||||
urls = BlinkURLHandler('test')
|
||||
self.assertEqual(urls.subdomain, 'rest-test')
|
||||
urls = BlinkURLHandler('test', legacy=True)
|
||||
self.assertEqual(urls.subdomain, 'rest.test')
|
||||
urls = BlinkURLHandler("test")
|
||||
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'
|
||||
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))
|
||||
|
||||
Reference in New Issue
Block a user