Skip to content

Commit

Permalink
Add tests for camera.uvc and fix bugs found in the process
Browse files Browse the repository at this point in the history
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.
  • Loading branch information
kk7ds committed Feb 22, 2016
1 parent d398832 commit 5905129
Show file tree
Hide file tree
Showing 4 changed files with 212 additions and 9 deletions.
5 changes: 4 additions & 1 deletion .coveragerc
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 13 additions & 7 deletions homeassistant/components/camera/uvc.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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):
Expand Down Expand Up @@ -93,7 +98,7 @@ def _login(self):
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
Expand All @@ -106,13 +111,14 @@ def _login(self):
_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

Expand Down
2 changes: 1 addition & 1 deletion requirements_all.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
194 changes: 194 additions & 0 deletions tests/components/camera/test_uvc.py
Original file line number Diff line number Diff line change
@@ -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()

0 comments on commit 5905129

Please sign in to comment.