Compare commits

..
6 Commits
Author SHA1 Message Date
KaifAhmad1andClaude Sonnet 4.6 f163ca24a5 fix(export): fix OWLExporter Turtle invalid syntax and silent data-property omission (#478)
- Add _ttl_block() helper to accumulate all predicate-object pairs before
  writing, producing a single valid Turtle subject block terminated by one
  period — eliminates the bug where rdfs:subClassOf / domain / range were
  appended after a closed '.' block
- Add missing data_properties loop to _export_owl_turtle so
  owl:DatatypeProperty declarations are no longer silently dropped
- Add _escape_ttl_str() to escape quotes, backslashes, newlines, carriage
  returns, and tabs inside Turtle string literals (rdfs:label, rdfs:comment,
  owl:versionInfo)
- Unify optional-field null checks to consistent x = prop.get(); if x: pattern
- Add 43 tests in tests/export/test_owl_exporter.py covering syntax validity,
  data properties, string escaping, null handling, and header output
- Update CHANGELOG.md with [Unreleased] entry

Closes #478

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-17 15:02:46 +05:30
Mohd Kaif a88300d74f Merge pull request #477 from Hawksight-AI/feat/node-distance-semantics-472
feat(explorer): add node distance semantics to PathResponse (#472)
2026-04-16 19:46:42 +05:30
KaifAhmad1 390152c78c feat(explorer): add node distance semantics to PathResponse (#472)
- Extend PathResponse with hop_count (len(path)-1) and distance_band
  ("direct"|"near"|"mid-range"|"distant") as first-class API fields
- Add classify_path_distance() to semantica/utils/helpers.py as the
  single source of truth for hop-count thresholds; both the route and
  the visualizer import from it, eliminating duplicate threshold logic
- Populate hop_count and distance_band in find_path route via
  classify_path_distance(); remove local _classify_distance() copy
- Add highlight_path: list[str] param to KGVisualizer.visualize_network;
  path edges rendered as a distance-aware orange trace (opacity and
  stroke width scale from direct→distant: 1.0/4px to 0.35/1.5px)
- Fix bidirectional edge lookup: only forward pairs (A→B) along the
  path are added to path_edge_set; reverse back-edges in directed
  graphs are no longer incorrectly highlighted
- Add logger.warning when highlight_path contains node IDs absent from
  the layout position map, surfacing silent no-op mismatches
- Extend frontend PathResponse type in GraphInspectorPanel.tsx and
  GraphWorkspaceShell.tsx with hop_count: number and distance_band
  literal union to match the updated API contract
- Add 10 new tests: 2 API-level and 8 unit tests covering all four
  band boundaries (0, 1, 2, 3, 4, 6, 7, 20 hops); 104 explorer tests
  pass, 0 failures introduced
- Update CHANGELOG.md
2026-04-16 17:59:50 +05:30
Mohd Kaif 17602812f9 Merge pull request #476 from Hawksight-AI/feat/bidirectional-path-finding-469
feat(explorer): Bidirectional Path Finding in Knowledge Explorer
2026-04-16 15:29:47 +05:30
KaifAhmad1andClaude Sonnet 4.6 523b02083f feat(explorer): add bidirectional path finding with directed=false param (#469)
- PathFinder.bfs_shortest_path() and dijkstra_shortest_path() gain a
  directed: bool = True parameter. When False, a temporary undirected
  view (graph.to_undirected()) is used for traversal only; the original
  directed edges are preserved and returned in the response.
- _make_undirected_view() helper added to PathFinder; falls back safely
  for non-NetworkX graph types.
- GET /api/graph/node/{id}/path exposes ?directed=false query param.
- PathResponse gains a directed: bool field echoing the mode used.
- Route now returns 404 on empty path (previously returned 200 with
  path: []).
- 21 new tests: 12 unit (TestBidirectionalPathFinding) + 9 API-level
  (TestBidirectionalPathRoute). All 120 tests pass.
- CHANGELOG updated under [Unreleased].

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-16 15:20:12 +05:30
Mohd Kaif 952a4530f5 Merge pull request #474 from Hawksight-AI/kg
feat(kg): Native `KnowledgeGraph` Support in `KGVisualizer`
2026-04-16 12:26:19 +05:30
12 changed files with 1042 additions and 81 deletions
+11 -1
View File
@@ -7,8 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
- **Enhancement: Native `KnowledgeGraph` type support in `KGVisualizer`** (PR `kg` by @KaifAhmad1, closes #471): Added `semantica/kg/knowledge_graph.py` — a formal `KnowledgeGraph` dataclass (`entities`, `relationships`, `metadata`) that is now the canonical in-memory type produced and consumed by the Semantica KG pipeline. Exported from `semantica.kg`. `KGVisualizer` gains `_convert_knowledge_graph()` — an authoritative, non-mutating conversion path from `KnowledgeGraph` to the internal dict format — and `_normalize_graph()` now routes `isinstance(graph, KnowledgeGraph)` through it as an explicit fast-path before duck-typing. All five public entry points (`visualize_network`, `visualize_communities`, `visualize_centrality`, `visualize_entity_types`, `visualize_relationship_matrix`) accept `KnowledgeGraph` directly; no manual conversion required. All existing callers passing dicts or duck-typed objects are unaffected. 15 new tests in `TestFormalKnowledgeGraphType` (conversion shape, non-mutation, determinism, routing, all five entry points, import availability).
- **Fix: `OWLExporter._export_owl_turtle` invalid Turtle syntax and silent data-property omission** (closes #478 by @KaifAhmad1):
- **Bug 1 — Invalid Turtle syntax**: `_export_owl_turtle` unconditionally wrote `rdfs:label` with a closing period (`.`), then appended `rdfs:subClassOf`, `rdfs:domain`, and `rdfs:range` triples after the closed block. Any RDF parser would reject the output. Fixed by introducing `_ttl_block(subject_uri, rdf_type, predicates)` — all predicate-object pairs for a subject are accumulated first, then joined with ` ;\n ` and terminated with a single ` .`, producing valid Turtle in all cases.
- **Bug 2 — Data properties silently dropped**: `_export_owl_turtle` had loops for `classes` and `object_properties` but no loop for `data_properties`, so all `owl:DatatypeProperty` declarations were silently omitted. Added the missing loop, mirroring the existing object-property loop.
- **String escaping**: User-provided strings (`name`, `description`, `comment`, version) were embedded directly into Turtle string literals without escaping. A class named `John"s Class` or a comment containing a backslash or newline produced unparseable output. Added `_escape_ttl_str()` static method (escapes `"`, `\`, `\n`, `\r`, `\t`) applied at every `rdfs:label`, `rdfs:comment`, and `owl:versionInfo` site.
- **Null-check consistency**: All optional field reads now use `x = prop.get("field"); if x:` uniformly — eliminates the mixed pattern of `.get()` guards followed by direct `[]` access.
- 43 tests added in `tests/export/test_owl_exporter.py` across five suites: `TestTurtleSyntaxValidity` (5), `TestDataPropertiesInTurtle` (8), `TestTurtleHeader` (4), `TestTurtleStringEscaping` (16), `TestNullFieldHandling` (7), plus `TestObjectPropertyListDomainRange` (2) and `TestEquivalentClass` (1).
- **Enhancement: Node distance semantics in path responses** (closes #472 by @KaifAhmad1): `PathResponse` now surfaces two new first-class fields — `hop_count: int` (equal to `len(path) - 1`; `0` for self-paths) and `distance_band: str` — so callers no longer need to count hops or implement band classification themselves. Four bands are defined: `"direct"` (01 hops), `"near"` (23), `"mid-range"` (46), `"distant"` (7+). The classification function `classify_path_distance()` lives in `semantica/utils/helpers.py` as the single source of truth; both the Explorer route and the visualizer import from it. `KGVisualizer.visualize_network()` gains an optional `highlight_path: list[str]` parameter: when provided, path edges are rendered as a separate orange trace with opacity and stroke width scaled to the distance band (direct: 1.0 / 4 px → distant: 0.35 / 1.5 px), while non-path edges render at reduced opacity underneath. Edge direction is respected — only the forward pairs `(A, B)` along the path are matched; reverse back-edges in directed graphs are not incorrectly highlighted. A logger warning is emitted when any node ID in `highlight_path` has no layout position, surfacing silent no-op mismatches. Frontend `PathResponse` type in `GraphInspectorPanel.tsx` and `GraphWorkspaceShell.tsx` extended with `hop_count: number` and `distance_band: "direct" | "near" | "mid-range" | "distant"`. All changes are additive; no existing fields removed. 10 new tests: 2 API-level (`test_response_includes_hop_count_and_distance_band`, `test_one_hop_path_is_direct`) and 8 unit tests covering all four band boundaries (0, 1, 2, 3, 4, 6, 7, 20 hops).
- **Enhancement: Bidirectional path finding in Knowledge Explorer** (closes #469 by @KaifAhmad1): Path queries in the Explorer were direction-sensitive — querying B→A when only the edge A→B existed always returned no result, because `PathFinder._get_neighbors()` called `graph.neighbors(node)` which on a `nx.DiGraph` yields only successors. Added a `directed: bool = True` parameter to `bfs_shortest_path()` and `dijkstra_shortest_path()`. When `directed=False` a lightweight undirected view is built via `graph.to_undirected()` for the traversal pass only; the original directed edges are preserved and returned in the response. A `_make_undirected_view()` helper encapsulates the conversion and falls back safely for non-NetworkX graph types. The `/api/graph/node/{id}/path` route exposes the parameter as a query string flag (`?directed=false`); `PathResponse` gains a `directed: bool` field that echoes the mode used. Default is `True`, so all existing callers are unaffected. The route also gained an empty-path 404 guard — previously a traversal that found no path returned `200` with `path: []` instead of `404`. 21 new tests: 12 unit tests in `TestBidirectionalPathFinding` (`tests/kg/test_path_finder.py`) and 9 API-level tests in `TestBidirectionalPathRoute` (`tests/explorer/test_explorer_api.py`).
- **Enhancement: Native `KnowledgeGraph` type support in `KGVisualizer`** (PR `kg` by @KaifAhmad1, closes #471): Added `semantica/kg/knowledge_graph.py` — a formal `KnowledgeGraph` dataclass (`entities`, `relationships`, `metadata`) that is now the canonical in-memory type produced and consumed by the Semantica KG pipeline. Exported from `semantica.kg`. `KGVisualizer` gains `_convert_knowledge_graph()` — an authoritative, non-mutating conversion path from `KnowledgeGraph` to the internal dict format — and `_normalize_graph()` now routes `isinstance(graph, KnowledgeGraph)` through it as an explicit fast-path before duck-typing. All five public entry points (`visualize_network`, `visualize_communities`, `visualize_centrality`, `visualize_entity_types`, `visualize_relationship_matrix`) accept `KnowledgeGraph` directly; no manual conversion required. All existing callers passing dicts or duck-typed objects are unaffected. 15 new tests in `TestFormalKnowledgeGraphType` (conversion shape, non-mutation, determinism, routing, all five entry points, import availability).
- **Fix: `KGVisualizer` now accepts `KnowledgeGraph` objects in all `visualize_*` methods** (PR `visualization` by @KaifAhmad1, closes #458): All five public methods (`visualize_network`, `visualize_communities`, `visualize_centrality`, `visualize_entity_types`, `visualize_relationship_matrix`) previously called `graph.get("entities", [])`, silently producing no output when passed a non-dict object. Added `_normalize_graph()` which duck-types the input — dicts pass through unchanged; any object exposing `.entities` / `.relationships` attributes (e.g. the result of `GraphBuilder.build()`) is converted to the canonical dict form; anything else raises a clear `ProcessingError` naming the offending type. 21 tests added in `tests/visualization/test_kg_visualizer_normalize_graph.py`.
- **Security: 12 vulnerability fixes across CRITICAL → LOW severity** (PR `security-enhancement` by @KaifAhmad1):
@@ -14,6 +14,8 @@ export type PathResponse = {
path: string[];
edge_ids?: string[];
total_weight: number;
hop_count: number;
distance_band: "direct" | "near" | "mid-range" | "distant";
};
export interface GraphInspectorPanelProps {
@@ -33,6 +33,8 @@ type LinkPrediction = {
type PathResponse = {
path: GraphPath;
total_weight: number;
hop_count: number;
distance_band: "direct" | "near" | "mid-range" | "distant";
};
type TemporalBounds = {
+12 -1
View File
@@ -8,6 +8,7 @@ from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from ...utils.helpers import classify_path_distance
from ..dependencies import get_session
from ..schemas import (
EdgeListResponse,
@@ -141,11 +142,14 @@ class _PathAlgorithm(str, Enum):
dijkstra = "dijkstra"
@router.get("/node/{node_id}/path", response_model=PathResponse)
async def find_path(
node_id: str,
target: str = Query(..., description="Target node ID"),
algorithm: _PathAlgorithm = Query(_PathAlgorithm.bfs, description="Algorithm: bfs or dijkstra"),
directed: bool = Query(True, description="If false, treat edges as undirected for traversal"),
session: GraphSession = Depends(get_session),
):
path_finder = session.path_finder
@@ -159,14 +163,18 @@ async def find_path(
else path_finder.bfs_shortest_path
)
try:
result = await asyncio.to_thread(path_fn, graph_dict, node_id, target)
result = await asyncio.to_thread(path_fn, graph_dict, node_id, target, directed=directed)
except Exception as exc:
raise HTTPException(status_code=404, detail=f"No path found from '{node_id}' to '{target}': {exc}")
path_nodes = result.get("path", []) if isinstance(result, dict) else (result or [])
if not path_nodes:
raise HTTPException(status_code=404, detail=f"No path found from '{node_id}' to '{target}'")
total_weight = result.get("total_weight", 0.0) if isinstance(result, dict) else 0.0
edge_ids = await asyncio.to_thread(session.resolve_path_edge_ids, path_nodes)
hop_count = len(path_nodes) - 1 if path_nodes else 0
return PathResponse(
source=node_id,
target=target,
@@ -174,6 +182,9 @@ async def find_path(
path=path_nodes,
edge_ids=edge_ids,
total_weight=total_weight,
directed=directed,
hop_count=hop_count,
distance_band=classify_path_distance(hop_count),
)
+3
View File
@@ -67,6 +67,9 @@ class PathResponse(BaseModel):
path: List[str]
edge_ids: List[str] = Field(default_factory=list)
total_weight: float = 0.0
directed: bool = True
hop_count: int = 0
distance_band: str = "direct"
class GraphStatsResponse(BaseModel):
+63 -40
View File
@@ -328,6 +328,18 @@ class OWLExporter:
lines.append("</rdf:RDF>")
return "\n".join(lines)
@staticmethod
def _escape_ttl_str(value: str) -> str:
"""Escape a string value for safe embedding in a Turtle string literal."""
return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n").replace("\r", "\\r").replace("\t", "\\t")
def _ttl_block(self, subject_uri: str, rdf_type: str, predicates: List[str]) -> str:
"""Build a valid Turtle subject block from accumulated predicate strings."""
stmt = f"<{subject_uri}> a {rdf_type}"
for pred in predicates:
stmt += f" ;\n {pred}"
return stmt + " ."
def _export_owl_turtle(self, ontology: Dict[str, Any], **options) -> str:
"""
Export ontology to OWL Turtle format.
@@ -342,6 +354,7 @@ class OWLExporter:
Returns:
String containing OWL Turtle serialization
"""
esc = self._escape_ttl_str
ontology_uri = ontology.get("uri") or self.ontology_uri
ontology_name = ontology.get("name", "SemanticaOntology")
version = ontology.get("version") or self.version
@@ -357,63 +370,73 @@ class OWLExporter:
lines.append("")
# Ontology declaration
lines.append(f"<{ontology_uri}> a owl:Ontology ;")
lines.append(f' rdfs:label "{ontology_name}" ;')
lines.append(f' owl:versionInfo "{version}" .')
if ontology.get("description"):
lines.append(f' rdfs:comment "{ontology.get("description")}" ;')
onto_predicates = [
f'rdfs:label "{esc(ontology_name)}"',
f'owl:versionInfo "{esc(version)}"',
]
description = ontology.get("description")
if description:
onto_predicates.append(f'rdfs:comment "{esc(description)}"')
lines.append(self._ttl_block(ontology_uri, "owl:Ontology", onto_predicates))
lines.append("")
# Classes
classes = ontology.get("classes", [])
for cls in classes:
for cls in ontology.get("classes", []):
class_uri = cls.get("uri") or cls.get("id", "")
class_name = cls.get("name") or cls.get("label", "")
lines.append(f"<{class_uri}> a owl:Class ;")
lines.append(f' rdfs:label "{class_name}" .')
if cls.get("comment"):
lines.append(f' rdfs:comment "{cls.get("comment")}" ;')
if cls.get("subClassOf"):
parent = cls.get("subClassOf")
lines.append(f" rdfs:subClassOf <{parent}> ;")
# Remove trailing semicolon and add period
if lines[-1].endswith(" ;"):
lines[-1] = lines[-1].rstrip(" ;") + " ."
else:
lines.append(" .")
predicates = [f'rdfs:label "{esc(class_name)}"']
comment = cls.get("comment")
if comment:
predicates.append(f'rdfs:comment "{esc(comment)}"')
sub_class = cls.get("subClassOf")
if sub_class:
predicates.append(f"rdfs:subClassOf <{sub_class}>")
equiv = cls.get("equivalentClass")
if equiv:
predicates.append(f"owl:equivalentClass <{equiv}>")
lines.append(self._ttl_block(class_uri, "owl:Class", predicates))
lines.append("")
# Object properties
object_properties = ontology.get("object_properties", [])
for prop in object_properties:
for prop in ontology.get("object_properties", []):
prop_uri = prop.get("uri") or prop.get("id", "")
prop_name = prop.get("name") or prop.get("label", "")
lines.append(f"<{prop_uri}> a owl:ObjectProperty ;")
lines.append(f' rdfs:label "{prop_name}" .')
if prop.get("domain"):
domain = prop.get("domain")
predicates = [f'rdfs:label "{esc(prop_name)}"']
comment = prop.get("comment")
if comment:
predicates.append(f'rdfs:comment "{esc(comment)}"')
domain = prop.get("domain")
if domain:
if isinstance(domain, list):
for d in domain:
lines.append(f" rdfs:domain <{d}> ;")
predicates.append(f"rdfs:domain <{d}>")
else:
lines.append(f" rdfs:domain <{domain}> ;")
if prop.get("range"):
range_val = prop.get("range")
predicates.append(f"rdfs:domain <{domain}>")
range_val = prop.get("range")
if range_val:
if isinstance(range_val, list):
for r in range_val:
lines.append(f" rdfs:range <{r}> ;")
predicates.append(f"rdfs:range <{r}>")
else:
lines.append(f" rdfs:range <{range_val}> ;")
predicates.append(f"rdfs:range <{range_val}>")
lines.append(self._ttl_block(prop_uri, "owl:ObjectProperty", predicates))
lines.append("")
if lines[-1].endswith(" ;"):
lines[-1] = lines[-1].rstrip(" ;") + " ."
# Data properties
for prop in ontology.get("data_properties", []):
prop_uri = prop.get("uri") or prop.get("id", "")
prop_name = prop.get("name") or prop.get("label", "")
predicates = [f'rdfs:label "{esc(prop_name)}"']
comment = prop.get("comment")
if comment:
predicates.append(f'rdfs:comment "{esc(comment)}"')
domain = prop.get("domain")
if domain:
predicates.append(f"rdfs:domain <{domain}>")
range_type = prop.get("range")
if range_type:
predicates.append(f"rdfs:range xsd:{range_type}")
lines.append(self._ttl_block(prop_uri, "owl:DatatypeProperty", predicates))
lines.append("")
return "\n".join(lines)
+38 -19
View File
@@ -104,7 +104,8 @@ class PathFinder:
source: str,
target: str,
weight_attribute: str = "weight",
default_weight: float = 1.0
default_weight: float = 1.0,
directed: bool = True
) -> List[str]:
"""
Find shortest path using Dijkstra's algorithm.
@@ -125,32 +126,34 @@ class PathFinder:
"""
try:
self.logger.info(f"Finding Dijkstra shortest path from {source} to {target}")
# Validate nodes exist
if not self._node_exists(graph, source):
raise ValueError(f"Source node {source} not found")
if not self._node_exists(graph, target):
raise ValueError(f"Target node {target} not found")
traversal_graph = graph if directed else self._make_undirected_view(graph)
# Dijkstra's algorithm
distances = {source: 0.0}
previous = {}
priority_queue = [(0.0, source)]
visited = set()
while priority_queue:
current_distance, current_node = heapq.heappop(priority_queue)
if current_node in visited:
continue
visited.add(current_node)
if current_node == target:
break
# Explore neighbors
for neighbor, edge_data in self._get_neighbors(graph, current_node):
for neighbor, edge_data in self._get_neighbors(traversal_graph, current_node):
if neighbor in visited:
continue
@@ -350,44 +353,48 @@ class PathFinder:
self,
graph: Any,
source: str,
target: str
target: str,
directed: bool = True
) -> List[str]:
"""
Find shortest path using BFS (unweighted).
Args:
graph: Graph object (NetworkX or similar)
source: Source node ID
target: Target node ID
directed: If False, treat the graph as undirected for traversal
Returns:
List of node IDs representing the shortest path
Raises:
ValueError: If source or target not found
"""
try:
self.logger.info(f"Finding BFS shortest path from {source} to {target}")
# Validate nodes exist
if not self._node_exists(graph, source):
raise ValueError(f"Source node {source} not found")
if not self._node_exists(graph, target):
raise ValueError(f"Target node {target} not found")
traversal_graph = graph if directed else self._make_undirected_view(graph)
# BFS algorithm
queue = deque([(source, [source])])
visited = {source}
while queue:
current, path = queue.popleft()
if current == target:
self.logger.info(f"Found BFS path of length {len(path)}")
return path
# Explore neighbors
for neighbor, _ in self._get_neighbors(graph, current):
for neighbor, _ in self._get_neighbors(traversal_graph, current):
if neighbor not in visited:
visited.add(neighbor)
queue.append((neighbor, path + [neighbor]))
@@ -564,6 +571,18 @@ class PathFinder:
return False
return False
def _make_undirected_view(self, graph: Any) -> Any:
"""Return an undirected view of the graph for bidirectional traversal.
For NetworkX directed graphs this calls ``to_undirected()``, which
preserves all edge attributes. For graph types that have no such
method the original object is returned as a fallback callers that
already expose undirected neighbors will still work correctly.
"""
if hasattr(graph, "to_undirected"):
return graph.to_undirected()
return graph
def _get_neighbors(self, graph: Any, node: str) -> List[Tuple[str, Any]]:
"""Get neighbors of a node with edge data."""
neighbors = []
+22
View File
@@ -562,3 +562,25 @@ def retry_on_error(
return wrapper
return decorator
def classify_path_distance(hop_count: int) -> str:
"""Classify a path hop count into a human-readable distance band.
Bands:
"direct" 01 hops (single edge or self)
"near" 23 hops (closely related)
"mid-range" 46 hops (reachable but separated)
"distant" 7+ hops (weakly coupled)
This is the single source of truth for distance-band thresholds used by
both the Explorer API (PathResponse.distance_band) and the KGVisualizer
(highlight_path edge styling).
"""
if hop_count <= 1:
return "direct"
if hop_count <= 3:
return "near"
if hop_count <= 6:
return "mid-range"
return "distant"
+81 -20
View File
@@ -61,6 +61,7 @@ try:
except Exception: # pragma: no cover
_KnowledgeGraph = None # type: ignore[assignment,misc]
from ..utils.helpers import classify_path_distance
from ..utils.progress_tracker import get_progress_tracker
from .utils.color_schemes import ColorPalette, ColorScheme
from .utils.export_formats import (
@@ -191,6 +192,7 @@ class KGVisualizer:
node_color_by: str = "type",
node_size_by: Optional[str] = None,
hover_data: Optional[List[str]] = None,
highlight_path: Optional[List[str]] = None,
**options,
) -> Optional[Any]:
"""
@@ -212,6 +214,9 @@ class KGVisualizer:
node_color_by: Property to map to node color (default: "type")
node_size_by: Property to map to node size (default: fixed)
hover_data: List of properties to show in hover tooltip
highlight_path: Optional ordered list of node IDs forming a path to
highlight with distance-aware edge styling (opacity and stroke
weight reflect hop count along the path).
**options: Additional visualization options
Returns:
@@ -261,13 +266,14 @@ class KGVisualizer:
tracking_id, message="Generating visualization..."
)
result = self._visualize_network_plotly(
nodes,
edges,
output,
file_path,
nodes,
edges,
output,
file_path,
node_color_by=node_color_by,
node_size_by=node_size_by,
hover_data=hover_data,
highlight_path=highlight_path,
**options
)
@@ -564,6 +570,21 @@ class KGVisualizer:
return edges
@staticmethod
def _path_edge_style(distance_band: str) -> Tuple[float, float]:
"""Return (opacity, width) for a path edge based on its distance band.
Bands come from ``classify_path_distance`` in ``utils.helpers`` the
single source of truth for hop-count thresholds.
"""
if distance_band == "direct":
return (1.0, 4.0)
if distance_band == "near":
return (0.85, 3.0)
if distance_band == "mid-range":
return (0.6, 2.0)
return (0.35, 1.5) # "distant"
def _visualize_network_plotly(
self,
nodes: List[Dict[str, Any]],
@@ -573,6 +594,7 @@ class KGVisualizer:
node_color_by: str = "type",
node_size_by: Optional[str] = None,
hover_data: Optional[List[str]] = None,
highlight_path: Optional[List[str]] = None,
**options,
) -> Optional[Any]:
"""Create Plotly network visualization."""
@@ -675,47 +697,73 @@ class KGVisualizer:
node_text.append(text)
# Prepare edge traces
edge_x = []
edge_y = []
# Build path edge lookup for highlight_path support
path_edge_set: set = set()
path_distance_band = "direct"
if highlight_path and len(highlight_path) >= 2:
path_hop_count = len(highlight_path) - 1
path_distance_band = classify_path_distance(path_hop_count)
# Only add the directed edges that actually form the path (A→B, not B→A).
# Adding the reverse would incorrectly highlight unrelated back-edges.
for i in range(path_hop_count):
path_edge_set.add((highlight_path[i], highlight_path[i + 1]))
# Warn if any path node has no layout position (silent highlight failure).
missing = [n for n in highlight_path if n not in pos]
if missing:
self.logger.warning(
"highlight_path contains node IDs not found in the graph: %s",
missing,
)
path_opacity, path_width = self._path_edge_style(path_distance_band)
# Prepare edge traces — split into background (non-path) and path edges
edge_x: List = []
edge_y: List = []
path_edge_x: List = []
path_edge_y: List = []
# Prepare edge label traces and annotations (for arrows)
edge_label_x = []
edge_label_y = []
edge_label_text = []
annotations = []
# Limit detailed edge rendering for performance if graph is too large
show_detailed_edges = len(edges) < 500
for edge in edges:
source_pos = pos.get(edge["source"])
target_pos = pos.get(edge["target"])
if source_pos and target_pos:
x0, y0 = source_pos
x1, y1 = target_pos
edge_x.extend([x0, x1, None])
edge_y.extend([y0, y1, None])
is_path_edge = (edge["source"], edge["target"]) in path_edge_set
if is_path_edge:
path_edge_x.extend([x0, x1, None])
path_edge_y.extend([y0, y1, None])
else:
edge_x.extend([x0, x1, None])
edge_y.extend([y0, y1, None])
if show_detailed_edges:
# Calculate midpoint for label
mx, my = (x0 + x1) / 2, (y0 + y1) / 2
if edge.get("label"):
edge_label_x.append(mx)
edge_label_y.append(my)
edge_label_text.append(edge["label"])
# Add arrow annotation
# Adjust arrow to point slightly before the node to avoid overlap with node marker
# This is approximate; precise calculation requires node size
annotations.append(
dict(
ax=x0, ay=y0, axref='x', ayref='y',
x=x1, y=y1, xref='x', yref='y',
arrowhead=2, arrowsize=1, arrowwidth=1,
arrowcolor="#888", opacity=0.6,
standoff=15 # Distance from target node
standoff=15
)
)
@@ -728,9 +776,22 @@ class KGVisualizer:
showlegend=False,
opacity=0.5
)
traces = [edge_trace]
# Overlay highlighted path edges with distance-aware styling
if path_edge_x:
path_trace = go.Scatter(
x=path_edge_x,
y=path_edge_y,
line=dict(width=path_width, color="#e05c00"),
hoverinfo="none",
mode="lines",
showlegend=False,
opacity=path_opacity,
)
traces.append(path_trace)
if show_detailed_edges and edge_label_text:
edge_label_trace = go.Scatter(
x=edge_label_x,
+175
View File
@@ -4,6 +4,7 @@ import json
from pathlib import Path
import uuid
import networkx as nx
import pytest
from semantica.context.context_graph import ContextGraph
@@ -638,3 +639,177 @@ class TestGenericGraphFileLoading:
assert repeat.status_code == 200
repeat_ids = [edge["id"] for edge in repeat.json()["edges"]]
assert repeat_ids == ["edge-alpha", "edge-beta"]
# ---------------------------------------------------------------------------
# Bidirectional path-finding tests (issue #469)
# ---------------------------------------------------------------------------
def _make_path_session() -> GraphSession:
"""Return a GraphSession whose build_graph_dict yields an nx.DiGraph with A→B only.
GraphSession wraps a ContextGraph (required by create_app), but we patch
build_graph_dict so PathFinder receives an actual NetworkX DiGraph the
graph type the Explorer is designed to traverse for path queries.
"""
cg = ContextGraph(advanced_analytics=False)
cg.add_node("A", node_type="entity", content="Node A")
cg.add_node("B", node_type="entity", content="Node B")
cg.add_edge("A", "B", edge_type="connects")
session = GraphSession(cg)
# Patch build_graph_dict to return the directed NetworkX graph that
# PathFinder needs. The ContextGraph dict format is not traversable by
# PathFinder; this mimics how a KG-backed session would expose the graph.
digraph = nx.DiGraph()
digraph.add_edge("A", "B")
session.build_graph_dict = lambda node_ids=None: digraph # type: ignore[method-assign]
return session
@pytest.fixture
def path_client():
session = _make_path_session()
app = create_app(session=session)
with TestClient(app) as c:
yield c
class TestBidirectionalPathRoute:
"""API-level tests for directed=true/false on GET /api/graph/node/{id}/path."""
# ------------------------------------------------------------------
# directed=true (default) — existing directed-only behaviour
# ------------------------------------------------------------------
def test_directed_true_forward_path_found(self, path_client):
"""A→B exists: forward query with directed=true must succeed."""
resp = path_client.get("/api/graph/node/A/path?target=B&directed=true")
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["A", "B"]
assert body["directed"] is True
def test_directed_true_reverse_returns_404(self, path_client):
"""Only A→B exists: reverse query with directed=true must return 404."""
resp = path_client.get("/api/graph/node/B/path?target=A&directed=true")
assert resp.status_code == 404
def test_default_param_reverse_returns_404(self, path_client):
"""Omitting directed= must preserve current directed behaviour (404 for reverse)."""
resp = path_client.get("/api/graph/node/B/path?target=A")
assert resp.status_code == 404
# ------------------------------------------------------------------
# directed=false — new undirected traversal
# ------------------------------------------------------------------
def test_directed_false_reverse_path_found(self, path_client):
"""directed=false must find B→A even though only A→B exists."""
resp = path_client.get("/api/graph/node/B/path?target=A&directed=false")
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["B", "A"]
assert body["directed"] is False
def test_directed_false_forward_path_found(self, path_client):
"""directed=false must not break the natural A→B direction."""
resp = path_client.get("/api/graph/node/A/path?target=B&directed=false")
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["A", "B"]
assert body["directed"] is False
# ------------------------------------------------------------------
# Algorithm variants
# ------------------------------------------------------------------
def test_dijkstra_directed_false_reverse(self, path_client):
resp = path_client.get(
"/api/graph/node/B/path?target=A&algorithm=dijkstra&directed=false"
)
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["B", "A"]
assert body["algorithm"] == "dijkstra"
assert body["directed"] is False
def test_dijkstra_directed_true_reverse_returns_404(self, path_client):
resp = path_client.get(
"/api/graph/node/B/path?target=A&algorithm=dijkstra&directed=true"
)
assert resp.status_code == 404
# ------------------------------------------------------------------
# PathResponse schema
# ------------------------------------------------------------------
def test_response_schema_includes_directed_field(self, path_client):
"""PathResponse must always include the directed field."""
resp = path_client.get("/api/graph/node/A/path?target=B")
assert resp.status_code == 200
body = resp.json()
assert "directed" in body
def test_response_directed_reflects_query_param(self, path_client):
resp_true = path_client.get("/api/graph/node/A/path?target=B&directed=true")
resp_false = path_client.get("/api/graph/node/A/path?target=B&directed=false")
assert resp_true.json()["directed"] is True
assert resp_false.json()["directed"] is False
# ------------------------------------------------------------------
# hop_count and distance_band — issue #472
# ------------------------------------------------------------------
def test_response_includes_hop_count_and_distance_band(self, path_client):
"""PathResponse must include hop_count and distance_band fields."""
resp = path_client.get("/api/graph/node/A/path?target=B")
assert resp.status_code == 200
body = resp.json()
assert "hop_count" in body
assert "distance_band" in body
def test_one_hop_path_is_direct(self, path_client):
"""A single-edge path (1 hop) must return distance_band='direct'."""
resp = path_client.get("/api/graph/node/A/path?target=B")
assert resp.status_code == 200
body = resp.json()
assert body["hop_count"] == 1
assert body["distance_band"] == "direct"
# ---------------------------------------------------------------------------
# _classify_distance unit tests — issue #472
# ---------------------------------------------------------------------------
from semantica.utils.helpers import classify_path_distance
class TestClassifyDistance:
"""Unit tests covering all four band boundaries."""
def test_zero_hops_is_direct(self):
assert classify_path_distance(0) == "direct"
def test_one_hop_is_direct(self):
assert classify_path_distance(1) == "direct"
def test_two_hops_is_near(self):
assert classify_path_distance(2) == "near"
def test_three_hops_is_near(self):
assert classify_path_distance(3) == "near"
def test_four_hops_is_mid_range(self):
assert classify_path_distance(4) == "mid-range"
def test_six_hops_is_mid_range(self):
assert classify_path_distance(6) == "mid-range"
def test_seven_hops_is_distant(self):
assert classify_path_distance(7) == "distant"
def test_large_hop_count_is_distant(self):
assert classify_path_distance(20) == "distant"
+549
View File
@@ -0,0 +1,549 @@
"""Tests for OWLExporter._export_owl_turtle fixes (issue #478).
Bug 1: invalid Turtle when subClassOf/domain/range present (predicates appended
after a closing period).
Bug 2: data_properties silently dropped from Turtle output.
"""
import pytest
from semantica.export import OWLExporter
# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def exporter():
return OWLExporter()
@pytest.fixture
def full_ontology():
return {
"uri": "http://example.org/onto",
"name": "TestOntology",
"description": "A test ontology",
"classes": [
{
"uri": "http://example.org/Person",
"name": "Person",
},
{
"uri": "http://example.org/Employee",
"name": "Employee",
"comment": "A person who is employed",
"subClassOf": "http://example.org/Person",
},
{
"uri": "http://example.org/Manager",
"name": "Manager",
"subClassOf": "http://example.org/Employee",
"equivalentClass": "http://example.org/Supervisor",
},
],
"object_properties": [
{
"uri": "http://example.org/worksFor",
"name": "worksFor",
"domain": "http://example.org/Employee",
"range": "http://example.org/Company",
},
{
"uri": "http://example.org/manages",
"name": "manages",
"comment": "manages a team",
"domain": ["http://example.org/Manager"],
"range": ["http://example.org/Employee"],
},
],
"data_properties": [
{
"uri": "http://example.org/hasAge",
"name": "hasAge",
"domain": "http://example.org/Person",
"range": "integer",
},
{
"uri": "http://example.org/hasName",
"name": "hasName",
"comment": "full name",
"domain": "http://example.org/Person",
"range": "string",
},
],
}
# ---------------------------------------------------------------------------
# Bug 1 — valid Turtle syntax
# ---------------------------------------------------------------------------
class TestTurtleSyntaxValidity:
"""Every subject block must have exactly one closing period at the end."""
def _blocks(self, turtle: str) -> list[str]:
"""Split output into non-empty logical blocks (separated by blank lines)."""
return [b.strip() for b in turtle.split("\n\n") if b.strip()]
def test_no_triple_after_period(self, exporter, full_ontology):
"""No predicate line may appear after a line that ends with ' .'."""
turtle = exporter._export_owl_turtle(full_ontology)
lines = turtle.splitlines()
for i, line in enumerate(lines):
stripped = line.rstrip()
if stripped.endswith(" .") and i + 1 < len(lines):
next_line = lines[i + 1].strip()
# next non-blank line must not be a predicate continuation
if next_line:
assert not next_line.startswith("rdfs:"), (
f"Predicate continuation after closing '.' at line {i + 1}: "
f"{lines[i]!r}{lines[i + 1]!r}"
)
def test_each_subject_block_ends_with_period(self, exporter, full_ontology):
"""Every subject block (class / property declaration) ends with exactly one '.'."""
turtle = exporter._export_owl_turtle(full_ontology)
blocks = self._blocks(turtle)
# skip the @prefix lines block and ontology declaration
subject_blocks = [b for b in blocks if b.startswith("<http://")]
for block in subject_blocks:
assert block.endswith("."), f"Block does not end with '.': {block!r}"
# Must not have a bare '.' on an interior line
interior_lines = block.splitlines()[:-1]
for ln in interior_lines:
assert not ln.rstrip().endswith(" ."), (
f"Premature closing period inside block: {ln!r}"
)
def test_class_with_subclassof_is_valid(self, exporter):
ontology = {
"uri": "http://example.org/onto",
"name": "T",
"classes": [
{
"uri": "http://example.org/Employee",
"name": "Employee",
"subClassOf": "http://example.org/Person",
}
],
"object_properties": [],
"data_properties": [],
}
turtle = exporter._export_owl_turtle(ontology)
# Must contain both predicates in the same block
assert 'rdfs:label "Employee"' in turtle
assert "rdfs:subClassOf <http://example.org/Person>" in turtle
# The subClassOf line must NOT come after a closing period
lines = turtle.splitlines()
for i, ln in enumerate(lines):
if "rdfs:subClassOf" in ln:
# Search backwards for the closest period-terminated line
for prev in reversed(lines[:i]):
prev_s = prev.rstrip()
if prev_s:
assert not prev_s.endswith(" ."), (
"rdfs:subClassOf appeared after a closed block"
)
break
def test_object_property_with_domain_range_is_valid(self, exporter):
ontology = {
"uri": "http://example.org/onto",
"name": "T",
"classes": [],
"object_properties": [
{
"uri": "http://example.org/worksFor",
"name": "worksFor",
"domain": "http://example.org/Employee",
"range": "http://example.org/Company",
}
],
"data_properties": [],
}
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:domain <http://example.org/Employee>" in turtle
assert "rdfs:range <http://example.org/Company>" in turtle
lines = turtle.splitlines()
for i, ln in enumerate(lines):
if "rdfs:domain" in ln or "rdfs:range" in ln:
for prev in reversed(lines[:i]):
prev_s = prev.rstrip()
if prev_s:
assert not prev_s.endswith(" ."), (
"domain/range appeared after a closed block"
)
break
def test_class_with_comment_subclassof_both_present(self, exporter):
ontology = {
"uri": "http://example.org/onto",
"name": "T",
"classes": [
{
"uri": "http://example.org/X",
"name": "X",
"comment": "some comment",
"subClassOf": "http://example.org/Y",
}
],
"object_properties": [],
"data_properties": [],
}
turtle = exporter._export_owl_turtle(ontology)
assert 'rdfs:comment "some comment"' in turtle
assert "rdfs:subClassOf <http://example.org/Y>" in turtle
# block must end with single period
block = [b for b in turtle.split("\n\n") if "owl:Class" in b][0].strip()
assert block.endswith(".")
assert block.count("\n.") == 0 # no bare period-only lines
# ---------------------------------------------------------------------------
# Bug 2 — data properties present in Turtle output
# ---------------------------------------------------------------------------
class TestDataPropertiesInTurtle:
def test_data_property_declared_as_datatypeproperty(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert "owl:DatatypeProperty" in turtle
def test_data_property_uri_present(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert "<http://example.org/hasAge>" in turtle
assert "<http://example.org/hasName>" in turtle
def test_data_property_label(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert 'rdfs:label "hasAge"' in turtle
assert 'rdfs:label "hasName"' in turtle
def test_data_property_domain(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert "rdfs:domain <http://example.org/Person>" in turtle
def test_data_property_range_uses_xsd_prefix(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert "rdfs:range xsd:integer" in turtle
assert "rdfs:range xsd:string" in turtle
def test_data_property_comment(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert 'rdfs:comment "full name"' in turtle
def test_data_properties_not_in_turtle_was_bug(self, exporter):
"""Regression: data_properties were silently dropped before the fix."""
ontology = {
"uri": "http://example.org/onto",
"name": "T",
"classes": [],
"object_properties": [],
"data_properties": [
{
"uri": "http://example.org/birthDate",
"name": "birthDate",
"range": "date",
}
],
}
turtle = exporter._export_owl_turtle(ontology)
assert "owl:DatatypeProperty" in turtle, (
"Data properties must appear in Turtle output (was silently dropped)"
)
assert "<http://example.org/birthDate>" in turtle
assert "rdfs:range xsd:date" in turtle
def test_data_property_block_ends_with_period(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
blocks = [b.strip() for b in turtle.split("\n\n") if "owl:DatatypeProperty" in b]
assert blocks, "Expected at least one DatatypeProperty block"
for block in blocks:
assert block.endswith("."), f"DatatypeProperty block missing closing '.': {block!r}"
# ---------------------------------------------------------------------------
# Namespace and ontology header
# ---------------------------------------------------------------------------
class TestTurtleHeader:
def test_prefix_declarations(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert "@prefix rdf:" in turtle
assert "@prefix rdfs:" in turtle
assert "@prefix owl:" in turtle
assert "@prefix xsd:" in turtle
def test_ontology_declaration(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert "a owl:Ontology" in turtle
assert 'rdfs:label "TestOntology"' in turtle
assert 'owl:versionInfo "1.0"' in turtle
def test_ontology_description_included(self, exporter, full_ontology):
turtle = exporter._export_owl_turtle(full_ontology)
assert 'rdfs:comment "A test ontology"' in turtle
def test_ontology_without_description(self, exporter):
ontology = {"uri": "http://example.org/onto", "name": "NoDesc",
"classes": [], "object_properties": [], "data_properties": []}
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:comment" not in turtle
# ---------------------------------------------------------------------------
# Object properties — list domain/range
# ---------------------------------------------------------------------------
class TestObjectPropertyListDomainRange:
def test_list_domain(self, exporter):
ontology = {
"uri": "http://example.org/onto", "name": "T",
"classes": [],
"object_properties": [
{
"uri": "http://example.org/p",
"name": "p",
"domain": ["http://example.org/A", "http://example.org/B"],
}
],
"data_properties": [],
}
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:domain <http://example.org/A>" in turtle
assert "rdfs:domain <http://example.org/B>" in turtle
def test_list_range(self, exporter):
ontology = {
"uri": "http://example.org/onto", "name": "T",
"classes": [],
"object_properties": [
{
"uri": "http://example.org/p",
"name": "p",
"range": ["http://example.org/X", "http://example.org/Y"],
}
],
"data_properties": [],
}
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:range <http://example.org/X>" in turtle
assert "rdfs:range <http://example.org/Y>" in turtle
# ---------------------------------------------------------------------------
# equivalentClass support (also tested under Bug 1 guard)
# ---------------------------------------------------------------------------
class TestEquivalentClass:
def test_equivalent_class_in_turtle(self, exporter):
ontology = {
"uri": "http://example.org/onto", "name": "T",
"classes": [
{
"uri": "http://example.org/Manager",
"name": "Manager",
"equivalentClass": "http://example.org/Supervisor",
}
],
"object_properties": [],
"data_properties": [],
}
turtle = exporter._export_owl_turtle(ontology)
assert "owl:equivalentClass <http://example.org/Supervisor>" in turtle
block = [b for b in turtle.split("\n\n") if "owl:Class" in b][0].strip()
assert block.endswith(".")
# ---------------------------------------------------------------------------
# String escaping in Turtle literals (issue #478 review — escape_001)
# ---------------------------------------------------------------------------
class TestTurtleStringEscaping:
"""User-provided strings must be escaped before embedding in Turtle literals."""
def _onto(self, **kwargs):
base = {"uri": "http://example.org/onto", "name": "T",
"classes": [], "object_properties": [], "data_properties": []}
base.update(kwargs)
return base
def test_escape_ttl_str_double_quote(self, exporter):
assert exporter._escape_ttl_str('say "hello"') == r'say \"hello\"'
def test_escape_ttl_str_backslash(self, exporter):
assert exporter._escape_ttl_str("C:\\path") == "C:\\\\path"
def test_escape_ttl_str_newline(self, exporter):
assert exporter._escape_ttl_str("line1\nline2") == "line1\\nline2"
def test_escape_ttl_str_carriage_return(self, exporter):
assert exporter._escape_ttl_str("a\rb") == "a\\rb"
def test_escape_ttl_str_tab(self, exporter):
assert exporter._escape_ttl_str("col1\tcol2") == "col1\\tcol2"
def test_escape_ttl_str_combined(self, exporter):
raw = 'back\\slash and "quote"\nnewline'
escaped = exporter._escape_ttl_str(raw)
assert '\\"' in escaped
assert "\\\\" in escaped
assert "\\n" in escaped
def test_ontology_name_with_quote_is_escaped(self, exporter):
ontology = self._onto(name='John"s Ontology')
turtle = exporter._export_owl_turtle(ontology)
assert 'rdfs:label "John\\"s Ontology"' in turtle
assert 'rdfs:label "John"s Ontology"' not in turtle
def test_ontology_description_with_quote_is_escaped(self, exporter):
ontology = self._onto(description='Describes "things"')
turtle = exporter._export_owl_turtle(ontology)
assert 'rdfs:comment "Describes \\"things\\""' in turtle
def test_class_name_with_quote_is_escaped(self, exporter):
ontology = self._onto(classes=[{
"uri": "http://example.org/C",
"name": 'My "Special" Class',
}])
turtle = exporter._export_owl_turtle(ontology)
assert r'rdfs:label "My \"Special\" Class"' in turtle
def test_class_comment_with_backslash_is_escaped(self, exporter):
ontology = self._onto(classes=[{
"uri": "http://example.org/C",
"name": "C",
"comment": "path is C:\\Users",
}])
turtle = exporter._export_owl_turtle(ontology)
assert r'rdfs:comment "path is C:\\Users"' in turtle
def test_class_comment_with_newline_is_escaped(self, exporter):
ontology = self._onto(classes=[{
"uri": "http://example.org/C",
"name": "C",
"comment": "line1\nline2",
}])
turtle = exporter._export_owl_turtle(ontology)
assert r'rdfs:comment "line1\nline2"' in turtle
def test_object_property_name_with_quote_is_escaped(self, exporter):
ontology = self._onto(object_properties=[{
"uri": "http://example.org/p",
"name": 'has"Value',
}])
turtle = exporter._export_owl_turtle(ontology)
assert r'rdfs:label "has\"Value"' in turtle
def test_object_property_comment_with_quote_is_escaped(self, exporter):
ontology = self._onto(object_properties=[{
"uri": "http://example.org/p",
"name": "p",
"comment": 'links "A" to "B"',
}])
turtle = exporter._export_owl_turtle(ontology)
assert r'rdfs:comment "links \"A\" to \"B\""' in turtle
def test_data_property_name_with_quote_is_escaped(self, exporter):
ontology = self._onto(data_properties=[{
"uri": "http://example.org/dp",
"name": 'the "name" prop',
"range": "string",
}])
turtle = exporter._export_owl_turtle(ontology)
assert r'rdfs:label "the \"name\" prop"' in turtle
def test_data_property_comment_with_quote_is_escaped(self, exporter):
ontology = self._onto(data_properties=[{
"uri": "http://example.org/dp",
"name": "dp",
"comment": 'see "spec" §3',
"range": "string",
}])
turtle = exporter._export_owl_turtle(ontology)
assert r'rdfs:comment "see \"spec\" §3"' in turtle
def test_plain_strings_unchanged(self, exporter):
"""Strings without special chars must pass through unchanged."""
ontology = self._onto(
name="MyOntology",
classes=[{"uri": "http://example.org/C", "name": "SafeName"}],
)
turtle = exporter._export_owl_turtle(ontology)
assert 'rdfs:label "MyOntology"' in turtle
assert 'rdfs:label "SafeName"' in turtle
# ---------------------------------------------------------------------------
# Null / missing optional fields — no KeyError raised (review null_check_001-3)
# ---------------------------------------------------------------------------
class TestNullFieldHandling:
"""Optional fields absent from dicts must not raise KeyError."""
def _onto(self, **kwargs):
base = {"uri": "http://example.org/onto", "name": "T",
"classes": [], "object_properties": [], "data_properties": []}
base.update(kwargs)
return base
def test_class_no_optional_fields(self, exporter):
ontology = self._onto(classes=[{"uri": "http://example.org/C", "name": "C"}])
turtle = exporter._export_owl_turtle(ontology)
assert "owl:Class" in turtle
def test_object_property_no_domain_no_range(self, exporter):
ontology = self._onto(object_properties=[{
"uri": "http://example.org/p", "name": "p"
}])
turtle = exporter._export_owl_turtle(ontology)
assert "owl:ObjectProperty" in turtle
assert "rdfs:domain" not in turtle
assert "rdfs:range" not in turtle
def test_data_property_no_domain_no_range(self, exporter):
ontology = self._onto(data_properties=[{
"uri": "http://example.org/dp", "name": "dp"
}])
turtle = exporter._export_owl_turtle(ontology)
assert "owl:DatatypeProperty" in turtle
assert "rdfs:domain" not in turtle
assert "rdfs:range" not in turtle
def test_data_property_none_domain(self, exporter):
"""Explicit None value for domain must not raise KeyError."""
ontology = self._onto(data_properties=[{
"uri": "http://example.org/dp", "name": "dp",
"domain": None, "range": "string",
}])
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:domain" not in turtle
def test_data_property_none_range(self, exporter):
"""Explicit None value for range must not raise KeyError."""
ontology = self._onto(data_properties=[{
"uri": "http://example.org/dp", "name": "dp",
"domain": "http://example.org/C", "range": None,
}])
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:range" not in turtle
def test_object_property_none_domain(self, exporter):
ontology = self._onto(object_properties=[{
"uri": "http://example.org/p", "name": "p",
"domain": None, "range": "http://example.org/X",
}])
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:domain" not in turtle
def test_object_property_none_range(self, exporter):
ontology = self._onto(object_properties=[{
"uri": "http://example.org/p", "name": "p",
"domain": "http://example.org/A", "range": None,
}])
turtle = exporter._export_owl_turtle(ontology)
assert "rdfs:range" not in turtle
+84
View File
@@ -821,3 +821,87 @@ class TestPathFinderEdgeCases:
paths = self.finder.all_shortest_paths(single_node_graph, "A")
assert len(paths) == 0 # No paths to other nodes
class TestBidirectionalPathFinding:
"""Tests for the directed=False undirected-traversal mode (issue #469)."""
def setup_method(self):
self.finder = PathFinder()
# Single directed edge A → B. Reverse query B → A has no directed path.
self.digraph = nx.DiGraph()
self.digraph.add_edge("A", "B")
# --- directed=True (default) preserves existing behaviour ---
def test_bfs_directed_true_reverse_returns_empty(self):
"""B→A should find nothing when directed=True (default)."""
path = self.finder.bfs_shortest_path(self.digraph, "B", "A", directed=True)
assert path == []
def test_dijkstra_directed_true_reverse_returns_empty(self):
"""B→A should find nothing when directed=True (default)."""
path = self.finder.dijkstra_shortest_path(self.digraph, "B", "A", directed=True)
assert path == []
def test_bfs_directed_true_default_arg(self):
"""Omitting directed= should behave the same as directed=True."""
path = self.finder.bfs_shortest_path(self.digraph, "B", "A")
assert path == []
def test_dijkstra_directed_true_default_arg(self):
path = self.finder.dijkstra_shortest_path(self.digraph, "B", "A")
assert path == []
# --- directed=False finds path against edge orientation ---
def test_bfs_directed_false_reverse_single_edge(self):
"""directed=False must find B→A even though only A→B exists."""
path = self.finder.bfs_shortest_path(self.digraph, "B", "A", directed=False)
assert path == ["B", "A"]
def test_dijkstra_directed_false_reverse_single_edge(self):
path = self.finder.dijkstra_shortest_path(self.digraph, "B", "A", directed=False)
assert path == ["B", "A"]
def test_bfs_directed_false_forward_still_works(self):
"""directed=False should not break the forward direction."""
path = self.finder.bfs_shortest_path(self.digraph, "A", "B", directed=False)
assert path == ["A", "B"]
def test_dijkstra_directed_false_forward_still_works(self):
path = self.finder.dijkstra_shortest_path(self.digraph, "A", "B", directed=False)
assert path == ["A", "B"]
# --- multi-hop path where one edge is against the query direction ---
def test_bfs_directed_false_multihop(self):
"""A→B, C→B graph: directed=False lets us find A→B→C (i.e. A→C via B)."""
g = nx.DiGraph()
g.add_edge("A", "B")
g.add_edge("C", "B") # oriented towards B, not away from it
# undirected view: A-B-C, so A→C path exists
path = self.finder.bfs_shortest_path(g, "A", "C", directed=False)
assert path[0] == "A" and path[-1] == "C"
assert "B" in path
def test_dijkstra_directed_false_multihop(self):
g = nx.DiGraph()
g.add_edge("A", "B")
g.add_edge("C", "B")
path = self.finder.dijkstra_shortest_path(g, "A", "C", directed=False)
assert path[0] == "A" and path[-1] == "C"
assert "B" in path
# --- PathResponse.directed field ---
def test_path_response_directed_field_exists(self):
"""PathResponse must carry a directed field."""
from semantica.explorer.schemas import PathResponse
r = PathResponse(source="A", target="B", algorithm="bfs", path=["A", "B"], directed=False)
assert r.directed is False
def test_path_response_directed_field_defaults_true(self):
from semantica.explorer.schemas import PathResponse
r = PathResponse(source="A", target="B", algorithm="bfs", path=["A", "B"])
assert r.directed is True