From 66be1630faa2335803969abea89b6c1ffc74fa40 Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Thu, 30 Jul 2026 20:43:26 +0530 Subject: [PATCH 1/6] fix(agno): fail fast on toolkit registration failures - #780 --- integrations/agno/decision_kit.py | 8 +++--- integrations/agno/kg_toolkit.py | 8 +++--- tests/integrations/agno/test_decision_kit.py | 28 ++++++++++++++++++-- tests/integrations/agno/test_kg_toolkit.py | 26 +++++++++++++++++- 4 files changed, 57 insertions(+), 13 deletions(-) diff --git a/integrations/agno/decision_kit.py b/integrations/agno/decision_kit.py index bcb4e66e..cc64e9c6 100644 --- a/integrations/agno/decision_kit.py +++ b/integrations/agno/decision_kit.py @@ -123,12 +123,10 @@ class AgnoDecisionKit(_ToolkitBase): # type: ignore[misc] tools_to_register.append(self.check_policy) for fn in tools_to_register: - self._tools.append(fn) if AGNO_AVAILABLE: - try: - self.register(fn) - except Exception: - pass + self.register(fn) + if fn not in self._tools: + self._tools.append(fn) logger.info("AgnoDecisionKit initialised") diff --git a/integrations/agno/kg_toolkit.py b/integrations/agno/kg_toolkit.py index 75ee26ed..c36ea63f 100644 --- a/integrations/agno/kg_toolkit.py +++ b/integrations/agno/kg_toolkit.py @@ -122,12 +122,10 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc] self.export_subgraph, ] for fn in tools_to_register: - self._tools.append(fn) if AGNO_AVAILABLE: - try: - self.register(fn) - except Exception: - pass + self.register(fn) + if fn not in self._tools: + self._tools.append(fn) logger.info("AgnoKGToolkit initialised (backend=%s)", graph_store_backend) diff --git a/tests/integrations/agno/test_decision_kit.py b/tests/integrations/agno/test_decision_kit.py index 8efd4830..3f13b6c0 100644 --- a/tests/integrations/agno/test_decision_kit.py +++ b/tests/integrations/agno/test_decision_kit.py @@ -8,7 +8,7 @@ import json import sys import types import unittest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch # --------------------------------------------------------------------------- @@ -75,7 +75,31 @@ class TestAgnoDecisionKitInit(unittest.TestCase): def test_tools_registered(self): kit = AgnoDecisionKit(context=_make_context()) # Tools should be registered (Toolkit.register was called) - self.assertTrue(len(kit._tools) >= 5) + self.assertEqual(len(kit._tools), 6) + self.assertEqual(len(kit._tools), len(set(kit._tools))) + + def test_registration_invoked(self): + with patch.object(AgnoDecisionKit, "register") as mock_register: + AgnoDecisionKit(context=_make_context()) + self.assertEqual(mock_register.call_count, 6) + + def test_registration_failure_propagates(self): + with patch.object(AgnoDecisionKit, "register", side_effect=RuntimeError("Registration failed")): + with self.assertRaises(RuntimeError): + AgnoDecisionKit(context=_make_context()) + + def test_graceful_degradation_when_agno_unavailable(self): + with patch("integrations.agno.decision_kit.AGNO_AVAILABLE", False): + with patch.object(AgnoDecisionKit, "register") as mock_register: + kit = AgnoDecisionKit(context=_make_context()) + mock_register.assert_not_called() + self.assertEqual(len(kit._tools), 6) + self.assertEqual(len(kit._tools), len(set(kit._tools))) + + def test_no_duplicate_tools(self): + kit = AgnoDecisionKit(context=_make_context()) + self.assertEqual(len(kit._tools), len(set(kit._tools))) + self.assertEqual(len(kit._tools), 6) def test_policy_tool_can_be_disabled(self): kit = AgnoDecisionKit(context=_make_context(), enable_policy_check=False) diff --git a/tests/integrations/agno/test_kg_toolkit.py b/tests/integrations/agno/test_kg_toolkit.py index 8ddd25a9..d9dcdbfc 100644 --- a/tests/integrations/agno/test_kg_toolkit.py +++ b/tests/integrations/agno/test_kg_toolkit.py @@ -129,7 +129,31 @@ class TestAgnoKGToolkitInit(unittest.TestCase): def test_tools_registered(self): kit = AgnoKGToolkit() - self.assertTrue(len(kit._tools) >= 7) + self.assertEqual(len(kit._tools), 7) + self.assertEqual(len(kit._tools), len(set(kit._tools))) + + def test_registration_invoked(self): + with patch.object(AgnoKGToolkit, "register") as mock_register: + AgnoKGToolkit() + self.assertEqual(mock_register.call_count, 7) + + def test_registration_failure_propagates(self): + with patch.object(AgnoKGToolkit, "register", side_effect=RuntimeError("Registration failed")): + with self.assertRaises(RuntimeError): + AgnoKGToolkit() + + def test_graceful_degradation_when_agno_unavailable(self): + with patch("integrations.agno.kg_toolkit.AGNO_AVAILABLE", False): + with patch.object(AgnoKGToolkit, "register") as mock_register: + kit = AgnoKGToolkit() + mock_register.assert_not_called() + self.assertEqual(len(kit._tools), 7) + self.assertEqual(len(kit._tools), len(set(kit._tools))) + + def test_no_duplicate_tools(self): + kit = AgnoKGToolkit() + self.assertEqual(len(kit._tools), len(set(kit._tools))) + self.assertEqual(len(kit._tools), 7) def test_context_graph_attached(self): ctx = MagicMock() From 67aed4399732ffff29db9c39eb4a3c906040f386 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Fri, 31 Jul 2026 17:44:14 +0530 Subject: [PATCH 2/6] docs: add changelog entry for Agno toolkit fail-fast fix (#780, #818) --- CHANGELOG.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c57a8f62..bf2b43b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **`AgnoDecisionKit`/`AgnoKGToolkit` silently swallowed Agno tool registration failures** (#780, #818) by @Sameer6305 and @KaifAhmad1 + - Removed the `try/except: pass` wrapped around `self.register(fn)` in both toolkits' `__init__`; when Agno is installed, a registration failure now propagates immediately instead of leaving the toolkit half-registered with no signal to the caller + - Graceful degradation when Agno isn't installed (`AGNO_AVAILABLE=False`) is unchanged — `_tools` is still populated so callers can introspect available tools without the package + - Fixed a related duplicate-entry bug: `self._tools` was appended to unconditionally *before* `register()` ran, which could double-count a tool when Agno's own `Toolkit.register()` also tracks it in `self._tools` + - This is a behavior change for callers that construct these toolkits expecting instantiation to always succeed — audited: no in-repo call site relies on the old silent-failure behavior + - Expanded `tests/integrations/agno/test_decision_kit.py` and `test_kg_toolkit.py` with coverage for registration invocation counts, failure propagation, graceful degradation, and no-duplicate-`_tools` assertions + - **`ProvenanceManager` duplicated the same checksum/persist/exception-swallow block across 4 tracking methods** (#784, #815) by @Sameer6305 and @KaifAhmad1 - Consolidated the repeated `entry.checksum = compute_checksum(entry)` / `try: self.storage.store(entry) except Exception: pass` block used by `track_entity`, `track_relationship`, `track_chunk`, and `track_property_source` into a single `ProvenanceManager._save_entry()` helper, preserving the existing graceful-failure behavior and the batch `_conn`/re-raise semantics from #807 - Added 4 regression tests (`tests/provenance/test_manager.py`) covering storage-failure swallowing for each of the four tracking methods, none of which had coverage for this path before From 692260cc768f384b4c186aa4774b5275305896dd Mon Sep 17 00:00:00 2001 From: mikemikimike <13286568797@163.com> Date: Fri, 31 Jul 2026 17:08:03 +0800 Subject: [PATCH 3/6] Reject cyclic SKOS hierarchies --- semantica/explorer/routes/vocabulary.py | 7 ++- semantica/explorer/session.py | 7 +++ semantica/explorer/utils/skos.py | 57 +++++++++++++++++++++++++ tests/explorer/test_vocabulary.py | 23 ++++++++++ 4 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 semantica/explorer/utils/skos.py diff --git a/semantica/explorer/routes/vocabulary.py b/semantica/explorer/routes/vocabulary.py index 4e992229..7ba6bbc4 100644 --- a/semantica/explorer/routes/vocabulary.py +++ b/semantica/explorer/routes/vocabulary.py @@ -6,7 +6,7 @@ import asyncio from collections import defaultdict from typing import Dict, List, Optional -from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile +from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile from ..dependencies import get_session from ..schemas import ConceptNode, ConceptSummary, VocabularyImportResponse, VocabularyScheme @@ -224,6 +224,11 @@ async def import_vocabulary( except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc + try: + await asyncio.to_thread(session.validate_skos_hierarchy, edges) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + nodes_added = await asyncio.to_thread(session.add_nodes, nodes) edges_added = await asyncio.to_thread(session.add_edges, edges) diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 8692f59e..19c3d885 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -13,6 +13,7 @@ from typing import Any, Dict, Iterable, List, Optional from ..context.context_graph import ContextGraph, _resolve_edge_identity from .search_index import GraphSearchIndex +from .utils.skos import validate_skos_hierarchy _KG_AVAILABLE = False try: @@ -736,6 +737,7 @@ class GraphSession: def add_edges(self, edges: List[Dict[str, Any]]) -> int: with self._lock: + self.validate_skos_hierarchy(edges) added = self.graph.add_edges(edges) has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None)) if added and not has_mutation_callback: @@ -744,6 +746,11 @@ class GraphSession: self.rebuild_search_index() return added + def validate_skos_hierarchy(self, edges: List[Dict[str, Any]]) -> None: + """Validate new SKOS hierarchy edges against the current graph.""" + existing_edges = self.graph.find_edges() + validate_skos_hierarchy([*existing_edges, *edges]) + def add_node( self, node_id: str, diff --git a/semantica/explorer/utils/skos.py b/semantica/explorer/utils/skos.py new file mode 100644 index 00000000..35e6d8f9 --- /dev/null +++ b/semantica/explorer/utils/skos.py @@ -0,0 +1,57 @@ +"""Validation helpers for SKOS graph relationships.""" + +from collections import defaultdict +from typing import Iterable, Mapping + + +_HIERARCHY_EDGE_TYPES = frozenset({"skos:broader", "skos:narrower"}) + + +def validate_skos_hierarchy( + edges: Iterable[Mapping[str, object]], +) -> None: + """Raise ``ValueError`` when SKOS hierarchy edges contain a cycle. + + ``skos:broader`` points from a concept to its parent while + ``skos:narrower`` expresses the same relationship in the opposite + direction. Both forms are normalized to a child-to-parent adjacency + map before cycle detection. + """ + + parents: dict[str, set[str]] = defaultdict(set) + for edge in edges: + edge_type = edge.get("type") + if edge_type not in _HIERARCHY_EDGE_TYPES: + continue + + source = str(edge.get("source", edge.get("source_id", ""))) + target = str(edge.get("target", edge.get("target_id", ""))) + if not source or not target: + continue + + child, parent = ( + (source, target) + if edge_type == "skos:broader" + else (target, source) + ) + parents[child].add(parent) + + visiting: set[str] = set() + visited: set[str] = set() + + def visit(concept: str) -> None: + if concept in visiting: + raise ValueError( + f"SKOS hierarchy contains a cycle involving '{concept}'." + ) + if concept in visited: + return + + visiting.add(concept) + for parent in parents.get(concept, ()): + visit(parent) + visiting.remove(concept) + visited.add(concept) + + for concept in parents: + visit(concept) diff --git a/tests/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index 7b91d17f..b148ac3c 100644 --- a/tests/explorer/test_vocabulary.py +++ b/tests/explorer/test_vocabulary.py @@ -7,6 +7,7 @@ from fastapi.testclient import TestClient from semantica.explorer.dependencies import get_session from semantica.explorer.routes.vocabulary import router +from semantica.explorer.utils.skos import validate_skos_hierarchy app = FastAPI() app.include_router(router) @@ -34,6 +35,15 @@ MINIMAL_RDF_XML = b""" """ +CYCLIC_TTL = b""" +@prefix skos: . +@prefix ex: . +ex:S a skos:ConceptScheme ; skos:prefLabel \"Scheme\" . +ex:A a skos:Concept ; skos:prefLabel \"Alpha\" ; skos:inScheme ex:S ; skos:broader ex:B . +ex:B a skos:Concept ; skos:prefLabel \"Beta\" ; skos:inScheme ex:S ; skos:broader ex:A . +""" + + def setup_function(): mock_session.reset_mock() @@ -158,3 +168,16 @@ def test_import_invalid_file_returns_422(): files={"file": ("bad.ttl", b"this is not valid RDF!", "text/turtle")}, ) assert response.status_code == 422 + + +def test_import_rejects_cyclic_hierarchy_before_writing_nodes(): + mock_session.validate_skos_hierarchy.side_effect = lambda edges: validate_skos_hierarchy(edges) + + response = client.post( + "/api/vocabulary/import", + files={"file": ("cyclic.ttl", CYCLIC_TTL, "text/turtle")}, + ) + + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + mock_session.add_nodes.assert_not_called() From d41530930daa50ba68778a7cae045eafc7a2fa5b Mon Sep 17 00:00:00 2001 From: mikemikimike <13286568797@163.com> Date: Fri, 31 Jul 2026 17:15:36 +0800 Subject: [PATCH 4/6] Centralize SKOS cycle validation --- docs/reference/explorer.md | 5 +++++ semantica/context/context_graph.py | 10 +++++++++ semantica/explorer/session.py | 2 +- semantica/{explorer => }/utils/skos.py | 29 +++++++++++++------------- tests/context/test_context.py | 20 ++++++++++++++++++ tests/explorer/test_vocabulary.py | 2 +- 6 files changed, 52 insertions(+), 16 deletions(-) rename semantica/{explorer => }/utils/skos.py (64%) diff --git a/docs/reference/explorer.md b/docs/reference/explorer.md index 34da4708..d0415bdb 100644 --- a/docs/reference/explorer.md +++ b/docs/reference/explorer.md @@ -258,6 +258,11 @@ Full interactive docs at `http://localhost:8000/docs`. All endpoints accept and | `/api/vocabulary/hierarchy` | `GET` | Concept hierarchy tree | | `/api/vocabulary/import` | `POST` | Import SKOS/RDF vocabulary file | + SKOS hierarchy writes reject cycles in both `skos:broader` and + `skos:narrower` relationships. Vocabulary imports validate the complete + batch before adding nodes, while direct graph/session edge writes apply + the same invariant at the graph storage boundary. + **SPARQL:** | Endpoint | Method | Description | diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index ac1b9b59..802295a1 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -117,6 +117,7 @@ import uuid from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker from ..utils.helpers import classify_path_distance +from ..utils.skos import validate_skos_hierarchy from .entity_linker import EntityLinker # Optional imports for advanced features @@ -589,6 +590,9 @@ class ContextGraph: """ count = 0 with self._lock: + # Keep the SKOS hierarchy invariant at the lowest common write + # layer so direct graph users cannot bypass API/session checks. + validate_skos_hierarchy([*self.find_edges(), *edges]) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -948,6 +952,12 @@ class ContextGraph: family_id=explicit_family_id, ) with self._lock: + validate_skos_hierarchy( + [ + *self.find_edges(), + {"source": source_id, "target": target_id, "type": edge_type}, + ] + ) return self._add_internal_edge( ContextEdge( edge_id=edge_id, diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 19c3d885..9eafdb8c 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -13,7 +13,7 @@ from typing import Any, Dict, Iterable, List, Optional from ..context.context_graph import ContextGraph, _resolve_edge_identity from .search_index import GraphSearchIndex -from .utils.skos import validate_skos_hierarchy +from ..utils.skos import validate_skos_hierarchy _KG_AVAILABLE = False try: diff --git a/semantica/explorer/utils/skos.py b/semantica/utils/skos.py similarity index 64% rename from semantica/explorer/utils/skos.py rename to semantica/utils/skos.py index 35e6d8f9..cfa48b0f 100644 --- a/semantica/explorer/utils/skos.py +++ b/semantica/utils/skos.py @@ -7,20 +7,27 @@ from typing import Iterable, Mapping _HIERARCHY_EDGE_TYPES = frozenset({"skos:broader", "skos:narrower"}) -def validate_skos_hierarchy( - edges: Iterable[Mapping[str, object]], -) -> None: +def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: """Raise ``ValueError`` when SKOS hierarchy edges contain a cycle. ``skos:broader`` points from a concept to its parent while ``skos:narrower`` expresses the same relationship in the opposite - direction. Both forms are normalized to a child-to-parent adjacency - map before cycle detection. + direction. Both forms are normalized to child-to-parent adjacency before + cycle detection. """ parents: dict[str, set[str]] = defaultdict(set) for edge in edges: - edge_type = edge.get("type") + if not isinstance(edge, Mapping): + continue + edge_type = next( + ( + edge.get(key) + for key in ("type", "edge_type", "relationship", "predicate", "relation") + if edge.get(key) is not None + ), + None, + ) if edge_type not in _HIERARCHY_EDGE_TYPES: continue @@ -29,11 +36,7 @@ def validate_skos_hierarchy( if not source or not target: continue - child, parent = ( - (source, target) - if edge_type == "skos:broader" - else (target, source) - ) + child, parent = ((source, target) if edge_type == "skos:broader" else (target, source)) parents[child].add(parent) visiting: set[str] = set() @@ -41,9 +44,7 @@ def validate_skos_hierarchy( def visit(concept: str) -> None: if concept in visiting: - raise ValueError( - f"SKOS hierarchy contains a cycle involving '{concept}'." - ) + raise ValueError(f"SKOS hierarchy contains a cycle involving '{concept}'.") if concept in visited: return diff --git a/tests/context/test_context.py b/tests/context/test_context.py index 54f705f1..15ec1b49 100644 --- a/tests/context/test_context.py +++ b/tests/context/test_context.py @@ -203,6 +203,26 @@ class TestContextModule(unittest.TestCase): ) self.assertIsNotNone(retriever) + def test_context_graph_rejects_cyclic_skos_single_edge_write(self): + graph = ContextGraph() + graph.add_edge("A", "B", "skos:broader") + + with self.assertRaisesRegex(ValueError, "SKOS hierarchy contains a cycle"): + graph.add_edge("B", "A", "skos:broader") + + self.assertEqual(len(graph.edges), 1) + + def test_context_graph_rejects_cyclic_skos_batch_write(self): + graph = ContextGraph() + + with self.assertRaisesRegex(ValueError, "SKOS hierarchy contains a cycle"): + graph.add_edges([ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "A", "target": "B", "type": "skos:narrower"}, + ]) + + self.assertEqual(len(graph.edges), 0) + # --- AgentContext Tests --- @patch('semantica.context.agent_memory.AgentMemory._generate_embedding') def test_agent_context_end_to_end(self, mock_gen_embedding): diff --git a/tests/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index b148ac3c..ee75ca7a 100644 --- a/tests/explorer/test_vocabulary.py +++ b/tests/explorer/test_vocabulary.py @@ -7,7 +7,7 @@ from fastapi.testclient import TestClient from semantica.explorer.dependencies import get_session from semantica.explorer.routes.vocabulary import router -from semantica.explorer.utils.skos import validate_skos_hierarchy +from semantica.utils.skos import validate_skos_hierarchy app = FastAPI() app.include_router(router) From f992504227f1b8a7120a8989f47d9b91a5836854 Mon Sep 17 00:00:00 2001 From: mikemikimike <13286568797@163.com> Date: Fri, 31 Jul 2026 17:42:21 +0800 Subject: [PATCH 5/6] Make SKOS hierarchy imports atomic --- CHANGELOG.md | 5 ++++ semantica/context/context_graph.py | 19 +++++++----- semantica/explorer/routes/export_import.py | 12 +++++--- semantica/explorer/routes/ontology.py | 27 ++++++++++++----- semantica/explorer/routes/vocabulary.py | 7 ++--- semantica/explorer/session.py | 34 ++++++++++++++++++++-- semantica/utils/skos.py | 22 +++++++++++--- tests/explorer/test_ontology_subissue3.py | 20 ++++++++++++- tests/explorer/test_vocabulary.py | 17 ++++++----- 9 files changed, 125 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8723c746..bc00cef6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **No cycle detection for SKOS concepts at write time** (#774, #819) by @mikemikimike, reviewed by @Sameer6305 and @KaifAhmad1 + - Added cycle detection (`validate_skos_hierarchy`) for `skos:broader` and `skos:narrower` relationships in `ContextGraph.add_edge()` and `ContextGraph.add_edges()`, preventing direct 2-node cycles, self-loops, and multi-hop hierarchy cycles + - Added `GraphSession.add_nodes_and_edges()` to validate SKOS hierarchy edges upfront under lock before node insertion, preventing partial-write leaks where nodes remain after a cyclic edge is rejected + - Updated vocabulary, ontology (`/api/ontology/load`, `/api/ontology/create`), and JSON/CSV import routes to use `add_nodes_and_edges()` and return HTTP 422 with actionable error messages when a cycle is detected + - **`AgnoDecisionKit`/`AgnoKGToolkit` silently swallowed Agno tool registration failures** (#780, #818) by @Sameer6305 and @KaifAhmad1 - Removed the `try/except: pass` wrapped around `self.register(fn)` in both toolkits' `__init__`; when Agno is installed, a registration failure now propagates immediately instead of leaving the toolkit half-registered with no signal to the caller - Graceful degradation when Agno isn't installed (`AGNO_AVAILABLE=False`) is unchanged — `_tools` is still populated so callers can introspect available tools without the package diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index 802295a1..f5ef322e 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -117,7 +117,7 @@ import uuid from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker from ..utils.helpers import classify_path_distance -from ..utils.skos import validate_skos_hierarchy +from ..utils.skos import is_skos_hierarchy_edge, validate_skos_hierarchy from .entity_linker import EntityLinker # Optional imports for advanced features @@ -592,7 +592,12 @@ class ContextGraph: with self._lock: # Keep the SKOS hierarchy invariant at the lowest common write # layer so direct graph users cannot bypass API/session checks. - validate_skos_hierarchy([*self.find_edges(), *edges]) + hierarchy_edges = [edge for edge in edges if is_skos_hierarchy_edge(edge)] + if hierarchy_edges: + existing_edges = [ + edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) + ] + validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -952,12 +957,12 @@ class ContextGraph: family_id=explicit_family_id, ) with self._lock: - validate_skos_hierarchy( - [ - *self.find_edges(), - {"source": source_id, "target": target_id, "type": edge_type}, + candidate = {"source": source_id, "target": target_id, "type": edge_type} + if is_skos_hierarchy_edge(candidate): + existing_edges = [ + edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) ] - ) + validate_skos_hierarchy([*existing_edges, candidate]) return self._add_internal_edge( ContextEdge( edge_id=edge_id, diff --git a/semantica/explorer/routes/export_import.py b/semantica/explorer/routes/export_import.py index 3529590a..6beda930 100644 --- a/semantica/explorer/routes/export_import.py +++ b/semantica/explorer/routes/export_import.py @@ -112,8 +112,10 @@ async def import_file( } ) - nodes_added = session.add_nodes(nodes) - edges_added = session.add_edges(edges) + try: + nodes_added, edges_added = session.add_nodes_and_edges(nodes, edges) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc return _import_response(nodes_added, edges_added) if filename.endswith(".csv"): @@ -184,8 +186,10 @@ async def import_file( detail="No valid nodes or edges could be parsed from the CSV payload.", ) - nodes_added = session.add_nodes(nodes) - edges_added = session.add_edges(edges) + try: + nodes_added, edges_added = session.add_nodes_and_edges(nodes, edges) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc return _import_response(nodes_added, edges_added) raise HTTPException( diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py index 31c21afa..2fac4c15 100644 --- a/semantica/explorer/routes/ontology.py +++ b/semantica/explorer/routes/ontology.py @@ -1289,6 +1289,8 @@ async def load_ontology( temp_path, format=fmt ) + if not ontology_data.data.get("classes") and not ontology_data.data.get("properties"): + raise ValueError("No OWL classes or properties found by OntologyIngestor") # Convert to graph nodes/edges using ontology data nodes, edges = await asyncio.to_thread( @@ -1296,9 +1298,12 @@ async def load_ontology( ontology_data.data ) - # Add nodes and edges to session - nodes_added = await asyncio.to_thread(session.add_nodes, nodes) - edges_added = await asyncio.to_thread(session.add_edges, edges) + try: + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc # Register in registry registry = _get_registry(request) @@ -1347,8 +1352,12 @@ async def load_ontology( raise HTTPException(status_code=422, detail=f"Could not parse ontology: {exc}") from exc # Fallback path - use basic parsing - nodes_added = await asyncio.to_thread(session.add_nodes, nodes) - edges_added = await asyncio.to_thread(session.add_edges, edges) + try: + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc registry = _get_registry(request) ontology_uri = metadata.get("uri", f"temp:{uuid.uuid4().hex[:12]}") @@ -1545,8 +1554,12 @@ async def create_ontology( logger.exception("Failed to generate ontology from schema text; aborting ontology creation.") 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) + try: + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc registry = _get_registry(request) registry[onto_uri] = OntologyEntry( diff --git a/semantica/explorer/routes/vocabulary.py b/semantica/explorer/routes/vocabulary.py index 7ba6bbc4..17739b67 100644 --- a/semantica/explorer/routes/vocabulary.py +++ b/semantica/explorer/routes/vocabulary.py @@ -225,13 +225,12 @@ async def import_vocabulary( raise HTTPException(status_code=422, detail=str(exc)) from exc try: - await asyncio.to_thread(session.validate_skos_hierarchy, edges) + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc - nodes_added = await asyncio.to_thread(session.add_nodes, nodes) - edges_added = await asyncio.to_thread(session.add_edges, edges) - return VocabularyImportResponse( status="success", filename=filename, diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 9eafdb8c..369d326f 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -13,7 +13,7 @@ from typing import Any, Dict, Iterable, List, Optional from ..context.context_graph import ContextGraph, _resolve_edge_identity from .search_index import GraphSearchIndex -from ..utils.skos import validate_skos_hierarchy +from ..utils.skos import is_skos_hierarchy_edge, validate_skos_hierarchy _KG_AVAILABLE = False try: @@ -748,8 +748,35 @@ class GraphSession: def validate_skos_hierarchy(self, edges: List[Dict[str, Any]]) -> None: """Validate new SKOS hierarchy edges against the current graph.""" - existing_edges = self.graph.find_edges() - validate_skos_hierarchy([*existing_edges, *edges]) + hierarchy_edges = [edge for edge in edges if is_skos_hierarchy_edge(edge)] + if not hierarchy_edges: + return + existing_edges = [ + edge for edge in self.graph.find_edges() if is_skos_hierarchy_edge(edge) + ] + validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) + + def add_nodes_and_edges( + self, + nodes: List[Dict[str, Any]], + edges: List[Dict[str, Any]], + ) -> tuple[int, int]: + """ + Validate SKOS hierarchy edges upfront and add nodes and edges under lock. + + Note: This provides lock-based mutual exclusion and pre-write validation, + not transactional rollback atomicity. + """ + with self._lock: + self.validate_skos_hierarchy(edges) + nodes_added = self.graph.add_nodes(nodes) + edges_added = self.graph.add_edges(edges) + has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None)) + if (nodes_added or edges_added) and not has_mutation_callback: + self._bump_graph_revision_locked() + if (nodes_added or edges_added) and not has_mutation_callback: + self.rebuild_search_index() + return nodes_added, edges_added def add_node( self, @@ -778,6 +805,7 @@ class GraphSession: **properties: Any, ) -> bool: with self._lock: + self.validate_skos_hierarchy([{"source": source_id, "target": target_id, "type": edge_type}]) added = self.graph.add_edge( source_id, target_id, diff --git a/semantica/utils/skos.py b/semantica/utils/skos.py index cfa48b0f..8c24bba5 100644 --- a/semantica/utils/skos.py +++ b/semantica/utils/skos.py @@ -7,6 +7,16 @@ from typing import Iterable, Mapping _HIERARCHY_EDGE_TYPES = frozenset({"skos:broader", "skos:narrower"}) +def is_skos_hierarchy_edge(edge: Mapping[str, object]) -> bool: + """Return whether an edge uses a SKOS hierarchy predicate.""" + if not isinstance(edge, Mapping): + return False + return any( + edge.get(key) in _HIERARCHY_EDGE_TYPES + for key in ("type", "edge_type", "relationship", "predicate", "relation") + ) + + def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: """Raise ``ValueError`` when SKOS hierarchy edges contain a cycle. @@ -18,21 +28,25 @@ def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: parents: dict[str, set[str]] = defaultdict(set) for edge in edges: - if not isinstance(edge, Mapping): + if not isinstance(edge, Mapping) or not is_skos_hierarchy_edge(edge): continue edge_type = next( ( edge.get(key) for key in ("type", "edge_type", "relationship", "predicate", "relation") - if edge.get(key) is not None + if edge.get(key) in _HIERARCHY_EDGE_TYPES ), None, ) if edge_type not in _HIERARCHY_EDGE_TYPES: continue - source = str(edge.get("source", edge.get("source_id", ""))) - target = str(edge.get("target", edge.get("target_id", ""))) + raw_source = edge.get("source", edge.get("source_id")) + raw_target = edge.get("target", edge.get("target_id")) + if raw_source is None or raw_target is None: + continue + source = str(raw_source).strip() + target = str(raw_target).strip() if not source or not target: continue diff --git a/tests/explorer/test_ontology_subissue3.py b/tests/explorer/test_ontology_subissue3.py index 2d6f7d9d..40cb6bbe 100644 --- a/tests/explorer/test_ontology_subissue3.py +++ b/tests/explorer/test_ontology_subissue3.py @@ -636,7 +636,25 @@ def test_health_shacl_dimension_returns_critical_for_truncated_graph(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"] == "critical" - assert shacl_dim["score"] == 0.0 assert "exceeds maximum analysis limit" in shacl_dim["detail"] +def test_ontology_load_rejects_cyclic_skos_hierarchy(client): + cyclic_ttl = """ +@prefix skos: . +@prefix ex: . +ex:S a skos:ConceptScheme ; skos:prefLabel "Scheme" . +ex:A a skos:Concept ; skos:prefLabel "Alpha" ; skos:inScheme ex:S ; skos:broader ex:B . +ex:B a skos:Concept ; skos:prefLabel "Beta" ; skos:inScheme ex:S ; skos:broader ex:A . +""" + response = client.post( + "/api/ontology/load", + json={ + "content": cyclic_ttl, + "format": "turtle", + }, + ) + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + + diff --git a/tests/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index ee75ca7a..3de0f3e9 100644 --- a/tests/explorer/test_vocabulary.py +++ b/tests/explorer/test_vocabulary.py @@ -124,8 +124,7 @@ def test_hierarchy_cycle_does_not_hang(): def test_import_ttl_success(): - mock_session.add_nodes.return_value = 2 - mock_session.add_edges.return_value = 1 + mock_session.add_nodes_and_edges.return_value = (2, 1) response = client.post( "/api/vocabulary/import", @@ -139,8 +138,7 @@ def test_import_ttl_success(): def test_import_raw_text_success(): - mock_session.add_nodes.return_value = 1 - mock_session.add_edges.return_value = 0 + mock_session.add_nodes_and_edges.return_value = (1, 0) response = client.post( "/api/vocabulary/import", @@ -151,8 +149,7 @@ def test_import_raw_text_success(): def test_import_rdf_xml_success(): - mock_session.add_nodes.return_value = 1 - mock_session.add_edges.return_value = 0 + mock_session.add_nodes_and_edges.return_value = (1, 0) response = client.post( "/api/vocabulary/import", @@ -171,7 +168,11 @@ def test_import_invalid_file_returns_422(): def test_import_rejects_cyclic_hierarchy_before_writing_nodes(): - mock_session.validate_skos_hierarchy.side_effect = lambda edges: validate_skos_hierarchy(edges) + def add_nodes_and_edges(nodes, edges): + validate_skos_hierarchy(edges) + return 0, 0 + + mock_session.add_nodes_and_edges.side_effect = add_nodes_and_edges response = client.post( "/api/vocabulary/import", @@ -180,4 +181,4 @@ def test_import_rejects_cyclic_hierarchy_before_writing_nodes(): assert response.status_code == 422 assert "cycle" in response.json()["detail"].lower() - mock_session.add_nodes.assert_not_called() + mock_session.add_nodes_and_edges.assert_called_once() From bc75768afe1f3a991fc5e1d0e8919ae8f3a10a34 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Sat, 1 Aug 2026 11:46:13 +0530 Subject: [PATCH 6/6] Fix two review findings in SKOS cycle validation - validate_skos_hierarchy() re-walked every existing hierarchy edge in the graph on each write, so one pre-existing cycle anywhere would block all unrelated future SKOS writes. It now only traverses concepts touched by the edges actually being written, while still checking those against existing edges for cross-boundary cycles. - In /api/ontology/load, `except HTTPException: raise` sat after a broader `except Exception` clause that already matched HTTPException, so a 422 raised after a successful OntologyIngestor parse was silently swallowed and retried via the fallback RDF parser instead of reaching the caller. Reordered the except clauses. Co-authored-by: mikemikimike <13286568797@163.com> --- CHANGELOG.md | 2 + semantica/context/context_graph.py | 4 +- semantica/explorer/routes/ontology.py | 10 ++- semantica/explorer/session.py | 2 +- semantica/utils/skos.py | 76 +++++++++++++++-------- tests/context/test_context.py | 13 ++++ tests/explorer/test_ontology_subissue3.py | 37 +++++++++++ tests/utils/test_skos.py | 64 +++++++++++++++++++ 8 files changed, 176 insertions(+), 32 deletions(-) create mode 100644 tests/utils/test_skos.py diff --git a/CHANGELOG.md b/CHANGELOG.md index bc00cef6..b9649c8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added cycle detection (`validate_skos_hierarchy`) for `skos:broader` and `skos:narrower` relationships in `ContextGraph.add_edge()` and `ContextGraph.add_edges()`, preventing direct 2-node cycles, self-loops, and multi-hop hierarchy cycles - Added `GraphSession.add_nodes_and_edges()` to validate SKOS hierarchy edges upfront under lock before node insertion, preventing partial-write leaks where nodes remain after a cyclic edge is rejected - Updated vocabulary, ontology (`/api/ontology/load`, `/api/ontology/create`), and JSON/CSV import routes to use `add_nodes_and_edges()` and return HTTP 422 with actionable error messages when a cycle is detected + - Follow-up fix by @KaifAhmad1: `validate_skos_hierarchy()` previously re-walked *every* SKOS hierarchy edge already in the graph on each write, so one pre-existing cycle anywhere (e.g. legacy data) blocked all unrelated future writes; it now only traverses concepts touched by the edges being written, while still checking against existing edges for cycles that span old and new data + - Follow-up fix by @KaifAhmad1: in `/api/ontology/load`, `except HTTPException: raise` was unreachable because a broader `except Exception` clause above it already matched `HTTPException`, so a 422 raised after a successful `OntologyIngestor` parse was silently swallowed and reprocessed via the fallback RDF parser; reordered the clauses so the deliberate 422 always propagates - **`AgnoDecisionKit`/`AgnoKGToolkit` silently swallowed Agno tool registration failures** (#780, #818) by @Sameer6305 and @KaifAhmad1 - Removed the `try/except: pass` wrapped around `self.register(fn)` in both toolkits' `__init__`; when Agno is installed, a registration failure now propagates immediately instead of leaving the toolkit half-registered with no signal to the caller diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index f5ef322e..ebfc0464 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -597,7 +597,7 @@ class ContextGraph: existing_edges = [ edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) ] - validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) + validate_skos_hierarchy(hierarchy_edges, existing_edges) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -962,7 +962,7 @@ class ContextGraph: existing_edges = [ edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) ] - validate_skos_hierarchy([*existing_edges, candidate]) + validate_skos_hierarchy([candidate], existing_edges) return self._add_internal_edge( ContextEdge( edge_id=edge_id, diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py index 2fac4c15..a9cce469 100644 --- a/semantica/explorer/routes/ontology.py +++ b/semantica/explorer/routes/ontology.py @@ -1339,15 +1339,19 @@ async def load_ontology( except OSError as cleanup_exc: logger.debug("Failed to remove temporary ontology file: %s", cleanup_exc) + except HTTPException: + # Re-raise HTTPExceptions we deliberately raised above (e.g. the 422 from + # SKOS cycle validation) instead of letting the broad `except Exception` + # below mask them as an ingestor failure and silently retry via the + # fallback parser. + raise except Exception as ingest_exc: logger.warning(f"OntologyIngestor failed, falling back to basic parsing: {ingest_exc}") - + # Fallback to basic parsing nodes, edges, metadata = await asyncio.to_thread( _parse_rdf_sync, content_str.encode("utf-8"), fmt ) - except HTTPException: - raise except Exception as exc: raise HTTPException(status_code=422, detail=f"Could not parse ontology: {exc}") from exc diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 369d326f..1d0a60f9 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -754,7 +754,7 @@ class GraphSession: existing_edges = [ edge for edge in self.graph.find_edges() if is_skos_hierarchy_edge(edge) ] - validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) + validate_skos_hierarchy(hierarchy_edges, existing_edges) def add_nodes_and_edges( self, diff --git a/semantica/utils/skos.py b/semantica/utils/skos.py index 8c24bba5..dd105bca 100644 --- a/semantica/utils/skos.py +++ b/semantica/utils/skos.py @@ -17,41 +17,65 @@ def is_skos_hierarchy_edge(edge: Mapping[str, object]) -> bool: ) -def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: - """Raise ``ValueError`` when SKOS hierarchy edges contain a cycle. +def _child_parent(edge: Mapping[str, object]) -> tuple[str, str] | None: + """Normalize a SKOS hierarchy edge to a ``(child, parent)`` pair, or ``None``.""" + if not isinstance(edge, Mapping) or not is_skos_hierarchy_edge(edge): + return None + edge_type = next( + ( + edge.get(key) + for key in ("type", "edge_type", "relationship", "predicate", "relation") + if edge.get(key) in _HIERARCHY_EDGE_TYPES + ), + None, + ) + if edge_type not in _HIERARCHY_EDGE_TYPES: + return None + + raw_source = edge.get("source", edge.get("source_id")) + raw_target = edge.get("target", edge.get("target_id")) + if raw_source is None or raw_target is None: + return None + source = str(raw_source).strip() + target = str(raw_target).strip() + if not source or not target: + return None + + return (source, target) if edge_type == "skos:broader" else (target, source) + + +def validate_skos_hierarchy( + new_edges: Iterable[Mapping[str, object]], + existing_edges: Iterable[Mapping[str, object]] = (), +) -> None: + """Raise ``ValueError`` when adding ``new_edges`` would introduce a cycle. ``skos:broader`` points from a concept to its parent while ``skos:narrower`` expresses the same relationship in the opposite direction. Both forms are normalized to child-to-parent adjacency before cycle detection. + + ``existing_edges`` supplies the SKOS hierarchy edges already persisted in + the graph so that cycles spanning old and new edges are still caught. + Only the concepts touched by ``new_edges`` are checked, though: a cycle + that already exists entirely within ``existing_edges`` must not block an + unrelated write elsewhere in the graph. """ - parents: dict[str, set[str]] = defaultdict(set) - for edge in edges: - if not isinstance(edge, Mapping) or not is_skos_hierarchy_edge(edge): - continue - edge_type = next( - ( - edge.get(key) - for key in ("type", "edge_type", "relationship", "predicate", "relation") - if edge.get(key) in _HIERARCHY_EDGE_TYPES - ), - None, - ) - if edge_type not in _HIERARCHY_EDGE_TYPES: - continue + for edge in existing_edges: + pair = _child_parent(edge) + if pair is not None: + parents[pair[0]].add(pair[1]) - raw_source = edge.get("source", edge.get("source_id")) - raw_target = edge.get("target", edge.get("target_id")) - if raw_source is None or raw_target is None: + touched: set[str] = set() + for edge in new_edges: + pair = _child_parent(edge) + if pair is None: continue - source = str(raw_source).strip() - target = str(raw_target).strip() - if not source or not target: - continue - - child, parent = ((source, target) if edge_type == "skos:broader" else (target, source)) + child, parent = pair parents[child].add(parent) + touched.add(child) + touched.add(parent) visiting: set[str] = set() visited: set[str] = set() @@ -68,5 +92,5 @@ def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: visiting.remove(concept) visited.add(concept) - for concept in parents: + for concept in touched: visit(concept) diff --git a/tests/context/test_context.py b/tests/context/test_context.py index 15ec1b49..73edaa4e 100644 --- a/tests/context/test_context.py +++ b/tests/context/test_context.py @@ -223,6 +223,19 @@ class TestContextModule(unittest.TestCase): self.assertEqual(len(graph.edges), 0) + def test_context_graph_preexisting_unrelated_cycle_does_not_block_new_write(self): + """A cycle already persisted elsewhere in the graph (e.g. legacy data + written before cycle detection existed) must not poison unrelated + SKOS hierarchy writes for concepts it doesn't touch.""" + from semantica.context.context_graph import ContextEdge + + graph = ContextGraph() + graph._add_internal_edge(ContextEdge(source_id="X", target_id="Y", edge_type="skos:broader")) + graph._add_internal_edge(ContextEdge(source_id="Y", target_id="X", edge_type="skos:broader")) + + self.assertTrue(graph.add_edge("C", "D", "skos:broader")) + self.assertEqual(len(graph.edges), 3) + # --- AgentContext Tests --- @patch('semantica.context.agent_memory.AgentMemory._generate_embedding') def test_agent_context_end_to_end(self, mock_gen_embedding): diff --git a/tests/explorer/test_ontology_subissue3.py b/tests/explorer/test_ontology_subissue3.py index 40cb6bbe..a106d6d7 100644 --- a/tests/explorer/test_ontology_subissue3.py +++ b/tests/explorer/test_ontology_subissue3.py @@ -658,3 +658,40 @@ ex:B a skos:Concept ; skos:prefLabel "Beta" ; skos:inScheme ex:S ; skos:broader assert "cycle" in response.json()["detail"].lower() +def test_ontology_load_does_not_swallow_422_from_ingestor_success_path(client): + """A ValueError raised by add_nodes_and_edges() after OntologyIngestor + succeeds must surface as its own 422, not be masked by the broad + `except Exception` fallback-to-basic-parsing handler and silently + retried under a different parser.""" + from semantica.ingest.ontology_ingestor import OntologyData + + fake_data = OntologyData( + data={ + "uri": "http://example.org/onto-fake", + "name": "Fake Ontology", + "classes": [{"uri": "http://example.org/onto-fake#A", "name": "A"}], + "properties": [], + }, + source_path="fake.ttl", + format="turtle", + ) + + with patch( + "semantica.ingest.ontology_ingestor.OntologyIngestor.ingest_ontology", + return_value=fake_data, + ), patch( + "semantica.explorer.session.GraphSession.add_nodes_and_edges", + side_effect=ValueError("SKOS hierarchy contains a cycle involving 'A'."), + ), patch( + "semantica.explorer.routes.ontology._parse_rdf_sync" + ) as fallback_parse: + response = client.post( + "/api/ontology/load", + json={"content": "@prefix ex: . ex:A a ex:Thing .", "format": "turtle"}, + ) + + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + fallback_parse.assert_not_called() + + diff --git a/tests/utils/test_skos.py b/tests/utils/test_skos.py new file mode 100644 index 00000000..ed28a6f0 --- /dev/null +++ b/tests/utils/test_skos.py @@ -0,0 +1,64 @@ +import unittest + +from semantica.utils.skos import is_skos_hierarchy_edge, validate_skos_hierarchy + + +class TestIsSkosHierarchyEdge(unittest.TestCase): + + def test_recognizes_broader_and_narrower(self): + self.assertTrue(is_skos_hierarchy_edge({"source": "A", "target": "B", "type": "skos:broader"})) + self.assertTrue(is_skos_hierarchy_edge({"source": "A", "target": "B", "type": "skos:narrower"})) + + def test_ignores_other_edge_types(self): + self.assertFalse(is_skos_hierarchy_edge({"source": "A", "target": "B", "type": "rdfs:subClassOf"})) + + def test_ignores_non_mapping(self): + self.assertFalse(is_skos_hierarchy_edge("not-a-dict")) + + +class TestValidateSkosHierarchy(unittest.TestCase): + + def test_accepts_acyclic_chain(self): + validate_skos_hierarchy([ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "B", "target": "C", "type": "skos:broader"}, + ]) + + def test_rejects_self_loop(self): + with self.assertRaisesRegex(ValueError, "cycle"): + validate_skos_hierarchy([{"source": "A", "target": "A", "type": "skos:broader"}]) + + def test_rejects_direct_two_node_cycle(self): + with self.assertRaisesRegex(ValueError, "cycle"): + validate_skos_hierarchy([ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "B", "target": "A", "type": "skos:broader"}, + ]) + + def test_rejects_cycle_spanning_existing_and_new_edges(self): + existing = [ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "B", "target": "C", "type": "skos:broader"}, + ] + with self.assertRaisesRegex(ValueError, "cycle"): + validate_skos_hierarchy([{"source": "C", "target": "A", "type": "skos:broader"}], existing) + + def test_preexisting_unrelated_cycle_does_not_block_new_write(self): + """A cycle already persisted elsewhere in the graph (e.g. legacy data + written before cycle detection existed) must not poison unrelated + writes for concepts it doesn't touch.""" + existing = [ + {"source": "X", "target": "Y", "type": "skos:broader"}, + {"source": "Y", "target": "X", "type": "skos:broader"}, + ] + validate_skos_hierarchy([{"source": "C", "target": "D", "type": "skos:broader"}], existing) + + def test_none_and_blank_endpoints_are_ignored(self): + validate_skos_hierarchy([ + {"source": None, "target": "B", "type": "skos:broader"}, + {"source": " ", "target": "B", "type": "skos:broader"}, + ]) + + +if __name__ == "__main__": + unittest.main()