def _validate_settings()

in stk-sample/lambda/stk-player-events-loader-mysql/package/mysqlx/__init__.py [0:0]
74 lines of code
37 McCabe index (conditional complexity)

def _validate_settings(settings):
    """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:
        raise InterfaceError("Invalid option(s): '{0}'"
                             "".format("', '".join(invalid_opts)))

    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:
        try:
            settings["ssl-mode"] = settings["ssl-mode"].lower()
            SSLMode.index(settings["ssl-mode"])
        except (AttributeError, ValueError):
            raise InterfaceError("Invalid SSL Mode '{0}'"
                                 "".format(settings["ssl-mode"]))
        if settings["ssl-mode"] == SSLMode.DISABLED and \
            any(key in settings for key in _SSL_OPTS):
            raise InterfaceError("SSL options used with ssl-mode 'disabled'")

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

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

    if "auth" in settings:
        try:
            settings["auth"] = settings["auth"].lower()
            Auth.index(settings["auth"])
        except (AttributeError, ValueError):
            raise InterfaceError("Invalid Auth '{0}'".format(settings["auth"]))

    if "compression" in settings:
        compression = settings["compression"].lower().strip()
        if compression not in ("preferred", "required", "disabled"):
            raise InterfaceError(
                "The connection property 'compression' acceptable values are: "
                "'preferred', 'required', or 'disabled'. The value '{0}' is "
                "not acceptable".format(settings["compression"]))
        settings["compression"] = compression

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

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

    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)