diff --git a/packages/linkml/src/linkml/generators/shacl/shacl_data_type.py b/packages/linkml/src/linkml/generators/shacl/shacl_data_type.py index 1210a31be2..7ffd11d45d 100644 --- a/packages/linkml/src/linkml/generators/shacl/shacl_data_type.py +++ b/packages/linkml/src/linkml/generators/shacl/shacl_data_type.py @@ -20,7 +20,7 @@ class ShaclDataType(DataType, Enum): FLOAT = ("float", URIRef("http://www.w3.org/2001/XMLSchema#float")) DOUBLE = ("double", URIRef("http://www.w3.org/2001/XMLSchema#double")) URI = ("uri", URIRef("http://www.w3.org/2001/XMLSchema#anyURI")) - CURIE = ("curi", URIRef("http://www.w3.org/2001/XMLSchema#string")) + CURIE = ("curie", URIRef("http://www.w3.org/2001/XMLSchema#string")) NCNAME = ("ncname", URIRef("http://www.w3.org/2001/XMLSchema#string")) OBJECT_IDENTIFIER = ("objectidentifier", URIRef("http://www.w3.org/ns/shex#iri")) NODE_IDENTIFIER = ("nodeidentifier", URIRef("http://www.w3.org/ns/shex#nonLiteral")) diff --git a/packages/linkml/src/linkml/generators/shaclgen.py b/packages/linkml/src/linkml/generators/shaclgen.py index eb6b48e9d6..9606d676cc 100644 --- a/packages/linkml/src/linkml/generators/shaclgen.py +++ b/packages/linkml/src/linkml/generators/shaclgen.py @@ -1,8 +1,9 @@ import logging import os +import re import string from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, fields import click from jsonasobj2 import JsonObj, as_dict @@ -16,7 +17,17 @@ from linkml.generators.shacl.shacl_ifabsent_processor import ShaclIfAbsentProcessor from linkml.utils.generator import Generator, shared_arguments from linkml.utils.language_tags import LanguageTagResolver -from linkml_runtime.linkml_model.meta import ClassDefinition, ElementName, PresenceEnum +from linkml_runtime.linkml_model.meta import ( + AnonymousClassExpression, + AnonymousSlotExpression, + ClassDefinition, + ClassRule, + Element, + ElementName, + PresenceEnum, + SlotDefinition, + SlotExpression, +) from linkml_runtime.utils.formatutils import underscore from linkml_runtime.utils.rdf_canonicalize import canonicalize_rdf_graph from linkml_runtime.utils.yamlutils import TypedNode, extended_float, extended_int, extended_str @@ -59,6 +70,28 @@ def _validate_message_template(template: str) -> None: ) +@dataclass +class _RuleSite: + """A rule as it is translated for the shape of a class. + + *cls* is the class whose shape the rule constrains, *owner* the class + that declares it, and *index* its 1-based position among the owner's + rules; *cls* differs from *owner* when the rule is inherited. + """ + + cls: ClassDefinition + owner: str + index: int + rule: ClassRule + + def __str__(self) -> str: + text = f"Rule {self.index} of class {self.owner!r}" + label = self.rule.title or self.rule.description + if label: + text += f" ({label})" + return text + + @dataclass class ShaclGenerator(Generator): """Generate SHACL (Shapes Constraint Language) shapes from a LinkML schema. @@ -152,11 +185,17 @@ class ShaclGenerator(Generator): SHACL-SPARQL constraints (``sh:SPARQLConstraint``) on the corresponding ``sh:NodeShape``. Currently two 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. - * *Exclusive value* — a precondition with ``equals_string`` on a slot and - a postcondition with ``maximum_cardinality`` on the *same* slot. + * *Presence implies value* — a precondition with ``value_presence: PRESENT`` + on a guard slot and a postcondition with ``equals_string`` or + ``equals_string_in`` on a target slot holding strings (an enum, or a type + with datatype ``xsd:string``) or booleans. The *boolean guard* is the + case ``equals_string: "true"`` on an ``xsd:boolean``-typed flag. + * *Exclusive value* — a precondition with ``has_member: {equals_string: V}`` + on a slot and a postcondition with ``maximum_cardinality`` on the *same* + slot. + + A rule that cannot be translated exactly is skipped with a warning. A + class shape carries the rules of the class's ancestors and mixins too. See `W3C SHACL §5 `_ and `linkml/linkml#2464 `_. @@ -203,6 +242,9 @@ def serialize(self, **args) -> str: def as_graph(self) -> Graph: sv = self.schemaview + # Problems with rules, collected while the rules are translated for + # every class that has them and reported once each at the end. + self._rule_problems: dict[tuple[str, int, str], tuple[str, list[str]]] = {} g = Graph() g.bind("sh", SH) @@ -442,6 +484,7 @@ def st_node_pv(p, v): if self.emit_rules: self._add_rules(g, class_uri_with_suffix, c) + self._report_rule_problems() return g LINKML_ANY_URI = "https://w3id.org/linkml/Any" @@ -451,227 +494,500 @@ def st_node_pv(p, v): # ------------------------------------------------------------------- def _add_rules(self, g: Graph, shape_uri: URIRef, cls: ClassDefinition) -> None: - """Emit ``sh:sparql`` constraints from LinkML ``rules:`` blocks. + """Emit ``sh:sparql`` constraints from the LinkML ``rules`` that apply to *cls*. Each recognised rule is converted into an ``sh:SPARQLConstraint`` - attached to *shape_uri*. Unrecognised patterns are logged at - ``DEBUG`` level and silently skipped. - - Currently recognised patterns: - - * **Boolean guard** — a *precondition* with - ``value_presence: PRESENT`` on a value slot and a *postcondition* - with ``equals_string: "true"`` on a boolean flag slot. - - * **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). + attached to *shape_uri*. Currently recognised patterns: + + * **Presence implies value** — a *precondition* with + ``value_presence: PRESENT`` on a guard slot and a *postcondition* + with ``equals_string`` or ``equals_string_in`` on a target slot. + When the guard is present, the target must be present and hold one + of the allowed values. The target must hold strings (an enum, or a + type with datatype ``xsd:string``) or booleans (``xsd:boolean``). + The **boolean guard** is the case ``equals_string: "true"`` on a + boolean flag. + + * **Exclusive value** — a *precondition* with ``has_member: + {equals_string: V}`` on a slot and a *postcondition* with + ``maximum_cardinality`` on the *same* slot. When V is one of the + slot's values, the slot holds at most that many values (typically 1 + for mutual exclusion). A bare ``equals_string: V`` precondition + applies to every value of the slot (``05validation.md``), so it is + not this pattern. + + Every emitted constraint translates its rule exactly. A rule that + cannot be translated exactly is skipped with a warning naming the + reason: an operator combination outside these patterns, a condition on + an unknown or identifier slot, a value the target's range cannot hold + (see :meth:`_value_terms`), or ``bidirectional``. Of a rule with ``elseconditions`` the forward + (if/then) direction is emitted, exactly, and a warning reports the + else branch as not enforced. + + A rule applies to "all members of this class" (metamodel ``rules``), + so the shape of a class carries the rules of its ancestors and mixins + as well, as it carries their slots, and as the JSON Schema generator + applies them. Each rule is translated in the context of *cls*, where + ``slot_usage`` may refine a slot the rule references. See `W3C SHACL §5 `_. """ - if not cls.rules: - return - sv = self.schemaview - for rule in cls.rules: - if getattr(rule, "deactivated", False): - continue - - if getattr(rule, "bidirectional", False): - logger.warning( - "Rule in class %r has bidirectional=true; " - "SHACL-SPARQL generation does not support bidirectional rules. " - "Skipping this rule entirely.", - cls.name, - ) - continue - - if getattr(rule, "open_world", False): - logger.warning( - "Rule in class %r has open_world=true; " - "SHACL operates under closed-world assumption. " - "The constraint is emitted but may not match open-world semantics.", - cls.name, - ) + for owner in sv.class_ancestors(cls.name): + for index, rule in enumerate(sv.get_class(owner).rules, start=1): + self._add_rule(g, shape_uri, _RuleSite(cls, owner, index, rule)) + + def _add_rule(self, g: Graph, shape_uri: URIRef, site: _RuleSite) -> None: + """Emit the ``sh:SPARQLConstraint`` of the rule at *site* on *shape_uri*, unless it is skipped.""" + rule = site.rule + if rule.deactivated: + return + if rule.bidirectional: + self._skip_rule(site, "bidirectional rules are not supported") + return + sparql_query = self._rule_to_sparql(site) + if sparql_query is None: + return + if rule.elseconditions is not None: + self._warn_rule( + site, "its elseconditions are not enforced; SHACL-SPARQL generation emits the if/then direction only" + ) + message = Literal(rule.description, lang=self._resolve_language(rule)) if rule.description else None + if self._has_sparql_constraint(g, shape_uri, Literal(sparql_query), message): + return - if getattr(rule, "elseconditions", 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.", - cls.name, - ) + constraint = BNode() + g.add((shape_uri, SH.sparql, constraint)) + g.add((constraint, RDF.type, SH.SPARQLConstraint)) + if message is not None: + g.add((constraint, SH.message, message)) + g.add((constraint, SH.select, Literal(sparql_query))) - sparql_query = self._rule_to_sparql(sv, cls, rule) - if sparql_query is None: - logger.debug( - "Skipping unsupported rule pattern in class %r: %s", - cls.name, - getattr(rule, "description", "(no description)"), - ) - continue + @staticmethod + def _has_sparql_constraint(g: Graph, shape_uri: URIRef, query: Literal, message: Literal | None) -> bool: + """Whether the shape *shape_uri* already carries the constraint with *query* and *message*. - constraint = BNode() - g.add((shape_uri, SH.sparql, constraint)) - g.add((constraint, RDF.type, SH.SPARQLConstraint)) + Classes that share a ``class_uri`` share one shape when shapes are named + by ``class_uri`` (the default), so a rule they all inherit would + otherwise be added once per class and reported twice. + """ + return any( + (constraint, SH.select, query) in g and set(g.objects(constraint, SH.message)) == {message} - {None} + for constraint in g.objects(shape_uri, SH.sparql) + ) - message = getattr(rule, "description", None) - if message: - g.add((constraint, SH.message, Literal(message))) + def _warn_rule(self, site: _RuleSite, problem: str) -> None: + """Record *problem* with the rule at *site* for the shape of ``site.cls``. - g.add((constraint, SH.select, Literal(sparql_query))) + A rule is translated once for every class that has it, declared or + inherited, so the same problem can arise several times; it is reported + once, by :meth:`_report_rule_problems`. + """ + _, classes = self._rule_problems.setdefault((site.owner, site.index, problem), (str(site), [])) + if site.cls.name not in classes: + classes.append(site.cls.name) + + def _report_rule_problems(self) -> None: + """Log each recorded rule problem once, naming the class shapes it affects + unless that is only the class declaring the rule.""" + for (owner, _, problem), (rule, classes) in self._rule_problems.items(): + shapes = "" if classes == [owner] else f" (in the shapes of {', '.join(map(repr, classes))})" + logger.warning("%s: %s%s.", rule, problem, shapes) + + def _skip_rule(self, site: _RuleSite, reason: str) -> None: + """Log that the rule at *site* is not translated, and why.""" + self._warn_rule(site, f"skipped, because {reason}") + + # 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). + # Derived from the metamodel: the metadata every ``element`` carries, minus + # anything that is a ``slot_expression`` operator. + _NON_OPERATOR_FIELDS = frozenset(f.name for f in fields(Element)) - frozenset( + f.name for f in fields(SlotExpression) + ) - def _rule_to_sparql(self, sv, cls: ClassDefinition, rule) -> str | None: - """Convert a ``ClassRule`` to a SPARQL SELECT query string. + # The lexical space of xsd:boolean and the value each lexical form maps to + # (XML Schema 1.1 Part 2 §3.3.2.2, ), + # as SPARQL boolean literals. + _XSD_BOOLEAN_LEXICAL = {"true": "true", "1": "true", "false": "false", "0": "false"} + + _NO_PATTERN = "its conditions match none of the translated patterns (presence implies value, exclusive value)" + + @classmethod + def _set_operator_fields( + cls, condition: SlotDefinition | AnonymousSlotExpression | AnonymousClassExpression + ) -> 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. + """ + return { + name + for name, value in vars(condition).items() + if not name.startswith("_") + and name not in cls._NON_OPERATOR_FIELDS + and value is not None + and not (isinstance(value, list | dict) and not value) + } - Returns ``None`` when the rule does not match any supported pattern. + def _rule_to_sparql(self, site: _RuleSite) -> str | None: + """Translate the rule at *site* to a SPARQL SELECT query. + + Returns ``None``, after a warning naming the reason, when the rule + matches no supported pattern exactly. 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, since dropping a term + would widen the precondition (false positives) or weaken the + postcondition (false negatives). """ - pre = getattr(rule, "preconditions", None) - post = getattr(rule, "postconditions", None) - if not pre or not post: + pre, post = site.rule.preconditions, site.rule.postconditions + if pre is None or post is None: + self._skip_rule(site, "only rules with preconditions and postconditions are translated") + return None + for side, expression in (("preconditions", pre), ("postconditions", post)): + untranslated = self._set_operator_fields(expression) - {"slot_conditions"} + if untranslated: + self._skip_rule(site, f"its {side} use {', '.join(sorted(untranslated))}") + return None + if len(pre.slot_conditions) != 1 or len(post.slot_conditions) != 1: + self._skip_rule(site, self._NO_PATTERN) return None - pre_slots = getattr(pre, "slot_conditions", None) or {} - post_slots = getattr(post, "slot_conditions", None) or {} + ((pre_name, pre_cond),) = pre.slot_conditions.items() + ((post_name, post_cond),) = post.slot_conditions.items() + pre_slot = self._rule_condition_slot(site, pre_name) + post_slot = self._rule_condition_slot(site, post_name) if pre_slot is not None else None + if pre_slot is None or post_slot is None: + return None + pre_ops = self._set_operator_fields(pre_cond) + post_ops = self._set_operator_fields(post_cond) + + # Presence implies value: "if the guard is present, the target must be + # present and hold one of the allowed values". The boolean guard is + # the case `equals_string: "true"` on a boolean flag. + if ( + pre_ops == {"value_presence"} + and pre_cond.value_presence == PresenceEnum(PresenceEnum.PRESENT) + and post_ops in ({"equals_string"}, {"equals_string_in"}) + ): + values = [post_cond.equals_string] if post_ops == {"equals_string"} else list(post_cond.equals_string_in) + terms = self._value_terms(site, post_slot, values) + if terms is None: + return None + return self._build_presence_implies_value_sparql(pre_slot, post_slot, terms, bool(site.rule.open_world)) + + # Exclusive value: "if V is one of the values of slot X, then X has at + # most N values". + if post_ops == {"maximum_cardinality"} and pre_slot.name == post_slot.name: + if pre_ops == {"equals_string"}: + self._skip_rule( + site, + f"its precondition equals_string on slot {pre_slot.name!r} applies to every value of the slot; " + f"to state that one of the values equals it, use has_member: {{equals_string: ...}}", + ) + return None + exclusive = self._exclusive_value(pre_cond, pre_ops) + if exclusive is not None: + terms = self._value_terms(site, pre_slot, [exclusive]) + if terms is None: + return None + return self._build_exclusive_value_sparql(pre_slot, terms[0], int(post_cond.maximum_cardinality)) + + self._skip_rule(site, self._NO_PATTERN) + return None + + def _exclusive_value(self, condition: SlotDefinition, operators: set[str]) -> str | None: + """The value V of an exclusive-value precondition, or ``None`` if *condition* is not one. + + Only ``has_member: {equals_string: V}`` states "V is one of the values" + (metamodel ``has_member``: "at least one member satisfying the + condition"). + """ + if operators == {"has_member"} and self._set_operator_fields(condition.has_member) == {"equals_string"}: + return condition.has_member.equals_string + return None - # 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)) + def _rule_slot(self, cls: ClassDefinition, slot_name: str) -> SlotDefinition | None: + """Resolve a rule condition's slot key to the slot it names, or ``None`` + when no such slot exists. - pre_cond = pre_slots[pre_slot_name] - post_cond = post_slots[post_slot_name] + 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. + """ + sv = self.schemaview + 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_condition_slot(self, site: _RuleSite, slot_name: str) -> SlotDefinition | None: + """The slot a condition of the rule at *site* names, or ``None`` (rule skipped) when it cannot be queried. + + An unknown name would make the query use a predicate no shape or data + uses. An identifier slot is the node's IRI, not a property arc (the + main slot loop does not require it either), so a condition on it can + never match. + """ + slot = self._rule_slot(site.cls, slot_name) + if slot is None: + self._skip_rule(site, f"its condition names {slot_name!r}, which is not a slot") + return None + if slot.identifier: + self._skip_rule( + site, f"its condition is on the identifier slot {slot_name!r}, which is the node IRI, not a property" + ) + return None + return slot - # 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" + def _slot_range(self, slot: SlotDefinition) -> ElementName | None: + """The range of *slot*, defaulting to the schema's ``default_range`` as an induced slot does.""" + return slot.range or self.schemaview.schema.default_range - if is_value_present and is_flag_true: - return self._build_boolean_guard_sparql(sv, cls, post_slot_name, pre_slot_name) + def _slot_iri(self, slot: SlotDefinition) -> str: + """The full IRI of *slot*, exactly as ``sh:path`` in the main slot loop renders it. - # Pattern: exclusive value - # preconditions: slot X has equals_string (a specific enum value) - # postconditions: same slot X has 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) + An induced slot carries its ``slot_usage`` overrides, so an overridden + ``slot_uri`` yields the same IRI as ``sh:path``; otherwise the query + would use a property the data never uses and never fire. + """ + sv = self.schemaview + if slot.name in sv.element_by_schema_map(): + return sv.get_uri(slot, expand=True) + return sv.expand_curie(f"{sv.schema.default_prefix}:{underscore(slot.name)}") - 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)) + def _type_uri(self, r: ElementName | None) -> str | None: + """The expanded datatype IRI of type range *r*, or ``None`` when *r* is not a type. + Resolved through the induced type, so a type derived with ``typeof`` + inherits the ``uri`` of its ancestor. A built-in type name in a schema + that does not import ``linkml:types`` resolves as the main slot loop + resolves it (:class:`ShaclDataType`). + """ + sv = self.schemaview + if r in sv.all_types(): + return sv.get_uri(sv.induced_type(r), expand=True) + builtin = next((t for t in ShaclDataType if t.linkml_type == r), None) + return str(builtin.uri_ref) if builtin is not None else None + + def _is_string_range(self, r: ElementName | None) -> bool: + """Whether a slot with range *r* holds strings, which ``equals_string`` compares against. + + True for an enum, whose permissible values are rendered as their + ``meaning`` IRI or as a plain literal (as :meth:`_add_enum` renders + them); for a type whose datatype is ``xsd:string``, whose values are + plain literals; and for no range at all, whose values are untyped and + compared as strings, as the JSON Schema generator compares them. A + type with any other datatype, including one derived from ``string`` + (``xsd:anyURI``, ``xsd:token``, ...), holds typed literals or IRIs + that a string literal does not match. + """ + if r is None or r in self.schemaview.all_enums(): + return True + return self._type_uri(r) == str(XSD.string) + + def _value_terms(self, site: _RuleSite, slot: SlotDefinition, values: list[str]) -> list[str] | None: + """The SPARQL terms of the equals_string(_in) *values* on *slot*, or ``None`` (rule skipped). + + The metamodel defines both operators for slots of range ``string``. + Enums, whose values are strings in instance data, and types with + datatype ``xsd:string`` are treated alike (:meth:`_string_value_term`). + A slot whose range is ``xsd:boolean``-typed holds booleans, as the + boolean guard compares them: each value must be a lexical form of + ``xsd:boolean`` and becomes the boolean it denotes, compared by value. + On any other range the RDF data holds typed literals + (``"3"^^xsd:integer``) or IRIs that no string equals, so the + constraint would report conforming data or never fire. + """ + r = self._slot_range(slot) + if self._is_string_range(r): + return [self._string_value_term(site, slot, value) for value in values] + if self._type_uri(r) == str(XSD.boolean): + terms = [self._XSD_BOOLEAN_LEXICAL.get(value) for value in values] + if None not in terms: + return terms + self._skip_rule( + site, + f"equals_string(_in) on the boolean slot {slot.name!r} with {values!r}, " + "which are not all xsd:boolean lexical forms (true, false, 1, 0)", + ) + return None + self._skip_rule( + site, + f"equals_string(_in) on slot {slot.name!r}, whose range {r!r} is neither an enum " + "nor a type with datatype xsd:string or xsd:boolean", + ) return None - def _build_boolean_guard_sparql(self, sv, cls: ClassDefinition, flag_slot_name: str, value_slot_name: str) -> str: - """Build a SPARQL SELECT query for the boolean-guard pattern. + def _string_value_term(self, site: _RuleSite, slot: SlotDefinition, value: str) -> str: + """The SPARQL term of the ``equals_string`` *value* of string-valued *slot*. + + A permissible value of an enum range is rendered as :meth:`_add_enum` + renders it, as the IRI of its ``meaning`` where it has one; anything + else is a string literal. A value that is not a permissible value of + an enum with static permissible values is reported: no valid value of + the slot can equal it. + """ + sv = self.schemaview + r = self._slot_range(slot) + if r in sv.all_enums(): + permissible_values = sv.get_enum(r).permissible_values + pv = permissible_values.get(value) + if pv is not None and pv.meaning: + return f"<{sv.expand_curie(pv.meaning)}>" + if pv is None and permissible_values: + self._warn_rule( + site, + f"it compares slot {slot.name!r} with {value!r}, which is not a permissible value of enum {r!r}", + ) + return self._sparql_string_literal(value) + + @staticmethod + def _sparql_string_literal(value: str) -> str: + """Render *value* as a SPARQL expression for that string. + + The characters the grammar forbids raw are escaped (`SPARQL 1.1 §19.7 + `_), so a value + with a double quote, backslash or newline neither breaks the + ``sh:select`` query nor injects into it. Codepoint escapes are replaced + before parsing (`§19.2 `_), + so an escaped backslash followed by ``u`` / ``U`` would be read as one: + the literal is split there and rejoined with ``CONCAT``. + """ + escaped = ( + str(value) + .replace("\\", "\\\\") + .replace('"', '\\"') + .replace("\n", "\\n") + .replace("\r", "\\r") + .replace("\t", "\\t") + ) + parts = re.split(r"(?<=\\)(?=[uU])", escaped) + if len(parts) == 1: + return f'"{escaped}"' + return "CONCAT(" + ", ".join(f'"{part}"' for part in parts) + ")" + + @staticmethod + def _sparql_is_one_of(var: str, terms: list[str]) -> str: + """A SPARQL expression that is true when *var* equals one of *terms*, and false otherwise. + + The test is spelled out as ``var = t1 || var = t2 || ...``, which is how + `SPARQL 1.1 §17.4.1.9 `_ + defines ``IN``. The ``=`` operator compares values, so a plain literal + also matches its RDF 1.1-identical ``xsd:string`` form (`RDF 1.1 + Concepts §3.3 `_), + which JSON-LD produces under ``"@type": "xsd:string"`` coercion; some + engines, rdflib among them, match ``IN`` and triple-pattern constants by + term identity and miss that form. ``COALESCE`` (§17.4.1.3) turns the + type error that ``RDFterm-equal`` (§17.4.1.7) raises for incomparable + literals into false: such a value equals none of *terms*, and an + unguarded error would drop the row and hide the violation. An unbound + *var* is false as well. + """ + disjunction = " || ".join(f"{var} = {term}" for term in terms) + return f"COALESCE( {disjunction}, false )" + + def _build_presence_implies_value_sparql( + self, guard: SlotDefinition, target: SlotDefinition, terms: list[str], open_world: bool + ) -> str: + """Build the SPARQL SELECT query of the presence-implies-value pattern. - The query detects violations where the value property is present - but the boolean flag is absent or not ``true``. + A focus node violates the rule when the *guard* slot is present and the + *target* slot is absent or holds a value that equals none of the SPARQL + *terms* (see :meth:`_sparql_is_one_of`). The boolean guard is this + pattern with the boolean ``true`` as the only term. With *open_world*, + "the postconditions may be omitted in instance data" (metamodel + ``open_world``), so an absent target is no violation. Conforms to `SHACL §5.3.1 - `_: - ``$this`` is pre-bound to each focus node. + `_ (``$this`` + is pre-bound to each focus node) and projects ``?path`` and ``?value`` + as result variables (`§5.3.2 + `_), so each + result names the target property and the offending value. """ - flag_uri = self._slot_uri(sv, flag_slot_name, cls) - value_uri = self._slot_uri(sv, value_slot_name, cls) - + guard_iri = self._slot_iri(guard) + target_iri = self._slot_iri(target) + is_allowed = self._sparql_is_one_of("?value", terms) + if open_world: + target_pattern = f"$this <{target_iri}> ?value ." + violation = f"!{is_allowed}" + else: + target_pattern = f"OPTIONAL {{ $this <{target_iri}> ?value . }}" + violation = f"!BOUND(?value) || !{is_allowed}" return ( - f"SELECT $this WHERE {{\n" - f" OPTIONAL {{ $this <{flag_uri}> ?flag . }}\n" - f" OPTIONAL {{ $this <{value_uri}> ?value . }}\n" - f" FILTER (\n" - f' ( !BOUND(?flag) || str(?flag) != "true" ) &&\n' - f" BOUND(?value)\n" - f" )\n" + f"SELECT DISTINCT $this (<{target_iri}> AS ?path) ?value WHERE {{\n" + f" $this <{guard_iri}> ?guard .\n" + f" {target_pattern}\n" + f" FILTER ( {violation} )\n" f"}}" ) - def _build_exclusive_value_sparql( - self, - sv, - cls: ClassDefinition, - slot_name: str, - value_name: str, - max_card: int, - ) -> str | None: - """Build a SPARQL SELECT query for the exclusive-value pattern. + def _build_exclusive_value_sparql(self, slot: SlotDefinition, term: str, max_card: int) -> str: + """Build the SPARQL SELECT query of the exclusive-value pattern. - Detects violations where a specific value is present in a multivalued - slot but the total number of values exceeds *max_card*. - - For the common case ``max_card == 1``, the query checks whether the - exclusive value coexists with any other value (simple existence test). - For ``max_card > 1``, a subquery counts all values and checks against - the limit. - - The exclusive value is resolved to its full IRI via the slot's enum - ``meaning`` field. If the slot is not an enum or the value has no - ``meaning``, the value is compared as a plain literal. + A focus node violates the rule when *slot* holds a value equal to the + SPARQL *term* (see :meth:`_sparql_is_one_of`) and more than *max_card* + values in total. For ``max_card == 1`` the query reports each other + value that coexists with the exclusive one. For ``max_card > 1`` a + subquery counts all values, testing ``HAVING`` on the ``COUNT`` + aggregate itself: expressions projected by ``SELECT`` are not visible + to ``HAVING`` (`SPARQL 1.1 §18.2.4.2 + `_), so a + projected alias would be unbound there and the constraint never fire. Conforms to `SHACL §5.3.1 - `_: - ``$this`` is pre-bound to each focus node. + `_ and + projects ``?path`` (and, for ``max_card == 1``, the coexisting value as + ``?value``) as result variables (`§5.3.2 + `_). """ - slot_uri = self._slot_uri(sv, slot_name, cls) - value_ref = self._resolve_enum_value_ref(sv, slot_name, value_name) - + iri = self._slot_iri(slot) + is_exclusive = self._sparql_is_one_of("?exclusive", [term]) if max_card == 1: + is_other_exclusive = self._sparql_is_one_of("?other", [term]) return ( - f"SELECT $this WHERE {{\n" - f" $this <{slot_uri}> {value_ref} .\n" - f" $this <{slot_uri}> ?other .\n" - f" FILTER (?other != {value_ref})\n" + f"SELECT DISTINCT $this (<{iri}> AS ?path) (?other AS ?value) WHERE {{\n" + f" $this <{iri}> ?exclusive .\n" + f" $this <{iri}> ?other .\n" + f" FILTER ( {is_exclusive} && !{is_other_exclusive} )\n" f"}}" ) - return ( - f"SELECT $this WHERE {{\n" - f" $this <{slot_uri}> {value_ref} .\n" + f"SELECT DISTINCT $this (<{iri}> AS ?path) WHERE {{\n" + f" $this <{iri}> ?exclusive .\n" + f" FILTER ( {is_exclusive} )\n" f" {{\n" - f" SELECT $this (COUNT(?val) AS ?count)\n" - f" WHERE {{ $this <{slot_uri}> ?val . }}\n" + f" SELECT $this\n" + f" WHERE {{ $this <{iri}> ?item . }}\n" f" GROUP BY $this\n" - f" HAVING (?count > {max_card})\n" + f" HAVING ( COUNT(?item) > {max_card} )\n" f" }}\n" f"}}" ) - def _resolve_enum_value_ref(self, sv, slot_name: str, value_name: str) -> 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``. - """ - slot = sv.get_slot(slot_name) - if slot: - range_name = slot.range - if range_name and range_name in sv.all_enums(): - enum = sv.get_enum(range_name) - pv = enum.permissible_values.get(value_name) - 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``. - """ - slot = sv.get_slot(slot_name) - if slot and 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)}") - def _add_class(self, func: Callable, r: ElementName) -> None: """Add a class/shape constraint for range class *r*. @@ -958,8 +1274,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 cf6764ad6d..1982600f59 100644 --- a/tests/linkml/test_generators/test_shaclgen.py +++ b/tests/linkml/test_generators/test_shaclgen.py @@ -1,9 +1,12 @@ +import json +import logging +import re from collections import Counter from typing import Any import pytest import rdflib -from rdflib import RDF, RDFS, SH, Literal, URIRef +from rdflib import RDF, RDFS, SH, XSD, Literal, URIRef from rdflib.collection import Collection from linkml.generators.shacl.shacl_data_type import ShaclDataType @@ -2108,7 +2111,7 @@ def test_rule_deactivated_skipped(): def test_rule_unsupported_pattern_skipped(): - """Unrecognised rule patterns are silently skipped (no sh:sparql emitted).""" + """Unrecognised rule patterns are skipped (no sh:sparql emitted).""" g = _parse_shacl(_UNSUPPORTED_RULE_SCHEMA_YAML) shape = URIRef("https://example.org/unsupported-test/TestClass") @@ -2423,7 +2426,7 @@ def test_rule_sparql_syntax_valid(): # =========================================================================== # # The "exclusive value" pattern translates a LinkML rule where: -# - preconditions: slot X has equals_string (a specific enum value name) +# - preconditions: slot X has has_member: {equals_string: V} # - postconditions: same slot X has maximum_cardinality N # # Semantics: "If value V is present in multivalued slot X, then X has at most @@ -2494,7 +2497,8 @@ def test_rule_sparql_syntax_valid(): preconditions: slot_conditions: edgeType: - equals_string: "EdgeNone" + has_member: + equals_string: "EdgeNone" postconditions: slot_conditions: edgeType: @@ -2510,7 +2514,8 @@ def test_rule_sparql_syntax_valid(): preconditions: slot_conditions: edgeType: - equals_string: "EdgeNone" + has_member: + equals_string: "EdgeNone" postconditions: slot_conditions: edgeType: @@ -2526,7 +2531,8 @@ def test_rule_sparql_syntax_valid(): preconditions: slot_conditions: priority: - equals_string: "High" + has_member: + equals_string: "High" postconditions: slot_conditions: priority: @@ -2543,7 +2549,8 @@ def test_rule_sparql_syntax_valid(): preconditions: slot_conditions: edgeType: - equals_string: "EdgeNone" + has_member: + equals_string: "EdgeNone" postconditions: slot_conditions: otherSlot: @@ -2585,13 +2592,14 @@ def test_exclusive_value_sparql_uses_enum_iri(): def test_exclusive_value_max_card_1_sparql_structure(): - """For maximum_cardinality: 1, SPARQL uses FILTER(?other != ). + """For maximum_cardinality: 1, SPARQL reports each ?other value coexisting with . The query pattern for N=1 is: - SELECT $this WHERE { - $this . + SELECT DISTINCT $this ( AS ?path) (?other AS ?value) WHERE { + $this ?exclusive . $this ?other . - FILTER (?other != ) + FILTER ( COALESCE( ?exclusive = , false ) && + !COALESCE( ?other = , false ) ) } This is more efficient than the COUNT-based approach for the common @@ -2616,13 +2624,14 @@ def test_exclusive_value_max_card_gt1_sparql_structure(): """For maximum_cardinality > 1, SPARQL uses COUNT-based subquery. The query pattern for N>1 is: - SELECT $this WHERE { - $this . + SELECT DISTINCT $this ( AS ?path) WHERE { + $this ?exclusive . + FILTER ( COALESCE( ?exclusive = , false ) ) { - SELECT $this (COUNT(?val) AS ?count) - WHERE { $this ?val . } + SELECT $this + WHERE { $this ?item . } GROUP BY $this - HAVING (?count > N) + HAVING ( COUNT(?item) > N ) } } """ @@ -2755,7 +2764,8 @@ def test_exclusive_value_coexists_with_boolean_guard(): preconditions: slot_conditions: status: - equals_string: "None" + has_member: + equals_string: "None" postconditions: slot_conditions: status: @@ -2972,3 +2982,1728 @@ def conforms(data_ttl: str) -> tuple[bool, str]: for value in ('"x1"', '"za"'): ok, _ = conforms(f"{prefixes}\nex:t2 a ex:SlotPatternWithAnyOf ; ex:tag {value} .") assert not ok, f"tag {value} violates the slot pattern or matches no branch and must be rejected" + + +# =========================================================================== +# 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 guard 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 guard slot is present, the target slot must be present +# and hold one of the allowed values", e.g. "if a document has a signature, +# its status must be Published". +# +# References: +# - W3C SHACL §5 +# - W3C SHACL §5.3.1 +# - W3C SHACL §5.3.2 +# =========================================================================== + +_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: + StatusEnum: + permissible_values: + Draft: + meaning: ex:Draft + Reviewed: + meaning: ex:Reviewed + Approved: + meaning: ex:Approved + Published: + meaning: ex:Published + + ModeEnum: + permissible_values: + Auto: + description: Automatic mode (no meaning IRI). + Manual: + description: Manual mode (no meaning IRI). + +slots: + status: + range: StatusEnum + slot_uri: ex:status + signature: + range: string + slot_uri: ex:signature + review_score: + range: float + slot_uri: ex:review_score + mode: + range: ModeEnum + slot_uri: ex:mode + manual_value: + range: decimal + slot_uri: ex:manual_value + +classes: + Document: + class_uri: ex:Document + slots: + - status + - signature + - review_score + rules: + - description: If a document has a signature, its status must be Published. + preconditions: + slot_conditions: + signature: + value_presence: PRESENT + postconditions: + slot_conditions: + status: + equals_string: "Published" + - description: If a document has a review score, its status must be Reviewed or Approved. + preconditions: + slot_conditions: + review_score: + value_presence: PRESENT + postconditions: + slot_conditions: + status: + equals_string_in: + - Reviewed + - Approved + + 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 _rule_results(schema: str, data: str | rdflib.Graph) -> tuple[bool, list[tuple]]: + """Validate *data* against the shapes generated for *schema*. + + *data* is Turtle or a graph. Returns whether the data conforms overall, + and the ``(focus node, result path, value)`` of each result that a rule + constraint (``sh:sparql``) reports, so that a property-shape result + (datatype, cardinality, ...) can neither mask nor stand in for what a rule + decides. + """ + import pyshacl + + shapes = rdflib.Graph().parse(data=ShaclGenerator(schema, mergeimports=False).serialize(), format="turtle") + data_graph = data if isinstance(data, rdflib.Graph) else rdflib.Graph().parse(data=data, format="turtle") + conforms, report, _ = pyshacl.validate(data_graph, shacl_graph=shapes, advanced=True) + results = [ + (report.value(result, SH.focusNode), report.value(result, SH.resultPath), report.value(result, SH.value)) + for result in report.subjects(SH.sourceConstraintComponent, SH.SPARQLConstraintComponent) + ] + return conforms, results + + +def _validate_rules(schema: str, data: str | rdflib.Graph) -> tuple[bool, set]: + """Whether *data* conforms to the shapes of *schema*, and the focus nodes its rule constraints report.""" + conforms, results = _rule_results(schema, data) + return conforms, {focus_node for focus_node, _, _ in results} + + +def _sparql_queries(g: rdflib.Graph, shape: URIRef) -> list[str]: + """The ``sh:select`` query of every ``sh:sparql`` constraint on *shape*.""" + return [str(query) for node in g.objects(shape, SH.sparql) for query in g.objects(node, SH.select)] + + +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.Document + 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 "?value = " in query, "presence-implies-value SPARQL must compare the target by value" + 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) + + signature_query = [q for q in _sparql_queries(g, EX_PIV.Document) if str(EX_PIV.signature) in q] + assert len(signature_query) == 1, "Expected exactly one signature rule" + query = signature_query[0] + + assert str(EX_PIV.status) in query + assert f"<{EX_PIV.Published}>" in query, f"Expected Published 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) + + review_query = [q for q in _sparql_queries(g, EX_PIV.Document) if str(EX_PIV.review_score) in q] + assert len(review_query) == 1, "Expected exactly one review_score rule" + query = review_query[0] + + assert f"<{EX_PIV.Reviewed}>" in query, f"Expected Reviewed IRI, got:\n{query}" + assert f"<{EX_PIV.Approved}>" in query, f"Expected Approved 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) + + queries = _sparql_queries(g, EX_PIV.Device) + assert len(queries) == 1 + query = queries[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) + + sparql_nodes = list(g.objects(EX_PIV.Document, SH.sparql)) + messages = [str(m) for node in sparql_nodes for m in g.objects(node, SH.message)] + + assert any("status must be Published" in m for m in messages), f"Expected message about Published, 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.Document, EX_PIV.Device): + for query in _sparql_queries(g, shape): + prepareQuery(query) + + +@pytest.mark.parametrize( + "instance,violates", + [ + pytest.param('ex:x a ex:Document ; ex:signature "s" ; ex:status ex:Published .', False, id="single-ok"), + pytest.param( + 'ex:x a ex:Document ; ex:review_score "4.5"^^xsd:float ; ex:status ex:Reviewed .', False, id="set-ok" + ), + pytest.param( + 'ex:x a ex:Document ; ex:review_score "3.0"^^xsd:float ; ex:status ex:Approved .', + False, + id="set-other-member-ok", + ), + pytest.param("ex:x a ex:Document ; ex:status ex:Draft .", False, id="unguarded"), + pytest.param('ex:x a ex:Document ; ex:signature "s" ; ex:status ex:Draft .', True, id="wrong-value"), + pytest.param( + 'ex:x a ex:Document ; ex:review_score "4.5"^^xsd:float ; ex:status ex:Published .', + True, + id="not-in-set", + ), + pytest.param('ex:x a ex:Document ; ex:signature "s" .', True, id="target-absent"), + pytest.param('ex:x a ex:Device ; ex:manual_value 1.5 ; ex:mode "Manual" .', False, id="literal-ok"), + pytest.param('ex:x a ex:Device ; ex:manual_value 1.5 ; ex:mode "Auto" .', True, id="literal-wrong"), + ], +) +def test_presence_implies_value_pyshacl_end_to_end(instance: str, violates: bool): + """End-to-end: pyshacl passes conforming instances and flags violations.""" + data = ( + "@prefix ex: .\n" + "@prefix xsd: .\n" + instance + ) + conforms, focus_nodes = _validate_rules(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML, data) + assert focus_nodes == ({EX_PIV.x} if violates else set()) + assert conforms is not violates + + +@pytest.mark.parametrize( + "instance,value", + [ + pytest.param('ex:signature "s" ; ex:status ex:Draft', EX_PIV.Draft, id="wrong-value"), + pytest.param('ex:signature "s"', EX_PIV.x, id="target-absent"), + ], +) +def test_presence_implies_value_result_names_path_and_value(instance, value): + """A result names the target property (``sh:resultPath``) and the + offending value (``sh:value``); when the target is absent, SHACL §5.3.2 + falls back to the focus node as the value.""" + data = f"@prefix ex: .\nex:x a ex:Document ; {instance} ." + _, results = _rule_results(_PRESENCE_IMPLIES_VALUE_SCHEMA_YAML, data) + assert results == [(EX_PIV.x, EX_PIV.status, value)] + + +# =========================================================================== +# equals_string / equals_string_in compare strings +# =========================================================================== +# +# The metamodel defines both operators for slots of range string: "the slot +# must have range string and the value of the slot must equal the specified +# value". Enums, whose values are strings in instance data, and types with +# datatype xsd:string are treated alike. A rule applying them to a slot of +# any other range is skipped with a warning: its RDF values are typed literals +# ("3"^^xsd:integer, false) or IRIs that equal no string, so a translated +# constraint would report conforming data or never fire. On a string-valued +# slot the value is compared with the SPARQL "=" operator, which matches the +# RDF 1.1-identical plain and xsd:string literal forms alike, and an +# incomparable value counts as unequal rather than as an error. +# +# References: +# - SPARQL 1.1 §17.4.1.3 +# - SPARQL 1.1 §17.4.1.9 +# - RDF 1.1 Concepts §3.3 +# =========================================================================== + +_RULE_TARGET_RANGE_SCHEMA_YAML = """ +id: https://example.org/rule-target-range +name: rule_target_range +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/rule-target-range/ + xsd: http://www.w3.org/2001/XMLSchema# +imports: + - linkml:types +default_prefix: ex +default_range: string +types: + label_string: + typeof: string + token_string: + typeof: string + uri: xsd:token + flag_boolean: + typeof: boolean +enums: + Code: + permissible_values: + alpha: + beta: + Color: + permissible_values: + Red: + meaning: ex:Red + Blue: + meaning: ex:Blue + Answer: + permissible_values: + "true": + meaning: ex:Yes + "false": + meaning: ex:No + Mixed: + permissible_values: + A: + meaning: ex:A + B: +classes: + Item: + class_uri: ex:Item + attributes: + id: + identifier: true + range: uriorcurie + Thing: + class_uri: ex:Thing + attributes: + guard: + multivalued: {guard_multivalued} + target: + range: {range} + multivalued: {multivalued} + rules: + - open_world: {open_world} + preconditions: + slot_conditions: + guard: {guard_condition} + postconditions: + slot_conditions: + target: {target_condition} + Sub: + is_a: Thing + class_uri: ex:Sub +""" + +EX_RTR = rdflib.Namespace("https://example.org/rule-target-range/") + +_RTR_PREFIXES = ( + "@prefix ex: .\n@prefix xsd: .\n" +) + + +def _rule_target_schema( + target_range: str, + value: str | list[str], + multivalued: bool = False, + *, + operator: str = "equals_string", + open_world: bool = False, + guard_multivalued: bool = False, + guard_extra: dict | None = None, + target_extra: dict | None = None, +) -> str: + """The rule-target schema: if ``guard`` is present, ``target`` (of *target_range*) must satisfy *operator*. + + *value* is the operand of *operator* (a list for ``equals_string_in``); + *guard_extra* and *target_extra* add operators to the two conditions. + ``Sub`` inherits the rule from ``Thing``. + """ + return _RULE_TARGET_RANGE_SCHEMA_YAML.format( + range=target_range, + multivalued=str(multivalued).lower(), + guard_multivalued=str(guard_multivalued).lower(), + open_world=str(open_world).lower(), + guard_condition=json.dumps({"value_presence": "PRESENT", **(guard_extra or {})}), + target_condition=json.dumps({operator: value, **(target_extra or {})}), + ) + + +@pytest.mark.parametrize( + "target_range,operator,value,conforming_term", + [ + pytest.param("integer", "equals_string", "3", "3", id="integer"), + pytest.param("integer", "equals_string_in", ["3", "4"], "3", id="integer-equals-string-in"), + pytest.param("float", "equals_string", "1.5", '"1.5"^^xsd:float', id="float"), + pytest.param("decimal", "equals_string", "1.5", "1.5", id="decimal"), + pytest.param("date", "equals_string", "2024-01-01", '"2024-01-01"^^xsd:date', id="date"), + pytest.param("uriorcurie", "equals_string", "ex:a", "ex:a", id="uriorcurie"), + pytest.param("token_string", "equals_string", "a", '"a"^^xsd:token', id="string-derived-xsd-token"), + pytest.param("Item", "equals_string", "ex:i1", "ex:i1 . ex:i1 a ex:Item", id="class"), + ], +) +def test_rule_equals_string_on_non_string_range_skipped(caplog, target_range, operator, value, conforming_term): + """equals_string on a slot that does not hold strings is skipped with a warning. + + The data holds a typed literal or an IRI, which equals no string: comparing + it with one would report every conforming instance. The conforming instance + must therefore pass, and the skip must be reported. + """ + schema = _rule_target_schema(target_range, value, operator=operator) + with caplog.at_level(logging.WARNING): + g = _parse_shacl(schema) + + assert _sparql_queries(g, EX_RTR.Thing) == [] + assert _sparql_queries(g, EX_RTR.Sub) == [] + assert any( + f"slot 'target', whose range '{target_range}' is neither an enum nor a type with datatype xsd:string or " + "xsd:boolean" in rec.message + for rec in caplog.records + ), caplog.text + conforms, focus_nodes = _validate_rules( + schema, f'{_RTR_PREFIXES}ex:x a ex:Thing ; ex:guard "g" ; ex:target {conforming_term} .' + ) + assert conforms and focus_nodes == set() + + +_STRING_TARGETS = [ + pytest.param("string", "x", '"x"', '"y"', id="string"), + pytest.param("label_string", "x", '"x"', '"y"', id="string-derived"), + pytest.param("curie", "ex:a", '"ex:a"', '"ex:b"', id="curie"), + pytest.param("Code", "alpha", '"alpha"', '"beta"', id="enum-literal"), + pytest.param("Color", "Red", "ex:Red", "ex:Blue", id="enum-meaning"), + pytest.param("Answer", "true", "ex:Yes", "ex:No", id="enum-meaning-named-true"), +] + + +@pytest.mark.parametrize("target_range,value,allowed,other", _STRING_TARGETS) +@pytest.mark.parametrize( + "body,violates", + [ + pytest.param('ex:guard "g" ; ex:target {allowed}', False, id="allowed"), + pytest.param('ex:guard "g" ; ex:target {allowed_xsd_string}', False, id="allowed-xsd-string"), + pytest.param('ex:guard "g" ; ex:target {other}', True, id="other"), + pytest.param('ex:guard "g" ; ex:target {allowed}, {other}', True, id="allowed-and-other"), + pytest.param('ex:guard "g"', True, id="target-absent"), + pytest.param("ex:target {other}", False, id="guard-absent"), + ], +) +@pytest.mark.parametrize("multivalued", [False, True]) +def test_rule_equals_string_on_string_range_compares_values( + target_range, value, allowed, other, body, violates, multivalued +): + """On a string-valued target, equals_string compares the value the range holds. + + A string type holds plain literals, an enum its meaning IRI or a plain + literal; the ``xsd:string``-typed form of a plain literal is the same RDF + 1.1 term and must also satisfy the rule. An enum value named ``true`` + with a meaning is an IRI like any other, not a boolean. + """ + allowed_xsd_string = f"{allowed}^^xsd:string" if allowed.startswith('"') else allowed + schema = _rule_target_schema(target_range, value, multivalued) + data = body.format(allowed=allowed, allowed_xsd_string=allowed_xsd_string, other=other) + + assert len(_sparql_queries(_parse_shacl(schema), EX_RTR.Thing)) == 1 + _, focus_nodes = _validate_rules(schema, f"{_RTR_PREFIXES}ex:x a ex:Thing ; {data} .") + assert focus_nodes == ({EX_RTR.x} if violates else set()) + + +_RULE_INSTANCES = [ + pytest.param({"guard": "g", "target": "{value}"}, True, True, id="allowed"), + pytest.param({"guard": "g", "target": "{other}"}, False, False, id="other"), + pytest.param({"guard": "g"}, False, True, id="target-absent"), + pytest.param({"target": "{other}"}, True, True, id="guard-absent"), +] +"""JSON instances of the rule-target schema, with their validity under closed- and open-world rules.""" + + +def _json_schema_and_shacl_verdicts( + schema: str, target_class: str, obj: dict, coerce_xsd_string: bool = False +) -> tuple[bool, bool]: + """Whether the JSON Schema and the SHACL rules generated for *schema* accept *obj*. + + *obj* is validated as is by the generated JSON Schema, and as RDF loaded + through the generated JSON-LD context by the generated SHACL shapes (rule + constraints only). With *coerce_xsd_string* the context types the + ``target`` slot ``xsd:string``, as JSON-LD 1.1 type coercion permits, + which yields the RDF 1.1-identical typed form of a plain literal. + """ + import jsonschema + + from linkml.generators.jsonldcontextgen import ContextGenerator + from linkml.generators.jsonschemagen import JsonSchemaGenerator + + json_schema = json.loads(JsonSchemaGenerator(schema, top_class=target_class).serialize()) + json_schema_valid = jsonschema.Draft7Validator(json_schema).is_valid(obj) + + context = json.loads(ContextGenerator(schema).serialize())["@context"] + if coerce_xsd_string and context["target"].get("@type") is None: + context["target"]["@type"] = "xsd:string" + document = {"@context": context, "@type": target_class, **obj} + _, focus_nodes = _validate_rules(schema, rdflib.Graph().parse(data=json.dumps(document), format="json-ld")) + return json_schema_valid, focus_nodes == set() + + +@pytest.mark.parametrize( + "target_range,value,other", + [ + pytest.param("string", "x", "y", id="string"), + pytest.param("curie", "ex:a", "ex:b", id="curie"), + pytest.param("Code", "alpha", "beta", id="enum-literal"), + pytest.param("Color", "Red", "Blue", id="enum-meaning"), + ], +) +@pytest.mark.parametrize("instance,valid,_valid_open_world", _RULE_INSTANCES) +@pytest.mark.parametrize("coerce_xsd_string", [False, True]) +def test_rule_equals_string_agrees_with_json_schema( + target_range, value, other, instance, valid, _valid_open_world, coerce_xsd_string +): + """The SHACL rule and the JSON Schema rule decide the same instance alike. + + One JSON object is checked by both generated artifacts (see + :func:`_json_schema_and_shacl_verdicts`), with and without ``xsd:string`` + coercion of the target, which the test adds to the generated context. An + enum whose values all have a meaning in one namespace is mapped to those + IRIs by the context and stays untyped. + """ + obj = {key: text.format(value=value, other=other) for key, text in instance.items()} + verdicts = _json_schema_and_shacl_verdicts( + _rule_target_schema(target_range, value), "Thing", obj, coerce_xsd_string + ) + assert verdicts == (valid, valid) + + +@pytest.mark.parametrize( + "target_range,value,other", + [ + pytest.param("string", "x", "y", id="string"), + pytest.param("Color", "Red", "Blue", id="enum-meaning"), + ], +) +@pytest.mark.parametrize("instance,valid_closed_world,valid_open_world", _RULE_INSTANCES) +@pytest.mark.parametrize("target_class", ["Thing", "Sub"]) +@pytest.mark.parametrize("open_world", [False, True]) +def test_rule_inheritance_and_open_world_agree_with_json_schema( + target_range, value, other, instance, valid_closed_world, valid_open_world, target_class, open_world +): + """A rule binds the instances of subclasses of its class, and ``open_world`` + lets the postcondition be omitted, in SHACL as in JSON Schema. + + The metamodel defines ``open_world`` as "the postconditions may be omitted + in instance data", so an absent target satisfies an open-world rule. + """ + obj = {key: text.format(value=value, other=other) for key, text in instance.items()} + schema = _rule_target_schema(target_range, value, open_world=open_world) + valid = valid_open_world if open_world else valid_closed_world + assert _json_schema_and_shacl_verdicts(schema, target_class, obj) == (valid, valid) + + +@pytest.mark.parametrize("flag_range", ["boolean", "flag_boolean"]) +@pytest.mark.parametrize("open_world", [False, True]) +@pytest.mark.parametrize("required", ["true", "1", "false", "0"]) +@pytest.mark.parametrize( + "flag,datatype", + [ + pytest.param("true", XSD.boolean, id="true"), + pytest.param("1", XSD.boolean, id="lexical-1"), + pytest.param("false", XSD.boolean, id="false"), + pytest.param("0", XSD.boolean, id="lexical-0"), + pytest.param("true", None, id="string-true"), + pytest.param(None, None, id="absent"), + ], +) +def test_rule_boolean_target_compared_by_value(flag_range, open_world, required, flag, datatype): + """equals_string on an xsd:boolean-typed slot (the boolean guard, for + "true") requires the boolean its value denotes in the lexical space of + xsd:boolean (XML Schema 1.1 Part 2 §3.3.2.2), whichever type the slot's + range derives it from. The flag is compared by value, so every lexical + form of a boolean satisfies the rule and a string "true" does not; an + open-world rule lets the flag be omitted. + + The data is built with unnormalised literals: rdflib would otherwise + rewrite ``"1"^^xsd:boolean`` to ``true`` while parsing, and a lexical + comparison would pass as well. + """ + schema = _rule_target_schema(flag_range, required, open_world=open_world) + required_value = required in ("true", "1") + queries = _sparql_queries(_parse_shacl(schema), EX_RTR.Thing) + assert len(queries) == 1 + assert f"?value = {str(required_value).lower()}" in queries[0], queries[0] + + data = rdflib.Graph() + data.add((EX_RTR.x, RDF.type, EX_RTR.Thing)) + data.add((EX_RTR.x, EX_RTR.guard, Literal("g"))) + if flag is not None: + data.add((EX_RTR.x, EX_RTR.target, Literal(flag, datatype=datatype, normalize=False))) + if flag is None: + violates = not open_world + else: + violates = datatype is None or (flag in ("true", "1")) != required_value + _, focus_nodes = _validate_rules(schema, data) + assert focus_nodes == ({EX_RTR.x} if violates else set()) + + +@pytest.mark.parametrize("value", ["yes", "True", "TRUE", " true", ""]) +def test_rule_boolean_target_non_lexical_value_skipped(caplog, value): + """A value outside the lexical space of xsd:boolean on a boolean slot equals + no boolean, so the rule is skipped with a warning. The lexical space is + exactly {true, false, 1, 0}: it is case-sensitive and has no whitespace.""" + with caplog.at_level(logging.WARNING): + g = _parse_shacl(_rule_target_schema("boolean", value)) + assert _sparql_queries(g, EX_RTR.Thing) == [] + assert any("not all xsd:boolean lexical forms" in rec.message for rec in caplog.records), caplog.text + + +@pytest.mark.parametrize( + "target,violates", + [ + pytest.param("ex:A", False, id="meaning-iri"), + pytest.param('"B"', False, id="literal"), + pytest.param('"B"^^xsd:string', False, id="literal-xsd-string"), + pytest.param('"A"', True, id="literal-of-value-with-meaning"), + pytest.param("ex:B", True, id="iri-of-value-without-meaning"), + pytest.param('"C"', True, id="other"), + ], +) +def test_rule_equals_string_in_mixes_meaning_iris_and_literals(target, violates): + """``equals_string_in`` over permissible values with and without a meaning + compares each as the term it is rendered as: its meaning IRI, or a literal.""" + schema = _rule_target_schema("Mixed", ["A", "B"], operator="equals_string_in") + _, focus_nodes = _validate_rules(schema, f'{_RTR_PREFIXES}ex:x a ex:Thing ; ex:guard "g" ; ex:target {target} .') + assert focus_nodes == ({EX_RTR.x} if violates else set()) + + +@pytest.mark.parametrize( + "target,violates", + [pytest.param('"x"', False, id="allowed"), pytest.param('"y"', True, id="other")], +) +def test_rule_multivalued_guard(target, violates): + """Several guard values trigger the rule once per focus node, like one.""" + schema = _rule_target_schema("string", "x", guard_multivalued=True) + data = f'{_RTR_PREFIXES}ex:x a ex:Thing ; ex:guard "g1", "g2", "g3" ; ex:target {target} .' + _, results = _rule_results(schema, data) + assert results == ([(EX_RTR.x, EX_RTR.target, rdflib.Literal("y"))] if violates else []) + + +def test_rule_query_yields_one_solution_per_focus_node(): + """Each solution of a SPARQL-based constraint is a validation result + (SHACL §5.3.2), so several guard values must not multiply the solutions. + pyshacl merges identical results; the query itself must not rely on it.""" + schema = _rule_target_schema("string", "x", guard_multivalued=True) + (query,) = _sparql_queries(_parse_shacl(schema), EX_RTR.Thing) + data = rdflib.Graph().parse( + data=f'{_RTR_PREFIXES}ex:x a ex:Thing ; ex:guard "g1", "g2", "g3" ; ex:target "y" .', format="turtle" + ) + assert len(list(data.query(query, initBindings={"this": EX_RTR.x}))) == 1 + + +@pytest.mark.parametrize("dawg_literal_collation", [False, True]) +def test_rule_incomparable_value_is_a_violation(monkeypatch, dawg_literal_collation): + """A value no allowed value can be compared with equals none of them. + + SPARQL ``=`` raises a type error for literals it cannot compare + (RDFterm-equal, SPARQL 1.1 §17.4.1.7), and a FILTER error drops the row, so + a spec-conformant engine would miss the violation unless the comparison + turns the error into false. ``rdflib.DAWG_LITERAL_COLLATION`` switches + rdflib from its lenient comparison to that spec-conformant one. + """ + monkeypatch.setattr(rdflib, "DAWG_LITERAL_COLLATION", dawg_literal_collation) + schema = _rule_target_schema("string", "x") + data = f'{_RTR_PREFIXES}ex:x a ex:Thing ; ex:guard "g" ; ex:target "x"^^ex:custom .' + _, focus_nodes = _validate_rules(schema, data) + assert focus_nodes == {EX_RTR.x} + + exclusive = _exclusive_schema("Tag", "alpha", 1) + data = ( + '@prefix ex: .\nex:x a ex:Thing ; ex:tags "alpha", "beta"^^ex:custom .' + ) + _, focus_nodes = _validate_rules(exclusive, data) + assert focus_nodes == {EX_EXR.x} + + +def test_rule_value_not_a_permissible_value_warns(caplog): + """An equals_string value that is not a permissible value of the enum is + reported: no valid value of the slot can equal it, so the rule rejects + every instance it applies to.""" + schema = _rule_target_schema("Color", "Purple") + with caplog.at_level(logging.WARNING): + queries = _sparql_queries(_parse_shacl(schema), EX_RTR.Thing) + + assert len(queries) == 1 + assert any("'Purple', which is not a permissible value of enum 'Color'" in rec.message for rec in caplog.records) + _, focus_nodes = _validate_rules(schema, f'{_RTR_PREFIXES}ex:x a ex:Thing ; ex:guard "g" ; ex:target ex:Red .') + assert focus_nodes == {EX_RTR.x} + + +@pytest.mark.parametrize( + "value", + [ + pytest.param('a"b', id="quote"), + pytest.param("a\\b", id="backslash"), + pytest.param("line\nbreak", id="newline"), + pytest.param("carriage\rreturn", id="carriage-return"), + pytest.param("tab\there", id="tab"), + pytest.param("a\\u0022b", id="backslash-u"), + pytest.param("a\\U00000022b", id="backslash-capital-u"), + pytest.param("check ✓", id="non-ascii"), + pytest.param("", id="empty"), + ], +) +def test_rule_string_value_round_trips(value): + """An equals_string value of any content is matched exactly: the query + parses, the value itself satisfies the rule, and any other value does not. + + Codepoint escapes are replaced before a SPARQL query is parsed (SPARQL 1.1 + §19.2), so a backslash followed by ``u`` in the value must not reach the + query as an escape sequence. + """ + from rdflib.plugins.sparql import prepareQuery + + schema = _rule_target_schema("string", value) + queries = _sparql_queries(_parse_shacl(schema), EX_RTR.Thing) + assert len(queries) == 1 + prepareQuery(queries[0]) + + for target, violates in ((value, False), (value + "!", True)): + data = rdflib.Graph() + data.add((EX_RTR.x, RDF.type, EX_RTR.Thing)) + data.add((EX_RTR.x, EX_RTR.guard, Literal("g"))) + data.add((EX_RTR.x, EX_RTR.target, Literal(target))) + _, focus_nodes = _validate_rules(schema, data) + assert focus_nodes == ({EX_RTR.x} if violates else set()), repr(target) + + +_EXCLUSIVE_RANGE_SCHEMA_YAML = """ +id: https://example.org/exclusive-range +name: exclusive_range +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/exclusive-range/ +imports: + - linkml:types +default_prefix: ex +default_range: string +enums: + Tag: + permissible_values: + alpha: + beta: + gamma: +classes: + Thing: + class_uri: ex:Thing + attributes: + tags: + range: {range} + multivalued: true + rules: + - preconditions: + slot_conditions: + tags: {precondition} + postconditions: + slot_conditions: + tags: + maximum_cardinality: {max_card} +""" + +EX_EXR = rdflib.Namespace("https://example.org/exclusive-range/") + + +def _exclusive_schema(tag_range: str, value: str, max_card: int, *, has_member: bool = True) -> str: + """The exclusive-value schema: if *value* is one of the ``tags``, there are at most *max_card* of them. + + The precondition is ``has_member: {equals_string: value}``, or with + ``has_member=False`` the bare ``equals_string: value``. + """ + precondition = {"has_member": {"equals_string": value}} if has_member else {"equals_string": value} + return _EXCLUSIVE_RANGE_SCHEMA_YAML.format( + range=tag_range, precondition=json.dumps(precondition), max_card=max_card + ) + + +@pytest.mark.parametrize( + "max_card,values,violates", + [ + pytest.param(1, '"alpha", "beta"', True, id="max1-plain"), + pytest.param(1, '"alpha"^^xsd:string, "beta"^^xsd:string', True, id="max1-xsd-string"), + pytest.param(1, '"alpha"', False, id="max1-alone"), + pytest.param(1, '"beta", "gamma"', False, id="max1-absent"), + pytest.param(2, '"alpha", "beta", "gamma"', True, id="max2-plain"), + pytest.param(2, '"alpha"^^xsd:string, "beta", "gamma"', True, id="max2-xsd-string"), + pytest.param(2, '"alpha", "beta"', False, id="max2-within"), + ], +) +def test_exclusive_value_matches_value_not_term(max_card, values, violates): + """The exclusive value is matched by value, so its RDF 1.1-identical + ``xsd:string``-typed form triggers the rule as the plain literal does, and + a ``maximum_cardinality`` above 1 is enforced.""" + schema = _exclusive_schema("Tag", "alpha", max_card) + data = ( + "@prefix ex: .\n" + "@prefix xsd: .\n" + f"ex:x a ex:Thing ; ex:tags {values} ." + ) + _, focus_nodes = _validate_rules(schema, data) + assert focus_nodes == ({EX_EXR.x} if violates else set()) + + +@pytest.mark.parametrize("max_card", [1, 2]) +def test_exclusive_value_result_names_path_and_value(max_card): + """A result names the slot. For ``maximum_cardinality: 1`` its value is + each value coexisting with the exclusive one; otherwise no value is + projected and SHACL §5.3.2 falls back to the focus node.""" + data = '@prefix ex: .\nex:x a ex:Thing ; ex:tags "alpha", "beta", "gamma" .' + _, results = _rule_results(_exclusive_schema("Tag", "alpha", max_card), data) + if max_card == 1: + assert sorted(results) == sorted( + [(EX_EXR.x, EX_EXR.tags, Literal("beta")), (EX_EXR.x, EX_EXR.tags, Literal("gamma"))] + ) + else: + assert results == [(EX_EXR.x, EX_EXR.tags, EX_EXR.x)] + + +@pytest.mark.parametrize("max_card", [1, 2]) +def test_exclusive_value_query_yields_distinct_solutions(max_card): + """The exclusive value in both RDF 1.1-identical forms binds twice; each + solution is a validation result (SHACL §5.3.2), so the query must not + multiply them.""" + (query,) = _sparql_queries(_parse_shacl(_exclusive_schema("Tag", "alpha", max_card)), EX_EXR.Thing) + data = rdflib.Graph().parse( + data="@prefix ex: .\n" + "@prefix xsd: .\n" + 'ex:x a ex:Thing ; ex:tags "alpha", "alpha"^^xsd:string, "beta" .', + format="turtle", + ) + assert len(list(data.query(query, initBindings={"this": EX_EXR.x}))) == 1 + + +def test_rule_warning_names_the_rule_by_title(caplog): + """A skipped rule is reported by its position and its ``title``, or its ``description`` without one.""" + rule = _rule({"target": {"equals_string": "x"}}, {"target": {"maximum_cardinality": 1}}) + with caplog.at_level(logging.WARNING): + _parse_shacl(_single_rule_schema({**rule, "title": "one target", "description": "a longer text"})) + _parse_shacl(_single_rule_schema({**rule, "description": "a longer text"})) + messages = [rec.message for rec in caplog.records if "skipped" in rec.message] + assert [m.split(":")[0] for m in messages] == [ + "Rule 1 of class 'Thing' (one target)", + "Rule 1 of class 'Thing' (a longer text)", + ] + + +@pytest.mark.parametrize("multivalued", [True, False]) +def test_exclusive_value_bare_equals_string_skipped(caplog, multivalued): + """A bare ``equals_string`` precondition applies to every value of the slot + (``05validation.md``), so it is not the exclusive-value pattern: the rule is + skipped with a warning that names ``has_member``, which states "one of the + values equals".""" + if multivalued: + schema, shape = _exclusive_schema("Tag", "alpha", 1, has_member=False), EX_EXR.Thing + else: + schema = _single_rule_schema(_rule({"target": {"equals_string": "x"}}, {"target": {"maximum_cardinality": 1}})) + shape = EX_SINGLE.Thing + with caplog.at_level(logging.WARNING): + assert _sparql_queries(_parse_shacl(schema), shape) == [] + assert any("every value" in rec.message and "has_member" in rec.message for rec in caplog.records), caplog.text + + +@pytest.mark.parametrize( + "precondition", + [ + pytest.param({"has_member": {"equals_string": "alpha", "pattern": "^a"}}, id="inside-has_member"), + pytest.param({"has_member": {"equals_string": "alpha"}, "minimum_cardinality": 2}, id="beside-has_member"), + ], +) +def test_exclusive_value_has_member_extra_operator_skipped(caplog, precondition): + """has_member is exact too: an operator next to its equals_string, or next + to has_member itself, makes the rule untranslatable.""" + precondition = json.dumps(precondition) + schema = _EXCLUSIVE_RANGE_SCHEMA_YAML.format(range="Tag", precondition=precondition, max_card=1) + with caplog.at_level(logging.WARNING): + assert _sparql_queries(_parse_shacl(schema), EX_EXR.Thing) == [] + assert any("match none of the translated patterns" in rec.message for rec in caplog.records), caplog.text + + +def test_exclusive_value_on_non_string_range_skipped(caplog): + """equals_string on a slot that does not hold strings is skipped with a + warning in the exclusive-value pattern too: the integer ``3`` never equals + the string "3", so the constraint would never fire.""" + schema = _exclusive_schema("integer", "3", 1) + with caplog.at_level(logging.WARNING): + g = _parse_shacl(schema) + + assert _sparql_queries(g, EX_EXR.Thing) == [] + assert any("whose range 'integer' is neither an enum" in rec.message for rec in caplog.records), caplog.text + + +def test_rule_non_operator_fields_are_the_metamodel_element_metadata(): + """The fields the operator accounting ignores are exactly the metamodel's + ``element`` slots that are not ``slot_expression`` slots.""" + from linkml_runtime.utils.formatutils import underscore + from linkml_runtime.utils.introspection import package_schemaview + + metamodel = package_schemaview("linkml_runtime.linkml_model.meta") + element_slots = {underscore(s) for s in metamodel.class_slots("element")} + operator_slots = {underscore(s) for s in metamodel.class_slots("slot_expression")} + + assert ShaclGenerator._NON_OPERATOR_FIELDS == element_slots - operator_slots + assert {"description", "annotations", "title"} <= ShaclGenerator._NON_OPERATOR_FIELDS + + +def test_rule_condition_metadata_does_not_block_translation(): + """Metadata on a condition does not change which instances satisfy it, + so it does not make an otherwise recognised rule untranslatable.""" + schema = _rule_target_schema( + "string", + "x", + guard_extra={"title": "Guard set"}, + target_extra={"description": "The target must then be x.", "comments": ["Metadata only."]}, + ) + assert len(_sparql_queries(_parse_shacl(schema), EX_RTR.Thing)) == 1 + + +# =========================================================================== +# Rule inheritance +# +# A rule applies to "all members of this class" (metamodel `rules`), so a +# class shape carries the rules of its ancestors and mixins, each translated +# in the class's own context. +# =========================================================================== + +_RULE_INHERITANCE_SCHEMA_YAML = """ +id: https://example.org/rule-inheritance +name: rule_inheritance +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/rule-inheritance/ +imports: + - linkml:types +default_prefix: ex +default_range: string +enums: + Color: + permissible_values: + Red: + meaning: ex:Red + Blue: + meaning: ex:Blue + Shade: + permissible_values: + Red: + meaning: ex:ShadeRed +slots: + guard: {} + target: + range: Color + code: {} + label: {} +classes: + Base: + class_uri: ex:Base + slots: [guard, target] + rules: + - preconditions: + slot_conditions: + guard: + value_presence: PRESENT + postconditions: + slot_conditions: + target: + equals_string: Red + Child: + is_a: Base + class_uri: ex:Child + GrandChild: + is_a: Child + class_uri: ex:GrandChild + Shaded: + is_a: Base + class_uri: ex:Shaded + slot_usage: + target: + range: Shade + Counted: + is_a: Base + class_uri: ex:Counted + slot_usage: + target: + range: integer + Labelled: + mixin: true + class_uri: ex:Labelled + slots: [code, label] + rules: + - preconditions: + slot_conditions: + code: + value_presence: PRESENT + postconditions: + slot_conditions: + label: + equals_string: x + Mixed: + class_uri: ex:Mixed + mixins: [Labelled] + SameUri: + is_a: Base + class_uri: ex:Base +""" + +EX_RI = rdflib.Namespace("https://example.org/rule-inheritance/") + + +def test_rule_inherited_by_descendant_shapes(caplog): + """The shape of a class carries the rules of its ancestors and mixins, each + translated in the class's own context: a ``slot_usage`` that narrows the + target to another enum resolves the value there, and one that narrows it to + a non-string range skips the rule for that class only. A class sharing its + ancestor's ``class_uri`` shares its shape, which carries the rule once.""" + with caplog.at_level(logging.WARNING): + g = _parse_shacl(_RULE_INHERITANCE_SCHEMA_YAML) + + for cls in ("Base", "Child", "GrandChild", "Shaded", "Labelled", "Mixed"): + assert len(_sparql_queries(g, EX_RI[cls])) == 1, cls + assert f"<{EX_RI.ShadeRed}>" in _sparql_queries(g, EX_RI.Shaded)[0] + assert _sparql_queries(g, EX_RI.Counted) == [] + assert any("'Counted'" in rec.message and "is neither an enum" in rec.message for rec in caplog.records) + + +@pytest.mark.parametrize( + "instance,violates", + [ + pytest.param('ex:x a ex:Child ; ex:guard "g" ; ex:target ex:Red .', False, id="child-ok"), + pytest.param('ex:x a ex:Child ; ex:guard "g" ; ex:target ex:Blue .', True, id="child-violates"), + pytest.param('ex:x a ex:GrandChild ; ex:guard "g" ; ex:target ex:Blue .', True, id="grandchild-violates"), + pytest.param('ex:x a ex:Shaded ; ex:guard "g" ; ex:target ex:ShadeRed .', False, id="narrowed-ok"), + pytest.param('ex:x a ex:Shaded ; ex:guard "g" ; ex:target ex:Red .', True, id="narrowed-violates"), + pytest.param('ex:x a ex:Mixed ; ex:code "c" ; ex:label "x" .', False, id="mixin-ok"), + pytest.param('ex:x a ex:Mixed ; ex:code "c" ; ex:label "y" .', True, id="mixin-violates"), + ], +) +def test_rule_inherited_rule_enforced_on_descendant_instances(instance, violates): + """An instance typed only with a descendant class is held to the inherited rule.""" + data = f"@prefix ex: .\n{instance}" + _, focus_nodes = _validate_rules(_RULE_INHERITANCE_SCHEMA_YAML, data) + assert focus_nodes == ({EX_RI.x} if violates else set()) + + +def test_rule_shared_shape_reports_once(): + """Two classes sharing a ``class_uri`` share one shape, so an instance + violating the rule both inherit gets one result, not one per class.""" + data = '@prefix ex: .\nex:x a ex:Base ; ex:guard "g" ; ex:target ex:Blue .' + _, results = _rule_results(_RULE_INHERITANCE_SCHEMA_YAML, data) + assert results == [(EX_RI.x, EX_RI.target, EX_RI.Blue)] + + +# =========================================================================== +# Slot resolution and value rendering +# +# A rule's SPARQL body must query the same IRI that ``sh:path`` emits for the +# slot, and resolve the slot's range in the class: +# 1. a slot_usage `slot_uri` (or enum `range`) override applies to the rule; +# 2. an alias-form key (``my_slot`` for slot ``my slot``) names the slot; +# 3. a slot without a range takes the schema's `default_range`; +# 4. built-in type names resolve without importing linkml:types; +# 5. a condition on an unknown or identifier slot makes the rule untranslatable. +# =========================================================================== + + +_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.""" + queries = _sparql_queries(_parse_shacl(_ENUM_NARROWING_SCHEMA_YAML), EX_EN.Scene) + assert len(queries) == 1 + query = queries[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}" + + +_SLOT_RESOLUTION_SCHEMA_YAML = """ +id: https://example.org/slot-resolution +name: slot_resolution +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/slot-resolution/ +imports: + - linkml:types +default_prefix: ex +default_range: string +slots: + guard: {} + my slot: {} + target: {} +classes: + Thing: + class_uri: ex:Thing + slots: [guard, my slot, target] + slot_usage: + target: + slot_uri: ex:narrowed_target + rules: + - preconditions: + slot_conditions: + guard: + value_presence: PRESENT + postconditions: + slot_conditions: + target: + equals_string: x + - preconditions: + slot_conditions: + my_slot: + value_presence: PRESENT + postconditions: + slot_conditions: + target: + equals_string: x +""" + +EX_SR = rdflib.Namespace("https://example.org/slot-resolution/") + + +def test_rule_slot_iris_match_sh_path(): + """Every predicate a rule queries is one that ``sh:path`` emits: a + slot_usage ``slot_uri`` override and an alias-form key resolve to the + slot's own IRI rather than a fabricated one the data never uses.""" + g = _parse_shacl(_SLOT_RESOLUTION_SCHEMA_YAML) + + paths = {str(path) for path in g.objects(None, SH.path)} + queries = _sparql_queries(g, EX_SR.Thing) + assert len(queries) == 2 + for query in queries: + assert set(re.findall(r"<([^>]+)>", query)) <= paths, query + assert f"<{EX_SR.narrowed_target}>" in query, query + assert any(f"<{EX_SR.my_slot}>" in query for query in queries) + + _, focus_nodes = _validate_rules( + _SLOT_RESOLUTION_SCHEMA_YAML, + "@prefix ex: .\n" + 'ex:x a ex:Thing ; ex:my_slot "s" ; ex:narrowed_target "y" .', + ) + assert focus_nodes == {EX_SR.x} + + +_RANGE_RESOLUTION_SCHEMA_YAML = """ +id: https://example.org/range-resolution +name: range_resolution +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/range-resolution/ +imports: + - linkml:types +default_prefix: ex +default_range: {default_range} +slots: + guard: + range: string + target: {target_slot} +classes: + Thing: + class_uri: ex:Thing + slots: {class_slots} + slot_usage: {slot_usage} + rules: + - preconditions: + slot_conditions: + guard: + value_presence: PRESENT + postconditions: + slot_conditions: + target: + equals_string: "3" +""" + +EX_RR = rdflib.Namespace("https://example.org/range-resolution/") + + +@pytest.mark.parametrize( + "default_range,target_slot,class_slots,slot_usage,translated", + [ + pytest.param("string", "{}", "[guard, target]", "{}", True, id="default-range-string"), + pytest.param("integer", "{}", "[guard, target]", "{}", False, id="default-range-integer"), + pytest.param( + "string", + "{range: integer}", + "[guard, target]", + "{target: {range: string}}", + True, + id="slot-usage-to-string", + ), + pytest.param( + "string", "{}", "[guard, target]", "{target: {range: integer}}", False, id="slot-usage-to-integer" + ), + pytest.param("string", "{}", "[guard]", "{}", True, id="slot-not-on-class-default-string"), + pytest.param("integer", "{}", "[guard]", "{}", False, id="slot-not-on-class-default-integer"), + ], +) +def test_rule_target_range_resolved_in_class_context(default_range, target_slot, class_slots, slot_usage, translated): + """Whether the target holds strings is decided by its range in the class: + the schema's ``default_range`` for a slot without one, as refined by + ``slot_usage``, also for a slot the class does not declare.""" + schema = _RANGE_RESOLUTION_SCHEMA_YAML.format( + default_range=default_range, target_slot=target_slot, class_slots=class_slots, slot_usage=slot_usage + ) + assert len(_sparql_queries(_parse_shacl(schema), EX_RR.Thing)) == (1 if translated else 0) + if translated: + prefix = "@prefix ex: .\n" + assert _validate_rules(schema, f'{prefix}ex:x a ex:Thing ; ex:guard "g" ; ex:target "3" .')[1] == set() + assert _validate_rules(schema, f'{prefix}ex:x a ex:Thing ; ex:guard "g" ; ex:target "4" .')[1] == {EX_RR.x} + + +EX_SINGLE = rdflib.Namespace("https://example.org/single-rule/") + + +def _single_rule_schema(rule: dict, attributes: dict | None = None, *, imports: bool = True) -> str: + """A schema whose class ``Thing`` has *attributes* and the one *rule* (JSON, which is YAML). + + By default the attributes are the string slots ``guard``, ``target`` and + ``other``, the boolean ``flag`` and the multivalued ``tags``. Without + *imports* the schema does not import ``linkml:types`` and has no + ``default_range``. + """ + schema = { + "id": "https://example.org/single-rule", + "name": "single_rule", + "prefixes": {"linkml": "https://w3id.org/linkml/", "ex": str(EX_SINGLE)}, + "default_prefix": "ex", + "classes": { + "Thing": { + "class_uri": "ex:Thing", + "attributes": attributes + or { + "guard": {"range": "string"}, + "target": {"range": "string"}, + "other": {"range": "string"}, + "flag": {"range": "boolean"}, + "tags": {"range": "string", "multivalued": True}, + }, + "rules": [rule], + } + }, + } + if imports: + schema["imports"] = ["linkml:types"] + schema["default_range"] = "string" + return json.dumps(schema) + + +def _rule(pre: dict, post: dict, **rule_fields) -> dict: + """A rule with the slot conditions *pre* and *post*.""" + return {"preconditions": {"slot_conditions": pre}, "postconditions": {"slot_conditions": post}, **rule_fields} + + +_GUARD_PRESENT = {"guard": {"value_presence": "PRESENT"}} + + +def _assert_skipped(caplog, schema: str, reason: str) -> None: + """Generating *schema* emits no rule constraint and warns that the rule was skipped for *reason*.""" + with caplog.at_level(logging.WARNING): + g = _parse_shacl(schema) + assert _sparql_queries(g, EX_SINGLE.Thing) == [] + assert any("skipped, because" in rec.message and reason in rec.message for rec in caplog.records), caplog.text + + +@pytest.mark.parametrize( + "target_range,value,allowed,other", + [ + pytest.param("boolean", "true", "true", "false", id="boolean-guard"), + pytest.param("string", "x", '"x"', '"y"', id="string"), + pytest.param("curie", "ex:a", '"ex:a"', '"ex:b"', id="curie"), + pytest.param(None, "x", '"x"', '"y"', id="no-range"), + ], +) +def test_rule_builtin_types_without_imports(target_range, value, allowed, other): + """In a schema that does not import ``linkml:types``, a built-in range name + resolves as the main slot loop resolves it, so the boolean guard and the + string comparison are both translated. A slot with no range at all (and + no ``default_range``) holds untyped values, compared as strings as the + JSON Schema generator compares them.""" + target = {} if target_range is None else {"range": target_range} + schema = _single_rule_schema( + _rule(_GUARD_PRESENT, {"target": {"equals_string": value}}), + {"guard": {"range": "string"}, "target": target}, + imports=False, + ) + assert len(_sparql_queries(_parse_shacl(schema), EX_SINGLE.Thing)) == 1 + prefix = "@prefix ex: .\n" + for term, violates in ((allowed, False), (other, True)): + data = f'{prefix}ex:x a ex:Thing ; ex:guard "g" ; ex:target {term} .' + assert _validate_rules(schema, data)[1] == ({EX_SINGLE.x} if violates else set()) + + +def test_rule_builtin_non_string_type_without_imports_skipped(caplog): + """A built-in non-string range without the ``linkml:types`` import is skipped like an imported one.""" + schema = _single_rule_schema( + _rule(_GUARD_PRESENT, {"target": {"equals_string": "3"}}), + {"guard": {"range": "string"}, "target": {"range": "integer"}}, + imports=False, + ) + _assert_skipped(caplog, schema, "is neither an enum nor a type with datatype xsd:string or xsd:boolean") + + +def test_curie_slot_without_imports_has_xsd_string_datatype(): + """A ``curie`` slot in a schema without the ``linkml:types`` import gets + ``sh:datatype xsd:string``, as with the import.""" + schema = _single_rule_schema( + _rule(_GUARD_PRESENT, {"target": {"equals_string": "ex:a"}}), + {"guard": {"range": "string"}, "target": {"range": "curie"}}, + imports=False, + ) + g = _parse_shacl(schema) + datatypes = { + o + for p in g.objects(EX_SINGLE.Thing, SH.property) + if (p, SH.path, EX_SINGLE.target) in g + for o in g.objects(p, SH.datatype) + } + assert datatypes == {XSD.string} + + +def test_rule_dynamic_enum_value_not_reported(caplog): + """An enum without static permissible values (a dynamic enum) can hold + values the schema does not list, so its equals_string value is not + reported as unknown.""" + + schema = json.loads(_single_rule_schema(_rule(_GUARD_PRESENT, {"target": {"equals_string": "GO:0008150"}}))) + schema["enums"] = {"Process": {"reachable_from": {"source_ontology": "obo:go", "source_nodes": ["GO:0008150"]}}} + schema["classes"]["Thing"]["attributes"]["target"] = {"range": "Process"} + with caplog.at_level(logging.WARNING): + queries = _sparql_queries(_parse_shacl(json.dumps(schema)), EX_SINGLE.Thing) + assert len(queries) == 1 and '"GO:0008150"' in queries[0] + assert not any("permissible value" in rec.message for rec in caplog.records), caplog.text + + +def test_rule_message_follows_default_language(): + """The rule description becomes ``sh:message``, tagged with the default + language like every other human-readable literal the generator emits.""" + rule = _rule(_GUARD_PRESENT, {"target": {"equals_string": "x"}}, description="The target must be x.") + g = _parse_shacl(_single_rule_schema(rule), default_language="en") + (node,) = g.objects(EX_SINGLE.Thing, SH.sparql) + assert set(g.objects(node, SH.message)) == {Literal("The target must be x.", lang="en")} + + +@pytest.mark.parametrize( + "descriptions,constraints", + [ + pytest.param(["Same.", "Same."], 1, id="identical-rules"), + pytest.param(["First.", "Second."], 2, id="different-messages"), + ], +) +def test_rule_identical_constraints_emitted_once(descriptions, constraints): + """A shape carries an identical constraint (query and message) once; rules + that differ in their message are kept apart.""" + schema = json.loads(_single_rule_schema(_rule(_GUARD_PRESENT, {"target": {"equals_string": "x"}}))) + rule = schema["classes"]["Thing"]["rules"][0] + schema["classes"]["Thing"]["rules"] = [{**rule, "description": text} for text in descriptions] + assert len(list(_parse_shacl(json.dumps(schema)).objects(EX_SINGLE.Thing, SH.sparql))) == constraints + + +def test_rule_problem_reported_once_for_inheriting_classes(caplog): + """A problem with a rule is reported once, naming the declaring class and + the rule's position, however many classes inherit it; and a skipped rule + with elseconditions is not also reported as partially emitted.""" + + schema = json.loads( + _single_rule_schema( + _rule(_GUARD_PRESENT, {"target": {"pattern": "^x"}}, elseconditions={"slot_conditions": {}}) + ) + ) + for name in ("C1", "C2", "C3"): + schema["classes"][name] = {"is_a": "Thing", "class_uri": f"ex:{name}"} + with caplog.at_level(logging.WARNING): + _parse_shacl(json.dumps(schema)) + messages = [rec.message for rec in caplog.records if "Rule 1 of class 'Thing'" in rec.message] + assert len(messages) == 1, messages + assert "skipped, because" in messages[0] and "(in the shapes of 'Thing', 'C1', 'C2', 'C3')" in messages[0] + assert not any("elseconditions" in rec.message for rec in caplog.records), caplog.text + + +def test_rule_problem_names_every_affected_shape_whatever_the_class_order(caplog): + """A problem with an inherited rule names every class shape it affects, so + the report does not depend on the order in which classes are declared.""" + schema = json.loads(_single_rule_schema(_rule(_GUARD_PRESENT, {"target": {"pattern": "^x"}}))) + schema["classes"] = {"Sub": {"is_a": "Thing", "class_uri": "ex:Sub"}, **schema["classes"]} + with caplog.at_level(logging.WARNING): + _parse_shacl(json.dumps(schema)) + messages = [rec.message for rec in caplog.records if "Rule 1 of class 'Thing'" in rec.message] + assert len(messages) == 1, messages + assert "'Sub'" in messages[0] and "'Thing'" in messages[0], messages[0] + + +def test_rule_problems_reported_per_rule_and_per_problem(caplog): + """Problems are kept apart by rule and by kind: two rules with the same + problem, and one rule with two problems, each produce their own warning. + Each warning names the rule by its position and description.""" + schema = json.loads(_single_rule_schema(_rule(_GUARD_PRESENT, {"target": {"pattern": "^x"}}))) + schema["enums"] = {"Color": {"permissible_values": {"Red": {}, "Blue": {}}}} + schema["classes"]["Thing"]["attributes"]["color"] = {"range": "Color"} + schema["classes"]["Thing"]["rules"] = [ + _rule(_GUARD_PRESENT, {"target": {"pattern": "^x"}}, description="First"), + _rule(_GUARD_PRESENT, {"target": {"pattern": "^y"}}, description="Second"), + _rule( + _GUARD_PRESENT, + {"color": {"equals_string": "Purple"}}, + description="Third", + elseconditions={"slot_conditions": {"color": {"equals_string": "Red"}}}, + ), + ] + with caplog.at_level(logging.WARNING): + _parse_shacl(json.dumps(schema)) + messages = [rec.message for rec in caplog.records if "of class 'Thing'" in rec.message] + assert len([m for m in messages if m.startswith("Rule 1 of class 'Thing' (First): skipped")]) == 1, messages + assert len([m for m in messages if m.startswith("Rule 2 of class 'Thing' (Second): skipped")]) == 1, messages + third = [m for m in messages if m.startswith("Rule 3 of class 'Thing' (Third)")] + assert len(third) == 2 and any("elseconditions" in m for m in third), messages + assert any("'Purple', which is not a permissible value" in m for m in third), messages + + +def test_rule_first_unknown_slot_is_the_reported_problem(caplog): + """A rule is skipped at its first untranslatable condition: with an + unknown guard and an unknown target, only the guard is reported.""" + schema = _single_rule_schema( + _rule({"no_such_guard": {"value_presence": "PRESENT"}}, {"no_such_target": {"equals_string": "x"}}) + ) + with caplog.at_level(logging.WARNING): + _parse_shacl(schema) + messages = [rec.message for rec in caplog.records if "Rule 1 of class 'Thing'" in rec.message] + assert len(messages) == 1 and "no_such_guard" in messages[0], messages + + +def test_rule_message_follows_rule_language(): + """A rule's own ``in_language`` takes precedence over the default language.""" + rule = _rule(_GUARD_PRESENT, {"target": {"equals_string": "x"}}, description="Das Ziel muss x sein.") + rule["in_language"] = "de" + g = _parse_shacl(_single_rule_schema(rule), default_language="en") + (node,) = g.objects(EX_SINGLE.Thing, SH.sparql) + assert set(g.objects(node, SH.message)) == {Literal("Das Ziel muss x sein.", lang="de")} + + +@pytest.mark.parametrize( + "rule", + [ + pytest.param( + _rule({"no_such_slot": {"value_presence": "PRESENT"}}, {"flag": {"equals_string": "true"}}), + id="boolean-guard-unknown-guard", + ), + pytest.param( + _rule({"no_such_slot": {"value_presence": "PRESENT"}}, {"target": {"equals_string": "x"}}), + id="presence-implies-value-unknown-guard", + ), + pytest.param( + _rule(_GUARD_PRESENT, {"no_such_slot": {"equals_string": "x"}}), + id="presence-implies-value-unknown-target", + ), + pytest.param( + _rule({"no_such_slot": {"equals_string": "x"}}, {"no_such_slot": {"maximum_cardinality": 1}}), + id="exclusive-value-unknown-slot", + ), + ], +) +def test_rule_unknown_slot_key_skipped(caplog, rule): + """A rule whose condition names a slot that does not exist is skipped: + a fabricated predicate would make a constraint that never fires (unknown + guard) or always fires (unknown target).""" + _assert_skipped(caplog, _single_rule_schema(rule), "'no_such_slot', which is not a slot") + + +@pytest.mark.parametrize( + "rule", + [ + pytest.param(_rule({"id": {"value_presence": "PRESENT"}}, {"target": {"equals_string": "x"}}), id="guard"), + pytest.param(_rule(_GUARD_PRESENT, {"id": {"equals_string": "x"}}), id="target"), + ], +) +def test_rule_identifier_slot_skipped(caplog, rule): + """The identifier is the node's IRI, not a property arc, so a condition on + it can never match a triple and the rule is skipped.""" + attributes = {"id": {"identifier": True, "range": "uriorcurie"}, "guard": {}, "target": {}} + _assert_skipped(caplog, _single_rule_schema(rule, attributes), "identifier slot 'id'") + + +_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 + + queries = _sparql_queries(_parse_shacl(_ESCAPING_SCHEMA_YAML), EX_ESC.Item) + assert len(queries) == 1 + query = queries[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}" + + +# =========================================================================== +# Operator exactness +# +# A rule is translated only when its conditions set exactly the operators a +# pattern translates. Each rule below would match a pattern if the converter +# dropped the extra operator, which would widen the precondition (false +# positives) or weaken the postcondition (false negatives). +# =========================================================================== + + +@pytest.mark.parametrize( + "extra", + [ + pytest.param({"pattern": "^x"}, id="pattern"), + pytest.param({"minimum_value": 0}, id="minimum_value"), + pytest.param({"maximum_value": 5}, id="maximum_value"), + pytest.param({"required": True}, id="required"), + pytest.param({"recommended": True}, id="recommended"), + pytest.param({"minimum_cardinality": 1}, id="minimum_cardinality"), + pytest.param({"exact_cardinality": 1}, id="exact_cardinality"), + pytest.param({"equals_number": 3}, id="equals_number"), + pytest.param({"has_member": {"equals_string": "x"}}, id="has_member"), + pytest.param({"any_of": [{"equals_string": "x"}]}, id="slot-any_of"), + ], +) +@pytest.mark.parametrize("condition", ["guard", "target"]) +def test_rule_extra_condition_operator_skipped(caplog, condition, extra): + """A guard or target condition that adds any operator to the ones a pattern + translates makes the rule untranslatable.""" + guard = {"value_presence": "PRESENT", **(extra if condition == "guard" else {})} + target = {"equals_string": "x", **(extra if condition == "target" else {})} + schema = _single_rule_schema(_rule({"guard": guard}, {"target": target})) + _assert_skipped(caplog, schema, "match none of the translated patterns") + + +@pytest.mark.parametrize("presence", ["ABSENT", "UNCOMMITTED"]) +def test_rule_guard_presence_other_than_present_skipped(caplog, presence): + """Only ``value_presence: PRESENT`` is a guard: reading ``ABSENT`` as a + guard would invert the trigger.""" + schema = _single_rule_schema(_rule({"guard": {"value_presence": presence}}, {"target": {"equals_string": "x"}})) + _assert_skipped(caplog, schema, "match none of the translated patterns") + + +@pytest.mark.parametrize("operator", ["any_of", "all_of", "exactly_one_of", "none_of"]) +@pytest.mark.parametrize("side", ["preconditions", "postconditions"]) +def test_rule_expression_level_operator_skipped(caplog, side, operator): + """An expression-level boolean operator on either side cannot be honoured + by any pattern; dropping it would widen or weaken the rule.""" + rule = _rule(_GUARD_PRESENT, {"target": {"equals_string": "x"}}) + rule[side][operator] = [{"slot_conditions": {"other": {"equals_string": "y"}}}] + _assert_skipped(caplog, _single_rule_schema(rule), f"its {side} use {operator}") + + +def test_rule_post_with_both_equals_forms_skipped(caplog): + """equals_string and equals_string_in set together is ambiguous — skip, + do not let one form silently win.""" + schema = _single_rule_schema(_rule(_GUARD_PRESENT, {"target": {"equals_string": "a", "equals_string_in": ["b"]}})) + _assert_skipped(caplog, schema, "match none of the translated patterns") + + +def test_rule_exclusive_value_extra_operator_skipped(caplog): + """The exclusive-value pattern is exact too: a precondition mixing + equals_string with an unsupported operator is skipped.""" + schema = _single_rule_schema( + _rule({"tags": {"equals_string": "x", "pattern": "^x"}}, {"tags": {"maximum_cardinality": 1}}) + ) + _assert_skipped(caplog, schema, "match none of the translated patterns") + + +_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 is compared as the string + "true", not as the boolean of the boolean guard.""" + queries = _sparql_queries(_parse_shacl(_STRING_TRUE_SCHEMA_YAML), URIRef("https://example.org/string-true/Conf")) + assert len(queries) == 1 + assert '?value = "true"' in queries[0], f"string-range 'true' must be a string comparison, got:\n{queries[0]}" + assert "?value = true" not in queries[0], "the boolean guard must not apply to a string slot" + + +@pytest.mark.parametrize( + "status,violates", + [ + pytest.param('"true"', False, id="plain"), + pytest.param('"true"^^xsd:string', False, id="xsd-string"), + pytest.param('"other"', True, id="other"), + ], +) +def test_rule_equals_true_on_string_slot_pyshacl_end_to_end(status, violates): + """status "true" (string) satisfies the rule in either RDF 1.1-identical form.""" + data = ( + "@prefix ex: .\n" + "@prefix xsd: .\n" + f'ex:x a ex:Conf ; ex:opt "x" ; ex:status {status} .' + ) + _, focus_nodes = _validate_rules(_STRING_TRUE_SCHEMA_YAML, data) + assert focus_nodes == ({URIRef("https://example.org/string-true/x")} if violates else set()) diff --git a/tests/linkml/test_validator/test_shacl_validation_plugin.py b/tests/linkml/test_validator/test_shacl_validation_plugin.py index 19d7f5d9be..ba912166b0 100644 --- a/tests/linkml/test_validator/test_shacl_validation_plugin.py +++ b/tests/linkml/test_validator/test_shacl_validation_plugin.py @@ -11,6 +11,9 @@ # not re-exported, so that pyshacl stays an optional dependency. from linkml.validator.plugins.shacl_validation_plugin import ShaclValidationPlugin from linkml.validator.report import Severity +from linkml.validator.validation_context import ValidationContext +from linkml_runtime.linkml_model import SchemaDefinition +from linkml_runtime.loaders import yaml_loader def test_conforming_instance_yields_no_results(validation_context): @@ -87,3 +90,88 @@ def test_generated_shapes_are_cached(validation_context): list(plugin.process({"id": "P:1", "name": "One"}, validation_context)) assert len(plugin._loaded_graphs) == 1 + + +_RULES_SCHEMA_YAML = """ +id: https://example.org/plugin-rules +name: plugin_rules +prefixes: + linkml: https://w3id.org/linkml/ + ex: https://example.org/plugin-rules/ +imports: + - linkml:types +default_prefix: ex +default_range: string +enums: + Color: + permissible_values: + Red: + meaning: ex:Red + Blue: + meaning: ex:Blue + Code: + permissible_values: + alpha: + beta: +classes: + Base: + attributes: + id: + identifier: true + guard: {} + label: {} + color: + range: Color + code: + range: Code + rules: + - preconditions: + slot_conditions: + guard: + value_presence: PRESENT + postconditions: + slot_conditions: + label: + equals_string: x + - preconditions: + slot_conditions: + guard: + value_presence: PRESENT + postconditions: + slot_conditions: + color: + equals_string: Red + - preconditions: + slot_conditions: + guard: + value_presence: PRESENT + postconditions: + slot_conditions: + code: + equals_string_in: [alpha] + Child: + is_a: Base +""" + + +@pytest.mark.parametrize("target_class", ["Base", "Child"]) +@pytest.mark.parametrize( + "instance,violates", + [ + pytest.param({"guard": "g", "label": "x", "color": "Red", "code": "alpha"}, False, id="conforming"), + pytest.param({"guard": "g", "label": "y", "color": "Red", "code": "alpha"}, True, id="string"), + pytest.param({"guard": "g", "label": "x", "color": "Blue", "code": "alpha"}, True, id="enum-meaning"), + pytest.param({"guard": "g", "label": "x", "color": "Red", "code": "beta"}, True, id="enum-literal"), + pytest.param({"guard": "g", "color": "Red", "code": "alpha"}, True, id="target-absent"), + pytest.param({"label": "y", "color": "Blue", "code": "beta"}, False, id="unguarded"), + ], +) +def test_rule_violation_is_reported(target_class, instance, violates): + """Rules translated to ``sh:sparql`` are enforced through the plugin, whose + data graph comes from the LinkML RDF dumper, on instances of the declaring + class and of its subclasses alike.""" + context = ValidationContext(yaml_loader.loads(_RULES_SCHEMA_YAML, SchemaDefinition), target_class) + results = list(ShaclValidationPlugin().process({"id": "ex:x", **instance}, context)) + + assert all(result.severity is Severity.ERROR for result in results) + assert any("SPARQLConstraintComponent" in result.message for result in results) is violates