def create_ruleset()

in python/qpid_dispatch_internal/policy/policy_local.py [0:0]

• 56 lines of code
• 25 McCabe index (conditional complexity)

    def create_ruleset(self, attributes):
        """
        Create or update named policy ruleset.
        @param[in] attributes: from config
        """
        warnings = []
        diag = []
        candidate = {}
        name = attributes[PolicyKeys.KW_VHOST_NAME]
        result = self._policy_compiler.compile_access_ruleset(name, attributes, candidate, warnings, diag)

        if not result:
            raise PolicyError("Policy '%s' is invalid: %s" % (name, diag[0]))
        if len(warnings) > 0:
            for warning in warnings:
                self._manager.log_warning(warning)

        # Reject if any vhost alias name conflicts
        if name in self._vhost_aliases:
            # hostname is an alias
            raise PolicyError(
                "Policy is creating vhost '%s' but that name is already an alias for vhost '%s'" % (name, self._vhost_aliases[name]))
        for vhost_alias in candidate[PolicyKeys.KW_VHOST_ALIASES]:
            # alias is a hostname
            if vhost_alias in self.rulesetdb.keys():
                raise PolicyError(
                    "Policy for vhost '%s' defines alias '%s' which conflicts with an existing vhost named '%s'" % (name, vhost_alias, vhost_alias))
        if name not in self.rulesetdb:
            # Creating new ruleset. Vhost aliases cannot overlap
            for vhost_alias in candidate[PolicyKeys.KW_VHOST_ALIASES]:
                if vhost_alias in self._vhost_aliases:
                    raise PolicyError(
                        "Policy for vhost '%s' alias '%s' conflicts with existing alias for vhost '%s'" % (name, vhost_alias, self._vhost_aliases[vhost_alias]))
        else:
            # Updating an existing ruleset.
            # Vhost aliases still cannot overlap but replacement is allowed
            for vhost_alias in candidate[PolicyKeys.KW_VHOST_ALIASES]:
                if vhost_alias in self._vhost_aliases and not self._vhost_aliases[vhost_alias] == name:
                    raise PolicyError(
                        "Policy for vhost '%s' alias '%s' conflicts with existing alias for vhost '%s'" % (name, vhost_alias, self._vhost_aliases[vhost_alias]))

        # Reject if parse tree optimized name collision
        # Coincidently add name and aliases to parse tree
        if self.use_hostname_patterns:
            agent = self._manager.get_agent()
            # construct a list of names to be added
            tnames = []
            tnames.append(name)
            tnames += candidate[PolicyKeys.KW_VHOST_ALIASES]
            # create a list of names to undo in case a subsequent name does not work
            snames = []
            for tname in tnames:
                if not agent.qd.qd_dispatch_policy_host_pattern_add(agent.dispatch, tname):
                    # undo the snames list
                    for sname in snames:
                        agent.qd.qd_dispatch_policy_host_pattern_del(agent.dispatch, sname)
                    raise PolicyError("Policy for vhost '%s' alias '%s' optimized pattern conflicts with existing pattern" % (name, tname))
                snames.append(tname)
        # Names pass administrative approval
        if name not in self.rulesetdb:
            # add new aliases
            for nname in candidate[PolicyKeys.KW_VHOST_ALIASES]:
                self._vhost_aliases[nname] = name
            if name not in self.statsdb:
                self.statsdb[name] = AppStats(name, self._manager, candidate)
            self._manager.log_info("Created policy rules for vhost %s" % name)
        else:
            # remove old aliases
            old_aliases = self.rulesetdb[name][PolicyKeys.KW_VHOST_ALIASES]
            for oname in old_aliases:
                del self._vhost_aliases[oname]
            # add new aliases
            for nname in candidate[PolicyKeys.KW_VHOST_ALIASES]:
                self._vhost_aliases[nname] = name
            self.statsdb[name].update_ruleset(candidate)
            self._manager.log_info("Updated policy rules for vhost %s" % name)
        # TODO: ruleset lock
        self.rulesetdb[name] = {}
        self.rulesetdb[name].update(candidate)