Add tests for element and type substitution

brunato · brunato · commit 3cc1ab6bbc14 · 2025-05-11T17:29:12.000+02:00
- Fix set schema in XPathContext
  - Add get_attribute_type() and get_child_type() to schema proxy
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -2,9 +2,9 @@
 CHANGELOG
 *********
 
-`v5.0.1`_ (2025-05-03)
+`v5.0.1`_ (2025-05-11)
 ======================
-* Fix XDM type labeling with substitutions (issue #89)
+* Fix XDM type labeling with element and xsi:type substitutions (issue #89)
 
 `v5.0.0`_ (2025-04-27)
 ======================
diff --git a/elementpath/schema_proxy.py b/elementpath/schema_proxy.py
@@ -9,11 +9,13 @@
 #
 from abc import ABCMeta, abstractmethod
 from collections.abc import Iterator
+from functools import lru_cache
 from typing import cast, Any, Optional, Union
 
 from elementpath.exceptions import ElementPathTypeError
 from elementpath.protocols import XsdTypeProtocol, XsdAttributeProtocol, \
-    XsdElementProtocol, XsdSchemaProtocol
+    XsdElementProtocol, XsdSchemaProtocol, XsdAttributeGroupProtocol
+from elementpath.namespaces import XSD_ANY_ATOMIC_TYPE
 from elementpath.datatypes import AtomicType
 from elementpath.etree import is_etree_element
 from elementpath.xpath_context import XPathSchemaContext
@@ -113,6 +115,16 @@ def find(self, path: str, namespaces: Optional[dict[str, str]] = None) \
             return self._base_element.find(path, namespaces=namespaces)
         return self._schema.find(path, namespaces)
 
+    @lru_cache(maxsize=None)
+    def cached_find(self, expanded_path: str) -> Optional[PathResult]:
+        """
+        Find a schema element or attribute using an expanded XPath expression.
+        Currently unused in the code, it may be deprecated in the future.
+        """
+        if self._base_element is not None:
+            return self._base_element.find(expanded_path)
+        return self._schema.find(expanded_path)
+
     @property
     def xsd_version(self) -> str:
         """The XSD version, returns '1.0' or '1.1'."""
@@ -131,49 +143,23 @@ def get_type(self, qname: str) -> Optional[XsdTypeProtocol]:
         """
         return self._schema.maps.types.get(qname)
 
-    def get_attribute(self, name: str, xsd_type: Optional[XsdTypeProtocol] = None) \
-            -> Optional[XsdAttributeProtocol]:
+    def get_attribute(self, qname: str) -> Optional[XsdAttributeProtocol]:
         """
-        Get the XSD attribute if it's matching in the provided scope, otherwise return None.
-
-        :param name: the name of the attribute to retrieve.
-        :param xsd_type: an optional XSD type that represents the scope where matching \
-        the attribute name. If not provided, the scope is assumed to be the global scope.
-        :returns: an object that represents an XSD attribute or `None`.
-        """
-        if xsd_type is None:
-            return self._schema.maps.attributes.get(name)
-        elif hasattr(xsd_type, 'attributes'):
-            if name in xsd_type.attributes:
-                return cast(XsdAttributeProtocol, xsd_type.attributes[name])
-            elif None not in xsd_type.attributes or \
-                    not xsd_type.attributes[None].is_matching(name):
-                return None
-
-        return self._schema.maps.attributes.get(name)
+        Get the XSD global attribute from the schema's scope.
 
-    def get_element(self, name: str, xsd_type: Optional[XsdTypeProtocol] = None) \
-            -> Optional[XsdElementProtocol]:
+        :param qname: the fully qualified name of the attribute to retrieve.
+        :returns: an object that represents an XSD type or `None`.
         """
-        Get the XSD element if it's matching in the provided scope, otherwise return None.
+        return self._schema.maps.attributes.get(qname)
 
-        :param name: the name of the element to match.
-        :param xsd_type: an optional XSD type that represents the scope where matching \
-        the element name. If not provided, the scope is assumed to be the global scope.
-        :returns: an object that represents an XSD element or `None`.
+    def get_element(self, qname: str) -> Optional[XsdElementProtocol]:
         """
-        if xsd_type is None:
-            return self._schema.maps.elements.get(name)
+        Get the XSD global element from the schema's scope.
 
-        content = xsd_type.model_group
-        if content is not None:
-            for xsd_element in content.iter_elements():
-                if xsd_element.is_matching(name):
-                    if xsd_element.name != name:
-                        # a wildcard or a substitute
-                        return self._schema.maps.elements.get(name)
-                    return xsd_element
-        return None
+        :param qname: the fully qualified name of the element to retrieve.
+        :returns: an object that represents an XSD type or `None`.
+        """
+        return self._schema.maps.elements.get(qname)
 
     def get_substitution_group(self, qname: str) -> Optional[set[XsdElementProtocol]]:
         """
@@ -185,6 +171,63 @@ def get_substitution_group(self, qname: str) -> Optional[set[XsdElementProtocol]
         """
         return self._schema.maps.substitution_groups.get(qname)
 
+    def get_attribute_type(self, name: str, parent_type: Optional[XsdTypeProtocol] = None) \
+            -> Optional[XsdTypeProtocol]:
+        """
+        Get the XSD attribute type if the provided name is matching in the scope,
+        otherwise return None.
+
+        :param name: the name of the attribute to retrieve.
+        :param parent_type: an optional XSD type that represents the scope where matching \
+        the attribute name. If not provided, the scope is assumed to be the global scope.
+        """
+        if parent_type is None:
+            try:
+                return self._schema.maps.attributes[name].type
+            except KeyError:
+                return None
+        elif hasattr(parent_type, 'attributes'):
+            attributes = cast(XsdAttributeGroupProtocol, parent_type.attributes)
+            if name in attributes:
+                return attributes[name].type
+            elif None in attributes and attributes[None].is_matching(name):
+                try:
+                    return self._schema.maps.attributes[name].type
+                except KeyError:
+                    return None
+            elif name.startswith('{http://www.w3.org/2001/XMLSchema-instance}'):
+                return self._schema.maps.types.get(XSD_ANY_ATOMIC_TYPE)
+
+        return None
+
+    def get_child_type(self, name: str, parent_type: Optional[XsdTypeProtocol] = None) \
+            -> Optional[XsdTypeProtocol]:
+        """
+        Get the child XSD type if the provided name is matching in the scope,
+        otherwise return None.
+
+        :param name: the name of the child element to match.
+        :param parent_type: an optional XSD type that represents the scope where matching \
+        the child element name. If `None` or not provided the scope is the schema.
+        """
+        if parent_type is None:
+            try:
+                return self._schema.maps.elements[name].type
+            except KeyError:
+                return None
+        elif (content := parent_type.model_group) is not None:
+            for xsd_element in content.iter_elements():
+                if xsd_element.is_matching(name):
+                    if xsd_element.name == name:
+                        return xsd_element.type
+                    else:
+                        # a wildcard or a substitute
+                        try:
+                            return self._schema.maps.elements[name].type
+                        except KeyError:
+                            return None
+        return None
+
     @abstractmethod
     def is_instance(self, obj: Any, type_qname: str) -> bool:
         """
diff --git a/elementpath/xpath1/xpath1_parser.py b/elementpath/xpath1/xpath1_parser.py
@@ -13,7 +13,7 @@
 import re
 from abc import ABCMeta
 from collections.abc import Callable, MutableMapping, Sequence
-from typing import cast, Any, ClassVar, Optional, Union, TYPE_CHECKING
+from typing import cast, Any, ClassVar, Optional, Union
 
 from elementpath.aliases import NamespacesType, NargsType
 from elementpath.exceptions import xpath_error, UnsupportedFeatureError, \
@@ -24,13 +24,11 @@
 from elementpath.tdop import Parser
 from elementpath.namespaces import XML_NAMESPACE, XSD_NAMESPACE, XPATH_FUNCTIONS_NAMESPACE
 from elementpath.sequence_types import match_sequence_type
+from elementpath.schema_proxy import AbstractSchemaProxy
 from elementpath.xpath_context import ContextType
 from elementpath.xpath_tokens import XPathTokenType, XPathToken, XPathAxis, \
     XPathFunction, ProxyToken
 
-if TYPE_CHECKING:
-    from elementpath.schema_proxy import AbstractSchemaProxy
-
 
 class XPath1Parser(Parser[XPathTokenType]):
     """
@@ -65,7 +63,7 @@ class XPath1Parser(Parser[XPathTokenType]):
     }
 
     # Class attributes for compatibility with XPath 2.0+
-    schema: Optional['AbstractSchemaProxy'] = None
+    schema: Optional[AbstractSchemaProxy] = None
     variable_types: Optional[dict[str, str]] = None
     document_types: Optional[dict[str, str]] = None
     collection_types: Optional[NamespacesType] = None
diff --git a/elementpath/xpath2/xpath2_parser.py b/elementpath/xpath2/xpath2_parser.py
@@ -14,7 +14,7 @@
 import locale
 from collections.abc import Callable, MutableMapping
 from urllib.parse import urlparse
-from typing import cast, Any, ClassVar, Optional, Union, TYPE_CHECKING
+from typing import cast, Any, ClassVar, Optional, Union
 
 from elementpath.aliases import NamespacesType, NargsType
 from elementpath.helpers import upper_camel_case, is_ncname, ordinal
@@ -28,11 +28,9 @@
 from elementpath.xpath_tokens import XPathToken, ProxyToken, XPathFunction, XPathConstructor
 from elementpath.xpath_context import XPathContext, XPathSchemaContext
 from elementpath.sequence_types import is_sequence_type, match_sequence_type
+from elementpath.schema_proxy import AbstractSchemaProxy
 from elementpath.xpath1 import XPath1Parser
 
-if TYPE_CHECKING:
-    from elementpath.schema_proxy import AbstractSchemaProxy
-
 
 class XPath2Parser(XPath1Parser):
     """
@@ -106,7 +104,7 @@ def __init__(self, namespaces: Optional[NamespacesType] = None,
                  default_namespace: Optional[str] = None,
                  function_namespace: Optional[str] = None,
                  xsd_version: Optional[str] = None,
-                 schema: Optional['AbstractSchemaProxy'] = None,
+                 schema: Optional[AbstractSchemaProxy] = None,
                  base_uri: Optional[str] = None,
                  variable_types: Optional[dict[str, str]] = None,
                  document_types: Optional[dict[str, str]] = None,
@@ -138,12 +136,13 @@ def __init__(self, namespaces: Optional[NamespacesType] = None,
         if function_namespace is not None:
             self.function_namespace = function_namespace
 
-        if schema is not None:
-            try:
-                schema.bind_parser(self)
-            except AttributeError:
-                msg = "argument 'schema' must be an instance of AbstractSchemaProxy"
-                raise ElementPathTypeError(msg)
+        if schema is None:
+            pass
+        elif not isinstance(schema, AbstractSchemaProxy):
+            msg = "argument 'schema' must be an instance of AbstractSchemaProxy"
+            raise ElementPathTypeError(msg)
+        else:
+            schema.bind_parser(self)
 
         if not variable_types:
             self.variable_types = {}
diff --git a/elementpath/xpath_context.py b/elementpath/xpath_context.py
@@ -144,9 +144,6 @@ def __init__(self,
             else:
                 self.item = self.root
 
-            if schema is not None:
-                self.root.apply_schema(schema)
-
         elif item is not None:
             self.root = None
             self.item = self.get_context_item(item, self.namespaces, uri, fragment)
@@ -180,7 +177,9 @@ def __init__(self,
                 if v is not None else v for k, v in documents.items()
             }
 
-        self.schema = schema
+        if schema is not None:
+            self.schema = schema
+
         self.variables = {}
         if variables is not None:
             for varname, value in variables.items():
@@ -243,11 +242,21 @@ def schema(self) -> Optional['AbstractSchemaProxy']:
     @schema.setter
     def schema(self, schema: Optional['AbstractSchemaProxy']) -> None:
         self._schema = schema
-        if schema is not None:
+        if schema is None:
             if self.root is not None:
+                self.root.clear_types()
+            elif isinstance(self.item, XPathNode):
+                self.item.clear_types()
+        elif hasattr(schema, 'is_assertion_based'):
+            if self.root is not None:
+                self.root.clear_types()
                 self.root.apply_schema(schema)
             elif isinstance(self.item, XPathNode):
+                self.item.clear_types()
                 self.item.apply_schema(schema)
+        else:
+            msg = f"{schema!r} is not an instance of AbstractSchemaProxy"
+            raise ElementPathTypeError(msg)
 
     def get_root(self, node: Any) -> Optional[RootNodeType]:
         if isinstance(self.root, (DocumentNode, ElementNode)):
@@ -622,9 +631,6 @@ class XPathSchemaContext(XPathContext):
     """
     root: ElementNode
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None:
-        super().__init__(*args, **kwargs)
-
     @property
     def schema(self) -> Optional['AbstractSchemaProxy']:
         return self._schema
diff --git a/elementpath/xpath_nodes.py b/elementpath/xpath_nodes.py
@@ -246,10 +246,15 @@ def is_list(self) -> Optional[bool]:
         return None
 
     def apply_schema(self, schema: 'AbstractSchemaProxy') -> None:
-        """set XSD types for elements and attribute nodes from schema proxy instance."""
+        """Set XSD types for elements and attribute nodes from schema proxy instance."""
         if self.parent is not None:
             self.parent.apply_schema(schema)
 
+    def clear_types(self) -> None:
+        """Clear XSD types for elements and attribute nodes."""
+        if self.parent is not None:
+            self.parent.clear_types()
+
     def match_name(self, name: str, default_namespace: Optional[str] = None) -> bool:
         """
         Returns `True` if the argument is matching the name of the node, `False` otherwise.
@@ -1199,8 +1204,7 @@ def apply_schema(self, schema: 'AbstractSchemaProxy') -> None:
             if self.obj.attrib:
                 for attr in self.attributes:
                     if isinstance(attr, TextAttributeNode):
-                        xsd_attribute = schema.get_attribute(attr.name, xsd_types[-1])
-                        attr.xsd_type = getattr(xsd_attribute, 'type', None)
+                        attr.xsd_type = schema.get_attribute_type(attr.name, xsd_types[-1])
         else:
             root_node: ParentNodeType = self
             while isinstance(root_node.parent, EtreeElementNode):
@@ -1222,23 +1226,22 @@ def apply_schema(self, schema: 'AbstractSchemaProxy') -> None:
                         elem.clear_types()
                         continue
                     else:
-                        elem.xsd_type = schema.get_type(type_name)
+                        xsd_type = schema.get_type(type_name)
                 else:
-                    xsd_element = schema.get_element(elem.name, xsd_types[-1])
-                    if xsd_element is None:
-                        elem.clear_types()
-                        continue
+                    xsd_type = schema.get_child_type(elem.name, xsd_types[-1])
 
-                    elem.xsd_type = xsd_element.type
+                if xsd_type is None:
+                    elem.clear_types()
+                    continue
 
+                elem.xsd_type = xsd_type
                 if elem.obj.attrib:
                     for attr in elem.attributes:
                         if isinstance(attr, TextAttributeNode):
-                            xsd_attribute = schema.get_attribute(attr.name, elem.xsd_type)
-                            attr.xsd_type = getattr(xsd_attribute, 'type', None)
+                            attr.xsd_type = schema.get_attribute_type(attr.name, xsd_type)
 
                 if len(elem.obj):
-                    xsd_types.append(elem.xsd_type)
+                    xsd_types.append(xsd_type)
                     iterators.append(children)
                     children = iter(elem)
                     break
diff --git a/publiccode.yml b/publiccode.yml
@@ -6,7 +6,7 @@ publiccodeYmlVersion: '0.4'
 name: elementpath
 url: 'https://github.com/sissaschool/elementpath'
 landingURL: 'https://github.com/sissaschool/elementpath'
-releaseDate: '2025-05-03'
+releaseDate: '2025-05-11'
 softwareVersion: v5.0.1
 developmentStatus: stable
 platforms:
diff --git a/tests/test_schema_proxy.py b/tests/test_schema_proxy.py
