def _validate_settings()

in mysqlx-connector-python/lib/mysqlx/connection.py [0:0]
107 lines of code
42 McCabe index (conditional complexity)

def _validate_settings(settings: Dict[str, Any]) -> None:
    """Validates the settings to be passed to a Session object
    the port values are converted to int if specified or set to 33060
    otherwise. The priority values for each router is converted to int
    if specified.

    Args:
        settings: dict containing connection settings.

    Raises:
        :class:`mysqlx.InterfaceError`: On any configuration issue.
    """
    invalid_opts = set(settings.keys()).difference(_SESS_OPTS)
    if invalid_opts:
        invalid_opts_list = "', '".join(invalid_opts)
        raise InterfaceError(f"Invalid option(s): '{invalid_opts_list}'")

    if "routers" in settings:
        for router in settings["routers"]:
            _validate_hosts(router, 33060)
    elif "host" in settings:
        _validate_hosts(settings)

    if "ssl-mode" in settings:
        ssl_mode = settings["ssl-mode"]
        try:
            settings["ssl-mode"] = SSLMode(
                ssl_mode.lower().strip() if isinstance(ssl_mode, str) else ssl_mode
            )
        except (AttributeError, ValueError) as err:
            raise InterfaceError(f"Invalid SSL Mode '{settings['ssl-mode']}'") from err
        if "ssl-ca" not in settings and settings["ssl-mode"] in [
            SSLMode.VERIFY_IDENTITY,
            SSLMode.VERIFY_CA,
        ]:
            raise InterfaceError("Cannot verify Server without CA")

    if "ssl-crl" in settings and "ssl-ca" not in settings:
        raise InterfaceError("CA Certificate not provided")

    if "ssl-key" in settings and "ssl-cert" not in settings:
        raise InterfaceError("Client Certificate not provided")

    if "ssl-ca" in settings and settings.get("ssl-mode") not in [
        SSLMode.VERIFY_IDENTITY,
        SSLMode.VERIFY_CA,
        SSLMode.DISABLED,
    ]:
        raise InterfaceError("Must verify Server if CA is provided")

    if "auth" in settings:
        auth = settings["auth"]
        try:
            settings["auth"] = Auth(
                auth.lower().strip() if isinstance(auth, str) else auth
            )
        except (AttributeError, ValueError) as err:
            raise InterfaceError(f"Invalid Auth '{settings['auth']}'") from err

    if "compression" in settings:
        compression = settings["compression"]
        try:
            settings["compression"] = Compression(
                compression.lower().strip()
                if isinstance(compression, str)
                else compression
            )
        except (AttributeError, ValueError) as err:
            raise InterfaceError(
                "The connection property 'compression' acceptable values are: "
                "'preferred', 'required', or 'disabled'. The value "
                f"'{settings['compression']}' is not acceptable"
            ) from err

    if "compression-algorithms" in settings:
        if isinstance(settings["compression-algorithms"], str):
            compression_algorithms = (
                settings["compression-algorithms"].strip().strip("[]")
            )
            if compression_algorithms:
                settings["compression-algorithms"] = compression_algorithms.split(",")
            else:
                settings["compression-algorithms"] = None
        elif not isinstance(settings["compression-algorithms"], (list, tuple)):
            raise InterfaceError(
                "Invalid type of the connection property 'compression-algorithms'"
            )
        if settings.get("compression") == Compression.DISABLED:
            settings["compression-algorithms"] = None

    if "connection-attributes" in settings:
        _validate_connection_attributes(settings)

    if "connect-timeout" in settings:
        try:
            if isinstance(settings["connect-timeout"], str):
                settings["connect-timeout"] = int(settings["connect-timeout"])
            if (
                not isinstance(settings["connect-timeout"], int)
                or settings["connect-timeout"] < 0
            ):
                raise ValueError
        except ValueError:
            raise TypeError(
                "The connection timeout value must be a positive "
                "integer (including 0)"
            ) from None

    if "dns-srv" in settings:
        if not isinstance(settings["dns-srv"], bool):
            raise InterfaceError("The value of 'dns-srv' must be a boolean")
        if settings.get("socket"):
            raise InterfaceError(
                "Using Unix domain sockets with DNS SRV lookup is not allowed"
            )
        if settings.get("port"):
            raise InterfaceError(
                "Specifying a port number with DNS SRV lookup is not allowed"
            )
        if settings.get("routers"):
            raise InterfaceError(
                "Specifying multiple hostnames with DNS SRV look up is not allowed"
            )
    elif "host" in settings and not settings.get("port"):
        settings["port"] = 33060

    if "tls-versions" in settings:
        _validate_tls_versions(settings)

    if "tls-ciphersuites" in settings:
        _validate_tls_ciphersuites(settings)