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)) == []