Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_set_invalid_osc(hass, calls):
"""Test set invalid oscillating when fan has valid osc."""
await _register_components(hass)
# Turn on fan
await common.async_turn_on(hass, _TEST_FAN)
# Set fan's osc to True
await common.async_oscillate(hass, _TEST_FAN, True)
# verify
assert hass.states.get(_OSC_INPUT).state == "True"
_verify(hass, STATE_ON, None, True, None)
# Set fan's osc to None
with pytest.raises(vol.Invalid):
await common.async_oscillate(hass, _TEST_FAN, None)
# verify osc is unchanged
assert hass.states.get(_OSC_INPUT).state == "True"
_verify(hass, STATE_ON, None, True, None)
def valid_topic(value: Any) -> str:
"""Validate that this is a valid topic name/filter."""
value = cv.string(value)
try:
raw_value = value.encode("utf-8")
except UnicodeError:
raise vol.Invalid("MQTT topic name/filter must be valid UTF-8 string.")
if not raw_value:
raise vol.Invalid("MQTT topic name/filter must not be empty.")
if len(raw_value) > 65535:
raise vol.Invalid(
"MQTT topic name/filter must not be longer than " "65535 encoded bytes."
)
if "\0" in value:
raise vol.Invalid("MQTT topic name/filter must not contain null " "character.")
return value
def validate_cold_white_colder(value):
cw = value[CONF_COLD_WHITE_COLOR_TEMPERATURE]
ww = value[CONF_WARM_WHITE_COLOR_TEMPERATURE]
if cw > ww:
raise vol.Invalid("Cold white color temperature cannot be higher than warm white")
if cw == ww:
raise vol.Invalid("Cold white color temperature cannot be the same as warm white")
return value
@app.errorhandler(voluptuous.Invalid)
def voluptuous_errors(error):
# FIXME(sileht): remove error at payload root
payload = voluptuous_error(error)
payload["errors"] = []
if isinstance(error, voluptuous.MultipleInvalid):
payload["errors"].extend(map(voluptuous_error, error.errors))
else:
payload["errors"].extend(voluptuous_error(error))
return flask.make_response(flask.jsonify(payload), 400)
def coerce_host_port(value):
"""Validate that provided value is either just host or host:port.
Returns (host, None) or (host, port) respectively.
"""
host, _, port = value.partition(':')
if not host:
raise vol.Invalid('host cannot be empty')
if port:
port = cv.port(port)
else:
port = None
return host, port
def Timespan(value):
try:
return utils.to_timespan(value)
except ValueError as e:
raise voluptuous.Invalid(e)
def is_persistence_file(value):
"""Validate that persistence file path ends in either .pickle or .json."""
if value.endswith(('.json', '.pickle')):
return value
else:
raise vol.Invalid(
'{} does not end in either `.json` or `.pickle`'.format(value))
def valid_topic(value: Any) -> str:
"""Validate that this is a valid topic name/filter."""
value = cv.string(value)
try:
raw_value = value.encode("utf-8")
except UnicodeError:
raise vol.Invalid("MQTT topic name/filter must be valid UTF-8 string.")
if not raw_value:
raise vol.Invalid("MQTT topic name/filter must not be empty.")
if len(raw_value) > 65535:
raise vol.Invalid(
"MQTT topic name/filter must not be longer than " "65535 encoded bytes."
)
if "\0" in value:
raise vol.Invalid("MQTT topic name/filter must not contain null " "character.")
return value
def validate(obj: Dict) -> Dict:
"""Test zero keys exist or one key exists in dict."""
if not isinstance(obj, dict):
raise vol.Invalid("expected dictionary")
if len(set(keys) & set(obj)) > 1:
raise vol.Invalid("must contain at most one of {}.".format(", ".join(keys)))
return obj
def f(v):
if str(v) in action_types:
return str(v)
else:
raise Invalid(msg or 120)
return f