Tests now working

- Simplified tests a bit
This commit is contained in:
Kevin Fronczak
2018-10-14 00:12:44 -04:00
parent 7fd822caf2
commit f7a5b5cc03
6 changed files with 155 additions and 315 deletions
+1
View File
@@ -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."""
+7 -6
View File
@@ -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)
+1 -69
View File
@@ -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()
@@ -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)
+67 -163
View File
@@ -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')
+58 -65
View File
@@ -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')