mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #804 from Sameer6305/fix/772-live-shacl-validation-v2

Browse Source 
fix(ontology): wire live SHACL validation into /shacl/validate and /health (closes #772) #803
This commit is contained in:
Mohd Kaif
2026-07-27 19:05:32 +05:30
committed by GitHub
parent 9eea49a070 35f8c0527c
commit 4f27d3dcae
3 changed files with 890 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+8
View File
@@ -28,6 +28,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- **`POST /shacl/validate` and the `/health` SHACL dimension never ran live SHACL validation** (#772, #804) by @Sameer6305 and @KaifAhmad1
- `/shacl/validate` had no data graph to validate submitted shapes against — only a Turtle syntax check. Added `_data_graph_turtle_for_uri()`, which serializes the loaded ontology's nodes/edges into an RDF/Turtle instance graph (CURIE resolution across owl/rdfs/skos/dct/dc, arbitrary node-property projection, typed individuals) and wires both `/shacl/validate` and the `/health` SHACL dimension to `OntologyEngine.validate_graph()` via pySHACL, returning real `conforms`/violations instead of a hardcoded `status="unavailable"` stub
- Fixed a cross-ontology namespace leak in `_node_belongs_to_ontology`: its prefix fallback (`_extract_namespace()`) split only on the last `/`, so sibling ontologies sharing a domain (e.g. `.../onto-a` and `.../onto-b`) could match entities across ontologies that shouldn't be related; fixed by comparing against the full URI stem via the new `_ontology_namespace()` helper
- Added resource guardrails to `/shacl/validate` to close a DoS risk flagged in review: a submitted-Turtle byte cap (`SEMANTICA_MAX_SHACL_TURTLE_BYTES`, default 256 KB), a parsed-triple cap (`SEMANTICA_MAX_SHACL_TRIPLES`, default 1,000), a validation timeout (`SEMANTICA_MAX_SHACL_TIMEOUT`, default 15s), and a global concurrency semaphore (`SEMANTICA_MAX_SHACL_CONCURRENCY`, default 4)
- Fixed `HealthDimension.status` being set to `"error"` on a real (non-`ImportError`) validation exception, which isn't a valid value on that model — Pydantic construction raised and turned the whole `/health` endpoint into a 422 on any real bug; now reports `status="critical"` (already a valid value) with a regression test forcing this exact path
- Follow-up review fixes: reverted an unrelated regression that had crept into this PR — `POST /api/ontology/create` had gone back to silently swallowing `OntologyEngine.from_data`/`from_text` failures into a near-empty "minimal" ontology instead of raising `HTTPException(500)`, undoing the earlier #770/#787 fix for the same endpoint (and breaking `TestOntologyCreateFailures`, which wasn't run before this PR's initial merge request); `sh:Warning`/`sh:Info`-severity pySHACL results were silently dropped from the `/shacl/validate` response — a shape using non-`Violation` severities could report `conforms=False` with an empty `violations` list and no explanation, so warnings/infos are now folded into the response's `violations` array; and `/health` was independently re-fetching and re-truncation-checking the same ontology's nodes/edges once for the generated SHACL shapes and once for the data graph — both now share a single fetch via `_fetch_analysis_graph()`
- New regression tests: `TestOntologyCreateFailures` (pre-existing, now passing again), `test_shacl_validate_surfaces_warning_severity_results`, `test_health_dedupes_node_edge_fetch`, plus the existing 26-test `tests/explorer/test_ontology_subissue3.py` suite (28/28 passing) and the pre-existing `tests/ontology/` suite (83/83 passing)
- **Neptune cookbook CloudFormation stack exposed the database port to the entire internet and had no network audit trail** ([code scanning alert #28](https://github.com/semantica-agi/semantica/security/code-scanning/28), [#26](https://github.com/semantica-agi/semantica/security/code-scanning/26), [#27](https://github.com/semantica-agi/semantica/security/code-scanning/27), `AC_AWS_0276`/`AC_AWS_0369`/`AC_AWS_0148`) by @KaifAhmad1
- `cookbook/introduction/neptune-setup.yaml`'s security group let anyone on `0.0.0.0/0` reach the Neptune Bolt/OpenCypher port (8182); it now requires a `ClientCidr` parameter (CIDR-validated, no default) so the stack can't be created without the deployer explicitly scoping access to their own IP or VPN/office range
- Added `AWS::EC2::FlowLog` plus a dedicated CloudWatch Logs group and IAM role so all traffic in the stack's VPC is now logged
semantica/explorer/routes/ontology.py
+493 -30
View File
@@ -36,6 +36,12 @@ _MAX_FETCH_BYTES = 20 * 1024 * 1024 # 20 MB
_MAX_ANALYSIS_NODES = 5_000 # cap for health/suggest/shacl node scans to avoid OOM
_MAX_ENTITIES_PER_SIDE = 500 # per-ontology cap for the O(n²) pairwise suggestion loop
class GraphTruncationError(Exception):
"""Raised when a graph exceeds _MAX_ANALYSIS_NODES so analysis would be truncated."""
pass
_CLASS_TYPES = frozenset({
"owl:Class", "rdfs:Class",
"http://www.w3.org/2002/07/owl#Class",
@@ -107,6 +113,43 @@ _INGEST_FORMAT_SUFFIXES: Dict[str, str] = {
}
# ---------------------------------------------------------------------------
# SHACL Validation Resource Guardrails
# ---------------------------------------------------------------------------
_MAX_SHACL_TURTLE_BYTES: int = int(
os.environ.get("SEMANTICA_MAX_SHACL_TURTLE_BYTES", "262144")
) # 256 KB
_MAX_SHACL_TRIPLES: int = int(
os.environ.get("SEMANTICA_MAX_SHACL_TRIPLES", "1000")
) # 1,000 triples
_MAX_SHACL_TIMEOUT_SECONDS: float = float(
os.environ.get("SEMANTICA_MAX_SHACL_TIMEOUT", "15.0")
)
_MAX_SHACL_CONCURRENCY: int = int(
os.environ.get("SEMANTICA_MAX_SHACL_CONCURRENCY", "4")
)
_shacl_validation_semaphore: Optional[
Tuple[asyncio.AbstractEventLoop, asyncio.Semaphore]
] = None
def _get_shacl_semaphore() -> asyncio.Semaphore:
global _shacl_validation_semaphore
try:
current_loop = asyncio.get_running_loop()
except RuntimeError:
current_loop = None
if (
_shacl_validation_semaphore is None
or _shacl_validation_semaphore[0] != current_loop
):
_shacl_validation_semaphore = (
current_loop,
asyncio.Semaphore(_MAX_SHACL_CONCURRENCY),
)
return _shacl_validation_semaphore[1]
# ---------------------------------------------------------------------------
# Schemas
# ---------------------------------------------------------------------------
@@ -579,7 +622,17 @@ def _as_uri_list(value: Any) -> List[str]:
if value is None or value == "":
return []
if isinstance(value, list):
return [str(item) for item in value if item]
result = []
for item in value:
if item is None or item == "":
continue
if isinstance(item, dict):
uri = item.get("uri") or item.get("id") or item.get("@id")
if uri:
result.append(str(uri))
else:
result.append(str(item))
return result
if isinstance(value, dict):
uri = value.get("uri") or value.get("id") or value.get("@id")
return [str(uri)] if uri else []
@@ -670,6 +723,14 @@ def _extract_namespace(uri: str) -> Optional[str]:
return None
def _ontology_namespace(uri: str) -> str:
if "#" in uri:
return uri.rsplit("#", 1)[0] + "#"
if uri.endswith("/"):
return uri
return uri.rstrip("#/") + "#"
def _alignment_id(source_uri: str, relation: str, target_uri: str) -> str:
key = f"{source_uri}|{relation}|{target_uri}"
return str(uuid.uuid5(uuid.NAMESPACE_OID, key))
@@ -697,8 +758,8 @@ def _node_belongs_to_ontology(node: Dict[str, Any], ontology_uri: str) -> bool:
return True
if _node_source_ontology(node) == ontology_uri:
return True
namespace = _extract_namespace(ontology_uri)
return bool(namespace and nid.startswith(namespace))
stem = ontology_uri.rstrip("#/")
return nid.startswith((stem + "#", stem + "/"))
def _is_ontology_entity(node: Dict[str, Any]) -> bool:
@@ -785,6 +846,17 @@ def _ontology_entities(nodes: List[Dict[str, Any]], ontology_uri: Optional[str]
return result
def _data_graph_entities(nodes: List[Dict[str, Any]], ontology_uri: Optional[str] = None) -> List[Dict[str, Any]]:
result = []
for node in nodes:
if _classify_node_type(node.get("type", "")) not in {"class", "property", "individual", "concept", "scheme"}:
continue
if ontology_uri and not _node_belongs_to_ontology(node, ontology_uri):
continue
result.append(node)
return result
def _ontology_dict_from_nodes(uri: str, name: str, nodes: List[Dict[str, Any]], edges: List[Dict[str, Any]]) -> Dict[str, Any]:
classes = []
properties = []
@@ -820,14 +892,14 @@ def _ontology_dict_from_nodes(uri: str, name: str, nodes: List[Dict[str, Any]],
return {
"name": name,
"namespace": _extract_namespace(uri) or uri.rstrip("#/") + "#",
"namespace": _ontology_namespace(uri),
"classes": classes,
"properties": properties,
}
def _basic_shacl_turtle(uri: str, name: str, nodes: List[Dict[str, Any]]) -> str:
namespace = _extract_namespace(uri) or uri.rstrip("#/") + "#"
namespace = _ontology_namespace(uri)
lines = [
"@prefix sh: <http://www.w3.org/ns/shacl#> .",
"@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .",
@@ -1401,8 +1473,7 @@ async def create_ontology(
except Exception as exc:
logger.exception("Failed to generate ontology from sample data; aborting ontology creation.")
logger.warning(f"OntologyEngine.from_data error: {exc}")
raise HTTPException(status_code=500, detail="Ontology generation failed")
raise HTTPException(status_code=500, detail=f"Ontology generation failed: {exc}") from exc
elif body.mode == "text" and body.schema_text:
try:
@@ -1472,8 +1543,7 @@ async def create_ontology(
except Exception as exc:
logger.exception("Failed to generate ontology from schema text; aborting ontology creation.")
logger.warning(f"OntologyEngine.from_text error: {exc}")
raise HTTPException(status_code=500, detail="Ontology generation failed")
raise HTTPException(status_code=500, detail=f"Ontology generation failed: {exc}") from exc
nodes_added = await asyncio.to_thread(session.add_nodes, nodes)
edges_added = await asyncio.to_thread(session.add_edges, edges)
@@ -2095,13 +2165,72 @@ async def ontology_health(
documentation_score = ((with_comment / total) * 80.0) + (20.0 if entry.version or entry.source_url else 0.0)
shacl_dimension = HealthDimension(
key="shacl",
label="SHACL Conformance",
score=0.0,
status="unavailable",
detail="Live SHACL validation is available in SHACL Studio when optional validation dependencies are installed.",
)
try:
health_nodes, health_edges = await _fetch_analysis_graph(session, uri, "shacl-health")
shacl_turtle, _ = await _generated_shacl_for_uri(
request, session, uri, nodes=health_nodes, edges=health_edges
)
data_graph_turtle = await _data_graph_turtle_for_uri(
request, session, uri, nodes=health_nodes, edges=health_edges
)
from ...ontology import OntologyEngine
engine = OntologyEngine()
report = await asyncio.to_thread(
engine.validate_graph,
data_graph_turtle,
shacl=shacl_turtle,
data_graph_format="turtle",
shacl_format="turtle",
)
warning_count = len(report.warnings)
if not report.conforms:
shacl_score = max(0.0, 100.0 - (report.violation_count * 20.0))
elif warning_count:
shacl_score = max(0.0, 100.0 - (warning_count * 5.0))
else:
shacl_score = 100.0
shacl_status = "ok" if report.conforms and not warning_count else "warning"
detail_parts = []
if not report.conforms:
detail_parts.append(f"{report.violation_count} SHACL violation(s)")
if warning_count:
detail_parts.append(f"{warning_count} SHACL warning(s)")
shacl_detail = (
"Graph conforms to all generated SHACL constraints."
if not detail_parts
else f"Graph has {', '.join(detail_parts)}."
)
shacl_dimension = HealthDimension(
key="shacl",
label="SHACL Conformance",
score=round(shacl_score, 1),
status=shacl_status,
detail=shacl_detail,
)
except GraphTruncationError as exc:
shacl_dimension = HealthDimension(
key="shacl",
label="SHACL Conformance",
score=0.0,
status="critical",
detail=str(exc),
)
except ImportError:
shacl_dimension = HealthDimension(
key="shacl",
label="SHACL Conformance",
score=0.0,
status="unavailable",
detail="Live SHACL validation is available in SHACL Studio when optional validation dependencies are installed.",
)
except Exception as exc:
shacl_dimension = HealthDimension(
key="shacl",
label="SHACL Conformance",
score=0.0,
status="critical",
detail=f"Live SHACL validation failed: {exc}",
)
dimensions = [
HealthDimension(
@@ -2147,19 +2276,46 @@ async def ontology_health(
)
async def _fetch_analysis_graph(
session: GraphSession,
uri: str,
log_tag: str,
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
"""Fetch nodes/edges for SHACL analysis, raising GraphTruncationError if oversized.
Shared by `_generated_shacl_for_uri` and `_data_graph_turtle_for_uri` so callers that
need both (e.g. `/health`) can fetch once and pass the results to both via `nodes=`/`edges=`.
"""
nodes, total_nodes = await asyncio.to_thread(session.get_nodes, skip=0, limit=_MAX_ANALYSIS_NODES)
edges, total_edges = await asyncio.to_thread(session.get_edges, skip=0, limit=_MAX_ANALYSIS_NODES)
if total_nodes > _MAX_ANALYSIS_NODES or total_edges > _MAX_ANALYSIS_NODES:
logger.warning(
"%s: graph for %s is too large to analyse (nodes=%d, edges=%d, limit=%d); skipping.",
log_tag, uri, total_nodes, total_edges, _MAX_ANALYSIS_NODES,
)
raise GraphTruncationError(
f"Graph size (nodes={total_nodes}, edges={total_edges}) exceeds maximum analysis limit ({_MAX_ANALYSIS_NODES}). "
"SHACL validation is skipped because the graph is too large to validate fully under current limits."
)
return nodes, edges
async def _generated_shacl_for_uri(
request: Request,
session: GraphSession,
uri: str,
quality_tier: str = "strict",
*,
nodes: Optional[List[Dict[str, Any]]] = None,
edges: Optional[List[Dict[str, Any]]] = None,
) -> tuple[str, List[ShaclShapeSummary]]:
registry = {entry.uri: entry for entry in await _registry_entries(request, session)}
entry = registry.get(uri)
if entry is None:
raise HTTPException(status_code=404, detail="Ontology not found in registry.")
nodes, _ = await asyncio.to_thread(session.get_nodes, skip=0, limit=_MAX_ANALYSIS_NODES)
edges, _ = await asyncio.to_thread(session.get_edges, skip=0, limit=_MAX_ANALYSIS_NODES)
if nodes is None or edges is None:
nodes, edges = await _fetch_analysis_graph(session, uri, "shacl-generate")
entities = _ontology_entities(nodes, uri)
ontology_dict = _ontology_dict_from_nodes(uri, entry.name, entities, edges)
@@ -2180,13 +2336,193 @@ async def _generated_shacl_for_uri(
return shacl_turtle, _summarize_shapes(shacl_turtle)
async def _data_graph_turtle_for_uri(
request: Request,
session: GraphSession,
uri: str,
*,
nodes: Optional[List[Dict[str, Any]]] = None,
edges: Optional[List[Dict[str, Any]]] = None,
) -> str:
try:
import rdflib
except ImportError as exc:
raise ImportError("rdflib is not installed.") from exc
registry = {entry.uri: entry for entry in await _registry_entries(request, session)}
if uri not in registry:
raise HTTPException(status_code=404, detail="Ontology not found in registry.")
if nodes is None or edges is None:
nodes, edges = await _fetch_analysis_graph(session, uri, "shacl-data-graph")
entities = _data_graph_entities(nodes, uri)
g = rdflib.Graph()
base_namespace = _ontology_namespace(uri)
OWL = rdflib.Namespace("http://www.w3.org/2002/07/owl#")
RDF = rdflib.RDF
RDFS = rdflib.RDFS
SKOS = rdflib.Namespace("http://www.w3.org/2004/02/skos/core#")
DCT = rdflib.Namespace("http://purl.org/dc/terms/")
DC = rdflib.Namespace("http://purl.org/dc/elements/1.1/")
SH = rdflib.Namespace("http://www.w3.org/ns/shacl#")
XSD = rdflib.XSD
ONTO = rdflib.Namespace(base_namespace)
g.bind("owl", OWL)
g.bind("rdf", RDF)
g.bind("rdfs", RDFS)
g.bind("skos", SKOS)
g.bind("dct", DCT)
g.bind("dc", DC)
g.bind("sh", SH)
g.bind("xsd", XSD)
g.bind("onto", ONTO)
def _resolve_uri(val: str, base_ns: str) -> rdflib.URIRef:
val_str = str(val).strip()
if val_str == "a":
return rdflib.RDF.type
if val_str.startswith(("http://", "https://", "urn:", "ftp://")):
return rdflib.URIRef(val_str)
prefix_map = {
"owl": "http://www.w3.org/2002/07/owl#",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#",
"skos": "http://www.w3.org/2004/02/skos/core#",
"dct": "http://purl.org/dc/terms/",
"dc": "http://purl.org/dc/elements/1.1/",
"sh": "http://www.w3.org/ns/shacl#",
"xsd": "http://www.w3.org/2001/XMLSchema#",
"onto": base_ns,
}
if ":" in val_str and not val_str.startswith("/"):
prefix, _, rest = val_str.partition(":")
if prefix in prefix_map:
return rdflib.URIRef(prefix_map[prefix] + rest)
return rdflib.URIRef(val_str)
if val_str.startswith("#"):
val_str = val_str.lstrip("#")
if base_ns.endswith(("/", "#")):
return rdflib.URIRef(base_ns + val_str.lstrip("/"))
return rdflib.URIRef(base_ns + "#" + val_str.lstrip("/"))
_shorthand_types = {
"class": "owl:Class",
"object_property": "owl:ObjectProperty",
"datatype_property": "owl:DatatypeProperty",
"annotation_property": "owl:AnnotationProperty",
"property": "rdf:Property",
"individual": "owl:NamedIndividual",
"concept": "skos:Concept",
"scheme": "skos:ConceptScheme",
"ontology": "owl:Ontology",
}
_uri_predicates = {
"rdf:type",
"a",
"rdfs:subClassOf",
"rdfs:domain",
"rdfs:range",
"owl:equivalentClass",
"owl:equivalentProperty",
"owl:sameAs",
"skos:exactMatch",
"skos:closeMatch",
"skos:broadMatch",
"skos:narrowMatch",
"skos:relatedMatch",
"sh:targetClass",
"sh:targetNode",
}
_literal_predicates = {
"rdfs:label",
"rdfs:comment",
"skos:definition",
"skos:prefLabel",
"skos:altLabel",
"dct:description",
"dct:title",
"dc:title",
"dc:description",
"version",
}
_skip_keys = {
"id",
"type",
"content",
"valid_from",
"valid_until",
"scheme_uri",
"ontology_uri",
"ontology",
}
entity_ids = set()
for node in entities:
nid_str = str(node.get("id", "")).strip()
if not nid_str:
continue
entity_ids.add(nid_str)
subj = _resolve_uri(nid_str, base_namespace)
node_type = node.get("type", "")
if node_type:
for t in _as_uri_list(node_type):
t_mapped = _shorthand_types.get(str(t).lower(), str(t))
g.add((subj, rdflib.RDF.type, _resolve_uri(t_mapped, base_namespace)))
content = str(node.get("content", "")).strip()
props = node.get("properties", {}) or {}
if content and "rdfs:label" not in props and "pref_label" not in props and "label" not in props:
g.add((subj, RDFS.label, rdflib.Literal(content)))
for key, val in props.items():
if key in _skip_keys or val is None or val == "":
continue
pred = _resolve_uri(str(key), base_namespace)
for item in _as_uri_list(val) if isinstance(val, (list, dict)) else [val]:
if item is None or item == "":
continue
if isinstance(item, dict):
item = item.get("uri") or item.get("id") or item.get("@id")
if not item:
continue
if str(key) in _uri_predicates or (
str(key) not in _literal_predicates
and str(item).strip().startswith(("http://", "https://", "urn:"))
):
g.add((subj, pred, _resolve_uri(str(item), base_namespace)))
else:
g.add((subj, pred, rdflib.Literal(str(item))))
for edge in edges:
source = str(edge.get("source", edge.get("source_id", ""))).strip()
target = str(edge.get("target", edge.get("target_id", ""))).strip()
pred_str = str(edge.get("type", "related_to")).strip()
if not source or not target:
continue
if source in entity_ids or target in entity_ids:
g.add((
_resolve_uri(source, base_namespace),
_resolve_uri(pred_str, base_namespace),
_resolve_uri(target, base_namespace),
))
return await asyncio.to_thread(g.serialize, format="turtle")
@router.post("/shacl/generate", response_model=ShaclGenerateResponse)
async def generate_shacl(
request: Request,
body: ShaclGenerateRequest,
session: GraphSession = Depends(get_session),
):
shacl_turtle, shapes = await _generated_shacl_for_uri(request, session, body.uri, body.quality_tier)
try:
shacl_turtle, shapes = await _generated_shacl_for_uri(request, session, body.uri, body.quality_tier)
except GraphTruncationError as exc:
raise HTTPException(status_code=413, detail=str(exc)) from exc
return ShaclGenerateResponse(
uri=body.uri,
shacl_turtle=shacl_turtle,
@@ -2201,7 +2537,10 @@ async def list_shacl_shapes(
uri: str = Query(...),
session: GraphSession = Depends(get_session),
):
shacl_turtle, shapes = await _generated_shacl_for_uri(request, session, uri)
try:
shacl_turtle, shapes = await _generated_shacl_for_uri(request, session, uri)
except GraphTruncationError as exc:
raise HTTPException(status_code=413, detail=str(exc)) from exc
return ShaclShapesResponse(
uri=uri,
shapes=shapes,
@@ -2211,15 +2550,43 @@ async def list_shacl_shapes(
@router.post("/shacl/validate", response_model=ShaclValidationResponse)
async def validate_shacl(body: ShaclValidateRequest):
async def validate_shacl(
request: Request,
body: ShaclValidateRequest,
session: GraphSession = Depends(get_session),
):
if not body.shacl_turtle.strip():
raise HTTPException(status_code=422, detail="SHACL Turtle cannot be empty.")
_shacl_bytes = body.shacl_turtle.encode("utf-8")
if len(_shacl_bytes) > _MAX_SHACL_TURTLE_BYTES:
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="error",
message=(
f"SHACL Turtle size ({len(_shacl_bytes)} bytes) "
f"exceeds maximum allowed size ({_MAX_SHACL_TURTLE_BYTES} bytes)."
),
violations=[],
)
# Syntax-check the submitted Turtle with rdflib before claiming anything about it.
try:
import rdflib # type: ignore
g = rdflib.Graph()
await asyncio.to_thread(g.parse, data=body.shacl_turtle, format="turtle")
if len(g) > _MAX_SHACL_TRIPLES:
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="error",
message=(
f"SHACL graph triple count ({len(g)}) "
f"exceeds maximum allowed limit ({_MAX_SHACL_TRIPLES})."
),
violations=[],
)
except ImportError:
pass # rdflib unavailable; skip syntax check
except Exception as exc:
@@ -2228,17 +2595,113 @@ async def validate_shacl(body: ShaclValidateRequest):
detail=f"Invalid Turtle syntax: {exc}",
) from exc
# Live data-graph validation requires pySHACL wired to OntologyEngine.validate_graph().
if not body.uri:
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="unavailable",
message=(
"Turtle parsed successfully. "
"No target ontology URI was provided — "
"specify 'uri' to execute live SHACL validation against an ontology data graph."
),
violations=[],
)
try:
data_graph_turtle = await _data_graph_turtle_for_uri(request, session, body.uri)
from ...ontology import OntologyEngine
engine = OntologyEngine()
semaphore = _get_shacl_semaphore()
async def _run_validation():
async with semaphore:
return await asyncio.to_thread(
engine.validate_graph,
data_graph_turtle,
shacl=body.shacl_turtle,
data_graph_format="turtle",
shacl_format="turtle",
)
report = await asyncio.wait_for(
_run_validation(),
timeout=_MAX_SHACL_TIMEOUT_SECONDS,
)
except (asyncio.TimeoutError, TimeoutError):
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="error",
message=(
f"SHACL validation timed out after {_MAX_SHACL_TIMEOUT_SECONDS} seconds."
),
violations=[],
)
except GraphTruncationError as exc:
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="unavailable",
message=str(exc),
violations=[],
)
except ImportError as exc:
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="unavailable",
message=(
"Turtle parsed successfully. "
f"Live SHACL validation is unavailable: {exc}"
),
violations=[],
)
except HTTPException:
raise
except Exception as exc:
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="error",
message=f"SHACL validation error: {exc}",
violations=[],
)
violations = [
ShaclViolation(
node=str(v.focus_node) if v.focus_node is not None else None,
path=str(v.result_path) if v.result_path is not None else None,
severity=str(v.severity or "Violation"),
message=str(
v.message
or v.explanation
or f"SHACL constraint violation ({v.constraint}) on {v.focus_node}"
),
focus_node=str(v.focus_node) if v.focus_node is not None else None,
source_shape=str(v.shape) if v.shape is not None else None,
)
for v in [*report.violations, *report.warnings, *report.infos]
]
_result_counts = []
if report.violations:
_result_counts.append(f"{len(report.violations)} violation(s)")
if report.warnings:
_result_counts.append(f"{len(report.warnings)} warning(s)")
if report.infos:
_result_counts.append(f"{len(report.infos)} info result(s)")
summary_msg = (
f"SHACL validation found {', '.join(_result_counts)}."
if _result_counts
else "Graph conforms to SHACL shapes."
)
return ShaclValidationResponse(
uri=body.uri,
conforms=False,
status="unavailable",
message=(
"Turtle parsed successfully. "
"Live graph validation is not yet wired to a data graph — "
"install semantica[shacl] and connect OntologyEngine.validate_graph() to enable full validation."
),
violations=[],
conforms=report.conforms,
status="success",
message=summary_msg,
violations=violations,
report_text=report.raw_report,
)
tests/explorer/test_ontology_subissue3.py
+389 -10
View File
@@ -1,5 +1,7 @@
"""Tests for Ontology Hub subissue 3 APIs."""
from unittest.mock import patch
import pytest
from semantica.context.context_graph import ContextGraph
@@ -147,16 +149,29 @@ def test_shacl_validate_returns_unavailable(client):
response = client.post(
"/api/ontology/shacl/validate",
json={
"uri": "http://example.org/onto-a",
"shacl_turtle": "@prefix sh: <http://www.w3.org/ns/shacl#> .",
},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "unavailable", "stub must not report conforms=True before validation is wired"
assert payload["status"] == "unavailable"
assert payload["conforms"] is False
assert isinstance(payload["violations"], list)
with patch.dict("sys.modules", {"pyshacl": None}):
res_no_pyshacl = client.post(
"/api/ontology/shacl/validate",
json={
"uri": "http://example.org/onto-a",
"shacl_turtle": "@prefix sh: <http://www.w3.org/ns/shacl#> .",
},
)
assert res_no_pyshacl.status_code == 200
payload_no_pyshacl = res_no_pyshacl.json()
assert payload_no_pyshacl["status"] == "unavailable"
assert payload_no_pyshacl["conforms"] is False
assert isinstance(payload_no_pyshacl["violations"], list)
def test_shacl_validate_rejects_empty_turtle(client):
response = client.post(
@@ -166,20 +181,135 @@ def test_shacl_validate_rejects_empty_turtle(client):
assert response.status_code == 422
def test_shacl_validate_detects_missing_required_property(client):
graph = client.app.state.session.graph
onto_uri = "http://example.org/onto-a"
person_a = "http://example.org/onto-a#Person"
person_inst = "http://example.org/onto-a#person-no-name"
graph.add_node(
person_inst,
node_type="owl:NamedIndividual",
content="Person Without Name",
scheme_uri=onto_uri,
**{
"rdf:type": person_a,
"rdfs:label": "Person Without Name",
},
)
shacl_turtle = """
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix onto: <http://example.org/onto-a#> .
onto:PersonNameShape a sh:NodeShape ;
sh:targetClass onto:Person ;
sh:property [
sh:path onto:name ;
sh:minCount 1 ;
sh:severity sh:Violation ;
sh:message "Person must have a name." ;
] .
"""
response = client.post(
"/api/ontology/shacl/validate",
json={
"uri": onto_uri,
"shacl_turtle": shacl_turtle,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "success"
assert payload["conforms"] is False
assert len(payload["violations"]) >= 1
violation = payload["violations"][0]
assert violation["severity"] == "Violation"
assert "person-no-name" in str(violation["focus_node"]) or "person-no-name" in str(violation["node"])
def test_shacl_validate_surfaces_warning_severity_results(client):
graph = client.app.state.session.graph
onto_uri = "http://example.org/onto-a"
person_a = "http://example.org/onto-a#Person"
person_inst = "http://example.org/onto-a#person-no-email"
graph.add_node(
person_inst,
node_type="owl:NamedIndividual",
content="Person Without Email",
scheme_uri=onto_uri,
**{
"rdf:type": person_a,
"rdfs:label": "Person Without Email",
},
)
shacl_turtle = """
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix onto: <http://example.org/onto-a#> .
onto:PersonEmailShape a sh:NodeShape ;
sh:targetClass onto:Person ;
sh:property [
sh:path onto:email ;
sh:minCount 1 ;
sh:severity sh:Warning ;
sh:message "Person should have an email." ;
] .
"""
response = client.post(
"/api/ontology/shacl/validate",
json={
"uri": onto_uri,
"shacl_turtle": shacl_turtle,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "success"
# pySHACL flips conforms=False for any result regardless of severity, but
# previously the response body gave no explanation for that: violations
# was always populated from report.violations only, so a report that was
# "non-conforming" purely due to a Warning-severity result rendered as
# conforms=False with an empty violations list. Warnings/infos are now
# folded into the violations array so the response is self-explanatory.
assert payload["conforms"] is False
assert len(payload["violations"]) >= 1
warning = payload["violations"][0]
assert warning["severity"] == "Warning"
assert "warning" in payload["message"].lower()
def test_health_dedupes_node_edge_fetch(client):
import semantica.explorer.routes.ontology as ont_mod
with patch.object(
ont_mod, "_fetch_analysis_graph", wraps=ont_mod._fetch_analysis_graph
) as fetch_spy:
response = client.get("/api/ontology/health?uri=http%3A%2F%2Fexample.org%2Fonto-a")
assert response.status_code == 200
# /health needs both the generated SHACL shapes and the data graph for the
# same uri; both used to independently re-fetch nodes/edges from the
# session. They now share a single fetch.
assert fetch_spy.call_count == 1
def test_health_returns_404_for_unknown_ontology(client):
response = client.get("/api/ontology/health?uri=http%3A%2F%2Fnot-loaded.example%2Fonto")
assert response.status_code == 404
def test_health_shacl_dimension_is_zero_when_unavailable(client):
payload = client.get("/api/ontology/health?uri=http%3A%2F%2Fexample.org%2Fonto-a").json()
shacl_dim = next(d for d in payload["dimensions"] if d["key"] == "shacl")
assert shacl_dim["status"] == "unavailable"
assert shacl_dim["score"] == 0.0
# Total score must NOT include the unavailable dimension in its average.
scoreable = [d for d in payload["dimensions"] if d["status"] != "unavailable"]
expected_total = round(sum(d["score"] for d in scoreable) / len(scoreable), 1)
assert payload["total_score"] == expected_total
with patch.dict("sys.modules", {"pyshacl": None}):
payload = client.get("/api/ontology/health?uri=http%3A%2F%2Fexample.org%2Fonto-a").json()
shacl_dim = next(d for d in payload["dimensions"] if d["key"] == "shacl")
assert shacl_dim["status"] == "unavailable"
assert shacl_dim["score"] == 0.0
# Total score must NOT include the unavailable dimension in its average.
scoreable = [d for d in payload["dimensions"] if d["status"] != "unavailable"]
expected_total = round(sum(d["score"] for d in scoreable) / len(scoreable), 1)
assert payload["total_score"] == expected_total
def test_delete_unknown_alignment_returns_404(client):
@@ -261,3 +391,252 @@ def test_health_alignment_coverage_uses_set_lookup(client):
payload = client.get("/api/ontology/health?uri=http%3A%2F%2Fexample.org%2Fonto-a").json()
alignment_dim = next(d for d in payload["dimensions"] if d["key"] == "alignment")
assert alignment_dim["score"] > 0.0, "alignment coverage must be non-zero after recording an alignment"
def test_suggest_alignments_unaffected_by_individuals(client):
graph = client.app.state.session.graph
graph.add_node(
"http://example.org/onto-a#individual-person",
node_type="owl:NamedIndividual",
content="Person",
scheme_uri="http://example.org/onto-a",
**{"rdf:type": "http://example.org/onto-a#Person", "rdfs:label": "Person"},
)
response = client.post(
"/api/ontology/suggest-alignments",
json={
"source_ontology_uri": "http://example.org/onto-a",
"target_ontology_uri": "http://example.org/onto-b",
"threshold": 0.35,
"limit": 5,
},
)
assert response.status_code == 200
suggestions = response.json()
assert suggestions
for s in suggestions:
assert "individual-person" not in s["source_uri"]
assert "individual-person" not in s["target_uri"]
def test_health_shacl_dimension_degrades_gracefully_on_real_error(client):
import semantica.explorer.routes.ontology as ont_mod
with patch.object(ont_mod, "_data_graph_turtle_for_uri", side_effect=RuntimeError("boom")):
response = client.get("/api/ontology/health?uri=http%3A%2F%2Fexample.org%2Fonto-a")
assert response.status_code == 200
payload = response.json()
shacl_dim = next(d for d in payload["dimensions"] if d["key"] == "shacl")
assert shacl_dim["status"] == "critical"
assert "boom" in shacl_dim["detail"]
@pytest.mark.asyncio
async def test_data_graph_turtle_resolves_onto_prefix_and_unknown_curies(client):
from unittest.mock import MagicMock
from semantica.explorer.routes.ontology import _data_graph_turtle_for_uri
graph = client.app.state.session.graph
onto_uri = "http://example.org/onto-a"
person_a = "http://example.org/onto-a#Person"
person_inst = "http://example.org/onto-a#person-with-name"
graph.add_node(
person_inst,
node_type="owl:NamedIndividual",
content="Person With Name",
scheme_uri=onto_uri,
**{
"rdf:type": person_a,
"onto:name": "Alice",
"custom:prop": "http://custom.example/val",
},
)
ttl = await _data_graph_turtle_for_uri(MagicMock(), client.app.state.session, onto_uri)
assert "http://example.org/onto-a#name" in ttl or "onto:name" in ttl
assert "http://example.org/#onto:name" not in ttl
assert "http://example.org/onto-a#custom:prop" not in ttl
def test_ontology_namespace_helper_handles_all_uri_forms():
from semantica.explorer.routes.ontology import _ontology_namespace
assert _ontology_namespace("http://example.org/onto-a") == "http://example.org/onto-a#"
assert _ontology_namespace("http://example.org/onto-a/") == "http://example.org/onto-a/"
assert _ontology_namespace("http://example.org/onto-a#") == "http://example.org/onto-a#"
assert _ontology_namespace("http://example.org/onto-a#schema") == "http://example.org/onto-a#"
@pytest.mark.asyncio
async def test_data_graph_turtle_preserves_slash_namespace_for_local_terms(client):
from semantica.explorer.routes.ontology import _data_graph_turtle_for_uri
from unittest.mock import MagicMock
graph = client.app.state.session.graph
onto_uri = "http://example.org/onto-slash/"
graph.add_node(
onto_uri,
node_type="owl:Ontology",
content="Slash Ontology",
**{"rdfs:label": "Slash Ontology", "uri": onto_uri},
)
person_a = "http://example.org/onto-slash/Person"
person_inst = "http://example.org/onto-slash/person-1"
graph.add_node(
person_inst,
node_type="owl:NamedIndividual",
content="Person 1",
scheme_uri=onto_uri,
**{
"rdf:type": person_a,
"onto:name": "Alice",
"name": "Alice Unprefixed",
},
)
ttl = await _data_graph_turtle_for_uri(MagicMock(), client.app.state.session, onto_uri)
assert "http://example.org/onto-slash/name" in ttl or "onto:name" in ttl
assert "http://example.org/onto-slash#name" not in ttl
assert "http://example.org/onto-slash/Person" in ttl or "onto:Person" in ttl
assert "http://example.org/onto-slash#Person" not in ttl
@pytest.mark.asyncio
async def test_data_graph_turtle_serializes_list_of_dicts_as_uri_references(client):
from semantica.explorer.routes.ontology import _data_graph_turtle_for_uri
from unittest.mock import MagicMock
graph = client.app.state.session.graph
onto_uri = "http://example.org/onto-jsonld/"
graph.add_node(
onto_uri,
node_type="owl:Ontology",
content="JSON-LD Ontology",
**{"rdfs:label": "JSON-LD Ontology", "uri": onto_uri},
)
node_inst = "http://example.org/onto-jsonld/item-1"
graph.add_node(
node_inst,
node_type="owl:Class",
content="Item 1",
scheme_uri=onto_uri,
**{
"rdfs:seeAlso": [
{"@id": "http://example.org/external/ref1"},
{"uri": "http://example.org/external/ref2"},
],
},
)
ttl = await _data_graph_turtle_for_uri(MagicMock(), client.app.state.session, onto_uri)
assert "<http://example.org/external/ref1>" in ttl
assert "<http://example.org/external/ref2>" in ttl
assert "{'" not in ttl and "'@id'" not in ttl
def test_validate_shacl_rejects_oversized_turtle(client):
shacl_turtle = """
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix onto: <http://example.org/onto-a#> .
onto:PersonShape a sh:NodeShape ;
sh:targetClass onto:Person .
"""
with patch("semantica.explorer.routes.ontology._MAX_SHACL_TURTLE_BYTES", 20):
response = client.post(
"/api/ontology/shacl/validate",
json={
"uri": "http://example.org/onto-a",
"shacl_turtle": shacl_turtle,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "error"
assert "exceeds maximum allowed size" in payload["message"]
def test_validate_shacl_rejects_too_many_triples(client):
shacl_turtle = """
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix onto: <http://example.org/onto-a#> .
onto:PersonShape a sh:NodeShape ;
sh:targetClass onto:Person .
"""
with patch("semantica.explorer.routes.ontology._MAX_SHACL_TRIPLES", 1):
response = client.post(
"/api/ontology/shacl/validate",
json={
"uri": "http://example.org/onto-a",
"shacl_turtle": shacl_turtle,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "error"
assert "exceeds maximum allowed limit" in payload["message"]
def test_validate_shacl_handles_timeout(client):
shacl_turtle = """
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix onto: <http://example.org/onto-a#> .
onto:PersonShape a sh:NodeShape ;
sh:targetClass onto:Person .
"""
import time
def slow_validate(*args, **kwargs):
time.sleep(0.3)
return MagicMock(conforms=True, violations=[])
with patch("semantica.explorer.routes.ontology._MAX_SHACL_TIMEOUT_SECONDS", 0.05), \
patch("semantica.ontology.OntologyEngine.validate_graph", side_effect=slow_validate):
response = client.post(
"/api/ontology/shacl/validate",
json={
"uri": "http://example.org/onto-a",
"shacl_turtle": shacl_turtle,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "error"
assert "timed out" in payload["message"]
def test_validate_shacl_returns_unavailable_for_truncated_graph(client):
shacl_turtle = """
@prefix sh: <http://www.w3.org/ns/shacl#> .
@prefix onto: <http://example.org/onto-a#> .
onto:PersonShape a sh:NodeShape ;
sh:targetClass onto:Person .
"""
with patch("semantica.explorer.routes.ontology._MAX_ANALYSIS_NODES", 0):
response = client.post(
"/api/ontology/shacl/validate",
json={
"uri": "http://example.org/onto-a",
"shacl_turtle": shacl_turtle,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "unavailable"
assert "exceeds maximum analysis limit" in payload["message"]
assert payload["conforms"] is False
def test_health_shacl_dimension_returns_critical_for_truncated_graph(client):
with patch("semantica.explorer.routes.ontology._MAX_ANALYSIS_NODES", 0):
payload = client.get("/api/ontology/health?uri=http%3A%2F%2Fexample.org%2Fonto-a").json()
shacl_dim = next(d for d in payload["dimensions"] if d["key"] == "shacl")
assert shacl_dim["status"] == "critical"
assert shacl_dim["score"] == 0.0
assert "exceeds maximum analysis limit" in shacl_dim["detail"]