Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_json_schema_object_or_array_of_object():
# Complex test where "test" accepts either an object or an array of that object
o = {"param1": "test1", Optional("param2"): "test2"}
s = Schema({"test": Or(o, [o])})
assert s.json_schema("my-id") == {
"$schema": "http://json-schema.org/draft-07/schema#",
"$id": "my-id",
"properties": {
"test": {
"anyOf": [
{
"additionalProperties": False,
"properties": {"param1": {"const": "test1"}, "param2": {"const": "test2"}},
"required": ["param1"],
"type": "object",
},
{
"type": "array",
"items": {
"additionalProperties": False,
def get_options(cls):
return {
"name": str,
ConfigOption("description", default=None): Or(str, None),
ConfigOption("logger_level", default=logger.TEST_INFO): int,
ConfigOption("file_log_level", default=logger.DEBUG): Or(
int, None
),
ConfigOption("runpath", default=default_runpath): Or(
None, str, lambda x: callable(x)
),
ConfigOption("path_cleanup", default=True): bool,
ConfigOption("all_tasks_local", default=False): bool,
ConfigOption(
"shuffle", default=[]
): list, # list of string choices
ConfigOption(
"shuffle_seed", default=float(random.randint(1, 9999))
): float,
ConfigOption("exporters", default=None): Use(get_exporters),
ConfigOption("stdout_style", default=defaults.STDOUT_STYLE): Style,
ConfigOption("report_dir", default=defaults.REPORT_DIR): Or(
str, None
),
ConfigOption("xml_dir", default=None): Or(str, None),
int, None
),
ConfigOption("runpath", default=default_runpath): Or(
None, str, lambda x: callable(x)
),
ConfigOption("path_cleanup", default=True): bool,
ConfigOption("all_tasks_local", default=False): bool,
ConfigOption(
"shuffle", default=[]
): list, # list of string choices
ConfigOption(
"shuffle_seed", default=float(random.randint(1, 9999))
): float,
ConfigOption("exporters", default=None): Use(get_exporters),
ConfigOption("stdout_style", default=defaults.STDOUT_STYLE): Style,
ConfigOption("report_dir", default=defaults.REPORT_DIR): Or(
str, None
),
ConfigOption("xml_dir", default=None): Or(str, None),
ConfigOption("pdf_path", default=None): Or(str, None),
ConfigOption("json_path", default=None): Or(str, None),
ConfigOption("http_url", default=None): Or(str, None),
ConfigOption("pdf_style", default=defaults.PDF_STYLE): Style,
ConfigOption("report_tags", default=[]): [
Use(tagging.validate_tag_value)
],
ConfigOption("report_tags_all", default=[]): [
Use(tagging.validate_tag_value)
],
ConfigOption("merge_scheduled_parts", default=False): bool,
ConfigOption("browse", default=False): bool,
ConfigOption("ui_port", default=None): Or(None, int),
def get_options(cls):
return {
ConfigOption('http_url'): is_valid_url,
ConfigOption('timeout', default=60):
Or(None, And(Use(int), lambda n: n > 0))
}
Tags.NAME_TAG: str,
Tags.FOLDER_TAG: str,
Optional(Tags.DEADLINE_TAG,
default=MAX_DATE_STR): str,
Tags.TASKS_TAG: [{
Tags.NAME_TAG: str,
Tags.LANGUAGE_TAG: Or(LangTags.CPP, LangTags.BASH),
Tags.FOLDER_TAG: str,
Optional(Tags.OUTPUT_TYPE_TAG,
default=OutputTags.STRING): Or(OutputTags.STRING,
OutputTags.NUMBER),
Optional(Tags.COMPILER_FLAGS_TAG, default="-Wall"): str,
Optional(Tags.BINARY_NAME_TAG, default="main"): str,
Optional(Tags.PIPE_TAG, default=""): str,
Optional(Tags.BUILD_TYPE_TAG,
default=BuildTags.CMAKE): Or(BuildTags.CMAKE,
BuildTags.SIMPLE),
Optional(Tags.INJECT_FOLDER_TAG): [str],
Optional(Tags.TESTS_TAG): [{
Tags.NAME_TAG: str,
Optional(Tags.INPUT_TAG): str,
Optional(Tags.INJECT_FOLDER_TAG): [str],
Optional(Tags.RUN_GTESTS_TAG, default=False): bool,
Optional(Tags.EXPECTED_OUTPUT_TAG): Or(str, float, int)
}]
}]
}]
})
yaml = YAML()
yaml.width = 4096 # big enough value to prevent wrapping
yaml.explicit_start = True
yaml.indent(mapping=2, sequence=4, offset=2)
instance_parameters = {
'region': And(str, Regex(r'^[a-z0-9-]+$')),
Optional('availabilityZone', default=''): And(str, Regex(r'^[a-z0-9-]+$')),
Optional('subnetId', default=''): And(str, Regex(r'^subnet-[a-z0-9]+$')),
'instanceType': And(str, And(is_valid_instance_type, error='Invalid instance type.')),
Optional('onDemandInstance', default=False): bool,
Optional('amiName', default=None): And(str, len, Regex(r'^[\w\(\)\[\]\s\.\/\'@-]{3,128}$')),
Optional('amiId', default=None): And(str, len, Regex(r'^ami-[a-z0-9]+$')),
Optional('rootVolumeSize', default=0): And(Or(int, str), Use(str),
Regex(r'^\d+$', error='Incorrect value for "rootVolumeSize".'),
Use(int),
And(lambda x: x > 0,
error='"rootVolumeSize" should be greater than 0 or should '
'not be specified.'),
),
Optional('maxPrice', default=0): And(Or(float, int, str), Use(str),
Regex(r'^\d+(\.\d{1,6})?$', error='Incorrect value for "maxPrice".'),
Use(float),
And(lambda x: x > 0, error='"maxPrice" should be greater than 0 or '
'should not be specified.'),
),
Optional('managedPolicyArns', default=[]): [str],
}
volumes_checks = [
And(lambda x: len(x) < 12, error='Maximum 11 volumes are supported at the moment.'),
]
instance_checks = [
And(lambda x: not (x['onDemandInstance'] and x['maxPrice']),
error='"maxPrice" cannot be specified for on-demand instances'),
And(lambda x: not (x['amiName'] and x['amiId']),
Or(str, int): Or(
{
Or(int, "default"): And(
bool, error=_("Rules must be either `true` or `false`.")
)
},
{},
error=_("Keys under command names must be IDs (numbers) or `default`."),
)
},
{},
error=_("Keys under `COMMAND` must be command names (strings)."),
),
UseOptional(COG): Or(
{
Or(str, int): Or(
{
Or(int, "default"): And(
bool, error=_("Rules must be either `true` or `false`.")
)
},
{},
error=_("Keys under cog names must be IDs or `default`."),
)
},
{},
error=_("Keys under `COG` must be cog names (strings)."),
),
},
{},
error=_("Top-level keys must be either `COG` or `COMMAND`."),
)
def validate_instance_parameters(params: dict):
from spotty.providers.gcp.config.instance_config import VOLUME_TYPE_DISK
instance_parameters = {
'zone': And(str, Regex(r'^[a-z0-9-]+$')),
'machineType': And(str, And(is_valid_machine_type, error='Invalid instance type.')),
Optional('gpu', default=None): {
'type': str,
Optional('count', default=1): int,
},
Optional('onDemandInstance', default=False): bool,
Optional('imageName', default=None): And(str, len, Regex(r'^[\w-]+$')),
Optional('imageUrl', default=None): And(str, len, Regex(IMAGE_URL_REGEX)),
Optional('bootDiskSize', default=0): And(Or(int, str), Use(str),
Regex(r'^\d+$', error='Incorrect value for "bootDiskSize".'),
Use(int),
And(lambda x: x > 0,
error='"rootVolumeSize" should be greater than 0 or should '
'not be specified.'),
),
}
instance_checks = [
And(lambda x: not x['gpu'] or is_gpu_machine_type(x['machineType']),
error='GPU cannot be attached to shared-core or memory-optimized machine types.'),
And(lambda x: not (x['imageName'] and x['imageUrl']),
error='"imageName" and "imageUrl" parameters cannot be used together.'),
]
schema = get_instance_parameters_schema(instance_parameters, VOLUME_TYPE_DISK, instance_checks, [])
PROVIDER_BUILD_SCHEMAS = {
'codebuild': Schema(DEFAULT_CODEBUILD_BUILD),
'jenkins': Schema(JENKINS_BUILD),
}
PROVIDER_DEPLOY_SCHEMAS = {
'cloudformation': Schema(DEFAULT_CLOUDFORMATION_DEPLOY),
's3': Schema(DEFAULT_S3_DEPLOY),
'codedeploy': Schema(DEFAULT_CODEDEPLOY_DEPLOY),
'lambda': Schema(DEFAULT_LAMBDA_INVOKE),
'service_catalog': Schema(DEFAULT_SERVICECATALOG_DEPLOY),
'codebuild': Schema(DEFAULT_CODEBUILD_BUILD),
}
PROVIDER_SCHEMA = {
'source': And(
{
'provider': Or('codecommit', 'github', 's3'),
'properties': dict,
},
lambda x: PROVIDER_SOURCE_SCHEMAS[x['provider']].validate(x), #pylint: disable=W0108
),
Optional('build'): And(
{
Optional('provider'): Or('codebuild', 'jenkins'),
Optional('enabled'): bool,
Optional('properties'): dict,
},
lambda x: PROVIDER_BUILD_SCHEMAS[x.get('provider', 'codebuild')].validate(x), #pylint: disable=W0108
),
Optional('deploy'): And(
{
'provider': Or(
'cloudformation', 's3', 'codedeploy', 'lambda',
def meta_schema(self) -> schema.Schema:
"""The schema for which the recipe configuration file (``recipe.toml``)
needs to comply for common part (*i.e.* the part of the configuration
common to all the recipes)"""
return schema.Schema({
"features": [
schema.Or(*(featclass.NAME for featclass in
features.RecipeFeature.__subclasses__())),
],
# All recipes have SDK except for SDK recipes
schema.Optional("sdk"): schema.Regex(RECIPE_IDENTIFIER_RE.pattern),
# other keys must be dict and will be validated by recipe features
# validation methods
str: dict,
})