mirror of
https://github.com/home-assistant/core.git
synced 2025-07-25 06:07:17 +00:00
improve isfile validation check (#3101)
This commit is contained in:
parent
87e332c777
commit
5036bb0bc6
@ -54,7 +54,6 @@ class FileNotificationService(BaseNotificationService):
|
|||||||
|
|
||||||
if self.add_timestamp:
|
if self.add_timestamp:
|
||||||
text = '{} {}\n'.format(dt_util.utcnow().isoformat(), message)
|
text = '{} {}\n'.format(dt_util.utcnow().isoformat(), message)
|
||||||
file.write(text)
|
|
||||||
else:
|
else:
|
||||||
text = '{}\n'.format(message)
|
text = '{}\n'.format(message)
|
||||||
file.write(text)
|
file.write(text)
|
||||||
|
@ -28,10 +28,10 @@ CONF_DEBUG = 'debug'
|
|||||||
CONF_SERVER = 'server'
|
CONF_SERVER = 'server'
|
||||||
|
|
||||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
|
||||||
vol.Required(CONF_RECIPIENT): cv.string,
|
vol.Required(CONF_RECIPIENT): vol.Email,
|
||||||
vol.Optional(CONF_SERVER, default='localhost'): cv.string,
|
vol.Optional(CONF_SERVER, default='localhost'): cv.string,
|
||||||
vol.Optional(CONF_PORT, default=25): cv.port,
|
vol.Optional(CONF_PORT, default=25): cv.port,
|
||||||
vol.Optional(CONF_SENDER): cv.string,
|
vol.Optional(CONF_SENDER): vol.Email,
|
||||||
vol.Optional(CONF_STARTTLS, default=False): cv.boolean,
|
vol.Optional(CONF_STARTTLS, default=False): cv.boolean,
|
||||||
vol.Optional(CONF_USERNAME): cv.string,
|
vol.Optional(CONF_USERNAME): cv.string,
|
||||||
vol.Optional(CONF_PASSWORD): cv.string,
|
vol.Optional(CONF_PASSWORD): cv.string,
|
||||||
|
@ -1,5 +1,6 @@
|
|||||||
"""Helpers for config validation using voluptuous."""
|
"""Helpers for config validation using voluptuous."""
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
|
import os
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
from typing import Any, Union, TypeVar, Callable, Sequence, List, Dict
|
from typing import Any, Union, TypeVar, Callable, Sequence, List, Dict
|
||||||
@ -65,9 +66,17 @@ def boolean(value: Any) -> bool:
|
|||||||
return bool(value)
|
return bool(value)
|
||||||
|
|
||||||
|
|
||||||
def isfile(value):
|
def isfile(value: Any) -> str:
|
||||||
"""Validate that the value is an existing file."""
|
"""Validate that the value is an existing file."""
|
||||||
return vol.IsFile('not a file')(value)
|
if value is None:
|
||||||
|
raise vol.Invalid('None is not file')
|
||||||
|
file_in = str(value)
|
||||||
|
|
||||||
|
if not os.path.isfile(file_in):
|
||||||
|
raise vol.Invalid('not a file')
|
||||||
|
if not os.access(file_in, os.R_OK):
|
||||||
|
raise vol.Invalid('file not readable')
|
||||||
|
return file_in
|
||||||
|
|
||||||
|
|
||||||
def ensure_list(value: Union[T, Sequence[T]]) -> List[T]:
|
def ensure_list(value: Union[T, Sequence[T]]) -> List[T]:
|
||||||
|
@ -59,7 +59,7 @@ class TestLocalCamera(unittest.TestCase):
|
|||||||
fp.flush()
|
fp.flush()
|
||||||
|
|
||||||
with mock.patch('os.access', return_value=False):
|
with mock.patch('os.access', return_value=False):
|
||||||
assert setup_component(self.hass, 'camera', {
|
assert not setup_component(self.hass, 'camera', {
|
||||||
'camera': {
|
'camera': {
|
||||||
'name': 'config_test',
|
'name': 'config_test',
|
||||||
'platform': 'local_file',
|
'platform': 'local_file',
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
|
import os
|
||||||
|
import tempfile
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
@ -59,6 +61,24 @@ def test_port():
|
|||||||
schema(value)
|
schema(value)
|
||||||
|
|
||||||
|
|
||||||
|
def test_isfile():
|
||||||
|
"""Validate that the value is an existing file."""
|
||||||
|
schema = vol.Schema(cv.isfile)
|
||||||
|
|
||||||
|
with tempfile.NamedTemporaryFile() as fp:
|
||||||
|
pass
|
||||||
|
|
||||||
|
for value in ('invalid', None, -1, 0, 80000, fp.name):
|
||||||
|
with pytest.raises(vol.Invalid):
|
||||||
|
schema(value)
|
||||||
|
|
||||||
|
with tempfile.TemporaryDirectory() as tmp_path:
|
||||||
|
tmp_file = os.path.join(tmp_path, "test.txt")
|
||||||
|
with open(tmp_file, "w") as tmp_handl:
|
||||||
|
tmp_handl.write("test file")
|
||||||
|
schema(tmp_file)
|
||||||
|
|
||||||
|
|
||||||
def test_url():
|
def test_url():
|
||||||
"""Test URL."""
|
"""Test URL."""
|
||||||
schema = vol.Schema(cv.url)
|
schema = vol.Schema(cv.url)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user