Streamline version (#1834)

* More streamline version handling

* Strict it up

* update tests
This commit is contained in:
Pascal Vizeli 2020-07-15 13:05:11 +02:00 committed by GitHub
parent bb64dca6e6
commit 5fdc340e58
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 77 additions and 72 deletions

View File

@ -98,6 +98,7 @@ from ..validate import (
network_port, network_port,
token, token,
uuid_match, uuid_match,
version_tag,
) )
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -180,7 +181,7 @@ def _simple_startup(value) -> str:
SCHEMA_ADDON_CONFIG = vol.Schema( SCHEMA_ADDON_CONFIG = vol.Schema(
{ {
vol.Required(ATTR_NAME): vol.Coerce(str), vol.Required(ATTR_NAME): vol.Coerce(str),
vol.Required(ATTR_VERSION): vol.Coerce(str), vol.Required(ATTR_VERSION): vol.All(version_tag, str),
vol.Required(ATTR_SLUG): vol.Coerce(str), vol.Required(ATTR_SLUG): vol.Coerce(str),
vol.Required(ATTR_DESCRIPTON): vol.Coerce(str), vol.Required(ATTR_DESCRIPTON): vol.Coerce(str),
vol.Required(ATTR_ARCH): [vol.In(ARCH_ALL)], vol.Required(ATTR_ARCH): [vol.In(ARCH_ALL)],

View File

@ -33,12 +33,12 @@ from ..const import (
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..exceptions import APIError from ..exceptions import APIError
from ..host.sound import StreamType from ..host.sound import StreamType
from ..validate import simple_version from ..validate import version_tag
from .utils import api_process, api_process_raw, api_validate from .utils import api_process, api_process_raw, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): simple_version}) SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
SCHEMA_VOLUME = vol.Schema( SCHEMA_VOLUME = vol.Schema(
{ {

View File

@ -19,12 +19,12 @@ from ..const import (
ATTR_VERSION_LATEST, ATTR_VERSION_LATEST,
) )
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..validate import simple_version from ..validate import version_tag
from .utils import api_process, api_validate from .utils import api_process, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): simple_version}) SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
class APICli(CoreSysAttributes): class APICli(CoreSysAttributes):

View File

@ -24,7 +24,7 @@ from ..const import (
) )
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..exceptions import APIError from ..exceptions import APIError
from ..validate import dns_server_list, simple_version from ..validate import dns_server_list, version_tag
from .utils import api_process, api_process_raw, api_validate from .utils import api_process, api_process_raw, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -32,7 +32,7 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
# pylint: disable=no-value-for-parameter # pylint: disable=no-value-for-parameter
SCHEMA_OPTIONS = vol.Schema({vol.Optional(ATTR_SERVERS): dns_server_list}) SCHEMA_OPTIONS = vol.Schema({vol.Optional(ATTR_SERVERS): dns_server_list})
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): simple_version}) SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
class APICoreDNS(CoreSysAttributes): class APICoreDNS(CoreSysAttributes):

View File

@ -33,7 +33,7 @@ from ..const import (
) )
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..exceptions import APIError from ..exceptions import APIError
from ..validate import complex_version, docker_image, network_port from ..validate import docker_image, network_port, version_tag
from .utils import api_process, api_process_raw, api_validate from .utils import api_process, api_process_raw, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -53,7 +53,7 @@ SCHEMA_OPTIONS = vol.Schema(
} }
) )
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): complex_version}) SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
class APIHomeAssistant(CoreSysAttributes): class APIHomeAssistant(CoreSysAttributes):

View File

@ -21,12 +21,12 @@ from ..const import (
) )
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..exceptions import APIError from ..exceptions import APIError
from ..validate import simple_version from ..validate import version_tag
from .utils import api_process, api_process_raw, api_validate from .utils import api_process, api_process_raw, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): simple_version}) SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
class APIMulticast(CoreSysAttributes): class APIMulticast(CoreSysAttributes):

View File

@ -8,12 +8,12 @@ import voluptuous as vol
from ..const import ATTR_BOARD, ATTR_BOOT, ATTR_VERSION, ATTR_VERSION_LATEST from ..const import ATTR_BOARD, ATTR_BOOT, ATTR_VERSION, ATTR_VERSION_LATEST
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..validate import complex_version from ..validate import version_tag
from .utils import api_process, api_validate from .utils import api_process, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): complex_version}) SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
class APIOS(CoreSysAttributes): class APIOS(CoreSysAttributes):

View File

@ -43,7 +43,7 @@ from ..const import (
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..exceptions import APIError from ..exceptions import APIError
from ..utils.validate import validate_timezone from ..utils.validate import validate_timezone
from ..validate import repositories, simple_version, wait_boot from ..validate import repositories, version_tag, wait_boot
from .utils import api_process, api_process_raw, api_validate from .utils import api_process, api_process_raw, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -61,7 +61,7 @@ SCHEMA_OPTIONS = vol.Schema(
} }
) )
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): simple_version}) SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
class APISupervisor(CoreSysAttributes): class APISupervisor(CoreSysAttributes):

View File

@ -3,11 +3,11 @@
import voluptuous as vol import voluptuous as vol
from ..const import ATTR_ACCESS_TOKEN, ATTR_IMAGE, ATTR_SERVERS, ATTR_VERSION from ..const import ATTR_ACCESS_TOKEN, ATTR_IMAGE, ATTR_SERVERS, ATTR_VERSION
from ..validate import dns_server_list, docker_image, simple_version, token from ..validate import dns_server_list, docker_image, token, version_tag
SCHEMA_DNS_CONFIG = vol.Schema( SCHEMA_DNS_CONFIG = vol.Schema(
{ {
vol.Optional(ATTR_VERSION): simple_version, vol.Optional(ATTR_VERSION): version_tag,
vol.Optional(ATTR_IMAGE): docker_image, vol.Optional(ATTR_IMAGE): docker_image,
vol.Optional(ATTR_SERVERS, default=list): dns_server_list, vol.Optional(ATTR_SERVERS, default=list): dns_server_list,
}, },
@ -16,17 +16,14 @@ SCHEMA_DNS_CONFIG = vol.Schema(
SCHEMA_AUDIO_CONFIG = vol.Schema( SCHEMA_AUDIO_CONFIG = vol.Schema(
{ {vol.Optional(ATTR_VERSION): version_tag, vol.Optional(ATTR_IMAGE): docker_image},
vol.Optional(ATTR_VERSION): simple_version,
vol.Optional(ATTR_IMAGE): docker_image,
},
extra=vol.REMOVE_EXTRA, extra=vol.REMOVE_EXTRA,
) )
SCHEMA_CLI_CONFIG = vol.Schema( SCHEMA_CLI_CONFIG = vol.Schema(
{ {
vol.Optional(ATTR_VERSION): simple_version, vol.Optional(ATTR_VERSION): version_tag,
vol.Optional(ATTR_IMAGE): docker_image, vol.Optional(ATTR_IMAGE): docker_image,
vol.Optional(ATTR_ACCESS_TOKEN): token, vol.Optional(ATTR_ACCESS_TOKEN): token,
}, },
@ -35,9 +32,6 @@ SCHEMA_CLI_CONFIG = vol.Schema(
SCHEMA_MULTICAST_CONFIG = vol.Schema( SCHEMA_MULTICAST_CONFIG = vol.Schema(
{ {vol.Optional(ATTR_VERSION): version_tag, vol.Optional(ATTR_IMAGE): docker_image},
vol.Optional(ATTR_VERSION): simple_version,
vol.Optional(ATTR_IMAGE): docker_image,
},
extra=vol.REMOVE_EXTRA, extra=vol.REMOVE_EXTRA,
) )

View File

@ -31,7 +31,7 @@ from ..const import (
SNAPSHOT_FULL, SNAPSHOT_FULL,
SNAPSHOT_PARTIAL, SNAPSHOT_PARTIAL,
) )
from ..validate import complex_version, docker_image, network_port, repositories from ..validate import docker_image, network_port, repositories, version_tag
ALL_FOLDERS = [FOLDER_HOMEASSISTANT, FOLDER_SHARE, FOLDER_ADDONS, FOLDER_SSL] ALL_FOLDERS = [FOLDER_HOMEASSISTANT, FOLDER_SHARE, FOLDER_ADDONS, FOLDER_SSL]
@ -58,7 +58,7 @@ SCHEMA_SNAPSHOT = vol.Schema(
vol.Inclusive(ATTR_CRYPTO, "encrypted"): CRYPTO_AES128, vol.Inclusive(ATTR_CRYPTO, "encrypted"): CRYPTO_AES128,
vol.Optional(ATTR_HOMEASSISTANT, default=dict): vol.Schema( vol.Optional(ATTR_HOMEASSISTANT, default=dict): vol.Schema(
{ {
vol.Optional(ATTR_VERSION): complex_version, vol.Optional(ATTR_VERSION): version_tag,
vol.Optional(ATTR_IMAGE): docker_image, vol.Optional(ATTR_IMAGE): docker_image,
vol.Optional(ATTR_BOOT, default=True): vol.Boolean(), vol.Optional(ATTR_BOOT, default=True): vol.Boolean(),
vol.Optional(ATTR_SSL, default=False): vol.Boolean(), vol.Optional(ATTR_SSL, default=False): vol.Boolean(),

View File

@ -54,25 +54,15 @@ sha256 = vol.Match(r"^[0-9a-f]{64}$")
token = vol.Match(r"^[0-9a-f]{32,256}$") token = vol.Match(r"^[0-9a-f]{32,256}$")
def simple_version(value: Union[str, int, None]) -> Optional[str]: def version_tag(value: Union[str, None, int, float]) -> Optional[str]:
"""Validate main version handling.""" """Validate main version handling."""
if not isinstance(value, (str, int)): if value is None:
return None
elif isinstance(value, int):
return str(value)
elif value.isnumeric() or value == "dev":
return value
return None
def complex_version(value: Union[str, None]) -> Optional[str]:
"""Validate main version handling."""
if not isinstance(value, str):
return None return None
try: try:
value = str(value)
pkg_version.parse(value) pkg_version.parse(value)
except pkg_version.InvalidVersion: except (pkg_version.InvalidVersion, TypeError):
raise vol.Invalid(f"Invalid version format {value}") raise vol.Invalid(f"Invalid version format {value}")
return value return value
@ -126,7 +116,7 @@ DOCKER_PORTS_DESCRIPTION = vol.Schema(
SCHEMA_HASS_CONFIG = vol.Schema( SCHEMA_HASS_CONFIG = vol.Schema(
{ {
vol.Optional(ATTR_UUID, default=lambda: uuid.uuid4().hex): uuid_match, vol.Optional(ATTR_UUID, default=lambda: uuid.uuid4().hex): uuid_match,
vol.Optional(ATTR_VERSION): complex_version, vol.Optional(ATTR_VERSION): version_tag,
vol.Optional(ATTR_IMAGE): docker_image, vol.Optional(ATTR_IMAGE): docker_image,
vol.Optional(ATTR_ACCESS_TOKEN): token, vol.Optional(ATTR_ACCESS_TOKEN): token,
vol.Optional(ATTR_BOOT, default=True): vol.Boolean(), vol.Optional(ATTR_BOOT, default=True): vol.Boolean(),
@ -149,13 +139,13 @@ SCHEMA_UPDATER_CONFIG = vol.Schema(
vol.Optional(ATTR_CHANNEL, default=UpdateChannels.STABLE): vol.Coerce( vol.Optional(ATTR_CHANNEL, default=UpdateChannels.STABLE): vol.Coerce(
UpdateChannels UpdateChannels
), ),
vol.Optional(ATTR_HOMEASSISTANT): complex_version, vol.Optional(ATTR_HOMEASSISTANT): vol.All(version_tag, str),
vol.Optional(ATTR_SUPERVISOR): simple_version, vol.Optional(ATTR_SUPERVISOR): vol.All(version_tag, str),
vol.Optional(ATTR_HASSOS): complex_version, vol.Optional(ATTR_HASSOS): vol.All(version_tag, str),
vol.Optional(ATTR_CLI): simple_version, vol.Optional(ATTR_CLI): vol.All(version_tag, str),
vol.Optional(ATTR_DNS): simple_version, vol.Optional(ATTR_DNS): vol.All(version_tag, str),
vol.Optional(ATTR_AUDIO): simple_version, vol.Optional(ATTR_AUDIO): vol.All(version_tag, str),
vol.Optional(ATTR_MULTICAST): simple_version, vol.Optional(ATTR_MULTICAST): vol.All(version_tag, str),
vol.Optional(ATTR_IMAGE, default=dict): vol.Schema( vol.Optional(ATTR_IMAGE, default=dict): vol.Schema(
{ {
vol.Optional(ATTR_HOMEASSISTANT): docker_image, vol.Optional(ATTR_HOMEASSISTANT): docker_image,
@ -177,7 +167,7 @@ SCHEMA_SUPERVISOR_CONFIG = vol.Schema(
{ {
vol.Optional(ATTR_TIMEZONE, default="UTC"): validate_timezone, vol.Optional(ATTR_TIMEZONE, default="UTC"): validate_timezone,
vol.Optional(ATTR_LAST_BOOT): vol.Coerce(str), vol.Optional(ATTR_LAST_BOOT): vol.Coerce(str),
vol.Optional(ATTR_VERSION, default=SUPERVISOR_VERSION): simple_version, vol.Optional(ATTR_VERSION, default=SUPERVISOR_VERSION): version_tag,
vol.Optional( vol.Optional(
ATTR_ADDONS_CUSTOM_LIST, ATTR_ADDONS_CUSTOM_LIST,
default=["https://github.com/hassio-addons/repository"], default=["https://github.com/hassio-addons/repository"],

View File

@ -1,68 +1,88 @@
"""Test validators.""" """Test validators."""
import pytest import pytest
import voluptuous.error import voluptuous as vol
import supervisor.validate from supervisor import validate
GOOD_V4 = [ DNS_GOOD_V4 = [
"dns://10.0.0.1", # random local "dns://10.0.0.1", # random local
"dns://254.254.254.254", # random high numbers "dns://254.254.254.254", # random high numbers
"DNS://1.1.1.1", # cloudflare "DNS://1.1.1.1", # cloudflare
"dns://9.9.9.9", # quad-9 "dns://9.9.9.9", # quad-9
] ]
GOOD_V6 = [ DNS_GOOD_V6 = [
"dns://2606:4700:4700::1111", # cloudflare "dns://2606:4700:4700::1111", # cloudflare
"DNS://2606:4700:4700::1001", # cloudflare "DNS://2606:4700:4700::1001", # cloudflare
] ]
BAD = ["hello world", "https://foo.bar", "", "dns://example.com"] DNS_BAD = ["hello world", "https://foo.bar", "", "dns://example.com"]
async def test_dns_url_v4_good(): async def test_dns_url_v4_good():
"""Test the DNS validator with known-good ipv6 DNS URLs.""" """Test the DNS validator with known-good ipv6 DNS URLs."""
for url in GOOD_V4: for url in DNS_GOOD_V4:
assert supervisor.validate.dns_url(url) assert validate.dns_url(url)
def test_dns_url_v6_good(): def test_dns_url_v6_good():
"""Test the DNS validator with known-good ipv6 DNS URLs.""" """Test the DNS validator with known-good ipv6 DNS URLs."""
for url in GOOD_V6: for url in DNS_GOOD_V6:
assert supervisor.validate.dns_url(url) assert validate.dns_url(url)
def test_dns_server_list_v4(): def test_dns_server_list_v4():
"""Test a list with v4 addresses.""" """Test a list with v4 addresses."""
assert supervisor.validate.dns_server_list(GOOD_V4) assert validate.dns_server_list(DNS_GOOD_V4)
def test_dns_server_list_v6(): def test_dns_server_list_v6():
"""Test a list with v6 addresses.""" """Test a list with v6 addresses."""
assert supervisor.validate.dns_server_list(GOOD_V6) assert validate.dns_server_list(DNS_GOOD_V6)
def test_dns_server_list_combined(): def test_dns_server_list_combined():
"""Test a list with both v4 and v6 addresses.""" """Test a list with both v4 and v6 addresses."""
combined = GOOD_V4 + GOOD_V6 combined = DNS_GOOD_V4 + DNS_GOOD_V6
# test the matches # test the matches
assert supervisor.validate.dns_server_list(combined) assert validate.dns_server_list(combined)
# test max_length is OK still # test max_length is OK still
assert supervisor.validate.dns_server_list(combined) assert validate.dns_server_list(combined)
# test that it fails when the list is too long # test that it fails when the list is too long
with pytest.raises(voluptuous.error.Invalid): with pytest.raises(vol.error.Invalid):
supervisor.validate.dns_server_list(combined + combined + combined + combined) validate.dns_server_list(combined + combined + combined + combined)
def test_dns_server_list_bad(): def test_dns_server_list_bad():
"""Test the bad list.""" """Test the bad list."""
# test the matches # test the matches
with pytest.raises(voluptuous.error.Invalid): with pytest.raises(vol.error.Invalid):
assert supervisor.validate.dns_server_list(BAD) assert validate.dns_server_list(DNS_BAD)
def test_dns_server_list_bad_combined(): def test_dns_server_list_bad_combined():
"""Test the bad list, combined with the good.""" """Test the bad list, combined with the good."""
combined = GOOD_V4 + GOOD_V6 + BAD combined = DNS_GOOD_V4 + DNS_GOOD_V6 + DNS_BAD
with pytest.raises(voluptuous.error.Invalid): with pytest.raises(vol.error.Invalid):
# bad list # bad list
assert supervisor.validate.dns_server_list(combined) assert validate.dns_server_list(combined)
def test_version_complex():
"""Test version simple with good version."""
for version in (
"landingpage",
"1c002dd",
"1.1.1",
"1.0",
"0.150.1",
"0.150.1b1",
"0.150.1.dev20200715",
"1",
"alpine-5.4",
1,
1.1,
):
assert validate.version_tag(version) == str(version)
assert validate.version_tag(None) is None