diff --git a/.coveragerc b/.coveragerc index 8a7da26d8ca..b0b7e63f043 100644 --- a/.coveragerc +++ b/.coveragerc @@ -106,7 +106,6 @@ omit = homeassistant/components/camera/bloomsky.py homeassistant/components/camera/ffmpeg.py homeassistant/components/camera/foscam.py - homeassistant/components/camera/generic.py homeassistant/components/camera/mjpeg.py homeassistant/components/camera/rpi_camera.py homeassistant/components/climate/eq3btsmart.py diff --git a/homeassistant/components/camera/generic.py b/homeassistant/components/camera/generic.py index 91712931c07..a03acc32eb8 100644 --- a/homeassistant/components/camera/generic.py +++ b/homeassistant/components/camera/generic.py @@ -8,23 +8,38 @@ import logging import requests from requests.auth import HTTPBasicAuth, HTTPDigestAuth +import voluptuous as vol + +from homeassistant.const import CONF_NAME, CONF_USERNAME, CONF_PASSWORD +from homeassistant.exceptions import TemplateError +from homeassistant.components.camera import PLATFORM_SCHEMA, Camera +from homeassistant.helpers import config_validation as cv, template -from homeassistant.components.camera import DOMAIN, Camera -from homeassistant.helpers import validate_config _LOGGER = logging.getLogger(__name__) +CONF_AUTHENTICATION = 'authentication' +CONF_STILL_IMAGE_URL = 'still_image_url' +CONF_LIMIT_REFETCH_TO_URL_CHANGE = 'limit_refetch_to_url_change' +DEFAULT_NAME = 'Generic Camera' BASIC_AUTHENTICATION = 'basic' DIGEST_AUTHENTICATION = 'digest' +PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ + # pylint: disable=no-value-for-parameter + vol.Required(CONF_STILL_IMAGE_URL): vol.Any(vol.Url(), cv.template), + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, + vol.Optional(CONF_USERNAME): cv.string, + vol.Optional(CONF_PASSWORD): cv.string, + vol.Optional(CONF_AUTHENTICATION, default=BASIC_AUTHENTICATION): + vol.In([BASIC_AUTHENTICATION, DIGEST_AUTHENTICATION]), + vol.Optional(CONF_LIMIT_REFETCH_TO_URL_CHANGE, default=False): cv.boolean, +}) + # pylint: disable=unused-argument def setup_platform(hass, config, add_devices_callback, discovery_info=None): """Setup a generic IP Camera.""" - if not validate_config({DOMAIN: config}, {DOMAIN: ['still_image_url']}, - _LOGGER): - return None - add_devices_callback([GenericCamera(config)]) @@ -35,36 +50,47 @@ class GenericCamera(Camera): def __init__(self, device_info): """Initialize a generic camera.""" super().__init__() - self._name = device_info.get('name', 'Generic Camera') - self._authentication = device_info.get('authentication', - BASIC_AUTHENTICATION) - self._username = device_info.get('username') - self._password = device_info.get('password') - self._still_image_url = device_info['still_image_url'] + self._name = device_info.get(CONF_NAME) + self._still_image_url = device_info[CONF_STILL_IMAGE_URL] + self._limit_refetch = device_info[CONF_LIMIT_REFETCH_TO_URL_CHANGE] + + username = device_info.get(CONF_USERNAME) + password = device_info.get(CONF_PASSWORD) + + if username and password: + if device_info[CONF_AUTHENTICATION] == DIGEST_AUTHENTICATION: + self._auth = HTTPDigestAuth(username, password) + else: + self._auth = HTTPBasicAuth(username, password) + else: + self._auth = None + + self._last_url = None + self._last_image = None def camera_image(self): """Return a still image response from the camera.""" - if self._username and self._password: - if self._authentication == DIGEST_AUTHENTICATION: - auth = HTTPDigestAuth(self._username, self._password) - else: - auth = HTTPBasicAuth(self._username, self._password) - try: - response = requests.get( - self._still_image_url, - auth=auth, - timeout=10) - except requests.exceptions.RequestException as error: - _LOGGER.error('Error getting camera image: %s', error) - return None - else: - try: - response = requests.get(self._still_image_url, timeout=10) - except requests.exceptions.RequestException as error: - _LOGGER.error('Error getting camera image: %s', error) - return None + try: + url = template.render(self.hass, self._still_image_url) + except TemplateError as err: + _LOGGER.error('Error parsing template %s: %s', + self._still_image_url, err) + return self._last_image - return response.content + if url == self._last_url and self._limit_refetch: + return self._last_image + + kwargs = {'timeout': 10, 'auth': self._auth} + + try: + response = requests.get(url, **kwargs) + except requests.exceptions.RequestException as error: + _LOGGER.error('Error getting camera image: %s', error) + return None + + self._last_url = url + self._last_image = response.content + return self._last_image @property def name(self): diff --git a/homeassistant/components/camera/local_file.py b/homeassistant/components/camera/local_file.py index 463bf3eca5a..f7df934d9de 100644 --- a/homeassistant/components/camera/local_file.py +++ b/homeassistant/components/camera/local_file.py @@ -3,48 +3,48 @@ import logging import os -from homeassistant.components.camera import Camera +import voluptuous as vol + +from homeassistant.const import CONF_NAME +from homeassistant.components.camera import Camera, PLATFORM_SCHEMA +from homeassistant.helpers import config_validation as cv + +CONF_FILE_PATH = 'file_path' +DEFAULT_NAME = 'Local File' + +PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ + vol.Required(CONF_FILE_PATH): cv.isfile, + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string +}) _LOGGER = logging.getLogger(__name__) def setup_platform(hass, config, add_devices, discovery_info=None): """Setup the Camera.""" - # check for missing required configuration variable - if config.get("file_path") is None: - _LOGGER.error("Missing required variable: file_path") - return False - - setup_config = ( - { - "name": config.get("name", "Local File"), - "file_path": config.get("file_path") - } - ) + file_path = config[CONF_FILE_PATH] # check filepath given is readable - if not os.access(setup_config["file_path"], os.R_OK): + if not os.access(file_path, os.R_OK): _LOGGER.error("file path is not readable") return False - add_devices([ - LocalFile(setup_config) - ]) + add_devices([LocalFile(config[CONF_NAME], file_path)]) class LocalFile(Camera): """Local camera.""" - def __init__(self, device_info): + def __init__(self, name, file_path): """Initialize Local File Camera component.""" super().__init__() - self._name = device_info["name"] - self._config = device_info + self._name = name + self._file_path = file_path def camera_image(self): """Return image response.""" - with open(self._config["file_path"], 'rb') as file: + with open(self._file_path, 'rb') as file: return file.read() @property diff --git a/tests/components/camera/test_generic.py b/tests/components/camera/test_generic.py new file mode 100644 index 00000000000..f57ef62c0d4 --- /dev/null +++ b/tests/components/camera/test_generic.py @@ -0,0 +1,115 @@ +"""The tests for UVC camera module.""" +import unittest +from unittest import mock + +import requests_mock +from werkzeug.test import EnvironBuilder + +from homeassistant.bootstrap import setup_component +from homeassistant.components.http import request_class + +from tests.common import get_test_home_assistant + + +class TestGenericCamera(unittest.TestCase): + """Test the generic camera platform.""" + + def setUp(self): + """Setup things to be run when tests are started.""" + self.hass = get_test_home_assistant() + self.hass.wsgi = mock.MagicMock() + self.hass.config.components.append('http') + + def tearDown(self): + """Stop everything that was started.""" + self.hass.stop() + + @requests_mock.Mocker() + def test_fetching_url(self, m): + """Test that it fetches the given url.""" + self.hass.wsgi = mock.MagicMock() + m.get('http://example.com', text='hello world') + + assert setup_component(self.hass, 'camera', { + 'camera': { + 'name': 'config_test', + 'platform': 'generic', + 'still_image_url': 'http://example.com', + 'username': 'user', + 'password': 'pass' + }}) + + image_view = self.hass.wsgi.mock_calls[0][1][0] + + builder = EnvironBuilder(method='GET') + Request = request_class() + request = Request(builder.get_environ()) + request.authenticated = True + resp = image_view.get(request, 'camera.config_test') + + assert m.call_count == 1 + assert resp.status_code == 200, resp.response + assert resp.response[0].decode('utf-8') == 'hello world' + + image_view.get(request, 'camera.config_test') + assert m.call_count == 2 + + @requests_mock.Mocker() + def test_limit_refetch(self, m): + """Test that it fetches the given url.""" + self.hass.wsgi = mock.MagicMock() + from requests.exceptions import Timeout + m.get('http://example.com/5a', text='hello world') + m.get('http://example.com/10a', text='hello world') + m.get('http://example.com/15a', text='hello planet') + m.get('http://example.com/20a', status_code=404) + + assert setup_component(self.hass, 'camera', { + 'camera': { + 'name': 'config_test', + 'platform': 'generic', + 'still_image_url': + 'http://example.com/{{ states.sensor.temp.state + "a" }}', + 'limit_refetch_to_url_change': True, + }}) + + image_view = self.hass.wsgi.mock_calls[0][1][0] + + builder = EnvironBuilder(method='GET') + Request = request_class() + request = Request(builder.get_environ()) + request.authenticated = True + + self.hass.states.set('sensor.temp', '5') + + with mock.patch('requests.get', side_effect=Timeout()): + resp = image_view.get(request, 'camera.config_test') + assert m.call_count == 0 + assert resp.status_code == 500, resp.response + + self.hass.states.set('sensor.temp', '10') + + resp = image_view.get(request, 'camera.config_test') + assert m.call_count == 1 + assert resp.status_code == 200, resp.response + assert resp.response[0].decode('utf-8') == 'hello world' + + resp = image_view.get(request, 'camera.config_test') + assert m.call_count == 1 + assert resp.status_code == 200, resp.response + assert resp.response[0].decode('utf-8') == 'hello world' + + self.hass.states.set('sensor.temp', '15') + + # Url change = fetch new image + resp = image_view.get(request, 'camera.config_test') + assert m.call_count == 2 + assert resp.status_code == 200, resp.response + assert resp.response[0].decode('utf-8') == 'hello planet' + + # Cause a template render error + self.hass.states.remove('sensor.temp') + resp = image_view.get(request, 'camera.config_test') + assert m.call_count == 2 + assert resp.status_code == 200, resp.response + assert resp.response[0].decode('utf-8') == 'hello planet' diff --git a/tests/components/camera/test_local_file.py b/tests/components/camera/test_local_file.py new file mode 100644 index 00000000000..d8ec570a8b0 --- /dev/null +++ b/tests/components/camera/test_local_file.py @@ -0,0 +1,69 @@ +"""The tests for UVC camera module.""" +from tempfile import NamedTemporaryFile +import unittest +from unittest import mock + +from werkzeug.test import EnvironBuilder + +from homeassistant.bootstrap import setup_component +from homeassistant.components.http import request_class + +from tests.common import get_test_home_assistant + + +class TestLocalCamera(unittest.TestCase): + """Test the generic camera platform.""" + + def setUp(self): + """Setup things to be run when tests are started.""" + self.hass = get_test_home_assistant() + self.hass.wsgi = mock.MagicMock() + self.hass.config.components.append('http') + + def tearDown(self): + """Stop everything that was started.""" + self.hass.stop() + + def test_loading_file(self): + """Test that it loads image from disk.""" + self.hass.wsgi = mock.MagicMock() + + with NamedTemporaryFile() as fp: + fp.write('hello'.encode('utf-8')) + fp.flush() + + assert setup_component(self.hass, 'camera', { + 'camera': { + 'name': 'config_test', + 'platform': 'local_file', + 'file_path': fp.name, + }}) + + image_view = self.hass.wsgi.mock_calls[0][1][0] + + builder = EnvironBuilder(method='GET') + Request = request_class() + request = Request(builder.get_environ()) + request.authenticated = True + resp = image_view.get(request, 'camera.config_test') + + assert resp.status_code == 200, resp.response + assert resp.response[0].decode('utf-8') == 'hello' + + def test_file_not_readable(self): + """Test local file will not setup when file is not readable.""" + self.hass.wsgi = mock.MagicMock() + + with NamedTemporaryFile() as fp: + fp.write('hello'.encode('utf-8')) + fp.flush() + + with mock.patch('os.access', return_value=False): + assert setup_component(self.hass, 'camera', { + 'camera': { + 'name': 'config_test', + 'platform': 'local_file', + 'file_path': fp.name, + }}) + + assert [] == self.hass.states.all()