Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def validate(doc):
base = v.Schema({
v.Optional('instance_type'): str,
v.Optional('username'): str,
v.Optional('connection'): str,
v.Optional('connection_timeout'): int,
v.Optional('inventory_args'): {str: v.Or(str, int, bool)},
v.Optional('port'): int,
v.Optional('associate_public_ip_address'): bool,
v.Optional('subnet'): str,
v.Optional('security_groups'): [str],
v.Optional('iam_instance_profile'): str,
v.Optional('tags'): {str: str},
v.Optional('user_data'): v.Or(
str,
{'file': str},
),
v.Optional('block_device_mappings'): [dict],
})
defaults = {
v.Optional('instance_type', default='t2.micro'): str,
v.Optional('username', default=DEFAULT_ANSIBLE_USER): str,
v.Optional('connection', default=DEFAULT_ANSIBLE_CONNECTION): str,
def validate_config(_config):
source_schema = voluptuous.Schema({
"module": voluptuous.And(six.string_types[0],
vu.NoSpaceCharacter()),
"min_event_per_burst": voluptuous.Or(float, int),
"sleep": voluptuous.And(
float, voluptuous.Range(
min=0, max=1, min_included=False, max_included=False)),
"transitions": {
"web_service": {
"run=>slow": {
voluptuous.And(vu.NumericString()): voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
"slow=>run": {
voluptuous.And(vu.NumericString()): voluptuous.And(
voluptuous.Or(int, float),
voluptuous.Range(min=0, max=1)),
},
"stop=>run": voluptuous.And(
def validate_config(_config):
log_reg_schema = voluptuous.Schema({
'module': voluptuous.And(six.string_types[0],
NoSpaceCharacter()),
'nb_samples': voluptuous.Or(float, int)
}, required=True)
return log_reg_schema(_config)
CONF_TRUSTED_NETWORKS = "trusted_networks"
CONF_TRUSTED_USERS = "trusted_users"
CONF_GROUP = "group"
CONF_ALLOW_BYPASS_LOGIN = "allow_bypass_login"
CONFIG_SCHEMA = AUTH_PROVIDER_SCHEMA.extend(
{
vol.Required(CONF_TRUSTED_NETWORKS): vol.All(cv.ensure_list, [ip_network]),
vol.Optional(CONF_TRUSTED_USERS, default={}): vol.Schema(
# we only validate the format of user_id or group_id
{
ip_network: vol.All(
cv.ensure_list,
[
vol.Or(
cv.uuid4_hex,
vol.Schema({vol.Required(CONF_GROUP): cv.uuid4_hex}),
)
],
)
}
),
vol.Optional(CONF_ALLOW_BYPASS_LOGIN, default=False): cv.boolean,
},
extra=vol.PREVENT_EXTRA,
)
class InvalidAuthError(HomeAssistantError):
"""Raised when try to access from untrusted networks."""
def validate_config(_config):
monasca_der_schema = voluptuous.Schema({
"module": voluptuous.And(six.string_types[0],
vu.NoSpaceCharacter()),
# Derivative period in multiple of batch interval
"period": voluptuous.And(
voluptuous.Or(float, int),
lambda i: i >= 0 and math.floor(i) == math.ceil(i))
}, required=True)
return monasca_der_schema(_config)
def validate_config(_config):
randomforest_schema = voluptuous.Schema({
'module': voluptuous.And(six.string_types[0],
NoSpaceCharacter()),
'nb_samples': voluptuous.Or(float, int)
}, required=True)
return randomforest_schema(_config)
def validate(doc):
base = v.Schema({
v.Optional('instance_type'): str,
v.Optional('username'): str,
v.Optional('connection'): str,
v.Optional('connection_timeout'): int,
v.Optional('inventory_args'): {str: v.Or(str, int, bool)},
v.Optional('port'): int,
v.Optional('associate_public_ip_address'): bool,
v.Optional('subnet'): str,
v.Optional('security_groups'): [str],
v.Optional('iam_instance_profile'): str,
v.Optional('tags'): {str: str},
v.Optional('user_data'): v.Or(
str,
{'file': str},
),
v.Optional('block_device_mappings'): [dict],
})
defaults = {
v.Optional('instance_type', default='t2.micro'): str,
v.Optional('username', default=DEFAULT_ANSIBLE_USER): str,
v.Optional('connection', default=DEFAULT_ANSIBLE_CONNECTION): str,
v.Optional('connection_timeout', default=600): int,
v.Optional('port', default=DEFAULT_ANSIBLE_PORT): int,
v.Optional('associate_public_ip_address', default=True): bool,
v.Optional('subnet', default=''): str,
v.Optional('security_groups', default=[]): [str],
v.Optional('iam_instance_profile', default=''): str,
v.Optional('tags', default={}): {str: str},
CONF_CHILDREN = "children"
CONF_COMMANDS = "commands"
CONF_SERVICE = "service"
CONF_SERVICE_DATA = "service_data"
OFF_STATES = [STATE_IDLE, STATE_OFF, STATE_UNAVAILABLE]
ATTRS_SCHEMA = cv.schema_with_slug_keys(cv.string)
CMD_SCHEMA = cv.schema_with_slug_keys(cv.SERVICE_SCHEMA)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_CHILDREN, default=[]): cv.entity_ids,
vol.Optional(CONF_COMMANDS, default={}): CMD_SCHEMA,
vol.Optional(CONF_ATTRS, default={}): vol.Or(
cv.ensure_list(ATTRS_SCHEMA), ATTRS_SCHEMA
),
vol.Optional(CONF_STATE_TEMPLATE): cv.template,
},
extra=vol.REMOVE_EXTRA,
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the universal media players."""
player = UniversalMediaPlayer(
hass,
config.get(CONF_NAME),
config.get(CONF_CHILDREN),
config.get(CONF_COMMANDS),
config.get(CONF_ATTRS),