From d39883211277b09be1f547690ba2d2a178a8e570 Mon Sep 17 00:00:00 2001 From: Dan Smith Date: Mon, 22 Feb 2016 13:17:53 -0800 Subject: [PATCH 1/2] Make UVC cameras honor the local camera password store, if set The NVR tells us the admin username, but not the password for the camera. Right now, we assume the default password, which obviously doesn't work for people that have changed it. The uvcclient library provides a way to set the cached admin password for a camera, which is stored in a client-specific location. We can utilize that to grab the password, falling back to the default if it's unset. With this, people just need to run a command per camera to set the admin password on their systems, if it has changed. --- homeassistant/components/camera/uvc.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/homeassistant/components/camera/uvc.py b/homeassistant/components/camera/uvc.py index a1ae128f4da..c69f1f18e1f 100644 --- a/homeassistant/components/camera/uvc.py +++ b/homeassistant/components/camera/uvc.py @@ -14,7 +14,7 @@ import requests from homeassistant.components.camera import DOMAIN, Camera from homeassistant.helpers import validate_config -REQUIREMENTS = ['uvcclient==0.6'] +REQUIREMENTS = ['uvcclient==0.8'] _LOGGER = logging.getLogger(__name__) @@ -81,18 +81,27 @@ class UnifiVideoCamera(Camera): def _login(self): from uvcclient import camera as uvc_camera + from uvcclient import store as uvc_store + caminfo = self._nvr.get_camera(self._uuid) if self._connect_addr: addrs = [self._connect_addr] else: addrs = [caminfo['host'], caminfo['internalHost']] + store = uvc_store.get_info_store() + password = store.get_camera_password(self._uuid) + if password is None: + _LOGGER.debug('Logging into camera %(name)s with default password', + dict(name=self._name)) + password = 'ubnt' + camera = None for addr in addrs: try: camera = uvc_camera.UVCCameraClient(addr, caminfo['username'], - 'ubnt') + password) camera.login() _LOGGER.debug('Logged into UVC camera %(name)s via %(addr)s', dict(name=self._name, addr=addr)) From 590512916a4f202d432cfa979384541653e960ea Mon Sep 17 00:00:00 2001 From: Dan Smith Date: Mon, 22 Feb 2016 14:06:06 -0800 Subject: [PATCH 2/2] Add tests for camera.uvc and fix bugs found in the process This adds tests for the uvc camera module. It's a good thing too, because I found a few bugs which are fixed here as well: - Graceful handling of non-integer port - Failure to take the first host that works when probing host,internalHost - Failure to detect if neither of them actually work This also converts the code to only call add_devices once with a listcomp. --- .coveragerc | 5 +- homeassistant/components/camera/uvc.py | 20 ++- requirements_all.txt | 2 +- tests/components/camera/test_uvc.py | 194 +++++++++++++++++++++++++ 4 files changed, 212 insertions(+), 9 deletions(-) create mode 100644 tests/components/camera/test_uvc.py diff --git a/.coveragerc b/.coveragerc index ba936e424e1..9eb9ee2f5b8 100644 --- a/.coveragerc +++ b/.coveragerc @@ -69,7 +69,10 @@ omit = homeassistant/components/binary_sensor/arest.py homeassistant/components/binary_sensor/rest.py homeassistant/components/browser.py - homeassistant/components/camera/* + homeassistant/components/camera/bloomsky.py + homeassistant/components/camera/foscam.py + homeassistant/components/camera/generic.py + homeassistant/components/camera/mjpeg.py homeassistant/components/device_tracker/actiontec.py homeassistant/components/device_tracker/aruba.py homeassistant/components/device_tracker/asuswrt.py diff --git a/homeassistant/components/camera/uvc.py b/homeassistant/components/camera/uvc.py index c69f1f18e1f..5a84c535540 100644 --- a/homeassistant/components/camera/uvc.py +++ b/homeassistant/components/camera/uvc.py @@ -26,8 +26,12 @@ def setup_platform(hass, config, add_devices, discovery_info=None): return None addr = config.get('nvr') - port = int(config.get('port', 7080)) key = config.get('key') + try: + port = int(config.get('port', 7080)) + except ValueError: + _LOGGER.error('Invalid port number provided') + return False from uvcclient import nvr nvrconn = nvr.UVCRemote(addr, port, key) @@ -43,10 +47,11 @@ def setup_platform(hass, config, add_devices, discovery_info=None): _LOGGER.error('Unable to connect to NVR: %s', str(ex)) return False - for camera in cameras: - add_devices([UnifiVideoCamera(nvrconn, - camera['uuid'], - camera['name'])]) + add_devices([UnifiVideoCamera(nvrconn, + camera['uuid'], + camera['name']) + for camera in cameras]) + return True class UnifiVideoCamera(Camera): @@ -93,7 +98,7 @@ class UnifiVideoCamera(Camera): password = store.get_camera_password(self._uuid) if password is None: _LOGGER.debug('Logging into camera %(name)s with default password', - dict(name=self._name)) + dict(name=self._name)) password = 'ubnt' camera = None @@ -106,13 +111,14 @@ class UnifiVideoCamera(Camera): _LOGGER.debug('Logged into UVC camera %(name)s via %(addr)s', dict(name=self._name, addr=addr)) self._connect_addr = addr + break except socket.error: pass except uvc_camera.CameraConnectError: pass except uvc_camera.CameraAuthError: pass - if not camera: + if not self._connect_addr: _LOGGER.error('Unable to login to camera') return None diff --git a/requirements_all.txt b/requirements_all.txt index b5736ef3801..c93dc535451 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -262,7 +262,7 @@ unifi==1.2.4 urllib3 # homeassistant.components.camera.uvc -uvcclient==0.6 +uvcclient==0.8 # homeassistant.components.verisure vsure==0.5.1 diff --git a/tests/components/camera/test_uvc.py b/tests/components/camera/test_uvc.py new file mode 100644 index 00000000000..1c87945cff2 --- /dev/null +++ b/tests/components/camera/test_uvc.py @@ -0,0 +1,194 @@ +""" +tests.components.camera.test_uvc +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +Tests for uvc camera module. +""" + +import socket +import unittest +from unittest import mock + +import requests +from uvcclient import camera +from uvcclient import nvr + +from homeassistant.components.camera import uvc + + +class TestUVCSetup(unittest.TestCase): + @mock.patch('uvcclient.nvr.UVCRemote') + @mock.patch.object(uvc, 'UnifiVideoCamera') + def test_setup_full_config(self, mock_uvc, mock_remote): + config = { + 'nvr': 'foo', + 'port': 123, + 'key': 'secret', + } + fake_cameras = [ + {'uuid': 'one', 'name': 'Front'}, + {'uuid': 'two', 'name': 'Back'}, + ] + hass = mock.MagicMock() + add_devices = mock.MagicMock() + mock_remote.return_value.index.return_value = fake_cameras + self.assertTrue(uvc.setup_platform(hass, config, add_devices)) + mock_remote.assert_called_once_with('foo', 123, 'secret') + add_devices.assert_called_once_with([ + mock_uvc.return_value, mock_uvc.return_value]) + mock_uvc.assert_has_calls([ + mock.call(mock_remote.return_value, 'one', 'Front'), + mock.call(mock_remote.return_value, 'two', 'Back'), + ]) + + @mock.patch('uvcclient.nvr.UVCRemote') + @mock.patch.object(uvc, 'UnifiVideoCamera') + def test_setup_partial_config(self, mock_uvc, mock_remote): + config = { + 'nvr': 'foo', + 'key': 'secret', + } + fake_cameras = [ + {'uuid': 'one', 'name': 'Front'}, + {'uuid': 'two', 'name': 'Back'}, + ] + hass = mock.MagicMock() + add_devices = mock.MagicMock() + mock_remote.return_value.index.return_value = fake_cameras + self.assertTrue(uvc.setup_platform(hass, config, add_devices)) + mock_remote.assert_called_once_with('foo', 7080, 'secret') + add_devices.assert_called_once_with([ + mock_uvc.return_value, mock_uvc.return_value]) + mock_uvc.assert_has_calls([ + mock.call(mock_remote.return_value, 'one', 'Front'), + mock.call(mock_remote.return_value, 'two', 'Back'), + ]) + + def test_setup_incomplete_config(self): + self.assertFalse(uvc.setup_platform( + None, {'nvr': 'foo'}, None)) + self.assertFalse(uvc.setup_platform( + None, {'key': 'secret'}, None)) + self.assertFalse(uvc.setup_platform( + None, {'port': 'invalid'}, None)) + + @mock.patch('uvcclient.nvr.UVCRemote') + def test_setup_nvr_errors(self, mock_remote): + errors = [nvr.NotAuthorized, nvr.NvrError, + requests.exceptions.ConnectionError] + config = { + 'nvr': 'foo', + 'key': 'secret', + } + for error in errors: + mock_remote.return_value.index.side_effect = error + self.assertFalse(uvc.setup_platform(None, config, None)) + + +class TestUVC(unittest.TestCase): + def setup_method(self, method): + self.nvr = mock.MagicMock() + self.uuid = 'uuid' + self.name = 'name' + self.uvc = uvc.UnifiVideoCamera(self.nvr, self.uuid, self.name) + self.nvr.get_camera.return_value = { + 'model': 'UVC Fake', + 'recordingSettings': { + 'fullTimeRecordEnabled': True, + }, + 'host': 'host-a', + 'internalHost': 'host-b', + 'username': 'admin', + } + + def test_properties(self): + self.assertEqual(self.name, self.uvc.name) + self.assertTrue(self.uvc.is_recording) + self.assertEqual('Ubiquiti', self.uvc.brand) + self.assertEqual('UVC Fake', self.uvc.model) + + @mock.patch('uvcclient.store.get_info_store') + @mock.patch('uvcclient.camera.UVCCameraClient') + def test_login(self, mock_camera, mock_store): + mock_store.return_value.get_camera_password.return_value = 'seekret' + self.uvc._login() + mock_camera.assert_called_once_with('host-a', 'admin', 'seekret') + mock_camera.return_value.login.assert_called_once_with() + + @mock.patch('uvcclient.store.get_info_store') + @mock.patch('uvcclient.camera.UVCCameraClient') + def test_login_no_password(self, mock_camera, mock_store): + mock_store.return_value.get_camera_password.return_value = None + self.uvc._login() + mock_camera.assert_called_once_with('host-a', 'admin', 'ubnt') + mock_camera.return_value.login.assert_called_once_with() + + @mock.patch('uvcclient.store.get_info_store') + @mock.patch('uvcclient.camera.UVCCameraClient') + def test_login_tries_both_addrs_and_caches(self, mock_camera, mock_store): + responses = [0] + + def fake_login(*a): + try: + responses.pop(0) + raise socket.error + except IndexError: + pass + + mock_store.return_value.get_camera_password.return_value = None + mock_camera.return_value.login.side_effect = fake_login + self.uvc._login() + self.assertEqual(2, mock_camera.call_count) + self.assertEqual('host-b', self.uvc._connect_addr) + + mock_camera.reset_mock() + self.uvc._login() + mock_camera.assert_called_once_with('host-b', 'admin', 'ubnt') + mock_camera.return_value.login.assert_called_once_with() + + @mock.patch('uvcclient.store.get_info_store') + @mock.patch('uvcclient.camera.UVCCameraClient') + def test_login_fails_both_properly(self, mock_camera, mock_store): + mock_camera.return_value.login.side_effect = socket.error + self.assertEqual(None, self.uvc._login()) + self.assertEqual(None, self.uvc._connect_addr) + + def test_camera_image_tries_login_bails_on_failure(self): + with mock.patch.object(self.uvc, '_login') as mock_login: + mock_login.return_value = False + self.assertEqual(None, self.uvc.camera_image()) + mock_login.assert_called_once_with() + + def test_camera_image_logged_in(self): + self.uvc._camera = mock.MagicMock() + self.assertEqual(self.uvc._camera.get_snapshot.return_value, + self.uvc.camera_image()) + + def test_camera_image_error(self): + self.uvc._camera = mock.MagicMock() + self.uvc._camera.get_snapshot.side_effect = camera.CameraConnectError + self.assertEqual(None, self.uvc.camera_image()) + + def test_camera_image_reauths(self): + responses = [0] + + def fake_snapshot(): + try: + responses.pop() + raise camera.CameraAuthError() + except IndexError: + pass + return 'image' + + self.uvc._camera = mock.MagicMock() + self.uvc._camera.get_snapshot.side_effect = fake_snapshot + with mock.patch.object(self.uvc, '_login') as mock_login: + self.assertEqual('image', self.uvc.camera_image()) + mock_login.assert_called_once_with() + self.assertEqual([], responses) + + def test_camera_image_reauths_only_once(self): + self.uvc._camera = mock.MagicMock() + self.uvc._camera.get_snapshot.side_effect = camera.CameraAuthError + with mock.patch.object(self.uvc, '_login') as mock_login: + self.assertRaises(camera.CameraAuthError, self.uvc.camera_image) + mock_login.assert_called_once_with()