first draft conversion of unittest syntax to pytest syntax for schema…

.github/workflows/main.yaml

@@ -14,53 +14,64 @@ jobs:
         os: [ubuntu-latest, windows-latest]
         python-version: ["3.8", "3.9", "3.10", "3.12"]
         exclude:
-        - os: windows-latest
-          python-version: "3.8"
+          # Test on Windows with only the oldest and newest Python versions
+          - os: windows-latest
+            python-version: "3.8"
+
+    # See https://github.com/snok/install-poetry#running-on-windows
+    defaults:
+      run:
+        shell: bash
 
     runs-on: ${{ matrix.os }}
 
     steps:
 
-      #----------------------------------------------
-      #          install poetry
-      #----------------------------------------------
-      - name: Install Poetry
-        run: pipx install poetry==1.4.0
-
       #----------------------------------------------
       #       check-out repo and set-up python
       #----------------------------------------------
+      - name: Install poetry
+        run: pipx install poetry==1.4.0
+
       - name: Check out repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Install dynamic versioning plugin
+        run: poetry self add "poetry-dynamic-versioning[plugin]"
 
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
+        id: setup-python
         with:
           python-version: ${{ matrix.python-version }}
           cache: 'poetry'
 
       #----------------------------------------------
-      #    install your root project, if required 
+      #    install dependencies
       #----------------------------------------------      
-      - name: Install library
-        run: poetry install --no-interaction
+      - name: Install dependencies
+        run: poetry install --no-interaction --all-extras
 
       #----------------------------------------------
-      #              coverage report   
+      #              coverage report
       #----------------------------------------------
       - name: Generate coverage results
         run: |
           poetry run coverage run -m pytest
           poetry run coverage xml
           poetry run coverage report -m
+        shell: bash
 
       #----------------------------------------------
       #           upload coverage results
       #----------------------------------------------
       - name: Upload coverage report
+        if: github.repository == 'linkml/linkml-runtime'
         uses: codecov/codecov-action@v3
         with:
           name: codecov-results-${{ matrix.os }}-${{ matrix.python-version }}
           token: ${{ secrets.CODECOV_TOKEN }}
           file: coverage.xml
-          fail_ci_if_error: false
+          fail_ci_if_error: false

Makefile

@@ -5,12 +5,13 @@
 
 # TODO: make this mechanism more robust
 MODEL_DIR = ../linkml-model/linkml_model/
+RUN=poetry run
 
 update_model:
 	cp -pr $(MODEL_DIR)/* linkml_runtime/linkml_model
 
 test:
-	poetry run python -m unittest discover
+	$(RUN) pytest
 
 # temporary measure until linkml-model is synced
 linkml_runtime/processing/validation_datamodel.py: linkml_runtime/processing/validation_datamodel.yaml

linkml_runtime/utils/schemaview.py

@@ -8,6 +8,7 @@
 from pathlib import Path
 from typing import Mapping, Tuple, TypeVar
 import warnings
+from pprint import pprint
 
 from linkml_runtime.utils.namespaces import Namespaces
 from deprecated.classic import deprecated
@@ -16,6 +17,8 @@
 from linkml_runtime.linkml_model.meta import *
 from linkml_runtime.exceptions import OrderingError
 from enum import Enum
+from linkml_runtime.linkml_model.meta import ClassDefinition, SlotDefinition, ClassDefinitionName
+from dataclasses import asdict, is_dataclass, fields
 
 logger = logging.getLogger(__name__)
 
@@ -35,9 +38,9 @@
 ENUM_NAME = Union[EnumDefinitionName, str]
 
 ElementType = TypeVar("ElementType", bound=Element)
-ElementNameType = TypeVar("ElementNameType", bound=Union[ElementName,str])
+ElementNameType = TypeVar("ElementNameType", bound=Union[ElementName, str])
 DefinitionType = TypeVar("DefinitionType", bound=Definition)
-DefinitionNameType = TypeVar("DefinitionNameType", bound=Union[DefinitionName,str])
+DefinitionNameType = TypeVar("DefinitionNameType", bound=Union[DefinitionName, str])
 ElementDict = Dict[ElementNameType, ElementType]
 DefDict = Dict[DefinitionNameType, DefinitionType]
 
@@ -52,7 +55,6 @@ class OrderedBy(Enum):
     """
 
 
-
 def _closure(f, x, reflexive=True, depth_first=True, **kwargs):
     if reflexive:
         rv = [x]
@@ -83,7 +85,7 @@ def load_schema_wrap(path: str, **kwargs):
     schema: SchemaDefinition
     schema = yaml_loader.load(path, target_class=SchemaDefinition, **kwargs)
     if "\n" not in path:
-    # if "\n" not in path and "://" not in path:
+        # if "\n" not in path and "://" not in path:
         # only set path if the input is not a yaml string or URL.
         # Setting the source path is necessary for relative imports;
         # while initializing a schema with a yaml string is possible, there
@@ -117,6 +119,43 @@ class SchemaUsage():
     inferred: bool = None
 
 
+def to_dict(obj):
+    """
+    Convert a LinkML element (such as ClassDefinition) to a dictionary.
+
+    :param obj: The LinkML class instance to convert.
+    :return: A dictionary representation of the class.
+    """
+    if is_dataclass(obj):
+        return asdict(obj)
+    elif isinstance(obj, list):
+        return [to_dict(item) for item in obj]
+    elif isinstance(obj, dict):
+        return {key: to_dict(value) for key, value in obj.items()}
+    else:
+        return obj
+
+
+def get_anonymous_class_definition(class_as_dict: ClassDefinition) -> AnonymousClassExpression:
+    """
+    Convert a ClassDefinition to an AnonymousClassExpression, typically for use in defining an Expression object
+    (e.g. SlotDefinition.range_expression). This method only fills out the fields that are present in the
+    AnonymousClassExpression class. #TODO: We should consider whether an Expression should share a common ancestor with
+    the Definition classes.
+
+    :param class_as_dict: The ClassDefinition to convert.
+    :return: An AnonymousClassExpression.
+    """
+    an_expr = AnonymousClassExpression()
+    valid_fields = {field.name for field in fields(an_expr)}
+    for k, v in class_as_dict.items():
+        if k in valid_fields:
+            setattr(an_expr, k, v)
+    for k, v in class_as_dict.items():
+        setattr(an_expr, k, v)
+    return an_expr
+
+
 @dataclass
 class SchemaView(object):
     """
@@ -221,7 +260,8 @@ def load_import(self, imp: str, from_schema: SchemaDefinition = None):
         return schema
 
     @lru_cache(None)
-    def imports_closure(self, imports: bool = True, traverse: Optional[bool] = None, inject_metadata=True) -> List[SchemaDefinitionName]:
+    def imports_closure(self, imports: bool = True, traverse: Optional[bool] = None, inject_metadata=True) -> List[
+        SchemaDefinitionName]:
         """
         Return all imports
 
@@ -289,7 +329,7 @@ def imports_closure(self, imports: bool = True, traverse: Optional[bool] = None,
             visited.add(sn)
 
         # filter duplicates, keeping first entry
-        closure = list({k:None for k in closure}.keys())
+        closure = list({k: None for k in closure}.keys())
 
         if inject_metadata:
             for s in self.schema_map.values():
@@ -395,7 +435,6 @@ def _order_inheritance(self, elements: DefDict) -> DefDict:
 
         return {s.name: s for s in slist}
 
-
     @lru_cache(None)
     def all_classes(self, ordered_by=OrderedBy.PRESERVE, imports=True) -> Dict[ClassDefinitionName, ClassDefinition]:
         """
@@ -840,15 +879,14 @@ def permissible_value_ancestors(self, permissible_value_text: str,
 
     @lru_cache(None)
     def permissible_value_descendants(self, permissible_value_text: str,
-                                    enum_name: ENUM_NAME,
-                                    reflexive=True,
-                                    depth_first=True) -> List[str]:
+                                      enum_name: ENUM_NAME,
+                                      reflexive=True,
+                                      depth_first=True) -> List[str]:
         """
         Closure of permissible_value_children method
         :enum
         """
 
-
         return _closure(lambda x: self.permissible_value_children(x, enum_name),
                         permissible_value_text,
                         reflexive=reflexive,
@@ -994,7 +1032,7 @@ def is_multivalued(self, slot_name: SlotDefinition) -> bool:
         :param slot_name: slot to test for multivalued
         :return boolean:
         """
-        induced_slot = self.induced_slot(slot_name)
+        induced_slot = self.induced_slot(slot_name.name)
         return True if induced_slot.multivalued else False
 
     @lru_cache(None)
@@ -1294,6 +1332,7 @@ def class_slots(self, class_name: CLASS_NAME, imports=True, direct=False, attrib
                 slots_nr.append(s)
         return slots_nr
 
+
     @lru_cache(None)
     def induced_slot(self, slot_name: SLOT_NAME, class_name: CLASS_NAME = None, imports=True,
                      mangle_name=False) -> SlotDefinition:
@@ -1307,6 +1346,7 @@ def induced_slot(self, slot_name: SLOT_NAME, class_name: CLASS_NAME = None, impo
         :param slot_name: slot to be queries
         :param class_name: class used as context
         :param imports: include imports closure
+        :param mangle_name: if True, the slot name will be mangled to include the class name
         :return: dynamic slot constructed by inference
         """
         if class_name:
@@ -1342,26 +1382,61 @@ def induced_slot(self, slot_name: SLOT_NAME, class_name: CLASS_NAME = None, impo
             for anc_sn in reversed(slot_anc_names):
                 anc_slot = self.get_slot(anc_sn, attributes=False)
                 for metaslot_name in SlotDefinition._inherited_slots:
+                    # getattr(x, 'y') is equivalent to x.y. None here means raise an error if x.y is not found
                     if getattr(anc_slot, metaslot_name, None):
+                        # setattr(x, 'y', v) is equivalent to ``x.y = v''
                         setattr(induced_slot, metaslot_name, copy(getattr(anc_slot, metaslot_name)))
         COMBINE = {
             'maximum_value': lambda x, y: min(x, y),
             'minimum_value': lambda x, y: max(x, y),
         }
         # iterate through all metaslots, and potentially populate metaslot value for induced slot
         for metaslot_name in self._metaslots_for_slot():
+
             # inheritance of slots; priority order
             #   slot-level assignment < ancestor slot_usage < self slot_usage
             v = getattr(induced_slot, metaslot_name, None)
             if not cls:
                 propagated_from = []
             else:
                 propagated_from = self.class_ancestors(class_name, reflexive=True, mixins=True)
+
             for an in reversed(propagated_from):
                 induced_slot.owner = an
                 a = self.get_class(an, imports)
+                # slot usage of the slot in the ancestor class, last ancestor iterated through here is "self"
+                # so that self.slot_usage overrides ancestor slot_usage at the conclusion of the loop.
                 anc_slot_usage = a.slot_usage.get(slot_name, {})
+                # slot name in the ancestor class
+                # getattr(x, 'y') is equivalent to x.y. None here means raise an error if x.y is not found
                 v2 = getattr(anc_slot_usage, metaslot_name, None)
+                # v2 is the value of the metaslot in slot_usage in the ancestor class, which in the loop, means that
+                # the class itself is the last slot_usage to be considered and applied.
+                if metaslot_name in ["any_of", "exactly_one_of"]:
+                    if anc_slot_usage != {}:
+                        for ao in anc_slot_usage.any_of:
+                            if ao.range is not None:
+                                ao_range = self.get_class(ao.range)
+                                if ao_range:
+                                    acd = get_anonymous_class_definition(to_dict(ao_range))
+                                    if induced_slot.range_expression is None:
+                                        induced_slot.range_expression = AnonymousClassExpression()
+                                    if induced_slot.range_expression.any_of is None:
+                                        induced_slot.range_expression.any_of = []
+                                    # Check for duplicates before appending
+                                    if acd not in induced_slot.range_expression.any_of:
+                                        induced_slot.range_expression.any_of.append(acd)
+                        for eoo in anc_slot_usage.exactly_one_of:
+                            if eoo.range is not None:
+                                eoo_range = self.get_class(eoo.range)
+                                acd = get_anonymous_class_definition(as_dict(eoo_range))
+                                if induced_slot.range_expression is None:
+                                    induced_slot.range_expression = AnonymousClassExpression()
+                                if induced_slot.range_expression.exactly_one_of is None:
+                                    induced_slot.range_expression.exactly_one_of = []
+                                # Check for duplicates before appending
+                                if acd not in induced_slot.range_expression.exactly_one_of:
+                                    induced_slot.range_expression.exactly_one_of.append(acd)
                 if v is None:
                     v = v2
                 else:
@@ -1371,7 +1446,6 @@ def induced_slot(self, slot_name: SLOT_NAME, class_name: CLASS_NAME = None, impo
                     else:
                         if v2 is not None:
                             v = v2
-                            logging.debug(f'{v} takes precedence over {v2} for {induced_slot.name}.{metaslot_name}')
             if v is None:
                 if metaslot_name == 'range':
                     v = self.schema.default_range
@@ -1390,6 +1464,30 @@ def induced_slot(self, slot_name: SLOT_NAME, class_name: CLASS_NAME = None, impo
             if induced_slot.name in c.slots or induced_slot.name in c.attributes:
                 if c.name not in induced_slot.domain_of:
                     induced_slot.domain_of.append(c.name)
+        if induced_slot.range is not None:
+            if induced_slot.range_expression is None:
+                induced_slot.range_expression = AnonymousClassExpression()
+                induced_slot.range_expression.any_of = []
+                induced_slot.range_expression.any_of.append(
+                    get_anonymous_class_definition(to_dict(self.get_class(induced_slot.range)))
+                )
+                return induced_slot
+            else:
+                any_of_ancestors = []
+                if induced_slot.range_expression.any_of is not None:
+                    for ao_range in induced_slot.range_expression.any_of:
+                        ao_range_class = self.get_class(ao_range.name)
+                        ao_anc = self.class_ancestors(ao_range_class.name)
+                        for a in ao_anc:
+                            if a not in any_of_ancestors:
+                                any_of_ancestors.append(a)
+                if induced_slot.range in any_of_ancestors:
+                    return induced_slot
+                else:
+                    induced_slot.range_expression.any_of.append(
+                        get_anonymous_class_definition(to_dict(self.get_class(induced_slot.range)))
+                    )
+                return induced_slot
         return induced_slot
 
     @lru_cache(None)
@@ -1516,7 +1614,7 @@ def is_inlined(self, slot: SlotDefinition, imports=True) -> bool:
                 return True
             elif slot.inlined_as_list:
                 return True
-            
+
             id_slot = self.get_identifier_slot(range, imports=imports)
             if id_slot is None:
                 # must be inlined as has no identifier
@@ -1560,7 +1658,7 @@ def slot_range_as_union(self, slot: SlotDefinition) -> List[ElementName]:
         """
         Returns all applicable ranges for a slot
 
-        Typically any given slot has exactly one range, and one metamodel element type,
+        Typically, any given slot has exactly one range, and one metamodel element type,
         but a proposed feature in LinkML 1.2 is range expressions, where ranges can be defined as unions
 
         :param slot:
@@ -1572,9 +1670,9 @@ def slot_range_as_union(self, slot: SlotDefinition) -> List[ElementName]:
             if x.range:
                 range_union_of.append(x.range)
         return range_union_of
-        
+
     def get_classes_by_slot(
-        self, slot: SlotDefinition, include_induced: bool = False
+            self, slot: SlotDefinition, include_induced: bool = False
     ) -> List[ClassDefinitionName]:
         """Get all classes that use a given slot, either as a direct or induced slot.
 
@@ -1891,4 +1989,4 @@ def materialize_derived_schema(self) -> SchemaDefinition:
             derived_schema.subsets[subset.name] = subset
         for enum in [deepcopy(e) for e in self.all_enums().values()]:
             derived_schema.enums[enum.name] = enum
-        return derived_schema
+        return derived_schema