From 1ea1ea2b93f806b31cb1408f6f891933f29b4178 Mon Sep 17 00:00:00 2001 From: Rayene Messaoud Date: Fri, 10 Jul 2026 22:08:24 +0200 Subject: [PATCH] feat(gen-shacl): add compositional fallback rule converters (M1-M5) Add a compositional fallback in _rule_to_sparql for rule-operator combinations outside the three named patterns (boolean guard, presence-implies-value, exclusive value). Tried only after the named patterns, so their output is unchanged. The fallback translates a conjunction of precondition slot conditions plus a single postcondition into one SELECT $this violation query; any unsupported operator makes the converter return None -- skip, never mis-translate. Supported combinations: - M1 conditional-required: equals_string / value_presence: PRESENT precondition + required: true postcondition (violation = FILTER NOT EXISTS on the target slot). - M2 conditional-absent: value_presence: ABSENT postcondition (violation = the forbidden slot is present). - M3 numeric threshold preconditions: minimum_value / maximum_value inclusive bounds on the trigger slot, rendered via _sparql_number. - M4 nested precondition: one hop into an inlined child object via range_expression.slot_conditions (e.g. sun_position.elevation <= 0); adds _member_conditions (shared inner-condition emitter) and _resolve_member_enum_ref, which resolves inner enum values against the container slot's range class (handles slot_usage-specialised enums). - M5 has_member list-membership: a multivalued slot must contain a member matching a nested range_expression (violation = FILTER NOT EXISTS over the members); reuses _member_conditions. Tests per converter: structural triple assertions, prepareQuery syntax validation, and pyshacl end-to-end (conforming instances pass, crafted violations fail) with advanced=True. Squashed from the five M1-M5 commits on feat/shacl-rule-converters (7566e30c, fac681e0, f8ea709d, 25f8cd28, 47769772). Signed-off-by: Carlo van Driesten --- .../linkml/src/linkml/generators/shaclgen.py | 199 +++++ tests/linkml/test_generators/test_shaclgen.py | 688 ++++++++++++++++++ 2 files changed, 887 insertions(+) diff --git a/packages/linkml/src/linkml/generators/shaclgen.py b/packages/linkml/src/linkml/generators/shaclgen.py index 0eef497306..e856774995 100644 --- a/packages/linkml/src/linkml/generators/shaclgen.py +++ b/packages/linkml/src/linkml/generators/shaclgen.py @@ -446,6 +446,12 @@ def _add_rules(self, g: Graph, shape_uri: URIRef, cls: ClassDefinition) -> None: 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 handled by a + small compositional fallback (:meth:`_compose_rule_sparql`) covering + conditional-required / conditional-absent postconditions, numeric + threshold and nested-object preconditions, and ``has_member`` list + membership. + See `W3C SHACL §5 `_. """ if not cls.rules: @@ -542,6 +548,199 @@ def _rule_to_sparql(self, sv, cls: ClassDefinition, rule) -> str | 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)) + # Fallback: a small compositional builder for operator combinations not + # covered by the three named patterns above (conditional-required, + # threshold preconditions, list membership, ...). Tried only after the + # named patterns, so their output is unchanged. + composed = self._compose_rule_sparql(sv, cls, rule) + if composed is not None: + return composed + + return None + + def _compose_rule_sparql(self, sv, cls: ClassDefinition, rule) -> str | None: + """Compose a SHACL-SPARQL violation query for rule shapes not covered + by the three named patterns. + + Translates a conjunction of *precondition* slot conditions and a single + *postcondition* slot condition into one ``SELECT $this`` query that + selects focus nodes which satisfy every precondition but violate the + postcondition. Supported operators grow incrementally in + :meth:`_precondition_patterns` and :meth:`_postcondition_violation`; + the method returns ``None`` (rule skipped, never mis-translated) as soon + as any operator is unsupported. + + A rule's ``postconditions`` are a conjunction, so violating a single + slot condition is sufficient; the single-postcondition case covers the + modeled cross-parameter rules. + + Conforms to `SHACL §5.3.1 + `_: ``$this`` + is pre-bound to each focus node. + """ + 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 {} + if not pre_slots or len(post_slots) != 1: + return None + + pre_lines = self._precondition_patterns(sv, cls, pre_slots) + if pre_lines is None: + return None + + post_slot_name, post_cond = next(iter(post_slots.items())) + violation = self._postcondition_violation(sv, cls, post_slot_name, post_cond) + if violation is None: + return None + + body = "\n".join(f" {line}" for line in (pre_lines + violation)) + return f"SELECT $this WHERE {{\n{body}\n}}" + + def _precondition_patterns(self, sv, cls: ClassDefinition, pre_slots) -> list[str] | None: + """Translate a conjunction of precondition slot conditions into SPARQL + graph patterns (plus ``FILTER`` lines) that bind focus nodes satisfying + every condition. + + Returns ``None`` if any condition uses an operator not handled here. + + Supported operators: ``value_presence: PRESENT``, ``equals_string``, + and the numeric thresholds ``minimum_value`` / ``maximum_value`` + (inclusive bounds, per the LinkML metamodel). + """ + lines: list[str] = [] + for i, (slot_name, cond) in enumerate(pre_slots.items()): + path = self._slot_uri(sv, slot_name, cls) + var = f"?pre{i}" + if getattr(cond, "value_presence", None) == PresenceEnum(PresenceEnum.PRESENT): + lines.append(f"$this <{path}> {var} .") + elif getattr(cond, "equals_string", None) is not None: + ref = self._resolve_enum_value_ref(sv, slot_name, cond.equals_string) + lines.append(f"$this <{path}> {var} .") + lines.append(f"FILTER ( {var} = {ref} )") + elif getattr(cond, "maximum_value", None) is not None: + lines.append(f"$this <{path}> {var} .") + lines.append(f"FILTER ( {var} <= {self._sparql_number(cond.maximum_value)} )") + elif getattr(cond, "minimum_value", None) is not None: + lines.append(f"$this <{path}> {var} .") + lines.append(f"FILTER ( {var} >= {self._sparql_number(cond.minimum_value)} )") + elif getattr(cond, "range_expression", None) is not None and getattr( + cond.range_expression, "slot_conditions", None + ): + # One-hop into an inlined child object: bind the child node and + # apply the inner slot conditions to it. + node = f"{var}_node" + lines.append(f"$this <{path}> {node} .") + inner = self._member_conditions(sv, cls, slot_name, node, cond.range_expression.slot_conditions) + if inner is None: + return None + lines.extend(inner) + else: + return None + return lines + + def _member_conditions( + self, sv, cls: ClassDefinition, container_slot_name: str, node_var: str, slot_conditions + ) -> list[str] | None: + """Constrain the object bound to *node_var* — an instance of the range + class of *container_slot_name* — by a set of inner slot conditions. + + Shared by the nested ``range_expression`` precondition (single inlined + child) and the ``has_member`` postcondition (a list member). Enum + values are resolved against the container slot's range class. Returns + ``None`` for unsupported inner operators. + """ + lines: list[str] = [] + for j, (inner_name, icond) in enumerate(slot_conditions.items()): + ipath = self._slot_uri(sv, inner_name, cls) + ivar = f"{node_var}_{j}" + if getattr(icond, "value_presence", None) == PresenceEnum(PresenceEnum.PRESENT): + lines.append(f"{node_var} <{ipath}> {ivar} .") + elif getattr(icond, "equals_string", None) is not None: + iref = self._resolve_member_enum_ref(sv, container_slot_name, inner_name, icond.equals_string) + lines.append(f"{node_var} <{ipath}> {ivar} .") + lines.append(f"FILTER ( {ivar} = {iref} )") + elif getattr(icond, "maximum_value", None) is not None: + lines.append(f"{node_var} <{ipath}> {ivar} .") + lines.append(f"FILTER ( {ivar} <= {self._sparql_number(icond.maximum_value)} )") + elif getattr(icond, "minimum_value", None) is not None: + lines.append(f"{node_var} <{ipath}> {ivar} .") + lines.append(f"FILTER ( {ivar} >= {self._sparql_number(icond.minimum_value)} )") + else: + return None + return lines + + def _resolve_member_enum_ref(self, sv, container_slot_name: str, inner_slot_name: str, value_name: str) -> str: + """Resolve an inner enum value to a SPARQL term using the *range class* + of the container slot. + + A slot such as ``type`` may be reused across classes with different + enum ranges (via ``slot_usage``); resolving through the container's + range class picks the correct enum. Falls back to + :meth:`_resolve_enum_value_ref` (and thence to a literal) when the + container is not a class, the inner slot is not on it, or the value has + no ``meaning``. + """ + container = sv.get_slot(container_slot_name) + range_class = container.range if container else None + if range_class and range_class in sv.all_classes() and inner_slot_name in sv.class_slots(range_class): + induced = sv.induced_slot(inner_slot_name, range_class) + if induced and induced.range in sv.all_enums(): + pv = sv.get_enum(induced.range).permissible_values.get(value_name) + if pv and pv.meaning: + return f"<{sv.expand_curie(pv.meaning)}>" + return self._resolve_enum_value_ref(sv, inner_slot_name, value_name) + + @staticmethod + def _sparql_number(value) -> str: + """Render a numeric threshold bound as a SPARQL numeric literal. + + LinkML parses ``minimum_value`` / ``maximum_value`` as ``int`` or + ``float`` (possibly the ``extended_int`` / ``extended_float`` runtime + subclasses); ``str`` yields a plain numeric token + (e.g. ``4000`` or ``0.0``) that SPARQL compares with numeric promotion + against ``xsd:float`` / ``xsd:decimal`` data values. + """ + return str(value) + + def _postcondition_violation(self, sv, cls: ClassDefinition, slot_name: str, cond) -> list[str] | None: + """Translate a single postcondition slot condition into SPARQL that + matches a *violation* of it. + + Returns ``None`` for operators not handled here. + + Supported operators: + + * ``required: true`` — violation = the target slot is absent on a focus + node that satisfies the preconditions. + * ``value_presence: ABSENT`` — violation = the target slot *is* present + (inapplicable-slot / conditional-absent). + * ``has_member`` with a nested ``range_expression`` — violation = *no* + member of the (multivalued) target slot matches the inner conditions + (list-membership; e.g. the light-group list must contain a + ``{group: Vehicle, type: front_fog_light}`` entry). + """ + path = self._slot_uri(sv, slot_name, cls) + if getattr(cond, "required", None) is True: + return [f"FILTER NOT EXISTS {{ $this <{path}> ?post . }}"] + if getattr(cond, "value_presence", None) == PresenceEnum(PresenceEnum.ABSENT): + return [f"$this <{path}> ?post ."] + has_member = getattr(cond, "has_member", None) + if ( + has_member is not None + and getattr(has_member, "range_expression", None) is not None + and getattr(has_member.range_expression, "slot_conditions", None) + ): + member_lines = [f"$this <{path}> ?mem ."] + inner = self._member_conditions(sv, cls, slot_name, "?mem", has_member.range_expression.slot_conditions) + if inner is None: + return None + member_lines.extend(inner) + block = " ".join(member_lines) + return [f"FILTER NOT EXISTS {{ {block} }}"] return None def _build_boolean_guard_sparql(self, sv, cls: ClassDefinition, flag_slot_name: str, value_slot_name: str) -> str: diff --git a/tests/linkml/test_generators/test_shaclgen.py b/tests/linkml/test_generators/test_shaclgen.py index f0ce86f51f..5e7249ab43 100644 --- a/tests/linkml/test_generators/test_shaclgen.py +++ b/tests/linkml/test_generators/test_shaclgen.py @@ -2806,3 +2806,691 @@ def test_presence_implies_value_pyshacl_end_to_end(): 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). +# =========================================================================== + +_CONDITIONAL_REQUIRED_SCHEMA_YAML = """ +id: https://example.org/conditional-required +name: conditional_required_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/conditional-required/ +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 + +slots: + sky_model: + range: SkyModelEnum + slot_uri: ex:sky_model + overcast_sky_illuminance: + range: float + slot_uri: ex:overcast_sky_illuminance + +classes: + Weather: + class_uri: ex:Weather + slots: + - sky_model + - overcast_sky_illuminance + rules: + - description: The MeasuredOvercastSky model requires the sky illuminance. + preconditions: + slot_conditions: + sky_model: + equals_string: MeasuredOvercastSky + postconditions: + slot_conditions: + overcast_sky_illuminance: + required: true +""" + +EX_CR = rdflib.Namespace("https://example.org/conditional-required/") + + +def test_conditional_required_generates_sparql(): + """equals_string precondition + required postcondition → one sh:sparql constraint.""" + g = _parse_shacl(_CONDITIONAL_REQUIRED_SCHEMA_YAML) + + shape = EX_CR.Weather + sparql_nodes = list(g.objects(shape, SH.sparql)) + assert len(sparql_nodes) == 1, f"Expected 1 sh:sparql constraint, got {len(sparql_nodes)}" + + node = sparql_nodes[0] + assert (node, RDF.type, SH.SPARQLConstraint) in g + query = str(list(g.objects(node, SH.select))[0]) + + assert "$this" in query, "SPARQL must use $this pre-bound variable (SHACL §5.3.1)" + assert "FILTER NOT EXISTS" in query, "required violation must use FILTER NOT EXISTS" + # precondition references the enum meaning IRI and the trigger slot + assert f"<{EX_CR.MeasuredOvercastSky}>" in query, f"precondition must use the enum IRI, got:\n{query}" + assert str(EX_CR.sky_model) in query + assert str(EX_CR.overcast_sky_illuminance) in query + + +def test_conditional_required_message_from_description(): + """Rule description is emitted as sh:message.""" + g = _parse_shacl(_CONDITIONAL_REQUIRED_SCHEMA_YAML) + messages = [str(m) for node in g.objects(EX_CR.Weather, SH.sparql) for m in g.objects(node, SH.message)] + assert any("requires the sky illuminance" in m for m in messages), messages + + +def test_conditional_required_sparql_syntax_valid(): + """Generated SPARQL must be syntactically valid.""" + from rdflib.plugins.sparql import prepareQuery + + g = _parse_shacl(_CONDITIONAL_REQUIRED_SCHEMA_YAML) + for node in g.objects(EX_CR.Weather, SH.sparql): + prepareQuery(str(list(g.objects(node, SH.select))[0])) + + +def test_conditional_required_pyshacl_end_to_end(): + """End-to-end: pyshacl passes conforming instances and flags the violation.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_CONDITIONAL_REQUIRED_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + + # Conforming: MeasuredOvercastSky WITH illuminance; ClearSky needs nothing. + conforming = """ + @prefix ex: . + @prefix xsd: . + + ex:wMeasured a ex:Weather ; + ex:sky_model ex:MeasuredOvercastSky ; + ex:overcast_sky_illuminance "4200.0"^^xsd:float . + + ex:wClear a ex:Weather ; + ex:sky_model ex:ClearSky . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Conforming instances should pass:\n{txt}" + + # Violating: MeasuredOvercastSky WITHOUT the required illuminance. + violating = """ + @prefix ex: . + + ex:wBad a ex:Weather ; + ex:sky_model ex:MeasuredOvercastSky . + """ + 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"MeasuredOvercastSky without illuminance should fail:\n{txt}" + + +# =========================================================================== +# 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. +# =========================================================================== + +_CONDITIONAL_ABSENT_SCHEMA_YAML = """ +id: https://example.org/conditional-absent +name: conditional_absent_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/conditional-absent/ +imports: + - linkml:types +default_prefix: ex +default_range: string + +enums: + SkyModelEnum: + permissible_values: + ClearSky: + meaning: ex:ClearSky + OvercastSky: + meaning: ex:OvercastSky + +slots: + sky_model: + range: SkyModelEnum + slot_uri: ex:sky_model + overcast_sky_illuminance: + range: float + slot_uri: ex:overcast_sky_illuminance + +classes: + Weather: + class_uri: ex:Weather + slots: + - sky_model + - overcast_sky_illuminance + rules: + - description: ClearSky makes overcast_sky_illuminance inapplicable. + preconditions: + slot_conditions: + sky_model: + equals_string: ClearSky + postconditions: + slot_conditions: + overcast_sky_illuminance: + value_presence: ABSENT +""" + +EX_CA = rdflib.Namespace("https://example.org/conditional-absent/") + + +def test_conditional_absent_generates_sparql(): + """equals_string precondition + value_presence ABSENT → one sh:sparql constraint.""" + g = _parse_shacl(_CONDITIONAL_ABSENT_SCHEMA_YAML) + + sparql_nodes = list(g.objects(EX_CA.Weather, SH.sparql)) + assert len(sparql_nodes) == 1, f"Expected 1 sh:sparql constraint, got {len(sparql_nodes)}" + + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "$this" in query + # violation = precondition holds AND the forbidden slot is present; the + # forbidden-slot triple must NOT be wrapped in NOT EXISTS. + assert "FILTER NOT EXISTS" not in query, f"conditional-absent must not use NOT EXISTS, got:\n{query}" + assert f"<{EX_CA.ClearSky}>" in query + assert str(EX_CA.overcast_sky_illuminance) in query + + +def test_conditional_absent_sparql_syntax_valid(): + """Generated SPARQL must be syntactically valid.""" + from rdflib.plugins.sparql import prepareQuery + + g = _parse_shacl(_CONDITIONAL_ABSENT_SCHEMA_YAML) + for node in g.objects(EX_CA.Weather, SH.sparql): + prepareQuery(str(list(g.objects(node, SH.select))[0])) + + +def test_conditional_absent_pyshacl_end_to_end(): + """End-to-end: pyshacl passes conforming instances and flags the violation.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_CONDITIONAL_ABSENT_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + + # Conforming: ClearSky without illuminance; OvercastSky may set illuminance. + conforming = """ + @prefix ex: . + @prefix xsd: . + + ex:wClear a ex:Weather ; + ex:sky_model ex:ClearSky . + + ex:wOvercast a ex:Weather ; + ex:sky_model ex:OvercastSky ; + ex:overcast_sky_illuminance "5000.0"^^xsd:float . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Conforming instances should pass:\n{txt}" + + # Violating: ClearSky WITH the inapplicable illuminance. + violating = """ + @prefix ex: . + @prefix xsd: . + + ex:wBad a ex:Weather ; + ex:sky_model ex:ClearSky ; + ex:overcast_sky_illuminance "5000.0"^^xsd:float . + """ + conforms, _, txt = pyshacl.validate( + data_graph=violating, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"ClearSky with illuminance should fail:\n{txt}" + + +# =========================================================================== +# 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. +# =========================================================================== + +_THRESHOLD_SCHEMA_YAML = """ +id: https://example.org/threshold +name: threshold_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/threshold/ +imports: + - linkml:types +default_prefix: ex +default_range: string + +slots: + meteorological_optical_range: + range: float + slot_uri: ex:meteorological_optical_range + fog_note: + range: string + slot_uri: ex:fog_note + +classes: + Weather: + class_uri: ex:Weather + slots: + - meteorological_optical_range + - fog_note + rules: + - description: In fog (optical range at or below 4000) a fog note is required. + preconditions: + slot_conditions: + meteorological_optical_range: + maximum_value: 4000 + postconditions: + slot_conditions: + fog_note: + required: true +""" + +EX_THR = rdflib.Namespace("https://example.org/threshold/") + + +def test_threshold_precondition_generates_sparql(): + """maximum_value precondition emits a numeric FILTER on the trigger slot.""" + g = _parse_shacl(_THRESHOLD_SCHEMA_YAML) + + sparql_nodes = list(g.objects(EX_THR.Weather, SH.sparql)) + assert len(sparql_nodes) == 1, f"Expected 1 sh:sparql constraint, got {len(sparql_nodes)}" + + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "<= 4000" in query, f"threshold must emit '<= 4000', got:\n{query}" + assert "FILTER NOT EXISTS" in query, "required postcondition violation must use NOT EXISTS" + assert str(EX_THR.meteorological_optical_range) in query + assert str(EX_THR.fog_note) in query + + +def test_threshold_precondition_sparql_syntax_valid(): + """Generated SPARQL must be syntactically valid.""" + from rdflib.plugins.sparql import prepareQuery + + g = _parse_shacl(_THRESHOLD_SCHEMA_YAML) + for node in g.objects(EX_THR.Weather, SH.sparql): + prepareQuery(str(list(g.objects(node, SH.select))[0])) + + +def test_threshold_precondition_pyshacl_end_to_end(): + """End-to-end: below-threshold requires the note; above-threshold does not.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_THRESHOLD_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + + # Conforming: foggy (400) with a note; clear (5000) needs nothing. + conforming = """ + @prefix ex: . + @prefix xsd: . + + ex:wFog a ex:Weather ; + ex:meteorological_optical_range "400.0"^^xsd:float ; + ex:fog_note "reduced visibility" . + + ex:wClear a ex:Weather ; + ex:meteorological_optical_range "5000.0"^^xsd:float . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Conforming instances should pass:\n{txt}" + + # Violating: foggy (400) without the required note. + violating = """ + @prefix ex: . + @prefix xsd: . + + ex:wBad a ex:Weather ; + ex:meteorological_optical_range "400.0"^^xsd:float . + """ + conforms, _, txt = pyshacl.validate( + data_graph=violating, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"Fog without the required note should fail:\n{txt}" + + +# =========================================================================== +# 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. +# =========================================================================== + +_NESTED_SCHEMA_YAML = """ +id: https://example.org/nested +name: nested_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/nested/ +imports: + - linkml:types +default_prefix: ex +default_range: string + +slots: + sun_position: + range: SunPosition + inlined: true + slot_uri: ex:sun_position + elevation: + range: float + slot_uri: ex:elevation + headlight_note: + range: string + slot_uri: ex:headlight_note + +classes: + SunPosition: + class_uri: ex:SunPosition + slots: + - elevation + Weather: + class_uri: ex:Weather + slots: + - sun_position + - headlight_note + rules: + - description: When the sun is at or below the horizon a headlight note is required. + preconditions: + slot_conditions: + sun_position: + range_expression: + slot_conditions: + elevation: + maximum_value: 0.0 + postconditions: + slot_conditions: + headlight_note: + required: true +""" + +EX_NEST = rdflib.Namespace("https://example.org/nested/") + + +def test_nested_precondition_generates_sparql(): + """A nested range_expression precondition emits a two-hop graph pattern.""" + g = _parse_shacl(_NESTED_SCHEMA_YAML) + + sparql_nodes = list(g.objects(EX_NEST.Weather, SH.sparql)) + assert len(sparql_nodes) == 1, f"Expected 1 sh:sparql constraint, got {len(sparql_nodes)}" + + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert str(EX_NEST.sun_position) in query, "must traverse the container slot" + assert str(EX_NEST.elevation) in query, "must traverse the inner slot" + assert "<= 0.0" in query, f"inner threshold must appear, got:\n{query}" + assert "FILTER NOT EXISTS" in query + assert str(EX_NEST.headlight_note) in query + + +def test_nested_precondition_sparql_syntax_valid(): + """Generated SPARQL must be syntactically valid.""" + from rdflib.plugins.sparql import prepareQuery + + g = _parse_shacl(_NESTED_SCHEMA_YAML) + for node in g.objects(EX_NEST.Weather, SH.sparql): + prepareQuery(str(list(g.objects(node, SH.select))[0])) + + +def test_nested_precondition_pyshacl_end_to_end(): + """End-to-end: sun below horizon requires the note; above horizon does not.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_NESTED_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + + # Conforming: night (elevation -90) with a note; day (45) needs nothing. + conforming = """ + @prefix ex: . + @prefix xsd: . + + ex:wNight a ex:Weather ; + ex:sun_position [ a ex:SunPosition ; ex:elevation "-90.0"^^xsd:float ] ; + ex:headlight_note "on" . + + ex:wDay a ex:Weather ; + ex:sun_position [ a ex:SunPosition ; ex:elevation "45.0"^^xsd:float ] . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Conforming instances should pass:\n{txt}" + + # Violating: night (elevation -90) without the required note. + violating = """ + @prefix ex: . + @prefix xsd: . + + ex:wBad a ex:Weather ; + ex:sun_position [ a ex:SunPosition ; ex:elevation "-90.0"^^xsd:float ] . + """ + conforms, _, txt = pyshacl.validate( + data_graph=violating, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert not conforms, f"Night without a headlight note should fail:\n{txt}" + + +# =========================================================================== +# 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. +# =========================================================================== + +_HAS_MEMBER_SCHEMA_YAML = """ +id: https://example.org/has-member +name: has_member_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/has-member/ +imports: + - linkml:types +default_prefix: ex +default_range: string + +enums: + LightGroupEnum: + permissible_values: + Vehicle: + meaning: ex:Vehicle + StreetLight: + meaning: ex:StreetLight + LightTypeEnum: + permissible_values: + low_beam_headlight: + meaning: ex:low_beam_headlight + front_fog_light: + meaning: ex:front_fog_light + +slots: + fog_declared: + range: string + slot_uri: ex:fog_declared + enabled_light_control_groups: + range: LightControlGroup + multivalued: true + inlined: true + inlined_as_list: true + slot_uri: ex:enabled_light_control_groups + group: + range: LightGroupEnum + slot_uri: ex:group + type: + range: LightTypeEnum + slot_uri: ex:type + +classes: + LightControlGroup: + class_uri: ex:LightControlGroup + slots: + - group + - type + Weather: + class_uri: ex:Weather + slots: + - fog_declared + - enabled_light_control_groups + rules: + - description: When fog is declared, a front fog light group must be enabled. + preconditions: + slot_conditions: + fog_declared: + value_presence: PRESENT + postconditions: + slot_conditions: + enabled_light_control_groups: + has_member: + range_expression: + slot_conditions: + group: + equals_string: Vehicle + type: + equals_string: front_fog_light +""" + +EX_HM = rdflib.Namespace("https://example.org/has-member/") + + +def test_has_member_generates_sparql(): + """has_member postcondition emits a FILTER NOT EXISTS over list members.""" + g = _parse_shacl(_HAS_MEMBER_SCHEMA_YAML) + + sparql_nodes = list(g.objects(EX_HM.Weather, SH.sparql)) + assert len(sparql_nodes) == 1, f"Expected 1 sh:sparql constraint, got {len(sparql_nodes)}" + + query = str(list(g.objects(sparql_nodes[0], SH.select))[0]) + assert "FILTER NOT EXISTS" in query, "list-membership violation must use FILTER NOT EXISTS" + assert str(EX_HM.enabled_light_control_groups) in query + assert str(EX_HM.group) in query and str(EX_HM.type) in query + # inner enum values resolve against the member class (LightControlGroup), + # so the reused `type` slot picks LightTypeEnum, not another enum. + assert f"<{EX_HM.Vehicle}>" in query, f"group value must be the enum IRI, got:\n{query}" + assert f"<{EX_HM.front_fog_light}>" in query, f"type value must be the enum IRI, got:\n{query}" + + +def test_has_member_sparql_syntax_valid(): + """Generated SPARQL must be syntactically valid.""" + from rdflib.plugins.sparql import prepareQuery + + g = _parse_shacl(_HAS_MEMBER_SCHEMA_YAML) + for node in g.objects(EX_HM.Weather, SH.sparql): + prepareQuery(str(list(g.objects(node, SH.select))[0])) + + +def test_has_member_pyshacl_end_to_end(): + """End-to-end: fog requires a front-fog-light member; otherwise it fails.""" + import pyshacl + + shacl_ttl = ShaclGenerator(_HAS_MEMBER_SCHEMA_YAML, mergeimports=False, emit_rules=True).serialize() + + # Conforming: fog declared WITH a front-fog-light group; and no fog at all. + conforming = """ + @prefix ex: . + + ex:wFog a ex:Weather ; + ex:fog_declared "yes" ; + ex:enabled_light_control_groups + [ a ex:LightControlGroup ; ex:group ex:Vehicle ; ex:type ex:front_fog_light ] . + + ex:wNoFog a ex:Weather . + """ + conforms, _, txt = pyshacl.validate( + data_graph=conforming, + shacl_graph=shacl_ttl, + data_graph_format="turtle", + shacl_graph_format="turtle", + advanced=True, + ) + assert conforms, f"Conforming instances should pass:\n{txt}" + + # Violating: fog declared but only a low-beam group (no front fog light). + violating = """ + @prefix ex: . + + ex:wBad a ex:Weather ; + ex:fog_declared "yes" ; + ex:enabled_light_control_groups + [ a ex:LightControlGroup ; ex:group ex:Vehicle ; ex:type ex:low_beam_headlight ] . + """ + 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"Fog without a front-fog-light group should fail:\n{txt}"