Python lxml.etree 模块，XPath() 实例源码

def populate_tag_for_multiple_nodes(tag, nodes, batch_size=DEFAULT_BATCH_SIZE):
    """Reevaluate a single tag for a multiple nodes.

    Presumably this tag's expression has recently changed. Use `populate_tags`
    when many nodes need reevaluating AND there are rack controllers available
    to which to farm-out work. Use this only when many nodes need reevaluating
    locally, i.e. when there are no rack controllers connected.
    """
    # Same expression, multuple documents: compile expression with XPath.
    xpath = etree.XPath(tag.definition, namespaces=tag_nsmap)
    # The XML details documents can be large so work in batches.
    for batch in gen_batches(nodes, batch_size):
        probed_details = get_probed_details(batch)
        probed_details_docs_by_node = {
            node: merge_details(probed_details[node.system_id])
            for node in batch
        }
        nodes_matching, nodes_nonmatching = classify(
            partial(try_match_xpath, xpath, logger=maaslog),
            probed_details_docs_by_node.items())
        tag.node_set.remove(*nodes_nonmatching)
        tag.node_set.add(*nodes_matching)

def test_DictCharWidget_renders_fieldset_with_label_and_field_names(self):
        names = [factory.make_string(), factory.make_string()]
        initials = []
        labels = [factory.make_string(), factory.make_string()]
        values = [factory.make_string(), factory.make_string()]
        widget = DictCharWidget(
            [widgets.TextInput, widgets.TextInput, widgets.CheckboxInput],
            names, initials, labels, skip_check=True)
        name = factory.make_string()
        html_widget = fromstring(
            '<root>' + widget.render(name, values) + '</root>')
        widget_names = XPath('fieldset/input/@name')(html_widget)
        widget_labels = XPath('fieldset/label/text()')(html_widget)
        widget_values = XPath('fieldset/input/@value')(html_widget)
        expected_names = [
            "%s_%s" % (name, widget_name) for widget_name in names]
        self.assertEqual(
            [expected_names, labels, values],
            [widget_names, widget_labels, widget_values])

def test_DictCharWidget_renders_with_empty_string_as_input_data(self):
        names = [factory.make_string(), factory.make_string()]
        initials = []
        labels = [factory.make_string(), factory.make_string()]
        widget = DictCharWidget(
            [widgets.TextInput, widgets.TextInput, widgets.CheckboxInput],
            names, initials, labels, skip_check=True)
        name = factory.make_string()
        html_widget = fromstring(
            '<root>' + widget.render(name, '') + '</root>')
        widget_names = XPath('fieldset/input/@name')(html_widget)
        widget_labels = XPath('fieldset/label/text()')(html_widget)
        expected_names = [
            "%s_%s" % (name, widget_name) for widget_name in names]
        self.assertEqual(
            [expected_names, labels],
            [widget_names, widget_labels])

def _details_prepare_merge(details):
    # We may mutate the details later, so copy now to prevent
    # affecting the caller's data.
    details = details.copy()

    # Prepare an nsmap in an OrderedDict. This ensures that lxml
    # serializes namespace declarations in a stable order.
    nsmap = OrderedDict((ns, ns) for ns in sorted(details))

    # Root everything in a namespace-less element. Setting the nsmap
    # here ensures that prefixes are preserved when dumping later.
    # This element will be replaced by the root of the lshw detail.
    # However, if there is no lshw detail, this root element shares
    # its tag with the tag of an lshw XML tree, so that XPath
    # expressions written with the lshw tree in mind will still work
    # without it, e.g. "/list//{lldp}something".
    root = etree.Element("list", nsmap=nsmap)

    # We have copied details, and root is new.
    return details, root

def _details_do_merge(details, root):
    # Merge the remaining details into the composite document.
    for namespace in sorted(details):
        xmldata = details[namespace]
        if xmldata is not None:
            try:
                detail = etree.fromstring(xmldata)
            except etree.XMLSyntaxError as e:
                maaslog.warning("Invalid %s details: %s", namespace, e)
            else:
                # Add the namespace to all unqualified elements.
                for elem in detail.iter("{}*"):
                    elem.tag = etree.QName(namespace, elem.tag)
                root.append(detail)

    # Re-home `root` in a new tree. This ensures that XPath
    # expressions like "/some-tag" work correctly. Without this, when
    # there's well-formed lshw data -- see the backward-compatibilty
    # hack futher up -- expressions would be evaluated from the first
    # root created in this function, even though that root is now the
    # parent of the current `root`.
    return etree.ElementTree(root)

def merge_details_cleanly(details):
    """Merge node details into a single XML document.

    `details` should be of the form::

      {"name": xml-as-bytes, "name2": xml-as-bytes, ...}

    where `name` is the namespace (and prefix) where each detail's XML
    should be placed in the composite document; elements in each
    detail document without a namespace are moved into that namespace.

    This is similar to `merge_details`, but the ``lshw`` detail is not
    treated specially. The result of this function is not compatible
    with XPath expressions created for old releases of MAAS.

    The returned document is always rooted with a ``list`` element.
    """
    details, root = _details_prepare_merge(details)
    return _details_do_merge(details, root)

def process_node_tags(
        rack_id, nodes, tag_name, tag_definition, tag_nsmap,
        client, batch_size=None):
    """Update the nodes for a new/changed tag definition.

    :param rack_id: System ID for the rack controller.
    :param nodes: List of nodes to process tags for.
    :param client: A `MAASClient` used to fetch the node's details via
        calls to the web API.
    :param tag_name: Name of the tag to update nodes for
    :param tag_definition: Tag definition
    :param batch_size: Size of batch
    """
    # We evaluate this early, so we can fail before sending a bunch of data to
    # the server
    xpath = etree.XPath(tag_definition, namespaces=tag_nsmap)
    system_ids = [
        node["system_id"]
        for node in nodes
    ]
    process_all(
        client, rack_id, tag_name, tag_definition, system_ids, xpath,
        batch_size=batch_size)

def match_xpath(xpath, doc):
    """Return a match of expression `xpath` against document `doc`.

    :type xpath: Either `unicode` or `etree.XPath`
    :type doc: Either `etree._ElementTree` or `etree.XPathDocumentEvaluator`

    :rtype: bool
    """
    is_xpath_compiled = is_compiled_xpath(xpath)
    is_doc_compiled = is_compiled_doc(doc)

    if is_xpath_compiled and is_doc_compiled:
        return doc(xpath.path)
    elif is_xpath_compiled:
        return xpath(doc)
    elif is_doc_compiled:
        return doc(xpath)
    else:
        return doc.xpath(xpath)

def try_match_xpath(xpath, doc, logger=logging):
    """See if the XPath expression matches the given XML document.

    Invalid XPath expressions are logged, and are returned as a
    non-match.

    :type xpath: Either `unicode` or `etree.XPath`
    :type doc: Either `etree._ElementTree` or `etree.XPathDocumentEvaluator`

    :rtype: bool
    """
    try:
        # Evaluating an XPath expression against a document with LXML
        # can return a list or a string, and perhaps other types.
        # Casting the return value into a boolean context appears to
        # be the most reliable way of detecting a match.
        return bool(match_xpath(xpath, doc))
    except etree.XPathEvalError as error:
        # Get a plaintext version of `xpath`.
        expr = xpath.path if is_compiled_xpath(xpath) else xpath
        logger.warning("Invalid expression '%s': %s", expr, str(error))
        return False

def __init__(self, content_type, *text_elements, **kwargs):
        """Building the extractor
        @param content_type: content_type of the part for which the extractor is defined
        @param text_elements: default text elements. See self.addTextElement(...)
        """
        self.content_type = content_type
        self.text_elts_xpaths = [etree.XPath('//' + te, namespaces=ns_map)
                                 for te in text_elements]
        if 'separator' in kwargs:
            self.separator = kwargs['separator']
        else:
            self.separator = ''
        return

def __init__(self, content_type, *text_elements, **kwargs):
        """Building the extractor
        @param content_type: content_type of the part for which the extractor is defined
        @param text_elements: default text elements. See self.addTextElement(...)
        """
        self.content_type = content_type
        self.text_elts_xpaths = [etree.XPath('//' + te, namespaces=ns_map)
                                 for te in text_elements]
        if 'separator' in kwargs:
            self.separator = kwargs['separator']
        else:
            self.separator = ''
        return

def precompile_xpaths(self):
        '''
        Add compiled versions of xpaths to items of self.configuration.labels.
        This avoids compilation each time each xpath is applied at each node, and saves
        a lot of time.
        '''
        try:
            assert self.configuration
        except AssertionError:
            self.logger.critical('no label configuration loaded, so cannot precompile xpaths')
            raise

        new_labels = []

        for feature_specification in self.configuration.labels:

            if feature_specification.has_key('xpath'):
                #osw# self.logger.debug('precompiling xpath %s' % feature_specification['xpath'] )
                compiled_xpath = etree.XPath(feature_specification['xpath'])
                ## overwrite the original string:
                feature_specification['xpath'] = compiled_xpath
                ## Note that it can be retrieved via path attribute: <COMPILEDXPATH>.path
            new_labels.append(feature_specification)


        ## set flag to use these instead of string xpaths when labels are made: 
        self.use_precompiled_xpaths = True

        self.configuration.labels = new_labels

def precompile_xpaths(self):
        '''
        Add compiled versions of xpaths to items of self.configuration.labels.
        This avoids compilation each time each xpath is applied at each node, and saves
        a lot of time.
        '''
        try:
            assert self.configuration
        except AssertionError:
            self.logger.critical('no label configuration loaded, so cannot precompile xpaths')
            raise

        new_labels = []

        for feature_specification in self.configuration.labels:

            if feature_specification.has_key('xpath'):
                #osw# self.logger.debug('precompiling xpath %s' % feature_specification['xpath'] )
                compiled_xpath = etree.XPath(feature_specification['xpath'])
                ## overwrite the original string:
                feature_specification['xpath'] = compiled_xpath
                ## Note that it can be retrieved via path attribute: <COMPILEDXPATH>.path
            new_labels.append(feature_specification)


        ## set flag to use these instead of string xpaths when labels are made: 
        self.use_precompiled_xpaths = True

        self.configuration.labels = new_labels

def GenerateContentInfos(factory, name, choices):
    choices_dict = {}
    for choice_name, infos in choices:
        if choice_name == "sequence":
            for element in infos["elements"]:
                if element["type"] == CHOICE:
                    element["elmt_type"] = GenerateContentInfos(factory, name, ComputeContentChoices(factory, name, element))
                elif element["name"] in choices_dict:
                    raise ValueError("'%s' element defined two times in choice" % choice_name)
                else:
                    choices_dict[element["name"]] = infos
        else:
            if choice_name in choices_dict:
                raise ValueError("'%s' element defined two times in choice" % choice_name)
            choices_dict[choice_name] = infos
    prefix = ("%s:" % factory.TargetNamespace
              if factory.TargetNamespace is not None else "")
    choices_xpath = "|".join(map(lambda x: prefix + x, choices_dict.keys()))

    def GetContentInitial():
        content_name, infos = choices[0]
        if content_name == "sequence":
            content_value = []
            for i in xrange(infos["minOccurs"]):
                for element_infos in infos["elements"]:
                    content_value.extend(GetElementInitialValue(factory, element_infos))
        else:
            content_value = GetElementInitialValue(factory, infos)
        return content_value

    return {
        "type": COMPLEXTYPE,
        "choices_xpath": etree.XPath(choices_xpath, namespaces=factory.NSMAP),
        "initial": GetContentInitial,
    }

# -------------------------------------------------------------------------------
#                           Structure extraction functions
# -------------------------------------------------------------------------------

def PLCOpen_XPath(xpath):
    return etree.XPath(xpath, namespaces=PLCOpenParser.NSMAP)

def test_mapping_assertion(self):
        """Tests the mappings for objects."""
        indicator = self.stix_package.indicators[0]

        for properties in indicator.typed_fields_with_attrnames():
            attr, tf = properties
            selector = attrmap.xmlfield(indicator, attr)
            self.assertTrue(selector, self.msg.format(indicator, attr))

            if selector == "Title":
                prefix = indicator._ID_PREFIX

                cntrl = generate_control_exp(prefix, selector)
                xpath = etree.XPath(cntrl, namespaces=self.PARSER._root.nsmap)

                result = xpath(self.PARSER._root)
                self.assertEqual(len(result), 1)

                name = xml.localname(result[0])

                self.assertEqual(name, selector)
                self.assertEqual(result[0].text, getattr(indicator, attr))

                apply_markings(self.stix_package, prefix, selector)

        # re-parse the document with marking changes.
        sio = StringIO(self.stix_package.to_xml().decode("utf-8"))
        self.PARSER = parser.MarkingParser(sio)
        package = self.PARSER.parse()

        self.assertEqual(len(package.indicators), 1)

        # See if the indicator was not marked.
        indicator = package.indicators[0]
        self.assertTrue(indicator in self.PARSER._entities)
        self.assertFalse(hasattr(indicator, api._ATTR_DATA_MARKINGS))

        title = indicator.title
        self.assertTrue(isinstance(title, (types.MarkableText, types.MarkableBytes)))
        self.assertEqual(len(title.__datamarkings__), 1)

def clean_definition(self):
        definition = self.cleaned_data['definition']
        if not definition:
            return ""
        try:
            etree.XPath(definition)
        except etree.XPathSyntaxError as e:
            raise ValidationError('Invalid xpath expression: %s' % (e,))
        return definition

def clean_definition(self):
        if self.is_defined:
            try:
                etree.XPath(self.definition)
            except etree.XPathSyntaxError as e:
                msg = 'Invalid XPath expression: %s' % (e,)
                raise ValidationError({'definition': [msg]})

def test_classify_evaluates_xpath(self):
        # Yay, something that doesn't need patching...
        xpath = etree.XPath('//node')
        xml = etree.fromstring
        node_details = [
            ('a', xml('<node />')),
            ('b', xml('<not-node />')),
            ('c', xml('<parent><node /></parent>')),
        ]
        self.assertEqual(
            (['a', 'c'], ['b']),
            tags.classify(xpath, node_details))

def test_logs_to_specified_logger(self):
        xpath = etree.XPath("/foo:bar")
        doc = etree.XML("<foo/>")
        root_logger = self.useFixture(FakeLogger())
        callers_logger = Mock()
        try_match_xpath(xpath, doc, callers_logger)
        self.assertEqual("", root_logger.output)
        self.assertThat(
            callers_logger.warning,
            MockCalledOnceWith(
                "Invalid expression '%s': %s",
                '/foo:bar', 'Undefined namespace prefix'))

def is_compiled_xpath(xpath):
    """Is `xpath` a compiled expression?"""
    return isinstance(xpath, etree.XPath)

def is_compiled_doc(doc):
    """Is `doc` a compiled XPath document evaluator?"""
    return isinstance(doc, etree.XPathDocumentEvaluator)

def __init__(self, content_type, *text_elements, **kwargs):
        """Building the extractor
        @param content_type: content_type of the part for which the extractor is defined
        @param text_elements: default text elements. See self.addTextElement(...)
        """
        self.content_type = content_type
        self.text_elts_xpaths = [etree.XPath('//' + te, namespaces=ns_map)
                                 for te in text_elements]
        if 'separator' in kwargs:
            self.separator = kwargs['separator']
        else:
            self.separator = ''
        return

def final_attribute_name(xpath):
    """
    Find the final text element of an xpath which we will assume is the name
    of an attribute.    

    TODO: find a better and less error-prone way to do this!
    """
    if type(xpath) == XPath: ## in case compiled:
        pathstring = xpath.path
    else:
        pathstring = xpath
    fragments = re.split("[/:@\(\)]+", pathstring)  
    return fragments[-1]

def __init__(self):
        sections_str = {"codefile_name": self.CODEFILE_NAME}
        if "includes" in self.SECTIONS_NAMES:
            sections_str["includes_section"] = SECTION_TAG_ELEMENT % "includes"
        else:
            sections_str["includes_section"] = ""
        sections_str["sections"] = "\n".join(
            [SECTION_TAG_ELEMENT % name
             for name in self.SECTIONS_NAMES if name != "includes"])

        self.CodeFileParser = GenerateParserFromXSDstring(
            CODEFILE_XSD % sections_str)
        self.CodeFileVariables = etree.XPath("variables/variable")

        filepath = self.CodeFileName()

        if os.path.isfile(filepath):
            xmlfile = open(filepath, 'r')
            codefile_xml = xmlfile.read()
            xmlfile.close()

            codefile_xml = codefile_xml.replace(
                '<%s>' % self.CODEFILE_NAME,
                '<%s xmlns:xhtml="http://www.w3.org/1999/xhtml">' % self.CODEFILE_NAME)
            for cre, repl in [
                    (re.compile("(?<!<xhtml:p>)(?:<!\[CDATA\[)"), "<xhtml:p><![CDATA["),
                    (re.compile("(?:]]>)(?!</xhtml:p>)"), "]]></xhtml:p>")]:
                codefile_xml = cre.sub(repl, codefile_xml)

            try:
                self.CodeFile, error = self.CodeFileParser.LoadXMLString(codefile_xml)
                if error is not None:
                    (fname, lnum, src) = ((self.CODEFILE_NAME,) + error)
                    self.GetCTRoot().logger.write_warning(XSDSchemaErrorMessage.format(a1=fname, a2=lnum, a3=src))
                self.CreateCodeFileBuffer(True)
            except Exception, exc:
                msg = _("Couldn't load confnode parameters {a1} :\n {a2}").format(a1=CTNName, a2=unicode(exc))
                self.GetCTRoot().logger.write_error(msg)
                self.GetCTRoot().logger.write_error(traceback.format_exc())
        else:
            self.CodeFile = self.CodeFileParser.CreateRoot()
            self.CreateCodeFileBuffer(False)
            self.OnCTNSave()

def css_to_func(css, flags, css_namespaces, lang):
    """Convert a css selector to an xpath, supporting pseudo elements."""
    from cssselect import parse, HTMLTranslator
    from cssselect.parser import FunctionalPseudoElement
    #  FIXME HACK need lessc to support functional-pseudo-selectors instead
    #  of marking as strings and stripping " here.
    if not (css):
        return None

    sel = parse(css.strip('" '))[0]
    xpath = HTMLTranslator().selector_to_xpath(sel)

    first_letter = False
    if sel.pseudo_element is not None:
        if type(sel.pseudo_element) == FunctionalPseudoElement:
            if sel.pseudo_element.name in ('attr', 'first-letter'):
                xpath += '/@' + sel.pseudo_element.arguments[0].value
                if sel.pseudo_element.name == 'first-letter':
                    first_letter = True
        elif type(sel.pseudo_element) == unicode:
            if sel.pseudo_element == 'first-letter':
                first_letter = True

    xp = etree.XPath(xpath, namespaces=css_namespaces)

    def toupper(u):
        """Use icu library for locale sensitive uppercasing (python2)."""
        loc = Locale(lang) if lang else Locale()
        return unicode(UnicodeString(u).toUpper(loc))

    def func(elem):
        res = xp(elem)
        if res:
            if etree.iselement(res[0]):
                res_str = etree.tostring(res[0], encoding='unicode',
                                         method="text")
            else:
                res_str = res[0]
            if first_letter:
                if res_str:
                    if flags and 'nocase' in flags:
                        return toupper(res_str[0])
                    else:
                        return res_str[0]
                else:
                    return res_str
            else:
                if flags and 'nocase' in flags:
                    return toupper(res_str)
                else:
                    return res_str

    return func