mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-09-13 04:04:09 +00:00
Merge branch 'main' into feat/claude-wrapper
This commit is contained in:
Zohaib Hassnain
@@ -213,6 +213,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
### Fixed
|
||||
|
||||
- **RETE engine matched every fact against every rule — `AlphaNode._matches()` and `BetaNode._can_join()` were placeholder stubs that always returned `True`** (closes #300)
|
||||
- `semantica/reasoning/rete_engine.py` shipped a Rete network whose per-condition alpha test and cross-condition beta join were both `return True` stubs, so `match_patterns()` fired every rule for every fact regardless of predicate, arity, or shared-variable consistency
|
||||
- New module-level `unify_condition()` reuses the regex-based approach from `Reasoner._match_pattern()`: a condition pattern like `Person(?x)` / `Parent(?x, ?y)` is compiled against a fact's `predicate(arg, ...)` string, `?var` becomes a named capture group, and a variable seen twice within one condition (e.g. `Loves(?x, ?x)`) becomes a backreference, so it only unifies when both positions hold the same value. Returns the bindings dict or `None`
|
||||
- Reworked propagation to carry partial-match **tokens** instead of bare facts: a new `Token` dataclass bundles the accumulated `facts` with the consistent `bindings`. `AlphaNode` emits a single-fact token per match; `BetaNode.join()` merges a left token with a right token, concatenating their facts in condition order and returning the merged token only when shared variables agree (conflicting values → `None`, no join). Terminal activations carry the full fact list and accumulated bindings through to the emitted match
|
||||
- This fixes a P1 chained-join defect: rules with three or more conditions (e.g. `Person(?x)`, `Parent(?x, ?y)`, `Located(?y, ?z)`) previously lost bindings and accumulated wrong facts at the third join, and a conflicting third condition could spuriously fire. Beta nodes now keep both `left_tokens` and `right_tokens` memories and join each new token against every token on the opposite side, so deep chains stay binding-consistent and third-level conflicts are correctly suppressed
|
||||
- Fixed an adjacent network-topology bug surfaced by the above: newly created beta nodes were never appended to their input nodes' `children`, so tokens could not propagate; propagation was reworked to support chained joins and to thread bindings end-to-end
|
||||
- Reconciled with the rule-actions/provenance layer (#1096) merged after this fix was opened: `execute_matches()` still dedupes and fires `Rule.actions`/legacy `handler` through a bound `Reasoner` via `_make_activation_key`, now sourced from the Token model's own `bindings` instead of the interim `_bindings_for_rule()` regex re-extraction, which is removed as redundant
|
||||
- New `tests/reasoning/test_rete_engine.py`: `unify_condition` unit cases (single/multi variable, literal args, predicate mismatch, repeated-variable equality), alpha match/reject, beta consistent-join vs conflict-reject, end-to-end rules (single-condition fires only the matching fact; multi-condition join fires only on consistent bindings), and a `TestThreeConditionChain` suite (valid three-condition match, third-level conflict suppression, insertion-order independence, `Match.facts` complete and in condition order, multiple left tokens joining one right fact, parity against `Reasoner._match_rule()`, and `reset()` clearing all token memory)
|
||||
|
||||
- **KG provenance tests asserted on generated ID strings instead of stored records, and `kg_provenance.py` was missed by the `utcnow` sweep** (closes #946) by @pravit-amp
|
||||
- The KG workflow and integration suites checked that a tracker call returned an ID matching a prefix (`assert cent_id.startswith("centrality_")`) without ever reading the record back, so an ID generator that returned a well-formed string and wrote nothing would have passed. Worse, some of those calls named tracker methods that do not exist anywhere in `semantica/` (`track_layer_analysis`, `track_centrality_score`), so the assertions were satisfied with no real interaction behind them
|
||||
- Those tests now read provenance back through `get_provenance()` and assert on algorithm metadata, and call the methods that actually persist records. Verified by mutation rather than by a green run alone: neutering the manager's storage write (`self.storage.store(...)` → no-op) fails 10 tests
|
||||
|
||||
@@ -41,69 +41,134 @@ from ..utils.logging import get_logger
|
||||
from ..utils.progress_tracker import get_progress_tracker
|
||||
from .reasoner import Fact, Rule, _make_activation_key
|
||||
|
||||
logger = get_logger("rete_engine")
|
||||
|
||||
def _extract_bindings(condition: Any, fact: Fact) -> Dict[str, Any]:
|
||||
"""Extract ``?var`` bindings by matching a condition pattern against a fact.
|
||||
|
||||
``condition`` is the pattern stored on the alpha node (typically a string
|
||||
like ``"Person(?x)"``); ``fact`` is the working-memory :class:`Fact`. The
|
||||
fact's canonical string form (``Predicate(arg1, arg2, ...)``) is matched
|
||||
against the pattern using the same ``?\\w+`` placeholder convention as the
|
||||
Reasoner, so downstream actions receive real bindings (e.g. ``{"x": "John"}``)
|
||||
instead of the empty dict that previously left ``?x`` placeholders
|
||||
unsubstituted.
|
||||
def _build_condition_regex(
|
||||
pattern: str,
|
||||
initial_bindings: Optional[Dict[str, str]] = None,
|
||||
) -> str:
|
||||
"""Build an anchored regex string for a condition pattern.
|
||||
|
||||
Returns an empty dict when the condition is not a string pattern or does
|
||||
not match -- callers treat that as "no bindings extracted".
|
||||
Splits the pattern on ``?var`` placeholders, escaping the literal
|
||||
segments so surrounding parentheses/commas match literally. Variables
|
||||
become named groups (or backreferences when repeated); variables already
|
||||
present in ``initial_bindings`` are inlined as their literal value.
|
||||
|
||||
Args:
|
||||
pattern: The condition pattern string (e.g. ``"Person(?x)"``).
|
||||
initial_bindings: Bindings already established upstream. Variables
|
||||
already bound are matched as literals rather than captured.
|
||||
|
||||
Returns:
|
||||
An anchored regex string (``^...$``) suitable for ``re.compile`` /
|
||||
``re.match``.
|
||||
"""
|
||||
if not isinstance(condition, str):
|
||||
return {}
|
||||
|
||||
segments = re.split(r"(\?\w+)", condition)
|
||||
bindings = initial_bindings or {}
|
||||
segments = re.split(r"(\?\w+)", pattern)
|
||||
seen_vars: Set[str] = set()
|
||||
p_regex = ""
|
||||
for seg in segments:
|
||||
if seg.startswith("?"):
|
||||
var_name = seg[1:]
|
||||
if var_name in seen_vars:
|
||||
if var_name in bindings:
|
||||
# Already bound — require the exact literal value.
|
||||
p_regex += re.escape(bindings[var_name])
|
||||
elif var_name in seen_vars:
|
||||
# Same variable used twice — enforce a backreference.
|
||||
p_regex += f"(?P={var_name})"
|
||||
else:
|
||||
p_regex += f"(?P<{var_name}>.+?)"
|
||||
seen_vars.add(var_name)
|
||||
else:
|
||||
p_regex += re.escape(seg)
|
||||
p_regex = f"^{p_regex}$"
|
||||
return f"^{p_regex}$"
|
||||
|
||||
|
||||
def unify_condition(
|
||||
condition: Any,
|
||||
fact: Fact,
|
||||
initial_bindings: Optional[Dict[str, str]] = None,
|
||||
) -> Optional[Dict[str, str]]:
|
||||
"""Unify a condition pattern against a fact.
|
||||
|
||||
A condition is a pattern string such as ``"Person(?x)"`` or
|
||||
``"knows(?x, ?y)"`` where tokens beginning with ``?`` are variables.
|
||||
The fact is rendered via its ``__str__`` representation
|
||||
(``predicate(arg1, arg2)``) and matched against the pattern.
|
||||
|
||||
This mirrors ``Reasoner._match_pattern`` but is self-contained so the
|
||||
RETE engine does not need a live ``Reasoner`` instance.
|
||||
|
||||
Args:
|
||||
condition: The condition pattern (string). Non-string conditions
|
||||
are stringified before matching.
|
||||
fact: The fact to test.
|
||||
initial_bindings: Bindings already established upstream. Variables
|
||||
already bound must match the corresponding literal in the fact.
|
||||
|
||||
Returns:
|
||||
A dict of variable bindings if the fact unifies with the condition,
|
||||
otherwise ``None``.
|
||||
"""
|
||||
bindings = dict(initial_bindings or {})
|
||||
pattern = condition if isinstance(condition, str) else str(condition)
|
||||
fact_str = str(fact)
|
||||
|
||||
# Build the anchored regex once (variables already bound are inlined as
|
||||
# literals). See ``_build_condition_regex`` for the segment handling.
|
||||
p_regex = _build_condition_regex(pattern, bindings)
|
||||
|
||||
try:
|
||||
match = re.match(p_regex, str(fact))
|
||||
except re.error:
|
||||
return {}
|
||||
match = re.match(p_regex, fact_str)
|
||||
except re.error as e:
|
||||
logger.warning(
|
||||
"unify_condition failed to compile/match condition "
|
||||
"%r (regex: %r) against fact %r: %s",
|
||||
pattern,
|
||||
p_regex,
|
||||
fact_str,
|
||||
e,
|
||||
)
|
||||
return None
|
||||
except Exception as e: # noqa: BLE001 - mirror Reasoner._match_pattern
|
||||
logger.warning(
|
||||
"unify_condition unexpected error matching condition "
|
||||
"%r (regex: %r) against fact %r: %s",
|
||||
pattern,
|
||||
p_regex,
|
||||
fact_str,
|
||||
e,
|
||||
)
|
||||
return None
|
||||
if not match:
|
||||
return {}
|
||||
return {k: v for k, v in match.groupdict().items() if v is not None}
|
||||
return None
|
||||
|
||||
|
||||
def _bindings_for_rule(rule: Rule, facts: List[Fact]) -> Dict[str, Any]:
|
||||
"""Merge ``?var`` bindings from matching a rule's conditions against facts.
|
||||
|
||||
Each fact is matched against every condition of the rule; the first
|
||||
condition that yields bindings for a fact contributes them. Bindings from
|
||||
all facts are merged so multi-condition (joined) rules receive the full
|
||||
variable environment. Later conflicting values do not overwrite earlier
|
||||
ones, preserving the binding that a join already validated.
|
||||
"""
|
||||
bindings: Dict[str, Any] = {}
|
||||
for fact in facts:
|
||||
for condition in rule.conditions:
|
||||
extracted = _extract_bindings(condition, fact)
|
||||
if not extracted:
|
||||
continue
|
||||
for key, value in extracted.items():
|
||||
bindings.setdefault(key, value)
|
||||
break
|
||||
for var, value in match.groupdict().items():
|
||||
if var in bindings and bindings[var] != value:
|
||||
return None # Binding conflict.
|
||||
bindings[var] = value
|
||||
return bindings
|
||||
|
||||
|
||||
@dataclass
|
||||
class Token:
|
||||
"""A partial match flowing through the Rete network.
|
||||
|
||||
A token represents an ordered collection of concrete facts that have
|
||||
been unified so far, together with the consistent variable bindings
|
||||
accumulated across those facts.
|
||||
|
||||
Alpha nodes emit single-fact tokens. Beta nodes merge a left token and
|
||||
a right token into a new token whose ``facts`` are the concatenation of
|
||||
both sides (preserving condition order) and whose ``bindings`` are the
|
||||
consistent union of both sides.
|
||||
"""
|
||||
|
||||
facts: List[Fact] = field(default_factory=list)
|
||||
bindings: Dict[str, str] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Match:
|
||||
"""Pattern match."""
|
||||
@@ -128,19 +193,68 @@ class AlphaNode(ReteNode):
|
||||
def __init__(self, node_id: str, condition: Any):
|
||||
super().__init__(node_id)
|
||||
self.condition = condition
|
||||
self.matches: List[Fact] = []
|
||||
# Single-fact tokens produced by unifying each matched fact with
|
||||
# this node's condition.
|
||||
self.tokens: List[Token] = []
|
||||
# Pre-compile the condition regex once. Alpha nodes never have
|
||||
# initial bindings, so the pattern is stable for the node's lifetime
|
||||
# and every incoming fact reuses this compiled matcher instead of
|
||||
# rebuilding it (avoids repeated regex construction overhead).
|
||||
pattern = condition if isinstance(condition, str) else str(condition)
|
||||
self._compiled: Optional[re.Pattern] = None
|
||||
try:
|
||||
self._compiled = re.compile(_build_condition_regex(pattern))
|
||||
except re.error as e:
|
||||
logger.warning(
|
||||
"AlphaNode %r failed to compile condition %r: %s; "
|
||||
"node will never match",
|
||||
node_id,
|
||||
pattern,
|
||||
e,
|
||||
)
|
||||
|
||||
def add_fact(self, fact: Fact) -> bool:
|
||||
"""Add fact if it matches condition."""
|
||||
if self._matches(fact):
|
||||
self.matches.append(fact)
|
||||
return True
|
||||
return False
|
||||
def add_fact(self, fact: Fact) -> Optional[Token]:
|
||||
"""Add fact if it matches the condition, returning its token.
|
||||
|
||||
def _matches(self, fact: Fact) -> bool:
|
||||
"""Check if fact matches condition."""
|
||||
# Simple matching - can be enhanced
|
||||
return True
|
||||
Returns the single-fact ``Token`` produced by unification when the
|
||||
fact matches, otherwise ``None``.
|
||||
"""
|
||||
bindings = self._matches(fact)
|
||||
if bindings is not None:
|
||||
token = Token(facts=[fact], bindings=dict(bindings))
|
||||
self.tokens.append(token)
|
||||
return token
|
||||
return None
|
||||
|
||||
def _matches(self, fact: Fact) -> Optional[Dict[str, str]]:
|
||||
"""Check if fact matches the alpha node condition.
|
||||
|
||||
Uses the pre-compiled regex built in ``__init__`` for performance,
|
||||
since RETE evaluates many facts against every alpha node.
|
||||
|
||||
Returns the variable bindings produced by unification if the fact
|
||||
matches, otherwise ``None``. An empty dict signals a match with no
|
||||
variables (still distinct from ``None``).
|
||||
"""
|
||||
if self._compiled is None:
|
||||
# Compilation failed at build time; treat as non-matching.
|
||||
return None
|
||||
fact_str = str(fact)
|
||||
try:
|
||||
match = self._compiled.match(fact_str)
|
||||
except Exception as e: # noqa: BLE001 - mirror unify_condition
|
||||
logger.warning(
|
||||
"AlphaNode %r unexpected error matching condition "
|
||||
"%r against fact %r: %s",
|
||||
self.node_id,
|
||||
self.condition,
|
||||
fact_str,
|
||||
e,
|
||||
)
|
||||
return None
|
||||
if not match:
|
||||
return None
|
||||
return match.groupdict()
|
||||
|
||||
|
||||
class BetaNode(ReteNode):
|
||||
@@ -150,19 +264,28 @@ class BetaNode(ReteNode):
|
||||
super().__init__(node_id)
|
||||
self.left = left
|
||||
self.right = right
|
||||
self.matches: List[Tuple[Fact, Fact]] = []
|
||||
# Token memories for each side. Incoming tokens are stored here so
|
||||
# that later-arriving tokens on the opposite side can be joined
|
||||
# against every token already seen (chained joins).
|
||||
self.left_tokens: List[Token] = []
|
||||
self.right_tokens: List[Token] = []
|
||||
|
||||
def join(self, left_fact: Fact, right_fact: Fact) -> bool:
|
||||
"""Join facts from left and right nodes."""
|
||||
if self._can_join(left_fact, right_fact):
|
||||
self.matches.append((left_fact, right_fact))
|
||||
return True
|
||||
return False
|
||||
def join(self, left_token: Token, right_token: Token) -> Optional[Token]:
|
||||
"""Join a left token with a right token.
|
||||
|
||||
def _can_join(self, left_fact: Fact, right_fact: Fact) -> bool:
|
||||
"""Check if facts can be joined."""
|
||||
# Simple join logic - can be enhanced
|
||||
return True
|
||||
Returns a new merged ``Token`` (facts concatenated in condition
|
||||
order, bindings unified) when the two tokens are consistent,
|
||||
otherwise ``None`` on a binding conflict.
|
||||
"""
|
||||
merged = dict(left_token.bindings)
|
||||
for var, value in right_token.bindings.items():
|
||||
if var in merged and merged[var] != value:
|
||||
return None # Binding conflict — cannot join.
|
||||
merged[var] = value
|
||||
return Token(
|
||||
facts=list(left_token.facts) + list(right_token.facts),
|
||||
bindings=merged,
|
||||
)
|
||||
|
||||
|
||||
class TerminalNode(ReteNode):
|
||||
@@ -248,12 +371,16 @@ class ReteEngine:
|
||||
self._add_rule_to_network(rule)
|
||||
|
||||
self.logger.info(
|
||||
f"Built Rete network with {len(self.network)} nodes for {len(rules)} rules"
|
||||
f"Built Rete network with {len(self.network)} nodes "
|
||||
f"for {len(rules)} rules"
|
||||
)
|
||||
self.progress_tracker.stop_tracking(
|
||||
tracking_id,
|
||||
status="completed",
|
||||
message=f"Built Rete network with {len(self.network)} nodes for {len(rules)} rules",
|
||||
message=(
|
||||
f"Built Rete network with {len(self.network)} nodes "
|
||||
f"for {len(rules)} rules"
|
||||
),
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
@@ -281,6 +408,10 @@ class ReteEngine:
|
||||
self.node_counter += 1
|
||||
beta_node = BetaNode(node_id, current, alpha_nodes[i])
|
||||
self.network[node_id] = beta_node
|
||||
# Wire the beta node as a child of both its inputs so facts
|
||||
# propagating from either side reach the join.
|
||||
current.children.append(beta_node)
|
||||
alpha_nodes[i].children.append(beta_node)
|
||||
current = beta_node
|
||||
final_node = current
|
||||
else:
|
||||
@@ -311,40 +442,58 @@ class ReteEngine:
|
||||
# Find matching alpha nodes
|
||||
for node_id, node in self.network.items():
|
||||
if isinstance(node, AlphaNode):
|
||||
if node.add_fact(fact):
|
||||
# Propagate to children
|
||||
self._propagate_from_alpha(node, fact)
|
||||
token = node.add_fact(fact)
|
||||
if token is not None:
|
||||
# Propagate the single-fact token to children.
|
||||
self._propagate_token(node, token)
|
||||
|
||||
def _propagate_from_alpha(self, alpha_node: AlphaNode, fact: Fact) -> None:
|
||||
"""Propagate from alpha node to children."""
|
||||
for child in alpha_node.children:
|
||||
def _propagate_token(self, source: ReteNode, token: Token) -> None:
|
||||
"""Propagate ``token`` (arriving from ``source``) to its children.
|
||||
|
||||
A ``Token`` carries the ordered facts and consistent bindings of a
|
||||
partial match. Beta children attempt joins and, on success, emit a
|
||||
new merged token downstream; terminal children turn the token into a
|
||||
rule activation using the token's complete facts and bindings.
|
||||
"""
|
||||
for child in source.children:
|
||||
if isinstance(child, BetaNode):
|
||||
# Join with matches from left side
|
||||
for left_fact in alpha_node.matches:
|
||||
if child.join(left_fact, fact):
|
||||
# Propagate to children
|
||||
for grandchild in child.children:
|
||||
if isinstance(grandchild, TerminalNode):
|
||||
facts = [left_fact, fact]
|
||||
match = Match(
|
||||
rule=grandchild.rule,
|
||||
facts=facts,
|
||||
bindings=_bindings_for_rule(
|
||||
grandchild.rule, facts
|
||||
),
|
||||
confidence=1.0,
|
||||
)
|
||||
grandchild.activate(match)
|
||||
self._propagate_to_beta(child, source, token)
|
||||
elif isinstance(child, TerminalNode):
|
||||
# Direct activation
|
||||
match = Match(
|
||||
rule=child.rule,
|
||||
facts=[fact],
|
||||
bindings=_bindings_for_rule(child.rule, [fact]),
|
||||
facts=list(token.facts),
|
||||
bindings=dict(token.bindings),
|
||||
confidence=1.0,
|
||||
)
|
||||
child.activate(match)
|
||||
|
||||
def _propagate_to_beta(
|
||||
self,
|
||||
beta: "BetaNode",
|
||||
source: ReteNode,
|
||||
token: Token,
|
||||
) -> None:
|
||||
"""Attempt joins at ``beta`` for a token arriving from one side.
|
||||
|
||||
The incoming token is stored in the corresponding side's memory,
|
||||
then joined against every token already recorded on the opposite
|
||||
side. Each successful join produces a new merged token that is
|
||||
propagated further downstream, enabling correct chained joins across
|
||||
three or more conditions.
|
||||
"""
|
||||
if source is beta.left:
|
||||
beta.left_tokens.append(token)
|
||||
for right_token in list(beta.right_tokens):
|
||||
merged = beta.join(token, right_token)
|
||||
if merged is not None:
|
||||
self._propagate_token(beta, merged)
|
||||
elif source is beta.right:
|
||||
beta.right_tokens.append(token)
|
||||
for left_token in list(beta.left_tokens):
|
||||
merged = beta.join(left_token, token)
|
||||
if merged is not None:
|
||||
self._propagate_token(beta, merged)
|
||||
|
||||
def match_patterns(self, facts: Optional[List[Fact]] = None) -> List[Match]:
|
||||
"""
|
||||
Match patterns using Rete algorithm.
|
||||
@@ -468,8 +617,11 @@ class ReteEngine:
|
||||
self.facts.clear()
|
||||
self.reset_action_history()
|
||||
for node in self.network.values():
|
||||
if isinstance(node, AlphaNode) or isinstance(node, BetaNode):
|
||||
node.matches.clear()
|
||||
if isinstance(node, AlphaNode):
|
||||
node.tokens.clear()
|
||||
elif isinstance(node, BetaNode):
|
||||
node.left_tokens.clear()
|
||||
node.right_tokens.clear()
|
||||
elif isinstance(node, TerminalNode):
|
||||
node.activations.clear()
|
||||
|
||||
|
||||
@@ -24,14 +24,19 @@ import math
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from pydantic import ValidationError
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the imports below, which need that extra.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
from semantica.context.context_graph import ContextGraph
|
||||
from semantica.explorer.app import create_app
|
||||
from semantica.explorer.routes.decisions import _node_to_decision
|
||||
from semantica.explorer.schemas import DecisionResponse
|
||||
from semantica.explorer.session import GraphSession
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
from pydantic import ValidationError # noqa: E402
|
||||
|
||||
from semantica.context.context_graph import ContextGraph # noqa: E402
|
||||
from semantica.explorer.app import create_app # noqa: E402
|
||||
from semantica.explorer.routes.decisions import _node_to_decision # noqa: E402
|
||||
from semantica.explorer.schemas import DecisionResponse # noqa: E402
|
||||
from semantica.explorer.session import GraphSession # noqa: E402
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -9,16 +9,15 @@ import networkx as nx
|
||||
import pytest
|
||||
|
||||
from semantica.context.context_graph import ContextGraph
|
||||
from semantica.explorer.app import create_app
|
||||
from semantica.explorer.session import GraphSession
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
try:
|
||||
from starlette.testclient import TestClient
|
||||
except ImportError:
|
||||
pytest.skip(
|
||||
"starlette TestClient is required for explorer tests. Install semantica[explorer].",
|
||||
allow_module_level=True,
|
||||
)
|
||||
from semantica.explorer.app import create_app # noqa: E402
|
||||
from semantica.explorer.session import GraphSession # noqa: E402
|
||||
|
||||
from starlette.testclient import TestClient # noqa: E402
|
||||
|
||||
|
||||
|
||||
@@ -1104,7 +1103,7 @@ class TestBidirectionalPathRoute:
|
||||
# _classify_distance unit tests — issue #472
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
from semantica.utils.helpers import classify_path_distance
|
||||
from semantica.utils.helpers import classify_path_distance # noqa: E402
|
||||
|
||||
|
||||
class _FakeSimilarity:
|
||||
|
||||
@@ -13,16 +13,15 @@ browsers can't set custom headers on a WebSocket handshake.
|
||||
import pytest
|
||||
|
||||
from semantica.context.context_graph import ContextGraph
|
||||
from semantica.explorer.app import create_app
|
||||
from semantica.explorer.session import GraphSession
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
try:
|
||||
from starlette.testclient import TestClient
|
||||
except ImportError:
|
||||
pytest.skip(
|
||||
"starlette TestClient is required for explorer tests. Install semantica[explorer].",
|
||||
allow_module_level=True,
|
||||
)
|
||||
from semantica.explorer.app import create_app # noqa: E402
|
||||
from semantica.explorer.session import GraphSession # noqa: E402
|
||||
|
||||
from starlette.testclient import TestClient # noqa: E402
|
||||
|
||||
|
||||
def _build_sample_graph() -> ContextGraph:
|
||||
|
||||
@@ -15,9 +15,14 @@ from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from semantica.context.context_graph import ContextGraph
|
||||
from semantica.explorer.app import create_app
|
||||
from semantica.explorer.session import GraphSession
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the explorer imports below, which pull fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
from semantica.context.context_graph import ContextGraph # noqa: E402
|
||||
from semantica.explorer.app import create_app # noqa: E402
|
||||
from semantica.explorer.session import GraphSession # noqa: E402
|
||||
|
||||
try:
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
@@ -27,7 +27,12 @@ import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from semantica.explorer.routes import ontology as ontology_mod
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
from semantica.explorer.routes import ontology as ontology_mod # noqa: E402
|
||||
|
||||
|
||||
def _start_local_server():
|
||||
|
||||
@@ -21,7 +21,12 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from semantica.explorer.routes import ontology as ontology_mod
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
from semantica.explorer.routes import ontology as ontology_mod # noqa: E402
|
||||
|
||||
|
||||
def _fake_getaddrinfo(host, *args, **kwargs):
|
||||
|
||||
@@ -6,17 +6,16 @@ from urllib.parse import quote
|
||||
import pytest
|
||||
|
||||
from semantica.context.context_graph import ContextGraph
|
||||
from semantica.explorer.app import create_app
|
||||
from semantica.explorer.routes.ontology import OntologyEntry
|
||||
from semantica.explorer.session import GraphSession
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
try:
|
||||
from starlette.testclient import TestClient
|
||||
except ImportError:
|
||||
pytest.skip(
|
||||
"starlette TestClient is required for explorer tests. Install semantica[explorer].",
|
||||
allow_module_level=True,
|
||||
)
|
||||
from semantica.explorer.app import create_app # noqa: E402
|
||||
from semantica.explorer.routes.ontology import OntologyEntry # noqa: E402
|
||||
from semantica.explorer.session import GraphSession # noqa: E402
|
||||
|
||||
from starlette.testclient import TestClient # noqa: E402
|
||||
|
||||
|
||||
def _build_ontology_graph() -> ContextGraph:
|
||||
|
||||
@@ -5,13 +5,18 @@ Tests for ProvenanceManager wiring into Explorer routes and application startup.
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from starlette.testclient import TestClient
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the starlette/explorer imports below, which need that extra.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
from semantica.context.context_graph import ContextGraph
|
||||
from semantica.explorer.app import create_app
|
||||
from semantica.explorer.session import GraphSession
|
||||
from semantica.provenance import ProvenanceManager
|
||||
from semantica.provenance.storage import SQLiteStorage
|
||||
from starlette.testclient import TestClient # noqa: E402
|
||||
|
||||
from semantica.context.context_graph import ContextGraph # noqa: E402
|
||||
from semantica.explorer.app import create_app # noqa: E402
|
||||
from semantica.explorer.session import GraphSession # noqa: E402
|
||||
from semantica.provenance import ProvenanceManager # noqa: E402
|
||||
from semantica.provenance.storage import SQLiteStorage # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -2,7 +2,14 @@
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
from semantica.explorer.routes.provenance import (
|
||||
import pytest
|
||||
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
from semantica.explorer.routes.provenance import ( # noqa: E402
|
||||
_add_chain_edges,
|
||||
_build_provenance,
|
||||
_render_markdown,
|
||||
|
||||
@@ -14,18 +14,17 @@ from unittest.mock import patch
|
||||
import pytest
|
||||
|
||||
from semantica.context.context_graph import ContextGraph
|
||||
from semantica.explorer.app import create_app
|
||||
from semantica.explorer.session import GraphSession
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
try:
|
||||
from starlette.testclient import TestClient
|
||||
except ImportError:
|
||||
pytest.skip(
|
||||
"starlette TestClient is required for explorer tests. Install semantica[explorer].",
|
||||
allow_module_level=True,
|
||||
)
|
||||
from semantica.explorer.app import create_app # noqa: E402
|
||||
from semantica.explorer.session import GraphSession # noqa: E402
|
||||
|
||||
import semantica.explorer.routes.sparql as sparql_mod
|
||||
from starlette.testclient import TestClient # noqa: E402
|
||||
|
||||
import semantica.explorer.routes.sparql as sparql_mod # noqa: E402
|
||||
|
||||
|
||||
def _build_sample_graph() -> ContextGraph:
|
||||
|
||||
@@ -2,12 +2,19 @@
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
import pytest
|
||||
|
||||
from semantica.explorer.dependencies import get_session
|
||||
from semantica.explorer.routes.vocabulary import router
|
||||
from semantica.utils.skos import validate_skos_hierarchy
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
|
||||
# must skip rather than fail collection when it is absent. The guard has to sit
|
||||
# above the import below, which pulls fastapi in transitively.
|
||||
pytest.importorskip("fastapi")
|
||||
|
||||
from fastapi import FastAPI # noqa: E402
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
|
||||
from semantica.explorer.dependencies import get_session # noqa: E402
|
||||
from semantica.explorer.routes.vocabulary import router # noqa: E402
|
||||
from semantica.utils.skos import validate_skos_hierarchy # noqa: E402
|
||||
|
||||
app = FastAPI()
|
||||
app.include_router(router)
|
||||
|
||||
@@ -0,0 +1,366 @@
|
||||
"""Tests for the RETE engine pattern matching (issue #300).
|
||||
|
||||
These tests verify that ``AlphaNode._matches`` and ``BetaNode._can_join`` no
|
||||
longer behave like the old always-``True`` stubs, and that the network as a
|
||||
whole only fires rules whose conditions actually unify with the facts.
|
||||
"""
|
||||
|
||||
import itertools
|
||||
import re
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from semantica.reasoning import rete_engine
|
||||
from semantica.reasoning.reasoner import Fact, Rule
|
||||
from semantica.reasoning.rete_engine import (
|
||||
AlphaNode,
|
||||
BetaNode,
|
||||
ReteEngine,
|
||||
unify_condition,
|
||||
)
|
||||
|
||||
|
||||
class TestUnifyCondition(unittest.TestCase):
|
||||
def test_single_variable_binds(self):
|
||||
fact = Fact("f1", "Person", ["John"])
|
||||
bindings = unify_condition("Person(?x)", fact)
|
||||
self.assertEqual(bindings, {"x": "John"})
|
||||
|
||||
def test_predicate_mismatch_returns_none(self):
|
||||
fact = Fact("f1", "Company", ["Google"])
|
||||
self.assertIsNone(unify_condition("Person(?x)", fact))
|
||||
|
||||
def test_two_arguments_bind(self):
|
||||
fact = Fact("f2", "Parent", ["John", "Mary"])
|
||||
bindings = unify_condition("Parent(?x, ?y)", fact)
|
||||
self.assertEqual(bindings, {"x": "John", "y": "Mary"})
|
||||
|
||||
def test_literal_argument_must_match(self):
|
||||
fact = Fact("f3", "Parent", ["John", "Mary"])
|
||||
self.assertIsNone(unify_condition("Parent(Bob, ?y)", fact))
|
||||
self.assertEqual(unify_condition("Parent(John, ?y)", fact), {"y": "Mary"})
|
||||
|
||||
def test_repeated_variable_requires_equal_values(self):
|
||||
loves_self = Fact("f4", "Loves", ["John", "John"])
|
||||
loves_other = Fact("f5", "Loves", ["John", "Mary"])
|
||||
self.assertEqual(unify_condition("Loves(?x, ?x)", loves_self), {"x": "John"})
|
||||
self.assertIsNone(unify_condition("Loves(?x, ?x)", loves_other))
|
||||
|
||||
def test_regex_error_logs_warning_and_returns_none(self):
|
||||
"""A regex compilation error is logged with context and yields None."""
|
||||
fact = Fact("f6", "Person", ["John"])
|
||||
with mock.patch.object(
|
||||
rete_engine.re,
|
||||
"match",
|
||||
side_effect=re.error("bad pattern"),
|
||||
), self.assertLogs("semantica.rete_engine", level="WARNING") as captured:
|
||||
result = unify_condition("Person(?x)", fact)
|
||||
self.assertIsNone(result)
|
||||
joined = "\n".join(captured.output)
|
||||
self.assertIn("Person(?x)", joined)
|
||||
self.assertIn("Person(John)", joined)
|
||||
self.assertIn("bad pattern", joined)
|
||||
|
||||
def test_unexpected_error_logs_warning_and_returns_none(self):
|
||||
"""An unexpected error is also logged and swallowed as None."""
|
||||
fact = Fact("f7", "Person", ["John"])
|
||||
with mock.patch.object(
|
||||
rete_engine.re,
|
||||
"match",
|
||||
side_effect=RuntimeError("boom"),
|
||||
), self.assertLogs("semantica.rete_engine", level="WARNING") as captured:
|
||||
result = unify_condition("Person(?x)", fact)
|
||||
self.assertIsNone(result)
|
||||
self.assertIn("boom", "\n".join(captured.output))
|
||||
|
||||
|
||||
class TestAlphaNode(unittest.TestCase):
|
||||
def test_matches_stores_bindings(self):
|
||||
node = AlphaNode("a1", "Person(?x)")
|
||||
fact = Fact("f1", "Person", ["John"])
|
||||
token = node.add_fact(fact)
|
||||
self.assertIsNotNone(token)
|
||||
assert token is not None # narrow type for the checker
|
||||
self.assertEqual(token.facts, [fact])
|
||||
self.assertEqual(token.bindings, {"x": "John"})
|
||||
self.assertIn(token, node.tokens)
|
||||
|
||||
def test_non_matching_fact_rejected(self):
|
||||
node = AlphaNode("a1", "Person(?x)")
|
||||
fact = Fact("f1", "Company", ["Google"])
|
||||
self.assertIsNone(node.add_fact(fact))
|
||||
self.assertEqual(node.tokens, [])
|
||||
|
||||
def test_uses_precompiled_regex(self):
|
||||
"""AlphaNode compiles its condition once and reuses it per fact."""
|
||||
node = AlphaNode("a1", "Person(?x)")
|
||||
self.assertIsNotNone(node._compiled)
|
||||
# Matching goes through the compiled matcher, not unify_condition.
|
||||
with mock.patch.object(rete_engine, "unify_condition") as unify:
|
||||
fact = Fact("f1", "Person", ["John"])
|
||||
token = node.add_fact(fact)
|
||||
unify.assert_not_called()
|
||||
self.assertIsNotNone(token)
|
||||
assert token is not None
|
||||
self.assertEqual(token.bindings, {"x": "John"})
|
||||
|
||||
def test_bad_condition_never_matches_and_logs(self):
|
||||
"""A condition that fails to compile logs a warning and never fires."""
|
||||
with mock.patch.object(
|
||||
rete_engine,
|
||||
"_build_condition_regex",
|
||||
return_value="(unbalanced",
|
||||
), self.assertLogs("semantica.rete_engine", level="WARNING") as captured:
|
||||
node = AlphaNode("bad", "Person(?x)")
|
||||
self.assertIsNone(node._compiled)
|
||||
self.assertIn("failed to compile", "\n".join(captured.output))
|
||||
fact = Fact("f1", "Person", ["John"])
|
||||
self.assertIsNone(node.add_fact(fact))
|
||||
self.assertEqual(node.tokens, [])
|
||||
|
||||
|
||||
class TestBetaNode(unittest.TestCase):
|
||||
def test_join_consistent_bindings(self):
|
||||
left = AlphaNode("a1", "Parent(?x, ?y)")
|
||||
right = AlphaNode("a2", "Person(?x)")
|
||||
beta = BetaNode("b1", left, right)
|
||||
|
||||
parent = Fact("f1", "Parent", ["John", "Mary"])
|
||||
person = Fact("f2", "Person", ["John"])
|
||||
left_token = left.add_fact(parent)
|
||||
right_token = right.add_fact(person)
|
||||
assert left_token is not None and right_token is not None
|
||||
|
||||
merged = beta.join(left_token, right_token)
|
||||
self.assertIsNotNone(merged)
|
||||
assert merged is not None # narrow type for the checker
|
||||
self.assertEqual(merged.bindings, {"x": "John", "y": "Mary"})
|
||||
# Facts are concatenated left-then-right in condition order.
|
||||
self.assertEqual(merged.facts, [parent, person])
|
||||
|
||||
def test_join_conflicting_bindings_rejected(self):
|
||||
left = AlphaNode("a1", "Parent(?x, ?y)")
|
||||
right = AlphaNode("a2", "Person(?x)")
|
||||
beta = BetaNode("b1", left, right)
|
||||
|
||||
parent = Fact("f1", "Parent", ["John", "Mary"])
|
||||
# ?x conflicts: John vs Alice
|
||||
person = Fact("f2", "Person", ["Alice"])
|
||||
left_token = left.add_fact(parent)
|
||||
right_token = right.add_fact(person)
|
||||
assert left_token is not None and right_token is not None
|
||||
|
||||
self.assertIsNone(beta.join(left_token, right_token))
|
||||
|
||||
|
||||
class TestReteEngineEndToEnd(unittest.TestCase):
|
||||
def test_only_matching_rule_fires(self):
|
||||
engine = ReteEngine()
|
||||
rule = Rule(
|
||||
rule_id="r1",
|
||||
name="person rule",
|
||||
conditions=["Person(?x)"],
|
||||
conclusion="Mortal(?x)",
|
||||
)
|
||||
engine.build_network([rule])
|
||||
|
||||
engine.add_fact(Fact("f1", "Person", ["John"]))
|
||||
engine.add_fact(Fact("f2", "Company", ["Google"])) # should NOT fire
|
||||
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(len(matches), 1)
|
||||
self.assertEqual(matches[0].bindings, {"x": "John"})
|
||||
|
||||
def test_multi_condition_join(self):
|
||||
engine = ReteEngine()
|
||||
rule = Rule(
|
||||
rule_id="r1",
|
||||
name="child rule",
|
||||
conditions=["Person(?x)", "Parent(?x, ?y)"],
|
||||
conclusion="Child(?y, ?x)",
|
||||
)
|
||||
engine.build_network([rule])
|
||||
|
||||
engine.add_fact(Fact("f1", "Person", ["John"]))
|
||||
engine.add_fact(Fact("f2", "Parent", ["John", "Mary"]))
|
||||
# Unrelated parent whose ?x does not match any Person -> no activation.
|
||||
engine.add_fact(Fact("f3", "Parent", ["Bob", "Sue"]))
|
||||
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(len(matches), 1)
|
||||
self.assertEqual(matches[0].bindings, {"x": "John", "y": "Mary"})
|
||||
|
||||
def test_no_activation_when_join_inconsistent(self):
|
||||
engine = ReteEngine()
|
||||
rule = Rule(
|
||||
rule_id="r1",
|
||||
name="child rule",
|
||||
conditions=["Person(?x)", "Parent(?x, ?y)"],
|
||||
conclusion="Child(?y, ?x)",
|
||||
)
|
||||
engine.build_network([rule])
|
||||
|
||||
engine.add_fact(Fact("f1", "Person", ["John"]))
|
||||
engine.add_fact(Fact("f2", "Parent", ["Alice", "Mary"])) # ?x mismatch
|
||||
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(matches, [])
|
||||
|
||||
|
||||
class TestThreeConditionChain(unittest.TestCase):
|
||||
"""Chained beta joins across three or more conditions (issue #300).
|
||||
|
||||
These exercise the Token model: a token must accumulate the ordered
|
||||
facts and the consistent bindings of every condition, so that deep
|
||||
chains neither drop bindings nor duplicate facts, and a conflict on the
|
||||
third condition correctly suppresses activation.
|
||||
"""
|
||||
|
||||
def _three_condition_rule(self):
|
||||
return Rule(
|
||||
rule_id="r1",
|
||||
name="location chain",
|
||||
conditions=[
|
||||
"Person(?x)",
|
||||
"Parent(?x, ?y)",
|
||||
"Located(?y, ?z)",
|
||||
],
|
||||
conclusion="LivesNear(?x, ?z)",
|
||||
)
|
||||
|
||||
def test_three_condition_valid_match(self):
|
||||
engine = ReteEngine()
|
||||
engine.build_network([self._three_condition_rule()])
|
||||
|
||||
engine.add_fact(Fact("f1", "Person", ["John"]))
|
||||
engine.add_fact(Fact("f2", "Parent", ["John", "Mary"]))
|
||||
engine.add_fact(Fact("f3", "Located", ["Mary", "Paris"]))
|
||||
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(len(matches), 1)
|
||||
self.assertEqual(
|
||||
matches[0].bindings,
|
||||
{"x": "John", "y": "Mary", "z": "Paris"},
|
||||
)
|
||||
|
||||
def test_three_condition_third_level_conflict(self):
|
||||
engine = ReteEngine()
|
||||
engine.build_network([self._three_condition_rule()])
|
||||
|
||||
engine.add_fact(Fact("f1", "Person", ["John"]))
|
||||
engine.add_fact(Fact("f2", "Parent", ["John", "Mary"]))
|
||||
# ?y is bound to Mary, so a Located fact about Bob must not join.
|
||||
engine.add_fact(Fact("f3", "Located", ["Bob", "Paris"]))
|
||||
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(matches, [])
|
||||
|
||||
def test_fact_insertion_order_independent(self):
|
||||
# Whatever order facts arrive, the same single match must result.
|
||||
base_facts = [
|
||||
Fact("f1", "Person", ["John"]),
|
||||
Fact("f2", "Parent", ["John", "Mary"]),
|
||||
Fact("f3", "Located", ["Mary", "Paris"]),
|
||||
]
|
||||
expected = {"x": "John", "y": "Mary", "z": "Paris"}
|
||||
|
||||
for order in itertools.permutations(base_facts):
|
||||
engine = ReteEngine()
|
||||
engine.build_network([self._three_condition_rule()])
|
||||
for fact in order:
|
||||
engine.add_fact(fact)
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(len(matches), 1, f"order={order}")
|
||||
self.assertEqual(matches[0].bindings, expected)
|
||||
|
||||
def test_match_facts_complete_in_condition_order(self):
|
||||
engine = ReteEngine()
|
||||
engine.build_network([self._three_condition_rule()])
|
||||
|
||||
person = Fact("f1", "Person", ["John"])
|
||||
parent = Fact("f2", "Parent", ["John", "Mary"])
|
||||
located = Fact("f3", "Located", ["Mary", "Paris"])
|
||||
engine.add_fact(person)
|
||||
engine.add_fact(parent)
|
||||
engine.add_fact(located)
|
||||
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(len(matches), 1)
|
||||
# All three facts preserved, in condition order, no duplicates.
|
||||
self.assertEqual(matches[0].facts, [person, parent, located])
|
||||
|
||||
def test_multiple_left_tokens_join_one_right_fact(self):
|
||||
# Two Person/Parent chains sharing the same Located(?y, ?z) fact.
|
||||
engine = ReteEngine()
|
||||
engine.build_network([self._three_condition_rule()])
|
||||
|
||||
engine.add_fact(Fact("f1", "Person", ["John"]))
|
||||
engine.add_fact(Fact("f2", "Parent", ["John", "Mary"]))
|
||||
engine.add_fact(Fact("f3", "Person", ["Alice"]))
|
||||
engine.add_fact(Fact("f4", "Parent", ["Alice", "Mary"]))
|
||||
# One right fact should join with both accumulated left tokens.
|
||||
engine.add_fact(Fact("f5", "Located", ["Mary", "Paris"]))
|
||||
|
||||
matches = engine.match_patterns()
|
||||
self.assertEqual(len(matches), 2)
|
||||
result = {m.bindings["x"]: m.bindings["z"] for m in matches}
|
||||
self.assertEqual(result, {"John": "Paris", "Alice": "Paris"})
|
||||
|
||||
def test_matches_reasoner_match_rule(self):
|
||||
from semantica.reasoning.reasoner import Reasoner
|
||||
|
||||
rule = self._three_condition_rule()
|
||||
facts = [
|
||||
Fact("f1", "Person", ["John"]),
|
||||
Fact("f2", "Parent", ["John", "Mary"]),
|
||||
Fact("f3", "Located", ["Mary", "Paris"]),
|
||||
]
|
||||
|
||||
# Reasoner works over stringified facts and returns
|
||||
# (conclusion, matched_facts, bindings) tuples from self.facts.
|
||||
reasoner = Reasoner()
|
||||
for fact in facts:
|
||||
reasoner.add_fact(str(fact))
|
||||
reasoner_matches = reasoner._match_rule(rule)
|
||||
|
||||
engine = ReteEngine()
|
||||
engine.build_network([rule])
|
||||
for fact in facts:
|
||||
engine.add_fact(fact)
|
||||
rete_matches = engine.match_patterns()
|
||||
|
||||
# Both engines must agree on the number of activations.
|
||||
self.assertEqual(len(rete_matches), len(reasoner_matches))
|
||||
self.assertEqual(len(rete_matches), 1)
|
||||
self.assertEqual(
|
||||
rete_matches[0].bindings,
|
||||
{"x": "John", "y": "Mary", "z": "Paris"},
|
||||
)
|
||||
# The RETE match must carry the instantiated conclusion facts too.
|
||||
conclusion, _, _ = reasoner_matches[0]
|
||||
self.assertEqual(conclusion, "LivesNear(John, Paris)")
|
||||
|
||||
def test_reset_clears_all_token_memory(self):
|
||||
engine = ReteEngine()
|
||||
engine.build_network([self._three_condition_rule()])
|
||||
|
||||
engine.add_fact(Fact("f1", "Person", ["John"]))
|
||||
engine.add_fact(Fact("f2", "Parent", ["John", "Mary"]))
|
||||
engine.add_fact(Fact("f3", "Located", ["Mary", "Paris"]))
|
||||
self.assertEqual(len(engine.match_patterns()), 1)
|
||||
|
||||
engine.reset()
|
||||
|
||||
# No stale facts, tokens or activations remain anywhere.
|
||||
self.assertEqual(engine.facts, [])
|
||||
for node in engine.network.values():
|
||||
if isinstance(node, AlphaNode):
|
||||
self.assertEqual(node.tokens, [])
|
||||
elif isinstance(node, BetaNode):
|
||||
self.assertEqual(node.left_tokens, [])
|
||||
self.assertEqual(node.right_tokens, [])
|
||||
self.assertEqual(engine.match_patterns(), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -24,9 +24,20 @@ import pytest
|
||||
# rdf:/rdfs: namespaces) and neither the code nor this test caught it,
|
||||
# since both had the same bug. Importing the real function makes that class
|
||||
# of drift impossible.
|
||||
from semantica.explorer.routes.sparql import _is_read_only_query
|
||||
# fastapi ships in the optional `explorer` extra, not in `dev`, so this import
|
||||
# fails on a plain dev install. Only the SPARQL class below needs it; the Cypher,
|
||||
# XXE, vector-serialization and SSRF classes in this module are independent, so
|
||||
# the skip is scoped to the one class rather than the whole file.
|
||||
try:
|
||||
from semantica.explorer.routes.sparql import _is_read_only_query
|
||||
except ImportError: # pragma: no cover - depends on the installed extras
|
||||
_is_read_only_query = None
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
_is_read_only_query is None,
|
||||
reason="requires semantica[explorer] (fastapi)",
|
||||
)
|
||||
class TestSparqlReadOnlyValidation:
|
||||
"""Regression tests for SPARQL injection prevention."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user