def parse_metadata(self, fn, key=None, base_url=None, fail_on_error=False, filter_invalid=True):
        """Parse a piece of XML and split it up into EntityDescriptor elements. Each such element
        is stored in the MDRepository instance.

:param fn: a file-like object containing SAML metadata
:param key: a certificate (file) or a SHA1 fingerprint to use for signature verification
:param base_url: use this base url to resolve relative URLs for XInclude processing
        """
        try:
            t = etree.parse(fn, base_url=base_url,
                            parser=etree.XMLParser(resolve_entities=False))
            t.xinclude()
            if filter_invalid:
                for e in t.findall('{%s}EntityDescriptor' % NS['md']):
                    if not schema().validate(e):
                        error = _e(schema().error_log, m=base_url)
                        log.debug("removing '%s': schema validation failed (%s)" % (
                            e.get('entityID'), error))
                        e.getparent().remove(e)
                        self.fire(type=EVENT_DROP_ENTITY, url=base_url,
                                  entityID=e.get('entityID'), error=error)
            else:
                # Having removed the invalid entities this should now never
                # happen...
                schema().assertValid(t)
        except DocumentInvalid, ex:
            traceback.print_exc()
            log.debug("schema validation failed on '%s': %s" % (
                base_url, _e(ex.error_log, m=base_url)))
            raise MetadataException("schema validation failed")

    def load_dir(self, directory, ext=".xml", url=None):
        """
:param directory: A directory to walk.
:param ext: Include files with this extension (default .xml)

Traverse a directory tree looking for metadata. Files ending in the specified extension are included. Directories
starting with '.' are excluded.
        """
        if url is None:
            url = directory
        log.debug("walking %s" % directory)
        if not directory in self.md:
            entities = []
            for top, dirs, files in os.walk(directory):
                for dn in dirs:
                    if dn.startswith("."):
                        dirs.remove(dn)
                for nm in files:
                    log.debug("found file %s" % nm)
                    if nm.endswith(ext):
                        fn = os.path.join(top, nm)
                        try:
                            t = self.parse_metadata(fn, fail_on_error=True)
                            # local metadata is assumed to be ok
                            entities.extend(self.entities(t))
                        except Exception, ex:
                            log.error(ex)
            self.import_metadata(self.entity_set(entities, url))

    def run(self, server):
        locked = False
        try:
            if self.lock.acquire(blocking=0):
                locked = True
                md = self.server.new_repository()
                for o in self.server.observers:
                    md.subscribe(o)

                for p in server.plumbings:
                    state = {'update': True, 'stats': {}}
                    p.process(md, state)
                    stats.update(state.get('stats', {}))
                if not md.sane():
                    log.error("update produced insane active repository - will try again later...")
                with server.lock.writelock:
                    log.debug("update produced new repository with %d entities" % md.index.size())
                    server.md = md
                    md.fire(type=EVENT_REPOSITORY_LIVE, size=md.index.size())
                    stats['Repository Update Time'] = datetime.now()
                    stats['Repository Size'] = md.index.size()
            else:
                log.error("another instance is running - will try again later...")
        except Exception, ex:
            traceback.print_exc(ex)

 def producer(q, resources, cache=self.metadata_cache_enabled):
     print resources
     for url, verify, id, tries in resources:
         log.debug("Starting fetcher for %s" % url)
         thread = URLFetch(url, verify, id, enable_cache=cache, tries=tries)
         thread.start()
         q.put(thread, True)

    def test_log_syslog(self):
        with patch('syslog.syslog', new=self.dummy_syslog):
            try:
                logger = logging.getLogger()
                old_handlers = []
                for hdl in logger.handlers:
                    logger.removeHandler(hdl)
                    old_handlers.append(hdl)
                test_handler = SysLogLibHandler("USER")
                logger.addHandler(test_handler)
                logger.setLevel(logging.WARNING)

                log.info("info")
                log.warn("warn")
                log.warning("warning")
                log.error("error")
                log.critical("critical")
                log.debug("debug")

                lines = self._syslog.getvalue().split("\n")

                assert("info" not in lines)
                assert("12:warn" in lines)
                assert("12:warning" in lines)
                assert("10:critical" in lines)
                assert("11:error" in lines)
                assert("debug" not in lines)
            finally:
                logger.removeHandler(test_handler)
                for hdl in old_handlers:
                    logger.addHandler(hdl)

    def test_log_plain(self):
        try:
            logfile = StringIO()
            logger = logging.getLogger()
            old_handlers = []
            for hdl in logger.handlers:
                logger.removeHandler(hdl)
                old_handlers.append(hdl)
            test_handler = logging.StreamHandler(logfile)
            logger.addHandler(test_handler)
            logger.setLevel(logging.WARNING)

            log.info("info")
            log.warn("warn")
            log.warning("warning")
            log.error("error")
            log.critical("critical")
            log.debug("debug")

            lines = logfile.getvalue().split("\n")

            assert("info" not in lines)
            assert("warn" in lines)
            assert("warning" in lines)
            assert("critical" in lines)
            assert("error" in lines)
            assert("debug" not in lines)
        finally:
            logger.removeHandler(test_handler)
            for hdl in old_handlers:
                logger.addHandler(hdl)

 def periodic(self, stats):
     now = _now()
     stats['Last Periodic Maintenance'] = now
     log.debug("periodic maintentance...")
     self.rc.zremrangebyscore("members", "-inf", now)
     self._drop_empty_av("collections", "members", now)
     self._drop_empty_av("attributes", "values", now)

    def lookup(self, key):
        log.debug("redis store lookup: %s" % key)
        if '+' in key:
            hk = hex_digest(key)
            if not self.rc.exists("%s#members" % hk):
                self.rc.zinterstore("%s#members" % hk, ["%s#members" % k for k in key.split('+')], 'min')
                self.rc.expire("%s#members" % hk, 30)  # XXX bad juju - only to keep clients from hammering
            return self.lookup(hk)

        m = re.match("^(.+)=(.+)$", key)
        if m:
            return self.lookup("{%s}%s" % (m.group(1), m.group(2)))

        m = re.match("^{(.+)}(.+)$", key)
        if m and ';' in m.group(2):
            hk = hex_digest(key)
            if not self.rc.exists("%s#members" % hk):
                self.rc.zunionstore("%s#members" % hk,
                                    ["{%s}%s#members" % (m.group(1), v) for v in m.group(2).split(';')], 'min')
                self.rc.expire("%s#members" % hk, 30)  # XXX bad juju - only to keep clients from hammering
            return self.lookup(hk)
        elif self.rc.exists("%s#alias" % key):
            return self.lookup(self.rc.get("%s#alias" % key))
        elif self.rc.exists("%s#metadata" % key):
            return [self._get_metadata(key)]
        else:
            return self._members(key)

    def _process(self, req):
        """The inner request pipeline processor.

        :param req: The request to run through the pipeline
        """
        log.debug('Processing \n%s' % self)
        for p in self.pipeline:
            try:
                pipe, opts, name, args = loader.load_pipe(p)
                #log.debug("traversing pipe %s,%s,%s using %s" % (pipe,name,args,opts))
                if type(args) is str or type(args) is unicode:
                    args = [args]
                if args is not None and type(args) is not dict and type(args) is not list and type(args) is not tuple:
                    raise PipeException("Unknown argument type %s" % repr(args))
                req.args = args
                req.name = name
                ot = pipe(req, *opts)
                if ot is not None:
                    req.t = ot
                    #log.debug("new state after %s: %s (done=%s)" % (pipe,req.state,req.done))
                if req.done:
                    break
            except PipeException, ex:
                log.error(ex)
                break

def when(req, condition, *values):
    """
Conditionally execute part of the pipeline.

:param req: The request
:param condition: The condition key
:param values: The condition values
:param opts: More Options (unused)
:return: None

The inner pipeline is executed if the at least one of the condition values is present for the specified key in
the request state.

**Examples**

.. code-block:: yaml

    - when foo
        - something
    - when bar bill
        - other

The condition operates on the state: if 'foo' is present in the state (with any value), then the something branch is
followed. If 'bar' is present in the state with the value 'bill' then the other branch is followed.
    """
    log.debug("condition key: %s" % repr(condition))
    c = req.state.get(condition, None)
    log.debug("condition %s" % repr(c))
    if c is not None:
        if not values or _any(values, c):
            return Plumbing(pipeline=req.args, id="%s.when" % req.plumbing.id)._process(req)
    return req.t

 def _drop_empty_av(self, attr, tag, ts):
     an = "#%s" % attr
     for c in self.rc.smembers(an):
         tn = "%s#members" % c
         self.rc.zremrangebyscore(tn, "-inf", ts)
         if not self.rc.zcard(tn) > 0:
             log.debug("dropping empty %s %s" % (attr, c))
             self.rc.srem(an, c)

 def default(self, *args, **kwargs):
     log.debug("request default: %s" % ",".join(args))
     if len(args) > 0 and args[0] in self.server.aliases:
         kwargs['pfx'] = args[0]
         if len(args) > 1:
             kwargs['path'] = args[1]
         return self.server.request(**kwargs)
     else:
         log.debug("not an alias: %s" % "/".join(args))
         kwargs['pfx'] = None
         kwargs['path'] = "/" + "/".join(args)
         return self.server.request(**kwargs)

def _select_args(req):
    args = req.args
    log.debug("selecting using args: %s" % args)
    if args is None and 'select' in req.state:
        args = [req.state.get('select')]
    if args is None:
        args = req.md.store.collections()
    if args is None or not args:
        args = req.md.store.lookup('entities')
    if args is None or not args:
        args = []

    return args

 def resolve(self, system_url, public_id, context):
     """
     Resolves URIs using the resource API
     """
     log.debug("resolve SYSTEM URL' %s' for '%s'" % (system_url, public_id))
     path = system_url.split("/")
     fn = path[len(path) - 1]
     if pkg_resources.resource_exists(__name__, fn):
         return self.resolve_file(pkg_resources.resource_stream(__name__, fn), context)
     elif pkg_resources.resource_exists(__name__, "schema/%s" % fn):
         return self.resolve_file(pkg_resources.resource_stream(__name__, "schema/%s" % fn), context)
     else:
         raise ValueError("Unable to locate %s" % fn)

    def parse_metadata(self,
                       fn,
                       key=None,
                       base_url=None,
                       fail_on_error=False,
                       filter_invalid=True,
                       validate=True,
                       post=None):
        """Parse a piece of XML and split it up into EntityDescriptor elements. Each such element
        is stored in the MDRepository instance.

:param fn: a file-like object containing SAML metadata
:param key: a certificate (file) or a SHA1 fingerprint to use for signature verification
:param base_url: use this base url to resolve relative URLs for XInclude processing
:param fail_on_error: (default: False)
:param filter_invalid: (default True) remove invalid EntityDescriptor elements rather than raise an errror
:param validate: (default: True) set to False to turn off all XML schema validation
:param post: A callable that will be called to modify the parse-tree before any validation
(but after xinclude processing)
        """
        try:
            t = etree.parse(fn, base_url=base_url, parser=etree.XMLParser(resolve_entities=False))
            t.xinclude()

            if key is not None:
                try:
                    log.debug("verifying signature using %s" % key)
                    refs = xmlsec.verified(t, key)
                    if len(refs) != 1:
                        raise MetadataException("XML metadata contains %d signatures - exactly 1 is required" % len(refs))
                    t = refs[0]  # prevent wrapping attacks
                except Exception, ex:
                    tb = traceback.format_exc()
                    print tb
                    log.error(ex)
                    return None

            if post is not None:
                t = post(t)

            if validate:
                if filter_invalid:
                    for e in t.findall('{%s}EntityDescriptor' % NS['md']):
                        if not schema().validate(e):
                            error = _e(schema().error_log, m=base_url)
                            log.debug("removing '%s': schema validation failed (%s)" % (e.get('entityID'), error))
                            e.getparent().remove(e)
                            self.fire(type=EVENT_DROP_ENTITY, url=base_url, entityID=e.get('entityID'), error=error)
                else:
                    # Having removed the invalid entities this should now never happen...
                    schema().assertValid(t)

    def _lookup(self, key):
        if key == 'entities' or key is None:
            return self.entities.values()
        if '+' in key:
            key = key.strip('+')
            #log.debug("lookup intersection of '%s'" % ' and '.join(key.split('+')))
            hits = None
            for f in key.split("+"):
                f = f.strip()
                if hits is None:
                    hits = set(self._lookup(f))
                else:
                    other = self._lookup(f)
                    hits.intersection_update(other)

                if not hits:
                    log.debug("empty intersection")
                    return []

            if hits is not None and hits:
                return list(hits)
            else:
                return []

        m = re.match("^(.+)=(.+)$", key)
        if m:
            return self._lookup("{%s}%s" % (m.group(1), m.group(2).rstrip("/")))

        m = re.match("^{(.+)}(.+)$", key)
        if m:
            res = set()
            for v in m.group(2).rstrip("/").split(';'):
                # log.debug("... adding %s=%s" % (m.group(1),v))
                res.update(self._get_index(m.group(1), v))
            return list(res)

        # log.debug("trying null index lookup %s" % key)
        l = self._get_index("null", key)
        if l:
            return list(l)

        # log.debug("trying main index lookup %s: " % key)
        if key in self.md:
            # log.debug("entities list %s: %s" % (key, self.md[key]))
            lst = []
            for entityID in self.md[key]:
                lst.extend(self.lookup(entityID))
            return lst

        return []

    def parse_metadata(self, fn, key=None, base_url=None, fail_on_error=False):
        """Parse a piece of XML and split it up into EntityDescriptor elements. Each such element
        is stored in the MDRepository instance.

:param fn: a file-like object containing SAML metadata
:param key: a certificate (file) or a SHA1 fingerprint to use for signature verification
:param base_url: use this base url to resolve relative URLs for XInclude processing
        """
        try:
            t = etree.parse(fn, base_url=base_url, parser=etree.XMLParser(resolve_entities=False))
            t.xinclude()
            schema().assertValid(t)
        except DocumentInvalid, ex:
            log.debug(_e(ex.error_log))
            raise ValueError("XML schema validation failed")

    def load_pipe(self, d):
        """Return a triple callable,name,args of the pipe specified by the object d.

        :param d: The following alternatives for d are allowed:

 - d is a string (or unicode) in which case the pipe is named d called with None as args.
 - d is a dict of the form {name: args} (i.e one key) in which case the pipe named *name* is called with args
 - d is an iterable (eg tuple or list) in which case d[0] is treated as the pipe name and d[1:] becomes the args
        """
        name = None
        args = None
        opts = []
        if type(d) is str or type(d) is unicode:
            name, opts = self._n(d)
        elif hasattr(d, '__iter__') and not type(d) is dict:
            if not len(d):
                raise PipeException("This does not look like a length of pipe... \n%s" % repr(d))
            name, opts = self._n(d[0])
        elif type(d) is dict:
            k = d.keys()[0]
            name, opts = self._n(k)
            args = d[k]
        else:
            raise PipeException("This does not look like a length of pipe... \n%s" % repr(d))

        if name is None:
            raise PipeException("Anonymous length of pipe... \n%s" % repr(d))

        mname = "pyff.pipes.builtins"
        fn = name
        if ':' in name:
            (mname, sep, fn) = name.rpartition(":")
        pm = mname
        if '.' in mname:
            (pm, sep, mn) = mname.rpartition('.')
            log.debug("importing %s from %s to find %s" % (mn, pm, fn))
        else:
            log.debug("importing %s from %s to find %s" % (mname, pm, fn))
        module = __import__(mname, fromlist=[pm])
        if hasattr(module, fn) and hasattr(getattr(module, fn), '__call__'):
            return getattr(module, fn), opts, fn, args
        elif hasattr(module, "_%s" % fn) and hasattr(getattr(module, "_%s" % fn), '__call__'):
            return getattr(module, "_%s" % fn), opts, fn, args
        else:
            raise PipeException("No such method %s in %s" % (fn, mname))

    def update(self, t, tid=None, ts=None, merge_strategy=None):  # TODO: merge ?
        log.debug("redis store update: %s: %s" % (t, tid))
        relt = root(t)
        ne = 0
        if ts is None:
            ts = int(_now() + 3600 * 24 * 4)  # 4 days is the arbitrary default expiration
        if relt.tag == "{%s}EntityDescriptor" % NS['md']:
            if tid is None:
                tid = relt.get('entityID')
            with self.rc.pipeline() as p:
                self.update_entity(relt, t, tid, ts, p)
                entity_id = relt.get("entityID")
                if entity_id is not None:
                    self.membership("entities", entity_id, ts, p)
                for ea, eav in entity_attribute_dict(relt).iteritems():
                    for v in eav:
                        # log.debug("%s=%s" % (ea, v))
                        self.membership("{%s}%s" % (ea, v), tid, ts, p)
                        p.zadd("%s#values" % ea, v, ts)
                    p.sadd("#attributes", ea)

                for hn in ('sha1', 'sha256', 'md5'):
                    tid_hash = hex_digest(tid, hn)
                    p.set("{%s}%s#alias" % (hn, tid_hash), tid)
                    if ts is not None:
                        p.expireat(tid_hash, ts)
                p.execute()
            ne += 1
        elif relt.tag == "{%s}EntitiesDescriptor" % NS['md']:
            if tid is None:
                tid = relt.get('Name')
            ts = self._expiration(relt)
            with self.rc.pipeline() as p:
                self.update_entity(relt, t, tid, ts, p)
                for e in iter_entities(t):
                    ne += self.update(e, ts=ts)
                    entity_id = e.get("entityID")
                    if entity_id is not None:
                        self.membership(tid, entity_id, ts, p)
                        self.membership("entities", entity_id, ts, p)
                p.execute()
        else:
            raise ValueError("Bad metadata top-level element: '%s'" % root(t).tag)

        return ne

    def run(self):

        def _parse_date(str):
            if str is None:
                return datetime.new()
            return datetime(*parsedate(str)[:6])

        self.start_time = clock()
        try:
            cache = httplib2.FileCache(".cache")
            headers = dict()
            if not self.enable_cache:
                headers['cache-control'] = 'no-cache'

            log.debug("fetching '%s'" % self.url)

            if self.url.startswith('file://'):
                path = self.url[7:]
                if not os.path.exists(path):
                    raise IOError("file not found: %s" % path)

                with open(path, 'r') as fd:
                    self.result = fd.read()
                    self.cached = False
                    self.date = datetime.now()
                    self.last_modified = datetime.fromtimestamp(os.stat(path).st_mtime)
            else:
                h = httplib2.Http(cache=cache, timeout=60,
                                  disable_ssl_certificate_validation=True)  # trust is done using signatures over here
                resp, content = h.request(self.url, headers=headers)
                self.resp = resp
                self.last_modified = _parse_date(resp.get('last-modified', resp.get('date', None)))
                self.date = _parse_date(resp['date'])
                if resp.status != 200:
                    raise IOError(resp.reason)
                self.result = content
                self.cached = resp.fromcache

            log.debug("got %d bytes from '%s'" % (len(self.result), self.url))
        except Exception, ex:
            #traceback.print_exc()
            #log.warn("unable to fetch '%s': %s" % (self.url, ex))
            self.ex = ex
            self.result = None