improve schema info collection combiner handling

asdf/_asdf.py

@@ -1378,6 +1378,10 @@ def schema_info(self, key="description", path=None, preserve_list=True, refresh_
         """
         Get a nested dictionary of the schema information for a given key, relative to the path.
 
+        This method will only return unambiguous info. If a property is subject to multiple
+        subschemas or contains ambiguous entries (multiple titles) no result will be returned
+        for that property.
+
         Parameters
         ----------
         key : str

asdf/_node_info.py

@@ -20,6 +20,67 @@ def _filter_tree(info, filters):
     return len(info.children) > 0 or all(f(info.node, info.identifier) for f in filters)
 
 
+def _get_matching_schema_property(schema, key):
+    if "properties" in schema:
+        props = schema["properties"]
+        if key in props:
+            return props[key]
+        if "patternProperties" in props:
+            patterns = props["patternProperties"]
+            for regex in patterns:
+                if re.search(regex, key):
+                    return patterns[regex]
+    return None
+
+
+def _get_subschema_for_property(schema, key):
+    # This does NOT handle $ref the expectation is that the schema
+    # is loaded with resolve_references=True
+    applicable = []
+
+    # first check properties and patternProperties
+    subschema = _get_matching_schema_property(schema, key)
+    if subschema is not None:
+        applicable.append(subschema)
+
+    # next handle schema combiners
+    if "not" in schema:
+        subschema = _get_subschema_for_property(schema["not"], key)
+        if subschema is not None:
+            # We can't resolve a valid subschema under a "not" since
+            # we'd have to know how to invert a schema
+            return None
+
+    for combiner in ("allOf", "oneOf", "anyOf"):
+        for combined_schema in schema.get(combiner, []):
+            subschema = _get_subschema_for_property(combined_schema, key)
+            if subschema is not None:
+                applicable.append(subschema)
+
+    # only return the subschema if we found exactly 1 applicable
+    if len(applicable) == 1:
+        return applicable[0]
+    return None
+
+
+def _get_schema_key(schema, key):
+    applicable = []
+    if key in schema:
+        applicable.append(schema[key])
+    # Here we don't consider any subschema under "not" to avoid
+    # false positives for keys like "type" etc.
+    for combiner in ("allOf", "oneOf", "anyOf"):
+        for combined_schema in schema.get(combiner, []):
+            possible = _get_schema_key(combined_schema, key)
+            if possible is not None:
+                applicable.append(possible)
+
+    # only return the property if we found exactly 1 applicable
+    if len(applicable) == 1:
+        return applicable[0]
+    return None
+
+
 def create_tree(key, node, identifier="root", filters=None, refresh_extension_manager=False):
     """
     Create a `NodeSchemaInfo` tree which can be filtered from a base node.
@@ -214,22 +275,12 @@ def parent_node(self):
 
     @property
     def info(self):
-        if self.schema is not None:
-            return self.schema.get(self.key, None)
-
-        return None
+        if self.schema is None:
+            return None
+        return _get_schema_key(self.schema, self.key)
 
     def get_schema_for_property(self, identifier):
-        subschema = self.schema.get("properties", {}).get(identifier, None)
-        if subschema is not None:
-            return subschema
-
-        subschema = self.schema.get("properties", {}).get("patternProperties", None)
-        if subschema:
-            for key in subschema:
-                if re.search(key, identifier):
-                    return subschema[key]
-        return {}
+        return _get_subschema_for_property(self.schema, identifier) or {}
 
     def set_schema_for_property(self, parent, identifier):
         """Extract a subschema from the parent for the identified property"""
@@ -241,7 +292,7 @@ def set_schema_from_node(self, node, extension_manager):
 
         tag_def = extension_manager.get_tag_definition(node._tag)
         schema_uri = tag_def.schema_uris[0]
-        schema = load_schema(schema_uri)
+        schema = load_schema(schema_uri, resolve_references=True)
 
         self.schema = schema
 

asdf/_tests/test_info.py

@@ -4,6 +4,7 @@
 import tempfile
 
 import numpy as np
+import pytest
 
 import asdf
 from asdf.extension import ExtensionManager, ExtensionProxy, ManifestExtension
@@ -168,8 +169,8 @@ def manifest_extension(tmp_path):
     description: Some silly description
     type: integer
     archive_catalog:
-        datatype: int
-        destination: [ScienceCommon.silly]
+      datatype: int
+      destination: [ScienceCommon.silly]
   clown:
     title: clown name
     description: clown description
@@ -231,14 +232,14 @@ def manifest_extension(tmp_path):
     title: Attribute1 Title
     type: string
     archive_catalog:
-        datatype: str
-        destination: [ScienceCommon.attribute1]
+      datatype: str
+      destination: [ScienceCommon.attribute1]
   attribute2:
     title: Attribute2 Title
     type: string
     archive_catalog:
-        datatype: str
-        destination: [ScienceCommon.attribute2]
+      datatype: str
+      destination: [ScienceCommon.attribute2]
 ...
 """
 
@@ -251,19 +252,29 @@ def manifest_extension(tmp_path):
 type: object
 title: object with info support 3 title
 description: object description
+allOf:
+    - $ref: drink_ref-1.0.0
+...
+"""
+    drink_ref_schema = """
+%YAML 1.1
+---
+$schema: "asdf://stsci.edu/schemas/asdf/asdf-schema-1.1.0"
+id: "asdf://somewhere.org/asdf/schemas/drink_ref-1.0.0"
 properties:
   attributeOne:
     title: AttributeOne Title
     description: AttributeOne description
     type: string
     archive_catalog:
-        datatype: str
-        destination: [ScienceCommon.attributeOne]
+      datatype: str
+      destination: [ScienceCommon.attributeOne]
   attributeTwo:
-    title: AttributeTwo Title
-    description: AttributeTwo description
-    type: string
-    archive_catalog:
+    allOf:
+    - title: AttributeTwo Title
+      description: AttributeTwo description
+      type: string
+      archive_catalog:
         datatype: str
         destination: [ScienceCommon.attributeTwo]
 ...
@@ -278,6 +289,9 @@ def manifest_extension(tmp_path):
     spath = tmp_path / "schemas" / "drink-1.0.0.yaml"
     with open(spath, "w") as fschema:
         fschema.write(drink_schema)
+    spath = tmp_path / "schemas" / "drink_ref-1.0.0.yaml"
+    with open(spath, "w") as fschema:
+        fschema.write(drink_ref_schema)
     os.mkdir(tmp_path / "manifests")
     mpath = str(tmp_path / "manifests" / "foo_manifest-1.0.yaml")
     with open(mpath, "w") as fmanifest:
@@ -702,3 +716,62 @@ def __str__(self):
     assert "(NewlineStr)\n" in captured.out
     assert "(CarriageReturnStr)\n" in captured.out
     assert "(NiceStr): nice\n" in captured.out
+
+
+@pytest.mark.parametrize(
+    "schema, expected",
+    [
+        ({"properties": {"foo": {"type": "object"}}}, {"type": "object"}),
+        ({"allOf": [{"properties": {"foo": {"type": "object"}}}]}, {"type": "object"}),
+        ({"oneOf": [{"properties": {"foo": {"type": "object"}}}]}, {"type": "object"}),
+        ({"anyOf": [{"properties": {"foo": {"type": "object"}}}]}, {"type": "object"}),
+    ],
+)
+def test_node_property(schema, expected):
+    ni = asdf._node_info.NodeSchemaInfo.from_root_node("title", "root", {}, schema)
+    assert ni.get_schema_for_property("foo") == expected
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [
+        {"not": {"properties": {"foo": {"type": "object"}}}},
+        {"properties": {"foo": {"type": "object"}}, "allOf": [{"properties": {"foo": {"type": "object"}}}]},
+        {"properties": {"foo": {"type": "object"}}, "anyOf": [{"properties": {"foo": {"type": "object"}}}]},
+        {"properties": {"foo": {"type": "object"}}, "oneOf": [{"properties": {"foo": {"type": "object"}}}]},
+        {
+            "allOf": [{"properties": {"foo": {"type": "object"}}}],
+            "anyOf": [{"properties": {"foo": {"type": "object"}}}],
+        },
+        {
+            "anyOf": [{"properties": {"foo": {"type": "object"}}}],
+            "oneOf": [{"properties": {"foo": {"type": "object"}}}],
+        },
+        {
+            "oneOf": [{"properties": {"foo": {"type": "object"}}}],
+            "allOf": [{"properties": {"foo": {"type": "object"}}}],
+        },
+    ],
+)
+def test_node_property_error(schema):
+    ni = asdf._node_info.NodeSchemaInfo.from_root_node("title", "root", {}, schema)
+    assert ni.get_schema_for_property("foo") == {}
+
+
+@pytest.mark.parametrize(
+    "schema, expected",
+    [
+        ({"title": "foo"}, "foo"),
+        ({"allOf": [{"title": "foo"}]}, "foo"),
+        ({"oneOf": [{"title": "foo"}]}, "foo"),
+        ({"anyOf": [{"title": "foo"}]}, "foo"),
+        ({"not": {"title": "foo"}}, None),
+        ({"allOf": [{"title": "foo"}, {"title": "bar"}]}, None),
+        ({"oneOf": [{"title": "foo"}, {"title": "bar"}]}, None),
+        ({"anyOf": [{"title": "foo"}, {"title": "bar"}]}, None),
+        ({"allOf": [{"title": "foo"}, {"title": "bar"}]}, None),
+    ],
+)
+def test_node_info(schema, expected):
+    ni = asdf._node_info.NodeSchemaInfo.from_root_node("title", "root", {}, schema)
+    assert ni.info == expected

changes/1875.bugfix.rst

@@ -0,0 +1 @@
+Improve ``schema_info`` handling of schemas with combiners (allOf, anyOf, etc).