278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""Tests camera and system functions."""
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from blinkpy.blinkpy import Blink
|
|
from blinkpy.helpers.util import BlinkURLHandler
|
|
from blinkpy.sync_module import BlinkSyncModule
|
|
from blinkpy.camera import BlinkCamera
|
|
|
|
|
|
@mock.patch("blinkpy.auth.Auth.query")
|
|
class TestBlinkSyncModule(unittest.TestCase):
|
|
"""Test BlinkSyncModule functions in blinkpy."""
|
|
|
|
def setUp(self):
|
|
"""Set up Blink module."""
|
|
self.blink = Blink(motion_interval=0)
|
|
self.blink.last_refresh = 0
|
|
self.blink.urls = 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},
|
|
{},
|
|
{},
|
|
None,
|
|
{"devicestatus": {}},
|
|
]
|
|
self.blink.sync["test"].network_info = {"network": {"armed": True}}
|
|
|
|
def tearDown(self):
|
|
"""Clean up after test."""
|
|
self.blink = None
|
|
self.camera = None
|
|
self.mock_start = None
|
|
|
|
def test_bad_status(self, mock_resp):
|
|
"""Check that we mark module unavaiable on bad status."""
|
|
self.blink.sync["test"].status = None
|
|
self.blink.sync["test"].available = True
|
|
self.assertFalse(self.blink.sync["test"].online)
|
|
self.assertFalse(self.blink.sync["test"].available)
|
|
|
|
def test_bad_arm(self, mock_resp):
|
|
"""Check that we mark module unavaiable if bad arm status."""
|
|
self.blink.sync["test"].network_info = None
|
|
self.blink.sync["test"].available = True
|
|
self.assertEqual(self.blink.sync["test"].arm, None)
|
|
self.assertFalse(self.blink.sync["test"].available)
|
|
self.blink.sync["test"].network_info = {}
|
|
self.blink.sync["test"].available = True
|
|
self.assertEqual(self.blink.sync["test"].arm, None)
|
|
self.assertFalse(self.blink.sync["test"].available)
|
|
|
|
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)
|
|
|
|
def test_get_events_fail(self, mock_resp):
|
|
"""Test handling of failed get events function."""
|
|
mock_resp.return_value = None
|
|
self.assertFalse(self.blink.sync["test"].get_events())
|
|
mock_resp.return_value = {}
|
|
self.assertFalse(self.blink.sync["test"].get_events())
|
|
|
|
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")
|
|
|
|
def test_get_camera_info_fail(self, mock_resp):
|
|
"""Test handling of failed get camera info function."""
|
|
mock_resp.return_value = None
|
|
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), [])
|
|
mock_resp.return_value = {}
|
|
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), [])
|
|
mock_resp.return_value = {"camera": None}
|
|
self.assertEqual(self.blink.sync["test"].get_camera_info("1"), [])
|
|
|
|
def test_get_network_info(self, mock_resp):
|
|
"""Test network retrieval."""
|
|
mock_resp.return_value = {"network": {"sync_module_error": False}}
|
|
self.assertTrue(self.blink.sync["test"].get_network_info())
|
|
mock_resp.return_value = {"network": {"sync_module_error": True}}
|
|
self.assertFalse(self.blink.sync["test"].get_network_info())
|
|
|
|
def test_get_network_info_failure(self, mock_resp):
|
|
"""Test failed network retrieval."""
|
|
mock_resp.return_value = {}
|
|
self.blink.sync["test"].available = True
|
|
self.assertFalse(self.blink.sync["test"].get_network_info())
|
|
self.assertFalse(self.blink.sync["test"].available)
|
|
self.blink.sync["test"].available = True
|
|
mock_resp.return_value = None
|
|
self.assertFalse(self.blink.sync["test"].get_network_info())
|
|
self.assertFalse(self.blink.sync["test"].available)
|
|
|
|
def test_check_new_videos_startup(self, mock_resp):
|
|
"""Test that check_new_videos does not block startup."""
|
|
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",
|
|
}
|
|
]
|
|
}
|
|
|
|
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.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"},
|
|
)
|
|
|
|
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",
|
|
}
|
|
]
|
|
}
|
|
|
|
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})
|
|
|
|
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",
|
|
}
|
|
]
|
|
}
|
|
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.assertTrue(sync_module.check_new_videos())
|
|
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": [
|
|
{
|
|
"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": "/foobar.mp4",
|
|
"created_at": "1970-01-01T00:00:01+00:00",
|
|
},
|
|
]
|
|
}
|
|
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})
|
|
expected_result = {
|
|
"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}
|
|
|
|
sync_module.motion["foo"] = True
|
|
self.assertFalse(sync_module.check_new_videos())
|
|
self.assertFalse(sync_module.motion["foo"])
|
|
|
|
sync_module.motion["foo"] = True
|
|
self.assertFalse(sync_module.check_new_videos())
|
|
self.assertFalse(sync_module.motion["foo"])
|
|
|
|
sync_module.motion["foo"] = True
|
|
self.assertFalse(sync_module.check_new_videos())
|
|
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")
|
|
|
|
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())
|
|
|
|
def test_summary_with_no_network_id(self, mock_resp):
|
|
"""Test handling of bad summary."""
|
|
self.mock_start[0]["syncmodule"] = None
|
|
mock_resp.side_effect = self.mock_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}
|
|
mock_resp.side_effect = self.mock_start
|
|
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.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})
|
|
|
|
def test_missing_camera_info(self, mock_resp):
|
|
"""Test missing key from camera info response."""
|
|
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})
|
|
|
|
def test_sync_attributes(self, mock_resp):
|
|
"""Test sync attributes."""
|
|
self.assertEqual(self.blink.sync["test"].attributes["name"], "test")
|
|
self.assertEqual(self.blink.sync["test"].attributes["network_id"], "1234")
|