mirror of https://github.com/home-assistant/core
Add template support to generic camera + local file tests (#2881)
* Add template support to generic camera * Add tests for local file
This commit is contained in:
Paulus Schoutsen
GitHub
parent
9cfad34866
commit
aa6a0523ef
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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'
|
||||
|
|
@ -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()
|
||||
Loading…
Reference in New Issue