1#@!#!123s

: / opt / cloudlinux / venv / lib64 / python3.11 / site-packages / clpackages /
Filename : cl_packages_arg_parse.py
back
# -*- coding: utf-8 -*-

# Command line arguments parser for cloudlinux-packages utility

# cloudlinux-packages Utility to control lvectl packages
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from docopt import DocoptExit, docopt
from schema import Schema, And, Or, Use, SchemaError

from clcommon.cpapi import is_panel_feature_supported
from clcommon.features import Feature

PROG_NAME = "cloudlinux-packages"

GET_USAGE_CMDS = (
    "[get] [--json] [--provider <str> | --for-reseller <str>] [--package <str>] "
    "[--limits=<keys>] [--human-readable-numbers]",
)

SET_USAGE_NO_LVE_CMDS = (
    "set [--json] (--provider <str> --package <str>) [--inodes <N,M>]",

    "set [--json] (--for-reseller <str> --package <str>) [--inodes <N,M>]",
)
SET_USAGE_LVE_CMDS = (
    "set [--json] (--provider <str> --package <str>) "
    "[--speed <str> --pmem <str> --vmem <str> --nproc <str> --maxEntryProcs <str> "
    "--io <str> --iops <str> --inodes <N,M> --mysql-cpu <int> --mysql-io <int>]",

    "set [--json] (--for-reseller <str> --package <str>) "
    "[--speed <str> --pmem <str> --vmem <str> --nproc <str> --maxEntryProcs <str> "
    "--io <str> --iops <str> --inodes <N,M>]",
)

USAGE_TEMPLATE = """
Usage:
    {get_usage}
    {set_usage}
    {prog_name} (-h | --help)
""".strip()

LIMITS_OPTIONS_LVE = """
    --speed <str>                       Limit CPU usage for LVE.
    --pmem <str>                        Limit physical memory usage for applications inside LVE.
    --vmem <str>                        Limit virtual memory for applications inside LVE.
    --nproc <str>                       Limit number of processes for LVE.
    --maxEntryProcs <str>               Limit number of entry processes for LVE.
    --io <str>                          Define io limits for LVE (KB/s).
    --iops <str>                        Limit io per second for LVE.
    --mysql-cpu <int>                   Set MySQL governor CPU limit (pct).
    --mysql-io <int>                    Set MySQL governor IO limit (read + write MB/s)
""".rstrip()  # strip only right side to preserve the newline symbol

OPTIONS_TEMPLATE = """
Options:
    --json                              Return data in JSON format.
    --provider <str>                    Show data only for specific reseller.
    --for-reseller <str>                Show data only for specific reseller.
    --package <str>                     Show data only for specific package. Use only with --provider
                                        WARNING: package name must be unicode-escaped string
    --limits <keys>                     Available keys: speed, nproc, pmem, vmem, maxEntryProcs, io, iops, inodes
    --human-readable-numbers            Return PMEM and VMEM limits in KBytes, MBytes or GBytes {limit_options}
    --inodes <N,M>                      Set inode limits. N - soft, M - hard.
    -h, --help                          Show this help message and exit
""".strip()

DOCSTRING_TEMPLATE = """Utility to get/set any Cloudlinux package limits

{usage}

{options}
"""


def _limits_keys_validate(keys):
    """
    Validate limits keys
    """
    avialable_keys_list = ["inodes"]
    if is_panel_feature_supported(Feature.LVE):
        avialable_keys_list += [
            "speed", "nproc", "pmem", "vmem", "maxEntryProcs", "io", "iops", "mysql-cpu", "mysql-io"
        ]

    return len(set(keys.split(",")) - set(avialable_keys_list)) == 0


def _get_commands_usage_template(commands):
    """
    Get usage for commands
    """
    return "\n    ".join(f"{{prog_name}} {cmd}" for cmd in commands).strip()


SCHEMA_NO_LVE = {
    "get": bool,
    "set": bool,
    "--json": And(bool, lambda x: x, error="use --json option, other modes currently unsupported"),
    "--provider": Or(None, str),
    "--for-reseller": Or(None, str),
    "--package": Or(None, str),
    "--limits": Or(None, _limits_keys_validate, error="Invalid keys"),
    "--inodes": Or(None, str),
    "--help": bool,
    "--human-readable-numbers": bool,
}

SCHEMA_LVE = SCHEMA_NO_LVE | {
    "--speed": Or(None, str),
    "--pmem": Or(None, str),
    "--vmem": Or(None, str),
    "--nproc": Or(None, str),
    "--maxEntryProcs": Or(None, str),
    "--io": Or(None, str),
    "--iops": Or(None, str),
    "--mysql-cpu": Or(None, And(Use(int), lambda x: x >= 0),
                    error="--mysql-cpu must be non-negative integer value"),
    "--mysql-io": Or(None, And(Use(int), lambda x: x >= 0),
                    error="--mysql-io must be non-negative integer value"),
}


def parse_cloudlinux_packages_opts(argv, _is_json_need=False):
    """
    Parse arguments for cloudlinux-packages command
    :param argv: sys.argv
    :param _is_json_need: sys.argv contains --json key
    :return cortege: (error_flag, s_message)
    """
    docstring = DOCSTRING_TEMPLATE.format(
        usage=USAGE_TEMPLATE.format(
            get_usage=_get_commands_usage_template(GET_USAGE_CMDS),
            set_usage=_get_commands_usage_template(
                SET_USAGE_LVE_CMDS if is_panel_feature_supported(Feature.LVE) else SET_USAGE_NO_LVE_CMDS,
            ),
            prog_name=PROG_NAME,
        ),
        options=OPTIONS_TEMPLATE.format(
            limit_options=LIMITS_OPTIONS_LVE if is_panel_feature_supported(Feature.LVE) else "",
        )
    ).format(prog_name=PROG_NAME)

    try:
        args = docopt(docstring, argv)
    except DocoptExit:
        s_error_string = 'ERROR: Invalid parameter passed'
        if not _is_json_need:
            s_error_string += "\n\n" + docstring
        return False, s_error_string
    # get mode by default
    if not args["get"] and not args["set"]:
        args["get"] = True
    if args['--for-reseller']:
        # --for-reseller option found, convert it to --provider
        args['--provider'] = args['--for-reseller']

    schema_dict = SCHEMA_LVE if is_panel_feature_supported(Feature.LVE) else SCHEMA_NO_LVE
    s = Schema(schema_dict)

    try:
        args = s.validate(args)
        status = True
    except SchemaError as e:
        args = str(e)
        status = False
    return status, args