diff --git a/CHANGELOG.md b/CHANGELOG.md index 8723c746..b9649c8b 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 +- **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 + - 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 - 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/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..ebfc0464 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 is_skos_hierarchy_edge, validate_skos_hierarchy from .entity_linker import EntityLinker # Optional imports for advanced features @@ -589,6 +590,14 @@ 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. + 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(hierarchy_edges, existing_edges) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -948,6 +957,12 @@ class ContextGraph: family_id=explicit_family_id, ) with self._lock: + 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([candidate], existing_edges) 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..a9cce469 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) @@ -1334,21 +1339,29 @@ 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 # 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 +1558,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 4e992229..17739b67 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,8 +224,12 @@ async def import_vocabulary( 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) + 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 return VocabularyImportResponse( status="success", diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 8692f59e..1d0a60f9 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 is_skos_hierarchy_edge, 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,38 @@ 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.""" + 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(hierarchy_edges, existing_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, node_id: str, @@ -771,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 new file mode 100644 index 00000000..dd105bca --- /dev/null +++ b/semantica/utils/skos.py @@ -0,0 +1,96 @@ +"""Validation helpers for SKOS graph relationships.""" + +from collections import defaultdict +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 _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 existing_edges: + pair = _child_parent(edge) + if pair is not None: + parents[pair[0]].add(pair[1]) + + touched: set[str] = set() + for edge in new_edges: + pair = _child_parent(edge) + if pair is None: + continue + child, parent = pair + parents[child].add(parent) + touched.add(child) + touched.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 touched: + visit(concept) diff --git a/tests/context/test_context.py b/tests/context/test_context.py index 54f705f1..73edaa4e 100644 --- a/tests/context/test_context.py +++ b/tests/context/test_context.py @@ -203,6 +203,39 @@ 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) + + 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 2d6f7d9d..a106d6d7 100644 --- a/tests/explorer/test_ontology_subissue3.py +++ b/tests/explorer/test_ontology_subissue3.py @@ -636,7 +636,62 @@ 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() + + +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/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index 7b91d17f..3de0f3e9 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.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() @@ -114,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", @@ -129,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", @@ -141,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", @@ -158,3 +165,20 @@ 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(): + 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", + files={"file": ("cyclic.ttl", CYCLIC_TTL, "text/turtle")}, + ) + + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + mock_session.add_nodes_and_edges.assert_called_once() 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()