diff --git a/packages/linkml/src/linkml/generators/shaclgen.py b/packages/linkml/src/linkml/generators/shaclgen.py index aa5452bb38..ac25d6774d 100644 --- a/packages/linkml/src/linkml/generators/shaclgen.py +++ b/packages/linkml/src/linkml/generators/shaclgen.py @@ -1,4 +1,5 @@ import logging +import math import os import string from collections.abc import Callable @@ -478,6 +479,14 @@ def _add_rules(self, g: Graph, shape_uri: URIRef, cls: ClassDefinition) -> None: cls.name, ) + if getattr(rule, "elseconditions", None) is not None: + logger.warning( + "Rule in class %r has elseconditions; " + "SHACL-SPARQL generation emits the forward (if/then) direction only. " + "The else branch is not enforced.", + cls.name, + ) + sparql_query = self._rule_to_sparql(sv, cls, rule) if sparql_query is None: logger.debug( @@ -497,22 +506,137 @@ def _add_rules(self, g: Graph, shape_uri: URIRef, cls: ClassDefinition) -> None: g.add((constraint, SH.select, Literal(sparql_query))) + # Fields on a slot condition / class expression that carry no constraint + # semantics: they never change which instances satisfy the condition, so + # they are ignored by the operator accounting below. Anything set on a + # condition that is neither here nor explicitly translated by a converter + # makes the rule untranslatable — the converters must SKIP such a rule + # rather than emit a query that silently drops a conjunct (which would + # widen the trigger or narrow the check: a mis-translation, not a skip). + _NON_OPERATOR_FIELDS = frozenset( + { + "name", + "description", + "title", + "deprecated", + "todos", + "notes", + "comments", + "examples", + "in_subset", + "from_schema", + "imported_from", + "source", + "in_language", + "see_also", + "deprecated_element_has_exact_replacement", + "deprecated_element_has_possible_replacement", + "aliases", + "structured_aliases", + "local_names", + "mappings", + "exact_mappings", + "close_mappings", + "related_mappings", + "narrow_mappings", + "broad_mappings", + "created_by", + "contributors", + "created_on", + "last_updated_on", + "modified_by", + "status", + "rank", + "categories", + "keywords", + "extensions", + "annotations", + "alt_descriptions", + "id_prefixes", + "id_prefixes_are_closed", + "definition_uri", + "conforms_to", + "implements", + "instantiates", + } + ) + + @classmethod + def _set_operator_fields(cls, cond) -> set[str]: + """Return the names of the constraint-bearing fields actually set on a + rule condition or class expression. + + A field counts as *set* when it is not ``None`` and not an empty + collection (SchemaView materialises unset multivalued fields as empty + lists / dicts). Scalars are never judged by truthiness, so legitimate + falsy constraints such as ``minimum_value: 0`` or + ``equals_string: ""`` still count as set. Metadata fields + (:data:`_NON_OPERATOR_FIELDS`) are excluded. + + The converters compare this set against the exact operator set they + translate and skip the rule on any mismatch, so an unrecognised (or + future-metamodel) operator can never be silently dropped. + """ + fields: set[str] = set() + for name, value in vars(cond).items(): + if name.startswith("_") or name in cls._NON_OPERATOR_FIELDS: + continue + if value is None: + continue + if isinstance(value, list | dict) and not value: + continue + if isinstance(value, JsonObj) and not as_dict(value): + continue + fields.add(name) + return fields + + def _rule_slot(self, sv, slot_name: str, cls: ClassDefinition): + """Resolve a rule condition's slot key to the slot it names, or ``None`` + when no such slot exists. + + Resolution order mirrors ``sh:path`` in the main slot loop: the induced + (class-specific) slot when the key names one of the class's slots, then + the underscored alias form (a rule key ``my_slot`` for a slot named + ``my slot`` — SchemaView normalises names the same way elsewhere), then + the base slot. Callers treat ``None`` as *unknown slot* and skip the + rule rather than fabricating a predicate no shape uses. + """ + class_slot_names = sv.class_slots(cls.name) + if slot_name in class_slot_names: + return sv.induced_slot(slot_name, cls.name) + canonical = next((s for s in class_slot_names if underscore(s) == underscore(slot_name)), None) + if canonical is not None: + return sv.induced_slot(canonical, cls.name) + return sv.get_slot(slot_name) + def _rule_to_sparql(self, sv, cls: ClassDefinition, rule) -> str | None: """Convert a ``ClassRule`` to a SPARQL SELECT query string. Returns ``None`` when the rule does not match any supported pattern. + Each pattern requires its conditions to set **exactly** the operators + it translates; a rule whose pre/postconditions carry anything more + (extra scalar operators, expression-level ``any_of``/``all_of``/ + ``none_of``/``exactly_one_of``, ...) is skipped rather than partially + translated. """ pre = getattr(rule, "preconditions", None) post = getattr(rule, "postconditions", None) if not pre or not post: return None - pre_slots = getattr(pre, "slot_conditions", None) or {} - post_slots = getattr(post, "slot_conditions", None) or {} + # Expression-level exactness: only a plain conjunction of slot + # conditions is translatable. An any_of/all_of/none_of/exactly_one_of + # branch cannot be honoured by any converter below; dropping it would + # widen the precondition (false positives) or weaken the postcondition + # (false negatives), so the whole rule is skipped. + if self._set_operator_fields(pre) != {"slot_conditions"}: + return None + if self._set_operator_fields(post) != {"slot_conditions"}: + return None + + pre_slots = pre.slot_conditions or {} + post_slots = post.slot_conditions or {} - # Pattern: boolean guard - # preconditions: exactly one slot with value_presence PRESENT - # postconditions: exactly one slot with equals_string "true" if len(pre_slots) == 1 and len(post_slots) == 1: pre_slot_name = next(iter(pre_slots)) post_slot_name = next(iter(post_slots)) @@ -520,33 +644,50 @@ def _rule_to_sparql(self, sv, cls: ClassDefinition, rule) -> str | None: pre_cond = pre_slots[pre_slot_name] post_cond = post_slots[post_slot_name] - is_value_present = getattr(pre_cond, "value_presence", None) == PresenceEnum(PresenceEnum.PRESENT) - is_flag_true = getattr(post_cond, "equals_string", None) == "true" + pre_ops = self._set_operator_fields(pre_cond) + post_ops = self._set_operator_fields(post_cond) - if is_value_present and is_flag_true: + is_value_present = pre_ops == {"value_presence"} and pre_cond.value_presence == PresenceEnum( + PresenceEnum.PRESENT + ) + + # Pattern: boolean guard + # preconditions: exactly one slot with (only) value_presence PRESENT + # postconditions: exactly one boolean-range slot with (only) + # equals_string "true". The range gate matters: on a non-boolean + # slot the string "true" must be compared as a string, which is + # the presence-implies-value pattern below — without the gate the + # boolean comparison mistranslates and flags conforming data. + if ( + is_value_present + and post_ops == {"equals_string"} + and post_cond.equals_string == "true" + and getattr(self._rule_slot(sv, post_slot_name, cls), "range", None) == "boolean" + ): return self._build_boolean_guard_sparql(sv, cls, post_slot_name, pre_slot_name) # Pattern: presence implies value (enum guard) - # preconditions: value slot with value_presence PRESENT - # postconditions: target slot with equals_string or equals_string_in + # preconditions: value slot with (only) value_presence PRESENT + # postconditions: target slot with (only) equals_string or (only) + # equals_string_in. # Semantics: "If the value slot is present, the target slot must be # present and hold one of the allowed values." Generalises the # boolean guard (equals_string "true") to arbitrary enum values. - post_equals = getattr(post_cond, "equals_string", None) - post_equals_in = getattr(post_cond, "equals_string_in", None) - if is_value_present and (post_equals is not None or post_equals_in): - allowed = list(post_equals_in) if post_equals_in else [post_equals] + if is_value_present and post_ops in ({"equals_string"}, {"equals_string_in"}): + if post_ops == {"equals_string_in"}: + allowed = list(post_cond.equals_string_in) + else: + allowed = [post_cond.equals_string] return self._build_presence_implies_value_sparql(sv, cls, pre_slot_name, post_slot_name, allowed) # Pattern: exclusive value - # preconditions: slot X has equals_string (a specific enum value) - # postconditions: same slot X has maximum_cardinality N + # preconditions: slot X with (only) equals_string (a specific enum value) + # postconditions: same slot X with (only) maximum_cardinality N # Semantics: "If value V is present in slot X, then X has at most N values." - pre_equals = getattr(pre_cond, "equals_string", None) - post_max_card = getattr(post_cond, "maximum_cardinality", None) - - if pre_equals is not None and post_max_card is not None and pre_slot_name == post_slot_name: - return self._build_exclusive_value_sparql(sv, cls, pre_slot_name, pre_equals, int(post_max_card)) + if pre_ops == {"equals_string"} and post_ops == {"maximum_cardinality"} and pre_slot_name == post_slot_name: + return self._build_exclusive_value_sparql( + sv, cls, pre_slot_name, pre_cond.equals_string, int(post_cond.maximum_cardinality) + ) # Fallback: a small compositional builder for operator combinations not # covered by the three named patterns above (conditional-required, @@ -613,25 +754,37 @@ def _scalar_filters(self, var: str, cond, resolve: Callable[[str], str]) -> list *resolve* maps an ``equals_string`` value to a SPARQL term (an enum ``meaning`` IRI or an escaped string literal). - Returns ``None`` when *cond* sets none of the recognised scalar - operators, so the caller skips a rule it cannot faithfully translate - rather than emitting an under-constrained (or vacuous) query. + Returns ``None`` when *cond* sets no recognised scalar operator, sets + any operator *outside* the recognised set (per + :meth:`_set_operator_fields` — partial translation would drop a + conjunct), sets ``value_presence`` to anything but ``PRESENT``, or + carries a non-numeric threshold bound. In every such case the caller + skips the rule it cannot faithfully translate rather than emitting an + under-constrained (or vacuous) query. """ + op_fields = self._set_operator_fields(cond) + if not op_fields or not op_fields <= {"value_presence", "equals_string", "minimum_value", "maximum_value"}: + return None # unsupported operator present (or none at all): skip + if "value_presence" in op_fields and cond.value_presence != PresenceEnum(PresenceEnum.PRESENT): + # ABSENT (or a future presence value) cannot be expressed as a + # triple binding + filter; translating the other operators anyway + # would invert the declared trigger. + return None + filters: list[str] = [] - recognized = getattr(cond, "value_presence", None) == PresenceEnum(PresenceEnum.PRESENT) - equals = getattr(cond, "equals_string", None) - if equals is not None: - filters.append(f"FILTER ( {var} = {resolve(equals)} )") - recognized = True - minimum = getattr(cond, "minimum_value", None) - if minimum is not None: - filters.append(f"FILTER ( {var} >= {self._sparql_number(minimum)} )") - recognized = True - maximum = getattr(cond, "maximum_value", None) - if maximum is not None: - filters.append(f"FILTER ( {var} <= {self._sparql_number(maximum)} )") - recognized = True - return filters if recognized else None + if "equals_string" in op_fields: + filters.append(f"FILTER ( {var} = {resolve(cond.equals_string)} )") + if "minimum_value" in op_fields: + minimum = self._sparql_number(cond.minimum_value) + if minimum is None: + return None + filters.append(f"FILTER ( {var} >= {minimum} )") + if "maximum_value" in op_fields: + maximum = self._sparql_number(cond.maximum_value) + if maximum is None: + return None + filters.append(f"FILTER ( {var} <= {maximum} )") + return filters def _precondition_patterns(self, sv, cls: ClassDefinition, pre_slots) -> list[str] | None: """Translate a conjunction of precondition slot conditions into SPARQL @@ -650,11 +803,19 @@ def _precondition_patterns(self, sv, cls: ClassDefinition, pre_slots) -> list[st lines: list[str] = [] for i, (slot_name, cond) in enumerate(pre_slots.items()): path = self._slot_uri(sv, slot_name, cls) + if path is None: + return None var = f"?pre{i}" - range_expr = getattr(cond, "range_expression", None) - if range_expr is not None and getattr(range_expr, "slot_conditions", None): + if self._set_operator_fields(cond) == {"range_expression"}: # One-hop into an inlined child object: bind the child node and - # apply the inner slot conditions to it. + # apply the inner slot conditions to it. The nested expression + # must itself be a plain conjunction of slot conditions; a + # condition mixing range_expression with scalar operators (or a + # nested any_of/...) is skipped rather than partially + # translated. + range_expr = cond.range_expression + if self._set_operator_fields(range_expr) != {"slot_conditions"}: + return None node = f"{var}_node" lines.append(f"$this <{path}> {node} .") inner = self._member_conditions(sv, cls, slot_name, node, range_expr.slot_conditions) @@ -678,19 +839,37 @@ def _member_conditions( class of *container_slot_name* — by a set of inner slot conditions. Shared by the nested ``range_expression`` precondition (single inlined - child) and the ``has_member`` postcondition (a list member). Enum - values are resolved against the container slot's range class, and (like - preconditions) combining operators on one condition emits all of them. - Returns ``None`` for unsupported inner operators. + child) and the ``has_member`` postcondition (a list member). Inner + slots live on the container slot's **range class**, so both their + property IRIs and their enum values are resolved in that class's + induced context (the container slot itself is induced against *cls*, + honouring a ``slot_usage`` range narrowing). Resolving against the + outer class instead would emit predicates the member nodes never carry + — the ``sh:path`` on the member shape and the SPARQL body would + diverge, making ``FILTER NOT EXISTS`` member checks vacuously true + (false positives) or preconditions never bind (false negatives). + + Like preconditions, combining operators on one condition emits all of + them. Returns ``None`` for unsupported inner operators or when the + container's range is not a class (inner conditions on a non-class + range cannot be resolved faithfully). """ + container = self._rule_slot(sv, container_slot_name, cls) + range_name = getattr(container, "range", None) + if not range_name or range_name not in sv.all_classes(): + return None + range_cls = sv.get_class(range_name) + lines: list[str] = [] for j, (inner_name, icond) in enumerate(slot_conditions.items()): - ipath = self._slot_uri(sv, inner_name, cls) + ipath = self._slot_uri(sv, inner_name, range_cls) + if ipath is None: + return None ivar = f"{node_var}_{j}" filters = self._scalar_filters( ivar, icond, - lambda v, inm=inner_name: self._resolve_member_enum_ref(sv, container_slot_name, inm, v), + lambda v, inm=inner_name: self._resolve_enum_value_ref(sv, inm, v, range_cls), ) if filters is None: return None @@ -698,37 +877,28 @@ class of *container_slot_name* — by a set of inner slot conditions. lines.extend(filters) return lines - def _resolve_member_enum_ref(self, sv, container_slot_name: str, inner_slot_name: str, value_name: str) -> str: - """Resolve an inner enum value to a SPARQL term using the *range class* - of the container slot. - - A slot such as ``type`` may be reused across classes with different - enum ranges (via ``slot_usage``); resolving through the container's - range class picks the correct enum. Falls back to - :meth:`_resolve_enum_value_ref` (and thence to a literal) when the - container is not a class, the inner slot is not on it, or the value has - no ``meaning``. - """ - container = sv.get_slot(container_slot_name) - range_class = container.range if container else None - if range_class and range_class in sv.all_classes() and inner_slot_name in sv.class_slots(range_class): - induced = sv.induced_slot(inner_slot_name, range_class) - if induced and induced.range in sv.all_enums(): - pv = sv.get_enum(induced.range).permissible_values.get(value_name) - if pv and pv.meaning: - return f"<{sv.expand_curie(pv.meaning)}>" - return self._resolve_enum_value_ref(sv, inner_slot_name, value_name) - @staticmethod - def _sparql_number(value) -> str: - """Render a numeric threshold bound as a SPARQL numeric literal. - - LinkML parses ``minimum_value`` / ``maximum_value`` as ``int`` or - ``float`` (possibly the ``extended_int`` / ``extended_float`` runtime - subclasses); ``str`` yields a plain numeric token - (e.g. ``4000`` or ``0.0``) that SPARQL compares with numeric promotion - against ``xsd:float`` / ``xsd:decimal`` data values. + def _sparql_number(value) -> str | None: + """Render a numeric threshold bound as a SPARQL numeric literal, or + ``None`` when the value is not a finite number (callers then skip the + rule). + + The metamodel range of ``minimum_value`` / ``maximum_value`` is + ``Anything``, so YAML strings, dates, booleans, ``.nan`` / ``.inf`` + all pass through SchemaView unchanged. Interpolating them raw is + unsound: ``"abc"`` yields unparsable SPARQL that poisons the whole + shapes graph at validation time, and a date like ``2020-01-01`` parses + as the arithmetic expression ``2020-01-01 = 2018`` and silently never + fires. Only ``int`` / finite ``float`` (including the + ``extended_int`` / ``extended_float`` runtime subclasses) are + rendered; their ``str`` yields a plain numeric token (e.g. ``4000`` or + ``0.0``) that SPARQL compares with numeric promotion against + ``xsd:float`` / ``xsd:decimal`` data values. """ + if isinstance(value, bool) or not isinstance(value, int | float): + return None + if isinstance(value, float) and not math.isfinite(value): + return None return str(value) @staticmethod @@ -756,7 +926,10 @@ def _postcondition_violation(self, sv, cls: ClassDefinition, slot_name: str, con """Translate a single postcondition slot condition into SPARQL that matches a *violation* of it. - Returns ``None`` for operators not handled here. + Returns ``None`` for operators not handled here, or when the condition + sets anything beyond the single operator a branch translates (dropping + a co-set operator would weaken the postcondition — the rule is skipped + instead). Supported operators: @@ -770,18 +943,22 @@ def _postcondition_violation(self, sv, cls: ClassDefinition, slot_name: str, con ``{group: Vehicle, type: front_fog_light}`` entry). """ path = self._slot_uri(sv, slot_name, cls) - if getattr(cond, "required", None) is True: + if path is None: + return None + op_fields = self._set_operator_fields(cond) + if op_fields == {"required"} and cond.required is True: return [f"FILTER NOT EXISTS {{ $this <{path}> ?post . }}"] - if getattr(cond, "value_presence", None) == PresenceEnum(PresenceEnum.ABSENT): + if op_fields == {"value_presence"} and cond.value_presence == PresenceEnum(PresenceEnum.ABSENT): return [f"$this <{path}> ?post ."] - has_member = getattr(cond, "has_member", None) - if ( - has_member is not None - and getattr(has_member, "range_expression", None) is not None - and getattr(has_member.range_expression, "slot_conditions", None) - ): + if op_fields == {"has_member"}: + has_member = cond.has_member + if self._set_operator_fields(has_member) != {"range_expression"}: + return None + range_expr = has_member.range_expression + if self._set_operator_fields(range_expr) != {"slot_conditions"}: + return None member_lines = [f"$this <{path}> ?mem ."] - inner = self._member_conditions(sv, cls, slot_name, "?mem", has_member.range_expression.slot_conditions) + inner = self._member_conditions(sv, cls, slot_name, "?mem", range_expr.slot_conditions) if inner is None: return None member_lines.extend(inner) @@ -789,11 +966,14 @@ def _postcondition_violation(self, sv, cls: ClassDefinition, slot_name: str, con return [f"FILTER NOT EXISTS {{ {block} }}"] return None - def _build_boolean_guard_sparql(self, sv, cls: ClassDefinition, flag_slot_name: str, value_slot_name: str) -> str: + def _build_boolean_guard_sparql( + self, sv, cls: ClassDefinition, flag_slot_name: str, value_slot_name: str + ) -> str | None: """Build a SPARQL SELECT query for the boolean-guard pattern. The query detects violations where the value property is present - but the boolean flag is absent or not ``true``. + but the boolean flag is absent or not ``true``. Returns ``None`` + (rule skipped) when either slot name resolves to no slot. Conforms to `SHACL §5.3.1 `_: @@ -801,6 +981,8 @@ def _build_boolean_guard_sparql(self, sv, cls: ClassDefinition, flag_slot_name: """ flag_uri = self._slot_uri(sv, flag_slot_name, cls) value_uri = self._slot_uri(sv, value_slot_name, cls) + if flag_uri is None or value_uri is None: + return None return ( f"SELECT $this WHERE {{\n" @@ -820,7 +1002,7 @@ def _build_presence_implies_value_sparql( value_slot_name: str, target_slot_name: str, allowed_values: list[str], - ) -> str: + ) -> str | None: """Build a SPARQL SELECT query for the presence-implies-value pattern. Detects violations where the *value slot* is present but the *target @@ -839,6 +1021,8 @@ def _build_presence_implies_value_sparql( """ value_uri = self._slot_uri(sv, value_slot_name, cls) target_uri = self._slot_uri(sv, target_slot_name, cls) + if value_uri is None or target_uri is None: + return None refs = ", ".join(self._resolve_enum_value_ref(sv, target_slot_name, v, cls) for v in allowed_values) return ( @@ -876,6 +1060,8 @@ def _build_exclusive_value_sparql( ``$this`` is pre-bound to each focus node. """ slot_uri = self._slot_uri(sv, slot_name, cls) + if slot_uri is None: + return None value_ref = self._resolve_enum_value_ref(sv, slot_name, value_name, cls) if max_card == 1: @@ -912,10 +1098,7 @@ def _resolve_enum_value_ref(self, sv, slot_name: str, value_name: str, cls: Clas (a class-specific enum) selects the correct permissible values instead of the base slot's enum. """ - if cls is not None and slot_name in sv.class_slots(cls.name): - slot = sv.induced_slot(slot_name, cls.name) - else: - slot = sv.get_slot(slot_name) + slot = self._rule_slot(sv, slot_name, cls) if cls is not None else sv.get_slot(slot_name) if slot: range_name = slot.range if range_name and range_name in sv.all_enums(): @@ -926,20 +1109,25 @@ def _resolve_enum_value_ref(self, sv, slot_name: str, value_name: str, cls: Clas return f"<{iri}>" return self._sparql_string_literal(value_name) - def _slot_uri(self, sv, slot_name: str, cls: ClassDefinition) -> str: - """Resolve a slot name to a full IRI string for use in SPARQL queries. + def _slot_uri(self, sv, slot_name: str, cls: ClassDefinition) -> str | None: + """Resolve a slot name to a full IRI string for use in SPARQL queries, + or ``None`` when the name resolves to no slot at all (callers then skip + the rule). Mirrors the resolution logic used for ``sh:path`` in the main slot loop, including the **induced** (class-specific) slot: a ``slot_usage`` override of ``slot_uri`` must yield the same IRI as ``sh:path``. Otherwise the SPARQL body would query a property the data never uses and - the constraint would silently never fire (a false negative). + the constraint would silently never fire (a false negative). A slot + that resolves but is not registered in the schema's element map falls + back to ``default_prefix:underscored_name``, again matching ``sh:path``; + an *unknown* name must NOT take that fallback — it would fabricate a + predicate no shape uses and emit a vacuous constraint. """ - if slot_name in sv.class_slots(cls.name): - slot = sv.induced_slot(slot_name, cls.name) - else: - slot = sv.get_slot(slot_name) - if slot and slot.name in sv.element_by_schema_map(): + slot = self._rule_slot(sv, slot_name, cls) + if slot is None: + return None + if slot.name in sv.element_by_schema_map(): return sv.get_uri(slot, expand=True) pfx = sv.schema.default_prefix return sv.expand_curie(f"{pfx}:{underscore(slot_name)}") diff --git a/tests/linkml/test_generators/test_shaclgen.py b/tests/linkml/test_generators/test_shaclgen.py index b0551ecc5e..ef5cd78403 100644 --- a/tests/linkml/test_generators/test_shaclgen.py +++ b/tests/linkml/test_generators/test_shaclgen.py @@ -2689,7 +2689,7 @@ def test_presence_implies_value_no_meaning_falls_back_to_literal(): query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) assert '"Manual"' in query, f"No-meaning enum should use literal '\"Manual\"', got:\n{query}" - assert "" not in query, "Should not emit as IRI when meaning is absent" + assert f"<{EX_PIV}Manual>" not in query, "Should not emit as IRI when meaning is absent" def test_presence_implies_value_message_from_description(): @@ -3825,3 +3825,855 @@ def test_rule_equals_string_special_chars_escaped(): prepareQuery(query) assert '\\"' in query, f"double quote must be escaped, got:\n{query}" assert "\\\\" in query, f"backslash must be escaped, got:\n{query}" + + +# =========================================================================== +# Audit-fix regression tests: operator exactness, nested-slot resolution, +# numeric bound gating, elseconditions warning +# =========================================================================== + +_PIV_EXTRA_PRE_SCHEMA_YAML = """ +id: https://example.org/piv-extra-pre +name: piv_extra_pre +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/piv-extra-pre/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + temp: + range: integer + slot_uri: ex:temp + mode: + range: string + slot_uri: ex:mode +classes: + Device: + class_uri: ex:Device + slots: [temp, mode] + rules: + - description: Above 100 the mode must be High (extra precondition operator). + preconditions: + slot_conditions: + temp: + value_presence: PRESENT + minimum_value: 100 + postconditions: + slot_conditions: + mode: + equals_string: "High" +""" + + +def test_rule_extra_precondition_operator_skipped(): + """A precondition combining PRESENT with a threshold must not dispatch to + presence-implies-value: dropping the threshold widens the trigger.""" + g = _parse_shacl(_PIV_EXTRA_PRE_SCHEMA_YAML) + shape = URIRef("https://example.org/piv-extra-pre/Device") + assert list(g.objects(shape, SH.sparql)) == [], "rule with an untranslated conjunct must be skipped" + + +def test_rule_extra_precondition_operator_pyshacl_end_to_end(): + """A device below the threshold satisfies the rule vacuously and must conform.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_PIV_EXTRA_PRE_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + data = """ + @prefix ex: . + + ex:cool a ex:Device ; ex:temp 50 ; ex:mode "Low" . + """ + conforms, _, txt = pyshacl.validate( + data_graph=data, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Below-threshold device must not be flagged:\n{txt}" + + +_POST_BOTH_EQUALS_SCHEMA_YAML = """ +id: https://example.org/post-both-equals +name: post_both_equals +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/post-both-equals/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + guard: + slot_uri: ex:guard + target: + slot_uri: ex:target +classes: + Thing: + class_uri: ex:Thing + slots: [guard, target] + rules: + - preconditions: + slot_conditions: + guard: + value_presence: PRESENT + postconditions: + slot_conditions: + target: + equals_string: "a" + equals_string_in: ["b", "c"] +""" + + +def test_rule_post_with_both_equals_forms_skipped(): + """equals_string and equals_string_in set together is ambiguous — skip, + do not let one form silently win.""" + g = _parse_shacl(_POST_BOTH_EQUALS_SCHEMA_YAML) + shape = URIRef("https://example.org/post-both-equals/Thing") + assert list(g.objects(shape, SH.sparql)) == [] + + +_MIXED_SCALAR_SCHEMA_YAML = """ +id: https://example.org/mixed-scalar +name: mixed_scalar +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/mixed-scalar/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + code: + slot_uri: ex:code + note: + slot_uri: ex:note +classes: + Obs: + class_uri: ex:Obs + slots: [code, note] + rules: + - preconditions: + slot_conditions: + code: + equals_string: fog + pattern: "^f" + postconditions: + slot_conditions: + note: + required: true +""" + + +def test_rule_recognized_plus_unrecognized_operator_skipped(): + """A condition mixing a supported operator (equals_string) with an + unsupported one (pattern) must skip — translating only the supported part + widens the trigger.""" + g = _parse_shacl(_MIXED_SCALAR_SCHEMA_YAML) + shape = URIRef("https://example.org/mixed-scalar/Obs") + assert list(g.objects(shape, SH.sparql)) == [] + + +_EXPR_ANY_OF_SCHEMA_YAML = """ +id: https://example.org/expr-any-of +name: expr_any_of +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/expr-any-of/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + code: + slot_uri: ex:code + other: + slot_uri: ex:other + note: + slot_uri: ex:note +classes: + Obs: + class_uri: ex:Obs + slots: [code, other, note] + rules: + - preconditions: + slot_conditions: + code: + equals_string: fog + any_of: + - slot_conditions: + other: + equals_string: x + - slot_conditions: + other: + equals_string: y + postconditions: + slot_conditions: + note: + required: true +""" + + +def test_rule_expression_level_any_of_skipped(): + """Expression-level any_of on the preconditions cannot be honoured by any + converter; dropping the branch widens the trigger, so the rule is skipped.""" + g = _parse_shacl(_EXPR_ANY_OF_SCHEMA_YAML) + shape = URIRef("https://example.org/expr-any-of/Obs") + assert list(g.objects(shape, SH.sparql)) == [] + + +def test_rule_expression_level_any_of_pyshacl_end_to_end(): + """An instance whose any_of branch is unmet satisfies the rule vacuously + and must conform.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_EXPR_ANY_OF_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + data = """ + @prefix ex: . + + ex:o a ex:Obs ; ex:code "fog" ; ex:other "z" . + """ + conforms, _, txt = pyshacl.validate( + data_graph=data, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Instance with unmet any_of branch must not be flagged:\n{txt}" + + +_POST_MIXED_SCHEMA_YAML = """ +id: https://example.org/post-mixed +name: post_mixed +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/post-mixed/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + guard: + slot_uri: ex:guard + target: + slot_uri: ex:target +classes: + Thing: + class_uri: ex:Thing + slots: [guard, target] + rules: + - preconditions: + slot_conditions: + guard: + equals_string: on + postconditions: + slot_conditions: + target: + required: true + pattern: "^x" +""" + + +def test_rule_post_mixed_operators_skipped(): + """A postcondition combining required with an untranslated operator must + skip — checking only required weakens the postcondition.""" + g = _parse_shacl(_POST_MIXED_SCHEMA_YAML) + shape = URIRef("https://example.org/post-mixed/Thing") + assert list(g.objects(shape, SH.sparql)) == [] + + +_ABSENT_COMBINED_SCHEMA_YAML = """ +id: https://example.org/absent-combined +name: absent_combined +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/absent-combined/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + count: + range: integer + slot_uri: ex:count + note: + slot_uri: ex:note +classes: + Obs: + class_uri: ex:Obs + slots: [count, note] + rules: + - preconditions: + slot_conditions: + count: + value_presence: ABSENT + minimum_value: 5 + postconditions: + slot_conditions: + note: + required: true +""" + + +def test_rule_absent_combined_with_bound_skipped(): + """value_presence ABSENT combined with another operator must skip: the + triple-binding translation would invert the declared trigger.""" + g = _parse_shacl(_ABSENT_COMBINED_SCHEMA_YAML) + shape = URIRef("https://example.org/absent-combined/Obs") + assert list(g.objects(shape, SH.sparql)) == [] + + +_STRING_TRUE_SCHEMA_YAML = """ +id: https://example.org/string-true +name: string_true +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/string-true/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + opt: + slot_uri: ex:opt + status: + range: string + slot_uri: ex:status +classes: + Conf: + class_uri: ex:Conf + slots: [opt, status] + rules: + - description: If opt is present, status must be the string "true". + preconditions: + slot_conditions: + opt: + value_presence: PRESENT + postconditions: + slot_conditions: + status: + equals_string: "true" +""" + + +def test_rule_equals_true_on_string_slot_uses_piv(): + """equals_string "true" on a NON-boolean slot must dispatch to + presence-implies-value (string comparison), not the boolean guard.""" + g = _parse_shacl(_STRING_TRUE_SCHEMA_YAML) + shape = URIRef("https://example.org/string-true/Conf") + sparql_nodes = list(g.objects(shape, SH.sparql)) + assert len(sparql_nodes) == 1 + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "NOT IN" in query, f"string-range 'true' must be a string comparison, got:\n{query}" + assert '"true"' in query, "the comparison term must be the string literal" + + +def test_rule_equals_true_on_string_slot_pyshacl_end_to_end(): + """status "true" (string) satisfies the rule; the boolean-guard hijack used + to flag it.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_STRING_TRUE_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + conforming = """ + @prefix ex: . + + ex:ok a ex:Conf ; ex:opt "x" ; ex:status "true" . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"status 'true' satisfies the rule and must conform:\n{txt}" + + violating = """ + @prefix ex: . + + ex:bad a ex:Conf ; ex:opt "x" ; ex:status "other" . + """ + conforms, _, txt = pyshacl.validate( + data_graph=violating, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"status 'other' violates the rule:\n{txt}" + + +_INNER_OVERRIDE_SCHEMA_YAML = """ +id: https://example.org/inner-override +name: inner_override +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/inner-override/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + sun_position: + range: SunPosition + inlined: true + slot_uri: ex:sun_position + elevation: + range: float + slot_uri: ex:elevation + headlight_note: + slot_uri: ex:headlight_note +classes: + SunPosition: + class_uri: ex:SunPosition + slots: [elevation] + slot_usage: + elevation: + slot_uri: ex:localElevation + Scene: + class_uri: ex:Scene + slots: [sun_position, headlight_note] + rules: + - description: Below the horizon a headlight note is required. + preconditions: + slot_conditions: + sun_position: + range_expression: + slot_conditions: + elevation: + maximum_value: 0 + postconditions: + slot_conditions: + headlight_note: + required: true +""" + + +def test_rule_nested_inner_slot_uri_resolved_on_range_class(): + """The inner slot of a nested precondition lives on the container's range + class; a slot_usage slot_uri override there must be honoured (sh:path / + SPARQL-body parity one hop down).""" + g = _parse_shacl(_INNER_OVERRIDE_SCHEMA_YAML) + shape = URIRef("https://example.org/inner-override/Scene") + sparql_nodes = list(g.objects(shape, SH.sparql)) + assert len(sparql_nodes) == 1 + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "https://example.org/inner-override/localElevation" in query, ( + f"inner slot must use the range class's induced slot_uri, got:\n{query}" + ) + assert "https://example.org/inner-override/elevation" not in query, ( + "the base slot_uri must not leak into the member pattern" + ) + + +def test_rule_nested_inner_slot_uri_override_pyshacl_end_to_end(): + """A night scene without the required note must be flagged — with the + base-URI mistranslation the constraint silently never fired.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_INNER_OVERRIDE_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + # The float is typed explicitly so the sh:datatype property constraint is + # satisfied and the outcome discriminates on the rule constraint alone. + violating = """ + @prefix ex: . + @prefix xsd: . + + ex:night a ex:Scene ; ex:sun_position ex:sp . + ex:sp a ex:SunPosition ; ex:localElevation "-5.0"^^xsd:float . + """ + conforms, _, txt = pyshacl.validate( + data_graph=violating, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"Night scene without headlight note must fail:\n{txt}" + + conforming = """ + @prefix ex: . + @prefix xsd: . + + ex:noon a ex:Scene ; ex:sun_position ex:sp2 . + ex:sp2 a ex:SunPosition ; ex:localElevation "45.0"^^xsd:float . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Daytime scene needs no headlight note:\n{txt}" + + +_INNER_COLLISION_SCHEMA_YAML = """ +id: https://example.org/inner-collision +name: inner_collision +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/inner-collision/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + marker_flag: + slot_uri: ex:marker_flag + items: + range: Item + multivalued: true + inlined: true + inlined_as_list: true + slot_uri: ex:items + type: + slot_uri: ex:defaultType +classes: + Item: + class_uri: ex:Item + slots: [type] + slot_usage: + type: + slot_uri: ex:itemType + Box: + class_uri: ex:Box + slots: [marker_flag, items, type] + slot_usage: + type: + slot_uri: ex:boxType + rules: + - description: A flagged box must contain a marker item. + preconditions: + slot_conditions: + marker_flag: + value_presence: PRESENT + postconditions: + slot_conditions: + items: + has_member: + range_expression: + slot_conditions: + type: + equals_string: marker +""" + + +def test_rule_has_member_inner_slot_not_shadowed_by_outer_class(): + """An inner slot name that also exists on the OUTER class with a different + slot_usage URI must still resolve against the member class.""" + g = _parse_shacl(_INNER_COLLISION_SCHEMA_YAML) + shape = URIRef("https://example.org/inner-collision/Box") + sparql_nodes = list(g.objects(shape, SH.sparql)) + assert len(sparql_nodes) == 1 + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "https://example.org/inner-collision/itemType" in query, ( + f"member condition must use the member class's slot URI, got:\n{query}" + ) + assert "boxType" not in query, "the outer class's slot_usage URI must not shadow the member's" + + +def test_rule_has_member_inner_slot_collision_pyshacl_end_to_end(): + """A conforming box (marker item present via the member class's predicate) + must conform — the outer-class shadowing made FILTER NOT EXISTS vacuous.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_INNER_COLLISION_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + conforming = """ + @prefix ex: . + + ex:b a ex:Box ; ex:marker_flag "y" ; ex:items ex:i1 . + ex:i1 a ex:Item ; ex:itemType "marker" . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Box with a marker item must conform:\n{txt}" + + +_CONTAINER_NARROWED_SCHEMA_YAML = """ +id: https://example.org/container-narrowed +name: container_narrowed +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/container-narrowed/ +imports: + - linkml:types +default_prefix: ex +default_range: string +enums: + BaseKindEnum: + permissible_values: + special: + meaning: ex:BASE_special + SpecialKindEnum: + permissible_values: + special: + meaning: ex:SPECIAL_special +slots: + part: + range: BasePart + inlined: true + slot_uri: ex:part + kind: + range: BaseKindEnum + slot_uri: ex:kind + label_note: + slot_uri: ex:label_note +classes: + BasePart: + class_uri: ex:BasePart + slots: [kind] + SpecialPart: + class_uri: ex:SpecialPart + is_a: BasePart + slot_usage: + kind: + range: SpecialKindEnum + Assembly: + class_uri: ex:Assembly + slots: [part, label_note] + slot_usage: + part: + range: SpecialPart + rules: + - description: A special part requires a label note. + preconditions: + slot_conditions: + part: + range_expression: + slot_conditions: + kind: + equals_string: special + postconditions: + slot_conditions: + label_note: + required: true +""" + + +def test_rule_container_range_narrowing_resolves_inner_enum(): + """A slot_usage range-narrowing of the CONTAINER slot must resolve inner + enum values against the narrowed range class's enum.""" + g = _parse_shacl(_CONTAINER_NARROWED_SCHEMA_YAML) + shape = URIRef("https://example.org/container-narrowed/Assembly") + sparql_nodes = list(g.objects(shape, SH.sparql)) + assert len(sparql_nodes) == 1 + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "SPECIAL_special" in query, f"inner enum must resolve via the narrowed range, got:\n{query}" + assert "BASE_special" not in query, "the base range's enum must not be used" + + +_NON_NUMERIC_BOUNDS_SCHEMA_YAML = """ +id: https://example.org/non-numeric-bounds +name: non_numeric_bounds +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/non-numeric-bounds/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + v: + range: integer + slot_uri: ex:v + note: + slot_uri: ex:note +classes: + Obs: + class_uri: ex:Obs + slots: [v, note] + rules: + - preconditions: + slot_conditions: + v: + minimum_value: "abc" + postconditions: + slot_conditions: + note: + required: true + - preconditions: + slot_conditions: + v: + minimum_value: 2020-01-01 + postconditions: + slot_conditions: + note: + required: true + - preconditions: + slot_conditions: + v: + maximum_value: .nan + postconditions: + slot_conditions: + note: + required: true + - preconditions: + slot_conditions: + v: + minimum_value: true + postconditions: + slot_conditions: + note: + required: true +""" + + +def test_rule_non_numeric_bounds_skipped(): + """Non-numeric threshold bounds (string, date, NaN, boolean) must skip the + rule: raw interpolation produced unparsable SPARQL (poisoning the whole + shapes graph) or silently-wrong arithmetic (2020-01-01 == 2018).""" + g = _parse_shacl(_NON_NUMERIC_BOUNDS_SCHEMA_YAML) + shape = URIRef("https://example.org/non-numeric-bounds/Obs") + assert list(g.objects(shape, SH.sparql)) == [] + + +def test_rule_non_numeric_bounds_shapes_graph_still_validates(): + """The generated shapes graph must remain usable by pyshacl — one bad bound + used to raise a ParseException for every validation run.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_NON_NUMERIC_BOUNDS_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + data = """ + @prefix ex: . + + ex:o a ex:Obs ; ex:v 1 . + """ + conforms, _, txt = pyshacl.validate( + data_graph=data, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Shapes graph must stay parseable and the data conform:\n{txt}" + + +def test_rule_with_elseconditions_warns(caplog): + """The unenforced else branch must be signalled, consistent with the + bidirectional/open_world warnings.""" + import logging + + with caplog.at_level(logging.WARNING, logger="linkml.generators.shaclgen"): + _parse_shacl(_ELSE_COND_SCHEMA_YAML) + assert any("elseconditions" in rec.message for rec in caplog.records), ( + "dropping the else branch must produce a warning" + ) + + +def test_has_member_zero_members_pyshacl_end_to_end(): + """A node meeting the precondition with ZERO members violates has_member + ('must contain a matching member'); locks the semantics in.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_HAS_MEMBER_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + violating = """ + @prefix ex: . + + ex:wZero a ex:Weather ; ex:fog_declared "fog" . + """ + conforms, _, txt = pyshacl.validate( + data_graph=violating, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"Zero members cannot contain the required member:\n{txt}" + + +_ALIAS_KEY_SCHEMA_YAML = """ +id: https://example.org/alias-key +name: alias_key +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/alias-key/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + my slot: + slot_uri: ex:customMySlot + note: + slot_uri: ex:note +classes: + Obs: + class_uri: ex:Obs + slots: ["my slot", note] + rules: + - description: Underscored alias key must resolve to the declared slot. + preconditions: + slot_conditions: + my_slot: + equals_string: trigger + postconditions: + slot_conditions: + note: + required: true +""" + + +def test_rule_alias_form_slot_key_resolves_override(): + """A rule key written `my_slot` for a slot named `my slot` must resolve to + that slot's URI (sh:path parity) instead of fabricating a default-prefix + predicate that makes the constraint vacuous.""" + g = _parse_shacl(_ALIAS_KEY_SCHEMA_YAML) + shape = URIRef("https://example.org/alias-key/Obs") + sparql_nodes = list(g.objects(shape, SH.sparql)) + assert len(sparql_nodes) == 1 + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "https://example.org/alias-key/customMySlot" in query, ( + f"alias-form key must resolve to the declared slot_uri, got:\n{query}" + ) + assert "https://example.org/alias-key/my_slot" not in query, ( + "the fabricated default-prefix predicate must not be emitted" + ) + + +_UNKNOWN_KEY_SCHEMA_YAML = """ +id: https://example.org/unknown-key +name: unknown_key +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/unknown-key/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + code: + slot_uri: ex:code + note: + slot_uri: ex:note +classes: + Obs: + class_uri: ex:Obs + slots: [code, note] + rules: + - description: A rule keyed on a nonexistent slot must be skipped. + preconditions: + slot_conditions: + no_such_slot: + equals_string: trigger + postconditions: + slot_conditions: + note: + required: true +""" + + +def test_rule_unknown_slot_key_skipped(): + """A rule whose condition keys a slot that does not exist must be skipped: + fabricating a default-prefix predicate would emit a constraint that can + never fire (or, for has_member, always fires).""" + g = _parse_shacl(_UNKNOWN_KEY_SCHEMA_YAML) + shape = URIRef("https://example.org/unknown-key/Obs") + assert list(g.objects(shape, SH.sparql)) == []