Viewing file:  validations.py (3.51 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
"""
This module contains validation schema for config file loaded values
"""
import json
from argparse import Namespace
from schema import Schema, And, Optional, Use, SchemaError, Or
from ..internal.exceptions import SSAError
cmdline_schema = Schema({
'command': And(str, lambda c: c in ('set-config', 'get-config',
'get-ssa-status', 'get-report',
'enable-ssa', 'disable-ssa',
'get-stat'),
error='Invalid command'),
Optional('domains_number'): And(Use(lambda n: int(n)),
lambda n: n >= 0,
error='number of domains should be a positive integer'),
Optional('urls_number'): And(Use(lambda n: int(n)),
lambda n: n > 0,
error='number of URLs should be a positive integer'),
Optional('requests_duration'): And(Use(lambda n: float(n)),
lambda n: n > 0.0,
error='request duration should be positive'),
Optional('request_number'): And(Use(lambda n: int(n)),
lambda n: n > 0,
error='number of requests should be a positive integer'),
Optional('time'): And(Use(lambda n: int(n)),
lambda n: 0 <= n <= 12,
error='time should be a positive integer between 1 and 12, or 0'),
Optional('correlation'): And(Use(lambda s: s.capitalize()), lambda s: s in ('On', 'Off'),
error='bad correlation value, should be On or Off'),
Optional('correlation_coefficient'): And(Use(lambda n: float(n)),
lambda n: 0.0 < n <= 1.0,
error='correlation coefficient could not be negative or greater than 1'),
Optional('ignore_list'): Use(lambda s: str(s)),
Optional('summary_notification_enabled'): And(
Use(lambda s: s.capitalize()), lambda s: s in ('On', 'Off'),
error='bad notification value, should be On or Off'),
})
autotracing_cmdline_schema = Schema({
'command': And(str, lambda c: c in ('enable', 'disable', 'status'),
error='Invalid autotracing command'),
Optional('user'): Or(str, None),
Optional('all'): bool,
Optional('list_disabled'): bool
})
def validate(data: dict) -> tuple:
"""
Validate given data.
Return: tuple with 2 elements: command string and list of the validated options
"""
try:
d = cmdline_schema.validate(data)
cmd = d['command']
del d['command']
return cmd, d
except SchemaError as e:
raise SSAError(f'Invalid configuration: {str(e)}')
def validate_autotracing(input_args: dict) -> Namespace:
"""
Validate given input with schema
Input arguments expected in s dict form
:param input_args: dict with input data
"""
try:
return Namespace(**(autotracing_cmdline_schema.validate(input_args)))
except SchemaError as e:
# logger.error('Input validation error', extra={'err': str(e)})
raise SystemExit(json.dumps({
'result': f'Input validation error: {str(e)}'
}))
|