From e161ce7c13901d6f0ec3437e23787ec297d3a3aa Mon Sep 17 00:00:00 2001 From: Rayene Messaoud Date: Fri, 11 Sep 2026 14:45:34 +0200 Subject: [PATCH] feat(gen-shacl): translate presence-implies-value rules to SHACL-SPARQL The rules-to-SHACL-SPARQL converter added in #3451 recognised a single named pattern. This adds the presence-implies-value pattern: a precondition asserting `value_presence: PRESENT` on one slot, and a postcondition constraining another slot with `equals_string` or `equals_string_in`. It reads as "if the guard slot is present, the target slot must be present and hold one of the allowed values", and generalises the existing boolean guard to arbitrary enum values. The boolean guard is now gated on the target slot's range actually being `boolean`. Without that gate a slot of range `string` carrying `equals_string: "true"` was translated as a boolean comparison, which does not match the string `"true"` in the data and so flagged conforming instances as violations. String-ranged slots now fall through to the presence-implies-value pattern and compare as strings. Pattern matching is exact: each converter requires its conditions to set precisely the operators it translates. A rule whose conditions carry anything further -- extra scalar operators, or expression-level any_of / all_of / none_of / exactly_one_of -- is skipped rather than partially translated, since dropping a term would either widen the precondition (false positives) or weaken the postcondition (false negatives). Slot resolution goes through induced slots so that `slot_usage` overrides, `slot_uri` overrides and alias-form keys resolve to the same IRI that `sh:path` emits. Co-authored-by: jdsika --- .../linkml/src/linkml/generators/shaclgen.py | 319 +++++- tests/linkml/test_generators/test_shaclgen.py | 925 ++++++++++++++++++ 2 files changed, 1204 insertions(+), 40 deletions(-) diff --git a/packages/linkml/src/linkml/generators/shaclgen.py b/packages/linkml/src/linkml/generators/shaclgen.py index 4731b9f0b8..325cd68c09 100644 --- a/packages/linkml/src/linkml/generators/shaclgen.py +++ b/packages/linkml/src/linkml/generators/shaclgen.py @@ -147,11 +147,15 @@ class ShaclGenerator(Generator): When ``True`` (default), recognised rule patterns are translated into SHACL-SPARQL constraints (``sh:SPARQLConstraint``) on the corresponding - ``sh:NodeShape``. Currently two patterns are recognised: + ``sh:NodeShape``. Currently three patterns are recognised: * *Boolean guard* — a precondition with ``value_presence: PRESENT`` on a value slot and a postcondition with ``equals_string: "true"`` on a boolean flag slot. + * *Presence implies value* — a precondition with ``value_presence: PRESENT`` + on a value slot and a postcondition with ``equals_string`` or + ``equals_string_in`` on a (typically enum-valued) target slot. This + generalises the boolean guard to arbitrary required values. * *Exclusive value* — a precondition with ``equals_string`` on a slot and a postcondition with ``maximum_cardinality`` on the *same* slot. @@ -429,12 +433,22 @@ def _add_rules(self, g: Graph, shape_uri: URIRef, cls: ClassDefinition) -> None: ``value_presence: PRESENT`` on a value slot and a *postcondition* with ``equals_string: "true"`` on a boolean flag slot. + * **Presence implies value** — a *precondition* with + ``value_presence: PRESENT`` on a value slot and a *postcondition* + with ``equals_string`` or ``equals_string_in`` on a target slot. + Enforces that when the value slot is present, the target slot must + be present and hold one of the allowed values (generalises the + boolean guard to enum-valued targets). + * **Exclusive value** — a *precondition* with ``equals_string`` on a slot and a *postcondition* with ``maximum_cardinality`` on the *same* slot. Enforces that when a specific value is present in a multivalued slot, the total number of values must not exceed the given cardinality (typically 1 for mutual exclusion). + Operator combinations outside these named patterns are not translated; + the rule is skipped rather than partially represented. + See `W3C SHACL §5 `_. """ if not cls.rules: @@ -462,11 +476,11 @@ def _add_rules(self, g: Graph, shape_uri: URIRef, cls: ClassDefinition) -> None: cls.name, ) - if getattr(rule, "elseconditions", None): + if getattr(rule, "elseconditions", None) is not None: logger.warning( "Rule in class %r has elseconditions; " - "only the forward (if/then) branch is emitted as sh:sparql. " - "The else branch cannot be represented in SHACL-SPARQL.", + "SHACL-SPARQL generation emits the forward (if/then) direction only. " + "The else branch is not enforced.", cls.name, ) @@ -489,22 +503,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)) @@ -512,31 +641,83 @@ 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] - # Note: PresenceEnum.PRESENT is a PermissibleValue, but parsed schemas - # return PresenceEnum instances — wrapping ensures type-compatible comparison. - 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 (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. + 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 return None - def _build_boolean_guard_sparql(self, sv, cls: ClassDefinition, flag_slot_name: str, value_slot_name: str) -> str: + @staticmethod + def _sparql_string_literal(value: str) -> str: + """Render *value* as a double-quoted SPARQL string literal, escaping the + characters the grammar forbids raw. + + ``equals_string`` / permissible-value names are schema-controlled but + may legitimately contain a double quote, backslash, or newline; without + escaping these would break the ``sh:select`` query (or allow SPARQL + injection). See `SPARQL 1.1 §19.7 escape sequences + `_. + """ + escaped = ( + str(value) + .replace("\\", "\\\\") + .replace('"', '\\"') + .replace("\n", "\\n") + .replace("\r", "\\r") + .replace("\t", "\\t") + ) + return f'"{escaped}"' + + 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 `_: @@ -544,6 +725,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" @@ -556,6 +739,44 @@ def _build_boolean_guard_sparql(self, sv, cls: ClassDefinition, flag_slot_name: f"}}" ) + def _build_presence_implies_value_sparql( + self, + sv, + cls: ClassDefinition, + value_slot_name: str, + target_slot_name: str, + allowed_values: list[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 + slot* is absent or holds a value outside the allowed set. This + generalises the boolean-guard pattern to enum-valued targets: it + supports a single required value (``equals_string``) or a set of + acceptable values (``equals_string_in``). + + Each allowed value is resolved via the target slot's enum ``meaning`` + to a full IRI; values without a ``meaning`` (or non-enum targets) fall + back to a plain string literal. + + Conforms to `SHACL §5.3.1 + `_: + ``$this`` is pre-bound to each focus node. + """ + 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 ( + f"SELECT $this WHERE {{\n" + f" $this <{value_uri}> ?value .\n" + f" OPTIONAL {{ $this <{target_uri}> ?target . }}\n" + f" FILTER ( !BOUND(?target) || ?target NOT IN ({refs}) )\n" + f"}}" + ) + def _build_exclusive_value_sparql( self, sv, @@ -583,7 +804,9 @@ def _build_exclusive_value_sparql( ``$this`` is pre-bound to each focus node. """ slot_uri = self._slot_uri(sv, slot_name, cls) - value_ref = self._resolve_enum_value_ref(sv, slot_name, value_name) + if slot_uri is None: + return None + value_ref = self._resolve_enum_value_ref(sv, slot_name, value_name, cls) if max_card == 1: return ( @@ -606,15 +829,20 @@ def _build_exclusive_value_sparql( f"}}" ) - def _resolve_enum_value_ref(self, sv, slot_name: str, value_name: str) -> str: + def _resolve_enum_value_ref(self, sv, slot_name: str, value_name: str, cls: ClassDefinition | None = None) -> str: """Resolve an enum value name to a SPARQL term (IRI or literal). Looks up the slot's range as an enum, finds the permissible value matching *value_name*, and returns its ``meaning`` as a full IRI - wrapped in angle brackets. Falls back to a quoted literal if the - slot is not an enum or the value lacks a ``meaning``. + wrapped in angle brackets. Falls back to an escaped quoted literal if + the slot is not an enum or the value lacks a ``meaning``. + + When *cls* is given and the slot is declared on it, the slot is resolved + in the class's induced context, so a range narrowed via ``slot_usage`` + (a class-specific enum) selects the correct permissible values instead + of the base slot's enum. """ - 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(): @@ -623,17 +851,27 @@ def _resolve_enum_value_ref(self, sv, slot_name: str, value_name: str) -> str: if pv and pv.meaning: iri = sv.expand_curie(pv.meaning) return f"<{iri}>" - return f'"{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. - - Mirrors the resolution logic used for ``sh:path`` in the main slot loop: - prefer ``sv.get_uri()`` for slots registered in the schema map, fall - back to ``default_prefix:underscored_name``. + return self._sparql_string_literal(value_name) + + 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). 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. """ - 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)}") @@ -924,8 +1162,9 @@ def add_simple_data_type(func: Callable, r: ElementName) -> None: show_default=True, help=( "Emit sh:sparql constraints from LinkML rules: blocks. " - "When enabled (default), recognised rule patterns (e.g. boolean-guard) " - "are translated into SHACL-SPARQL constraints on the corresponding " + "When enabled (default), recognised rule patterns (boolean-guard, " + "presence-implies-value, exclusive-value) are translated into " + "SHACL-SPARQL constraints on the corresponding " "sh:NodeShape. Use --no-emit-rules to suppress rule generation." ), ) diff --git a/tests/linkml/test_generators/test_shaclgen.py b/tests/linkml/test_generators/test_shaclgen.py index 8604f712de..2fb7594212 100644 --- a/tests/linkml/test_generators/test_shaclgen.py +++ b/tests/linkml/test_generators/test_shaclgen.py @@ -2784,3 +2784,928 @@ def test_exclusive_value_coexists_with_boolean_guard(): has_boolean = any("BOUND" in q for q in queries) assert has_exclusive, "Expected one exclusive-value SPARQL constraint" assert has_boolean, "Expected one boolean-guard SPARQL constraint" + + +# =========================================================================== +# Presence-implies-value pattern tests (enum guard) +# =========================================================================== +# +# The "presence implies value" pattern generalises the boolean guard to +# enum-valued targets. It translates a LinkML rule where: +# - preconditions: a value slot has value_presence: PRESENT +# - postconditions: a target slot has equals_string (single required value) +# or equals_string_in (a set of acceptable values) +# +# Semantics: "If the value slot is present, the target slot must be present +# and hold one of the allowed values." The motivating use case is the aiSim +# environment model, e.g. "if texture_sky_color is set, sky_model must be +# TextureSky" and "if overcast_sky_illuminance is set, sky_model must be an +# overcast model". +# +# References: +# - W3C SHACL §5 +# - W3C SHACL §5.3.1 +# =========================================================================== + +_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML = """ +id: https://example.org/presence-implies-value +name: presence_implies_value_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/presence-implies-value/ +imports: + - linkml:types +default_prefix: ex +default_range: string + +enums: + SkyModelEnum: + permissible_values: + ClearSky: + meaning: ex:ClearSky + OvercastSky: + meaning: ex:OvercastSky + MeasuredOvercastSky: + meaning: ex:MeasuredOvercastSky + TextureSky: + meaning: ex:TextureSky + + ModeEnum: + permissible_values: + Auto: + description: Automatic mode (no meaning IRI). + Manual: + description: Manual mode (no meaning IRI). + +slots: + sky_model: + range: SkyModelEnum + slot_uri: ex:sky_model + texture_sky_color: + range: string + slot_uri: ex:texture_sky_color + overcast_sky_illuminance: + range: float + slot_uri: ex:overcast_sky_illuminance + mode: + range: ModeEnum + slot_uri: ex:mode + manual_value: + range: decimal + slot_uri: ex:manual_value + +classes: + Weather: + class_uri: ex:Weather + slots: + - sky_model + - texture_sky_color + - overcast_sky_illuminance + rules: + - description: If texture_sky_color is provided, sky_model must be TextureSky. + preconditions: + slot_conditions: + texture_sky_color: + value_presence: PRESENT + postconditions: + slot_conditions: + sky_model: + equals_string: "TextureSky" + - description: If overcast_sky_illuminance is provided, sky_model must be an overcast model. + preconditions: + slot_conditions: + overcast_sky_illuminance: + value_presence: PRESENT + postconditions: + slot_conditions: + sky_model: + equals_string_in: + - OvercastSky + - MeasuredOvercastSky + + Device: + class_uri: ex:Device + slots: + - mode + - manual_value + rules: + - description: If manual_value is provided, mode must be Manual (literal fallback). + preconditions: + slot_conditions: + manual_value: + value_presence: PRESENT + postconditions: + slot_conditions: + mode: + equals_string: "Manual" +""" + +EX_PIV = rdflib.Namespace("https://example.org/presence-implies-value/") + + +def test_presence_implies_value_generates_sparql(): + """Presence-implies-value rules produce sh:sparql constraints on the NodeShape.""" + g = _parse_shacl(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML) + + shape = EX_PIV.Weather + sparql_nodes = list(g.objects(shape, SH.sparql)) + assert len(sparql_nodes) == 2, f"Expected 2 sh:sparql constraints, got {len(sparql_nodes)}" + + for node in sparql_nodes: + assert (node, RDF.type, SH.SPARQLConstraint) in g + selects = list(g.objects(node, SH.select)) + assert len(selects) == 1, "Each constraint must have exactly one sh:select" + query = str(selects[0]) + assert "$this" in query, "SPARQL must use $this pre-bound variable" + assert "NOT IN" in query, "presence-implies-value SPARQL must use NOT IN membership test" + assert "FILTER" in query, "SPARQL must have a FILTER clause" + + +def test_presence_implies_value_single_uses_enum_iri(): + """A single equals_string target resolves to the enum meaning IRI.""" + g = _parse_shacl(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML) + + shape = EX_PIV.Weather + sparql_nodes = list(g.objects(shape, SH.sparql)) + queries = [str(list(g.objects(n, SH.select))[0]) for n in sparql_nodes] + + texture_query = [q for q in queries if "texture_sky_color" in q] + assert len(texture_query) == 1, "Expected exactly one texture_sky_color rule" + query = texture_query[0] + + # value slot and target slot URIs both present + assert str(EX_PIV.texture_sky_color) in query + assert str(EX_PIV.sky_model) in query + # target value resolves to the TextureSky meaning IRI in angle brackets + assert f"<{EX_PIV.TextureSky}>" in query, f"Expected TextureSky IRI, got:\n{query}" + + +def test_presence_implies_value_set_uses_all_iris(): + """equals_string_in resolves every allowed value to its enum meaning IRI.""" + g = _parse_shacl(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML) + + shape = EX_PIV.Weather + sparql_nodes = list(g.objects(shape, SH.sparql)) + queries = [str(list(g.objects(n, SH.select))[0]) for n in sparql_nodes] + + overcast_query = [q for q in queries if "overcast_sky_illuminance" in q] + assert len(overcast_query) == 1, "Expected exactly one overcast rule" + query = overcast_query[0] + + assert f"<{EX_PIV.OvercastSky}>" in query, f"Expected OvercastSky IRI, got:\n{query}" + assert f"<{EX_PIV.MeasuredOvercastSky}>" in query, f"Expected MeasuredOvercastSky IRI, got:\n{query}" + + +def test_presence_implies_value_no_meaning_falls_back_to_literal(): + """When the target enum value lacks a meaning IRI, it is compared as a literal.""" + g = _parse_shacl(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML) + + shape = EX_PIV.Device + 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 '"Manual"' in query, f"No-meaning enum should use literal '\"Manual\"', got:\n{query}" + 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(): + """Rule description is emitted as sh:message on the SPARQLConstraint.""" + g = _parse_shacl(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML) + + shape = EX_PIV.Weather + sparql_nodes = list(g.objects(shape, SH.sparql)) + messages = [str(m) for node in sparql_nodes for m in g.objects(node, SH.message)] + + assert any("sky_model must be TextureSky" in m for m in messages), ( + f"Expected message about TextureSky, got: {messages}" + ) + + +def test_presence_implies_value_sparql_syntax_valid(): + """Generated SPARQL for presence-implies-value rules must be syntactically valid.""" + from rdflib.plugins.sparql import prepareQuery + + g = _parse_shacl(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML) + + for shape in (EX_PIV.Weather, EX_PIV.Device): + sparql_nodes = list(g.objects(shape, SH.sparql)) + for node in sparql_nodes: + query_text = str(list(g.objects(node, SH.select))[0]) + prepareQuery(query_text) + + +def test_presence_implies_value_pyshacl_end_to_end(): + """End-to-end: pyshacl passes conforming instances and flags violations.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + + # Conforming: guarded slots paired with an allowed sky_model; and an + # unguarded instance (no texture/overcast) is unaffected by the rules. + conforming_data = """ + @prefix ex: . + @prefix xsd: . + + ex:wTexture a ex:Weather ; + ex:texture_sky_color "0,0,0" ; + ex:sky_model ex:TextureSky . + + ex:wOvercast a ex:Weather ; + ex:overcast_sky_illuminance "5000.0"^^xsd:float ; + ex:sky_model ex:OvercastSky . + + ex:wMeasured a ex:Weather ; + ex:overcast_sky_illuminance "4200.0"^^xsd:float ; + ex:sky_model ex:MeasuredOvercastSky . + + ex:wClear a ex:Weather ; + ex:sky_model ex:ClearSky . + """ + + conforms, _, results_text = pyshacl.validate( + data_graph=conforming_data, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Conforming instances should pass SHACL validation:\n{results_text}" + + # Violating: texture_sky_color present but sky_model is ClearSky (not TextureSky). + violating_wrong_value = """ + @prefix ex: . + + ex:wBad a ex:Weather ; + ex:texture_sky_color "0,0,0" ; + ex:sky_model ex:ClearSky . + """ + conforms, _, results_text = pyshacl.validate( + data_graph=violating_wrong_value, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"Wrong-value instance should fail SHACL validation:\n{results_text}" + + # Violating: overcast_sky_illuminance present but sky_model is TextureSky + # (not in the allowed overcast set). + violating_not_in_set = """ + @prefix ex: . + @prefix xsd: . + + ex:wBad2 a ex:Weather ; + ex:overcast_sky_illuminance "5000.0"^^xsd:float ; + ex:sky_model ex:TextureSky . + """ + conforms, _, results_text = pyshacl.validate( + data_graph=violating_not_in_set, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"Not-in-set instance should fail SHACL validation:\n{results_text}" + + # Violating: texture_sky_color present but sky_model entirely absent. + violating_missing_target = """ + @prefix ex: . + + ex:wBad3 a ex:Weather ; + ex:texture_sky_color "0,0,0" . + """ + conforms, _, results_text = pyshacl.validate( + data_graph=violating_missing_target, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"Missing-target instance should fail SHACL validation:\n{results_text}" + + +# =========================================================================== +# Compositional fallback: conditional-required pattern (M1) +# =========================================================================== +# +# Rule shape: +# - preconditions: slot X has equals_string V +# - postconditions: slot Y has required: true +# +# Semantics: "If X = V, then Y must be present." Emitted as an +# sh:SPARQLConstraint whose SELECT matches focus nodes where the precondition +# holds but the required slot is absent (FILTER NOT EXISTS). +# =========================================================================== + + +# =========================================================================== +# Compositional fallback: conditional-absent pattern (M2) +# =========================================================================== +# +# Rule shape: +# - preconditions: slot X has equals_string V +# - postconditions: slot Y has value_presence: ABSENT +# +# Semantics: "If X = V, then Y must NOT be present" (inapplicable slot). +# Emitted as an sh:SPARQLConstraint whose SELECT matches focus nodes where the +# precondition holds and the forbidden slot is present. +# =========================================================================== + + +# =========================================================================== +# Compositional fallback: numeric threshold precondition (M3) +# =========================================================================== +# +# Rule shape: +# - preconditions: slot X has maximum_value N (or minimum_value) +# - postconditions: slot Y has required: true +# +# Semantics: "If X <= N, then Y must be present." The threshold becomes a +# SPARQL FILTER; combined here with the M1 required violation. +# =========================================================================== + + +# =========================================================================== +# Compositional fallback: nested range_expression precondition (M4) +# =========================================================================== +# +# Rule shape: +# - preconditions: slot X (inlined child) has range_expression on an inner +# slot (e.g. sun_position.elevation <= 0) +# - postconditions: slot Y has required: true +# +# Semantics: "If the child's inner value satisfies the condition, then Y must +# be present." The SPARQL binds the child node with one extra hop. +# =========================================================================== + + +# =========================================================================== +# Compositional fallback: has_member list-membership postcondition (M5) +# =========================================================================== +# +# Rule shape: +# - preconditions: any supported precondition (here value_presence PRESENT) +# - postconditions: multivalued slot has_member with a nested +# range_expression constraining the member's inner slots +# +# Semantics: "If the precondition holds, the list must contain a member +# matching the inner conditions." Violation = no such member (FILTER NOT +# EXISTS over the members). Inner enum values resolve against the member +# class (LightControlGroup), which disambiguates the reused `type` slot. +# =========================================================================== + + +# =========================================================================== +# Rule-converter robustness regressions (review hardening) +# +# These guard three defects found while reviewing the rule converters: +# 1. A single precondition combining minimum_value + maximum_value dropped +# all but the first bound (silent under-constraint / false positives). +# 2. A slot_usage `slot_uri` (or enum `range`) override made the SPARQL body +# query the *base* IRI while `sh:path` used the *induced* IRI, so the +# constraint silently never fired (false negative). +# 3. An `equals_string` value containing a quote/backslash produced invalid, +# unparsable SPARQL (broken artifact / injection). +# =========================================================================== + + +_ENUM_NARROWING_SCHEMA_YAML = """ +id: https://example.org/enum-narrowing +name: enum_narrowing_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/enum-narrowing/ +imports: + - linkml:types +default_prefix: ex +default_range: string + +enums: + BaseMode: + permissible_values: + Active: + meaning: ex:GLOBAL_Active + SceneMode: + permissible_values: + Active: + meaning: ex:LOCAL_Active + +slots: + activator: + range: string + slot_uri: ex:activator + mode: + range: BaseMode + slot_uri: ex:mode + +classes: + Scene: + class_uri: ex:Scene + slots: + - activator + - mode + slot_usage: + mode: + range: SceneMode + rules: + - description: If an activator is present the mode must be Active. + preconditions: + slot_conditions: + activator: + value_presence: PRESENT + postconditions: + slot_conditions: + mode: + equals_string: Active +""" + +EX_EN = rdflib.Namespace("https://example.org/enum-narrowing/") + + +def test_rule_enum_range_narrowed_by_slot_usage(): + """A slot_usage range override to a class-specific enum must resolve the + value's meaning against the induced (narrowed) enum, not the base range.""" + g = _parse_shacl(_ENUM_NARROWING_SCHEMA_YAML) + + nodes = list(g.objects(EX_EN.Scene, SH.sparql)) + assert len(nodes) == 1 + query = str(list(g.objects(nodes[0], SH.select))[0]) + assert str(EX_EN.LOCAL_Active) in query, f"must resolve the narrowed enum meaning, got:\n{query}" + assert "GLOBAL_Active" not in query, f"must not resolve the base enum meaning, got:\n{query}" + + +_ESCAPING_SCHEMA_YAML = """ +id: https://example.org/escaping +name: escaping_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/escaping/ +imports: + - linkml:types +default_prefix: ex +default_range: string + +slots: + trigger: + range: string + slot_uri: ex:trigger + label: + range: string + slot_uri: ex:label + +classes: + Item: + class_uri: ex:Item + slots: + - trigger + - label + rules: + - description: If a trigger is present the label must equal the quoted marker. + preconditions: + slot_conditions: + trigger: + value_presence: PRESENT + postconditions: + slot_conditions: + label: + equals_string: 'a"b\\\\c' +""" + +EX_ESC = rdflib.Namespace("https://example.org/escaping/") + + +def test_rule_equals_string_special_chars_escaped(): + """An equals_string value with a quote and backslash must be escaped so the + generated SPARQL stays syntactically valid (no injection / broken query).""" + from rdflib.plugins.sparql import prepareQuery + + g = _parse_shacl(_ESCAPING_SCHEMA_YAML) + nodes = list(g.objects(EX_ESC.Item, SH.sparql)) + assert len(nodes) == 1 + query = str(list(g.objects(nodes[0], SH.select))[0]) + + # Would raise ParseException on the unescaped `... = "a"b\c"` form. + 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}" + + +_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)) == []