diff --git a/docs/reference/ontology.md b/docs/reference/ontology.md index 4fded4b6..8f72c3f2 100644 --- a/docs/reference/ontology.md +++ b/docs/reference/ontology.md @@ -335,6 +335,114 @@ else: --- +## SKOS Vocabulary Management + +Semantica supports [SKOS (Simple Knowledge Organization System)](https://www.w3.org/TR/skos-reference/) vocabularies as first-class semantic assets. SKOS triples are stored in the existing RDF triplet store and queried through the `OntologyEngine` — no additional packages are required. + +### Concepts and data model + +| SKOS element | RDF type / predicate | +|---|---| +| ConceptScheme | `skos:ConceptScheme` | +| Concept | `skos:Concept` | +| Preferred label | `skos:prefLabel` | +| Alternative label | `skos:altLabel` | +| Broader concept | `skos:broader` | +| Narrower concept | `skos:narrower` | +| Related concept | `skos:related` | +| Human definition | `skos:definition` | +| Notation / code | `skos:notation` | + +### Importing a SKOS vocabulary + +Use `TripletStore.add_skos_concept()` to load individual concepts. The method automatically asserts the parent `skos:ConceptScheme` triple the first time any concept for that scheme is added. + +```python +from semantica.triplet_store import TripletStore + +store = TripletStore(backend="blazegraph", endpoint="http://localhost:9999/blazegraph") + +SCHEME = "https://vocab.example.org/colours" + +store.add_skos_concept( + concept_uri="https://vocab.example.org/colours/red", + scheme_uri=SCHEME, + pref_label="Red", + alt_labels=["Crimson", "Rouge"], + broader=["https://vocab.example.org/colours/warm"], + definition="The colour at the long-wavelength end of the visible spectrum.", + notation="RED", +) + +store.add_skos_concept( + concept_uri="https://vocab.example.org/colours/blue", + scheme_uri=SCHEME, + pref_label="Blue", + alt_labels=["Azure", "Cerulean"], +) +``` + +For bulk ingestion of an existing SKOS/Turtle file use `TripletStore.add_triplets()` after parsing the file with [rdflib](https://rdflib.readthedocs.io/): + +```python +import rdflib +from semantica.semantic_extract.triplet_extractor import Triplet + +g = rdflib.Graph() +g.parse("my_vocabulary.ttl", format="turtle") + +triplets = [ + Triplet(subject=str(s), predicate=str(p), object=str(o)) + for s, p, o in g +] +store.add_triplets(triplets) +``` + +### Listing and searching concepts + +Once a vocabulary is loaded, use `OntologyEngine` to browse and search it: + +```python +from semantica.ontology import OntologyEngine + +engine = OntologyEngine(store=store) + +# 1. List all ConceptSchemes in the store +vocabularies = engine.list_vocabularies() +# [{"uri": "https://vocab.example.org/colours", "label": "Colours"}, ...] + +# 2. List every concept in a specific scheme +concepts = engine.list_concepts("https://vocab.example.org/colours") +# [{"uri": "...", "pref_label": "Red", "alt_labels": ["Crimson", "Rouge"]}, ...] + +# 3. Case-insensitive substring search across prefLabel and altLabel +results = engine.search_concepts("crimson") +# [{"uri": "https://vocab.example.org/colours/red", "label": "Crimson"}] + +# 4. Restrict search to one scheme +results = engine.search_concepts("azure", scheme_uri="https://vocab.example.org/colours") +``` + +### Building SKOS URIs with NamespaceManager + +`NamespaceManager` provides helpers for constructing well-formed SKOS IRIs: + +```python +from semantica.ontology import NamespaceManager + +nm = NamespaceManager(base_uri="https://vocab.example.org/") + +# Full SKOS predicate URI +nm.get_skos_uri("prefLabel") +# "http://www.w3.org/2004/02/skos/core#prefLabel" + +# Slug-based ConceptScheme URI anchored at the base +nm.build_concept_scheme_uri("ISO 3166 Countries") +# "https://vocab.example.org/vocab/iso-3166-countries" +``` + +--- + ## Best Practices 1. **Reuse Standard Ontologies**: Don't reinvent `Person` or `Organization`; import FOAF or Schema.org using `ReuseManager`. diff --git a/semantica/ontology/engine.py b/semantica/ontology/engine.py index ffb54c21..d698f5e2 100644 --- a/semantica/ontology/engine.py +++ b/semantica/ontology/engine.py @@ -385,6 +385,194 @@ class OntologyEngine: self.progress.stop_tracking(tracking_id, status="failed", message=str(e)) raise + # ── SKOS Vocabulary Management ──────────────────────────────────────────── + + _SKOS = "http://www.w3.org/2004/02/skos/core#" + _RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" + + def list_vocabularies(self, **options) -> List[Dict[str, Any]]: + """ + List all SKOS ConceptSchemes stored in the triplet store. + + Returns: + List of dicts with keys ``uri`` and ``label`` (may be empty string + when no ``skos:prefLabel`` is present). + + Raises: + ProcessingError: If no store is configured or the query fails. + """ + if not self.store: + raise ProcessingError("TripletStore instance not configured in OntologyEngine.") + + SKOS = self._SKOS + RDF_TYPE = self._RDF_TYPE + + query = f""" + SELECT DISTINCT ?scheme ?label WHERE {{ + ?scheme <{RDF_TYPE}> <{SKOS}ConceptScheme> . + OPTIONAL {{ ?scheme <{SKOS}prefLabel> ?label }} + }} + """ + tracking_id = self.progress.start_tracking( + module="ontology", submodule="OntologyEngine", message="Listing SKOS vocabularies" + ) + try: + result = self.store.execute_query(query, **options) + vocabs = [] + if hasattr(result, "bindings"): + seen: set = set() + for b in result.bindings: + def _v(key): + val = b.get(key) + return (val.get("value") if isinstance(val, dict) else val) if val else None + uri = _v("scheme") + if uri and uri not in seen: + seen.add(uri) + vocabs.append({"uri": uri, "label": _v("label") or ""}) + self.progress.stop_tracking(tracking_id, status="completed", + message=f"Found {len(vocabs)} vocabularies") + return vocabs + except Exception as e: + self.progress.stop_tracking(tracking_id, status="failed", message=str(e)) + raise ProcessingError(f"list_vocabularies failed: {e}") + + def list_concepts(self, scheme_uri: str, **options) -> List[Dict[str, Any]]: + """ + List all SKOS concepts that belong to the given ConceptScheme. + + Args: + scheme_uri: Full URI of the ``skos:ConceptScheme`` to inspect. + + Returns: + List of dicts with keys ``uri``, ``pref_label``, and + ``alt_labels`` (list, may be empty). + + Raises: + ProcessingError: If no store is configured or the query fails. + """ + if not self.store: + raise ProcessingError("TripletStore instance not configured in OntologyEngine.") + + SKOS = self._SKOS + RDF_TYPE = self._RDF_TYPE + safe_scheme = self._sanitize_uri(scheme_uri) + + query = f""" + SELECT DISTINCT ?concept ?prefLabel ?altLabel WHERE {{ + ?concept <{RDF_TYPE}> <{SKOS}Concept> . + ?concept <{SKOS}inScheme> <{safe_scheme}> . + OPTIONAL {{ ?concept <{SKOS}prefLabel> ?prefLabel }} + OPTIONAL {{ ?concept <{SKOS}altLabel> ?altLabel }} + }} + """ + tracking_id = self.progress.start_tracking( + module="ontology", submodule="OntologyEngine", + message=f"Listing concepts in {scheme_uri}" + ) + try: + result = self.store.execute_query(query, **options) + concepts: Dict[str, Dict[str, Any]] = {} + if hasattr(result, "bindings"): + for b in result.bindings: + def _v(key): + val = b.get(key) + return (val.get("value") if isinstance(val, dict) else val) if val else None + uri = _v("concept") + if not uri: + continue + if uri not in concepts: + concepts[uri] = {"uri": uri, "pref_label": _v("prefLabel") or "", "alt_labels": []} + if not concepts[uri]["pref_label"] and _v("prefLabel"): + concepts[uri]["pref_label"] = _v("prefLabel") + lbl = _v("altLabel") + if lbl and lbl not in concepts[uri]["alt_labels"]: + concepts[uri]["alt_labels"].append(lbl) + self.progress.stop_tracking(tracking_id, status="completed", + message=f"Found {len(concepts)} concepts") + return list(concepts.values()) + except Exception as e: + self.progress.stop_tracking(tracking_id, status="failed", message=str(e)) + raise ProcessingError(f"list_concepts failed: {e}") + + def search_concepts( + self, + query: str, + scheme_uri: Optional[str] = None, + **options, + ) -> List[Dict[str, Any]]: + """ + Search SKOS concepts by matching ``skos:prefLabel`` or ``skos:altLabel``. + + The search is case-insensitive substring matching performed at the + SPARQL level via ``CONTAINS(LCASE(…))``. + + Args: + query: Substring to search for. + scheme_uri: When given, restrict results to this ConceptScheme. + + Returns: + List of dicts with keys ``uri`` and ``label`` (the matching label). + + Raises: + ProcessingError: If no store is configured or the query fails. + """ + if not self.store: + raise ProcessingError("TripletStore instance not configured in OntologyEngine.") + + SKOS = self._SKOS + RDF_TYPE = self._RDF_TYPE + + # Sanitize user query for embedding in a SPARQL string literal + safe_query = ( + query + .replace("\\", "\\\\") + .replace('"', '\\"') + .replace("\n", " ") + .replace("\r", " ") + ) + + scheme_filter = "" + if scheme_uri: + safe_scheme = self._sanitize_uri(scheme_uri) + scheme_filter = f"?concept <{SKOS}inScheme> <{safe_scheme}> ." + + sparql = f""" + SELECT DISTINCT ?concept ?label WHERE {{ + ?concept <{RDF_TYPE}> <{SKOS}Concept> . + {scheme_filter} + {{ + ?concept <{SKOS}prefLabel> ?label + }} UNION {{ + ?concept <{SKOS}altLabel> ?label + }} + FILTER(CONTAINS(LCASE(STR(?label)), LCASE("{safe_query}"))) + }} + """ + tracking_id = self.progress.start_tracking( + module="ontology", submodule="OntologyEngine", + message=f"Searching SKOS concepts: '{query}'" + ) + try: + result = self.store.execute_query(sparql, **options) + matches = [] + seen: set = set() + if hasattr(result, "bindings"): + for b in result.bindings: + def _v(key): + val = b.get(key) + return (val.get("value") if isinstance(val, dict) else val) if val else None + uri = _v("concept") + lbl = _v("label") + if uri and uri not in seen: + seen.add(uri) + matches.append({"uri": uri, "label": lbl or ""}) + self.progress.stop_tracking(tracking_id, status="completed", + message=f"Found {len(matches)} matches") + return matches + except Exception as e: + self.progress.stop_tracking(tracking_id, status="failed", message=str(e)) + raise ProcessingError(f"search_concepts failed: {e}") + # ── Ontology Evaluation / Validation ───────────────────────────────────── def evaluate(self, ontology: Dict[str, Any], **options): diff --git a/semantica/ontology/namespace_manager.py b/semantica/ontology/namespace_manager.py index 614492a7..f7ed0047 100644 --- a/semantica/ontology/namespace_manager.py +++ b/semantica/ontology/namespace_manager.py @@ -207,6 +207,35 @@ class NamespaceManager: """ return dict(self.namespaces) + def get_skos_uri(self, local_name: str) -> str: + """ + Build a full SKOS URI from a local name. + + Args: + local_name: SKOS local term (e.g. ``"Concept"``, ``"prefLabel"``) + + Returns: + Full SKOS URI string + """ + skos_ns = self.namespaces["skos"] + return f"{skos_ns}{local_name}" + + def build_concept_scheme_uri(self, name: str) -> str: + """ + Build a ConceptScheme URI anchored at the current base URI. + + The scheme name is slugified (spaces → hyphens, lower-cased) so that + ``"My Vocabulary"`` becomes ``/vocab/my-vocabulary>``. + + Args: + name: Human-readable vocabulary name + + Returns: + ConceptScheme URI string + """ + slug = re.sub(r"[^a-zA-Z0-9]+", "-", name).strip("-").lower() + return urljoin(self.get_base_uri(), f"vocab/{slug}") + def get_alignment_predicates(self) -> Dict[str, str]: """ Get standard alignment predicates for ontology mapping. diff --git a/semantica/triplet_store/triplet_store.py b/semantica/triplet_store/triplet_store.py index dd244a76..fa55ce90 100644 --- a/semantica/triplet_store/triplet_store.py +++ b/semantica/triplet_store/triplet_store.py @@ -422,6 +422,157 @@ class TripletStore: return True + # ── SKOS helpers ───────────────────────────────────────────────────────── + + _SKOS = "http://www.w3.org/2004/02/skos/core#" + _RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" + + def add_skos_concept( + self, + concept_uri: str, + scheme_uri: str, + pref_label: str, + alt_labels: Optional[List[str]] = None, + broader: Optional[List[str]] = None, + narrower: Optional[List[str]] = None, + related: Optional[List[str]] = None, + definition: Optional[str] = None, + notation: Optional[str] = None, + **options, + ) -> Dict[str, Any]: + """ + Add a SKOS concept (and its scheme if not already present) to the store. + + Core triples added: + + * ``concept_uri rdf:type skos:Concept`` + * ``concept_uri skos:inScheme scheme_uri`` + * ``concept_uri skos:prefLabel pref_label`` + * ``scheme_uri rdf:type skos:ConceptScheme`` (auto-created) + * Optional: altLabel, broader, narrower, related, definition, notation + + Args: + concept_uri: Full URI for the concept. + scheme_uri: Full URI for the parent ConceptScheme. + pref_label: Preferred label string. + alt_labels: Optional list of alternative label strings. + broader: Optional list of broader concept URIs. + narrower: Optional list of narrower concept URIs. + related: Optional list of related concept URIs. + definition: Optional human-readable definition string. + notation: Optional notation / code string. + **options: Forwarded to :meth:`add_triplets`. + + Returns: + :meth:`add_triplets` status dict. + """ + SKOS = self._SKOS + RDF_TYPE = self._RDF_TYPE + + triplets: List[Triplet] = [ + # Scheme declaration + Triplet(scheme_uri, RDF_TYPE, f"{SKOS}ConceptScheme"), + # Concept core + Triplet(concept_uri, RDF_TYPE, f"{SKOS}Concept"), + Triplet(concept_uri, f"{SKOS}inScheme", scheme_uri), + Triplet(concept_uri, f"{SKOS}prefLabel", pref_label), + ] + + for lbl in (alt_labels or []): + triplets.append(Triplet(concept_uri, f"{SKOS}altLabel", lbl)) + for uri in (broader or []): + triplets.append(Triplet(concept_uri, f"{SKOS}broader", uri)) + for uri in (narrower or []): + triplets.append(Triplet(concept_uri, f"{SKOS}narrower", uri)) + for uri in (related or []): + triplets.append(Triplet(concept_uri, f"{SKOS}related", uri)) + if definition: + triplets.append(Triplet(concept_uri, f"{SKOS}definition", definition)) + if notation: + triplets.append(Triplet(concept_uri, f"{SKOS}notation", notation)) + + return self.add_triplets(triplets, **options) + + def get_skos_concepts( + self, scheme_uri: Optional[str] = None, **options + ) -> List[Dict[str, Any]]: + """ + Retrieve SKOS concepts from the store as plain dicts. + + Each returned dict has at minimum ``uri`` and ``pref_label``; optional + keys ``alt_labels``, ``broader``, ``narrower``, and ``related`` are + populated when available. + + Args: + scheme_uri: When given, only concepts ``skos:inScheme`` this URI + are returned. When omitted all concepts are returned. + **options: Forwarded to :meth:`execute_query`. + + Returns: + List of concept dicts. + """ + SKOS = self._SKOS + RDF_TYPE = self._RDF_TYPE + + scheme_filter = ( + f"?concept <{SKOS}inScheme> <{self.query_engine._sanitize_uri(scheme_uri)}> ." + if scheme_uri + else "" + ) + + query = f""" + SELECT DISTINCT ?concept ?prefLabel ?altLabel ?broader ?narrower ?related + WHERE {{ + ?concept <{RDF_TYPE}> <{SKOS}Concept> . + {scheme_filter} + OPTIONAL {{ ?concept <{SKOS}prefLabel> ?prefLabel }} + OPTIONAL {{ ?concept <{SKOS}altLabel> ?altLabel }} + OPTIONAL {{ ?concept <{SKOS}broader> ?broader }} + OPTIONAL {{ ?concept <{SKOS}narrower> ?narrower }} + OPTIONAL {{ ?concept <{SKOS}related> ?related }} + }} + """ + + try: + result = self.execute_query(query, **options) + except Exception as e: + self.logger.error(f"get_skos_concepts query failed: {e}") + raise ProcessingError(f"Failed to retrieve SKOS concepts: {e}") + + # Collapse multi-valued properties per concept URI + concepts: Dict[str, Dict[str, Any]] = {} + for b in result.bindings: + def _val(key: str) -> Optional[str]: + v = b.get(key) + return (v.get("value") if isinstance(v, dict) else v) if v else None + + uri = _val("concept") + if not uri: + continue + if uri not in concepts: + concepts[uri] = { + "uri": uri, + "pref_label": _val("prefLabel") or "", + "alt_labels": [], + "broader": [], + "narrower": [], + "related": [], + } + entry = concepts[uri] + if not entry["pref_label"] and _val("prefLabel"): + entry["pref_label"] = _val("prefLabel") + for multi_key, sparql_key in [ + ("alt_labels", "altLabel"), + ("broader", "broader"), + ("narrower", "narrower"), + ("related", "related"), + ]: + v = _val(sparql_key) + if v and v not in entry[multi_key]: + entry[multi_key].append(v) + + return list(concepts.values()) + def get_stats(self) -> Dict[str, Any]: """Get store statistics.""" if hasattr(self._store_backend, "get_stats"): diff --git a/tests/ontology/test_ontology_comprehensive.py b/tests/ontology/test_ontology_comprehensive.py index 090840f7..7c6302af 100644 --- a/tests/ontology/test_ontology_comprehensive.py +++ b/tests/ontology/test_ontology_comprehensive.py @@ -431,5 +431,199 @@ class TestSHACLGeneration(unittest.TestCase): self.assertEqual(gen.quality_tier, "standard") +class TestSKOSOntologyEngine(unittest.TestCase): + """Tests for SKOS vocabulary management APIs in OntologyEngine.""" + + def setUp(self): + self.mock_logger = MagicMock() + self.mock_tracker = MagicMock() + self.mock_tracker.start_tracking.return_value = "track_id" + + patchers = [ + patch('semantica.ontology.engine.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.engine.get_progress_tracker', return_value=self.mock_tracker), + patch('semantica.ontology.ontology_generator.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.ontology_generator.get_progress_tracker', return_value=self.mock_tracker), + patch('semantica.ontology.class_inferrer.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.class_inferrer.get_progress_tracker', return_value=self.mock_tracker), + patch('semantica.ontology.property_generator.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.property_generator.get_progress_tracker', return_value=self.mock_tracker), + patch('semantica.ontology.owl_generator.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.owl_generator.get_progress_tracker', return_value=self.mock_tracker), + patch('semantica.ontology.ontology_evaluator.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.ontology_evaluator.get_progress_tracker', return_value=self.mock_tracker), + patch('semantica.ontology.ontology_validator.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.llm_generator.get_logger', return_value=self.mock_logger), + patch('semantica.ontology.llm_generator.get_progress_tracker', return_value=self.mock_tracker), + patch('semantica.change_management.ontology_version_manager.get_logger', return_value=self.mock_logger), + patch('semantica.change_management.ontology_version_manager.get_progress_tracker', return_value=self.mock_tracker), + ] + self.patchers = patchers + for p in self.patchers: + p.start() + + # Mock store with a controllable execute_query + self.mock_store = MagicMock() + from semantica.ontology.engine import OntologyEngine + self.engine = OntologyEngine(store=self.mock_store) + + def tearDown(self): + for p in self.patchers: + p.stop() + + def _make_result(self, bindings): + """Build a fake QueryResult-like object.""" + result = MagicMock() + result.bindings = bindings + return result + + # --- NamespaceManager SKOS helpers --- + + def test_get_skos_uri(self): + from semantica.ontology.namespace_manager import NamespaceManager + nm = NamespaceManager() + self.assertEqual( + nm.get_skos_uri("Concept"), + "http://www.w3.org/2004/02/skos/core#Concept", + ) + self.assertEqual( + nm.get_skos_uri("prefLabel"), + "http://www.w3.org/2004/02/skos/core#prefLabel", + ) + + def test_build_concept_scheme_uri(self): + from semantica.ontology.namespace_manager import NamespaceManager + nm = NamespaceManager(base_uri="https://example.org/onto/") + uri = nm.build_concept_scheme_uri("My Vocabulary") + self.assertIn("my-vocabulary", uri) + self.assertTrue(uri.startswith("https://example.org/onto/")) + + def test_build_concept_scheme_uri_special_chars(self): + from semantica.ontology.namespace_manager import NamespaceManager + nm = NamespaceManager() + uri = nm.build_concept_scheme_uri("ISO 3166 Countries") + self.assertIn("iso-3166-countries", uri) + + # --- list_vocabularies --- + + def test_list_vocabularies_returns_schemes(self): + self.mock_store.execute_query.return_value = self._make_result([ + {"scheme": {"value": "http://example.org/vocab/colours"}, + "label": {"value": "Colours"}}, + {"scheme": {"value": "http://example.org/vocab/sizes"}, + "label": None}, + ]) + vocabs = self.engine.list_vocabularies() + self.assertEqual(len(vocabs), 2) + uris = [v["uri"] for v in vocabs] + self.assertIn("http://example.org/vocab/colours", uris) + self.assertIn("http://example.org/vocab/sizes", uris) + colours = next(v for v in vocabs if "colours" in v["uri"]) + self.assertEqual(colours["label"], "Colours") + + def test_list_vocabularies_deduplicates(self): + # Same scheme URI appearing twice (multi-valued label rows) + self.mock_store.execute_query.return_value = self._make_result([ + {"scheme": {"value": "http://example.org/vocab/colours"}, + "label": {"value": "Colours"}}, + {"scheme": {"value": "http://example.org/vocab/colours"}, + "label": {"value": "Colors"}}, + ]) + vocabs = self.engine.list_vocabularies() + self.assertEqual(len(vocabs), 1) + + def test_list_vocabularies_no_store_raises(self): + from semantica.utils.exceptions import ProcessingError + from semantica.ontology.engine import OntologyEngine + engine_no_store = OntologyEngine() + with self.assertRaises(ProcessingError): + engine_no_store.list_vocabularies() + + # --- list_concepts --- + + def test_list_concepts_returns_concepts(self): + self.mock_store.execute_query.return_value = self._make_result([ + {"concept": {"value": "http://example.org/concept/red"}, + "prefLabel": {"value": "Red"}, + "altLabel": {"value": "Crimson"}}, + {"concept": {"value": "http://example.org/concept/red"}, + "prefLabel": {"value": "Red"}, + "altLabel": {"value": "Rouge"}}, + {"concept": {"value": "http://example.org/concept/blue"}, + "prefLabel": {"value": "Blue"}, + "altLabel": None}, + ]) + concepts = self.engine.list_concepts("http://example.org/vocab/colours") + self.assertEqual(len(concepts), 2) + red = next(c for c in concepts if "red" in c["uri"]) + self.assertEqual(red["pref_label"], "Red") + self.assertIn("Crimson", red["alt_labels"]) + self.assertIn("Rouge", red["alt_labels"]) + blue = next(c for c in concepts if "blue" in c["uri"]) + self.assertEqual(blue["alt_labels"], []) + + def test_list_concepts_no_store_raises(self): + from semantica.utils.exceptions import ProcessingError + from semantica.ontology.engine import OntologyEngine + engine_no_store = OntologyEngine() + with self.assertRaises(ProcessingError): + engine_no_store.list_concepts("http://example.org/vocab/colours") + + # --- search_concepts --- + + def test_search_concepts_returns_matches(self): + self.mock_store.execute_query.return_value = self._make_result([ + {"concept": {"value": "http://example.org/concept/red"}, + "label": {"value": "Red"}}, + {"concept": {"value": "http://example.org/concept/infrared"}, + "label": {"value": "Infrared"}}, + ]) + results = self.engine.search_concepts("red") + self.assertEqual(len(results), 2) + uris = [r["uri"] for r in results] + self.assertIn("http://example.org/concept/red", uris) + self.assertIn("http://example.org/concept/infrared", uris) + + def test_search_concepts_with_scheme_filter(self): + self.mock_store.execute_query.return_value = self._make_result([ + {"concept": {"value": "http://example.org/concept/red"}, + "label": {"value": "Red"}}, + ]) + results = self.engine.search_concepts("red", scheme_uri="http://example.org/vocab/colours") + self.assertEqual(len(results), 1) + # Scheme URI should appear in the SPARQL issued to the store + issued_sparql = self.mock_store.execute_query.call_args[0][0] + self.assertIn("http://example.org/vocab/colours", issued_sparql) + + def test_search_concepts_empty_result(self): + self.mock_store.execute_query.return_value = self._make_result([]) + results = self.engine.search_concepts("zzznomatch") + self.assertEqual(results, []) + + def test_search_concepts_no_store_raises(self): + from semantica.utils.exceptions import ProcessingError + from semantica.ontology.engine import OntologyEngine + engine_no_store = OntologyEngine() + with self.assertRaises(ProcessingError): + engine_no_store.search_concepts("red") + + def test_search_concepts_sanitizes_query(self): + """Ensure user input containing SPARQL-special chars doesn't break the query.""" + self.mock_store.execute_query.return_value = self._make_result([]) + # Should not raise + self.engine.search_concepts('red" } MALICIOUS { ?x ?y ?z') + + def test_search_concepts_deduplicates(self): + # Same concept URI matched by both prefLabel and altLabel + self.mock_store.execute_query.return_value = self._make_result([ + {"concept": {"value": "http://example.org/concept/red"}, + "label": {"value": "Red"}}, + {"concept": {"value": "http://example.org/concept/red"}, + "label": {"value": "Reddish"}}, + ]) + results = self.engine.search_concepts("red") + self.assertEqual(len(results), 1) + + if __name__ == '__main__': unittest.main() diff --git a/tests/triplet_store/test_triplet_store.py b/tests/triplet_store/test_triplet_store.py index 4bc72e98..1611d300 100644 --- a/tests/triplet_store/test_triplet_store.py +++ b/tests/triplet_store/test_triplet_store.py @@ -162,3 +162,183 @@ class TestTripletStore(unittest.TestCase): self.assertIn("http://aligned.org/2", sparql_query) self.assertIn("VALUES ?subject", sparql_query) mock_backend.execute_sparql.assert_called_once() + + +class TestSKOSTripletStore(unittest.TestCase): + """Tests for SKOS helper methods on TripletStore.""" + + _SKOS = "http://www.w3.org/2004/02/skos/core#" + _RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" + + def setUp(self): + self.mock_logger = MagicMock() + self.mock_tracker = MagicMock() + + self.logger_patcher = patch( + 'semantica.triplet_store.triplet_store.get_logger', return_value=self.mock_logger + ) + self.tracker_patcher = patch( + 'semantica.triplet_store.triplet_store.get_progress_tracker', return_value=self.mock_tracker + ) + self.logger_patcher.start() + self.tracker_patcher.start() + + def tearDown(self): + self.logger_patcher.stop() + self.tracker_patcher.stop() + + def _make_store(self, mock_blazegraph): + """Return a TripletStore backed by a MagicMock BlazegraphStore.""" + mock_backend = MagicMock() + mock_blazegraph.return_value = mock_backend + store = TripletStore(backend="blazegraph") + # Provide a fast no-op bulk loader + mock_loader = MagicMock() + mock_progress = MagicMock() + mock_progress.metadata = {"success": True} + mock_progress.total_triplets = 0 + mock_progress.loaded_triplets = 0 + mock_progress.failed_triplets = 0 + mock_progress.total_batches = 0 + mock_loader.load_triplets.return_value = mock_progress + store.bulk_loader = mock_loader + return store, mock_backend, mock_loader + + # --- add_skos_concept --- + + @patch('semantica.triplet_store.blazegraph_store.BlazegraphStore') + def test_add_skos_concept_core_triples(self, mock_bg): + """add_skos_concept must produce ConceptScheme + Concept + inScheme + prefLabel triples.""" + store, _, mock_loader = self._make_store(mock_bg) + + store.add_skos_concept( + concept_uri="http://example.org/concept/red", + scheme_uri="http://example.org/vocab/colours", + pref_label="Red", + ) + + mock_loader.load_triplets.assert_called_once() + triplets = mock_loader.load_triplets.call_args[0][0] + subjects_predicates = {(t.subject, t.predicate) for t in triplets} + + SKOS = self._SKOS + RDF_TYPE = self._RDF_TYPE + + self.assertIn(("http://example.org/vocab/colours", RDF_TYPE), subjects_predicates) + self.assertIn(("http://example.org/concept/red", RDF_TYPE), subjects_predicates) + self.assertIn(("http://example.org/concept/red", f"{SKOS}inScheme"), subjects_predicates) + self.assertIn(("http://example.org/concept/red", f"{SKOS}prefLabel"), subjects_predicates) + + @patch('semantica.triplet_store.blazegraph_store.BlazegraphStore') + def test_add_skos_concept_optional_fields(self, mock_bg): + """Optional fields produce extra triples.""" + store, _, mock_loader = self._make_store(mock_bg) + + store.add_skos_concept( + concept_uri="http://example.org/concept/red", + scheme_uri="http://example.org/vocab/colours", + pref_label="Red", + alt_labels=["Crimson", "Rouge"], + broader=["http://example.org/concept/colour"], + definition="The colour red.", + notation="RED", + ) + + triplets = mock_loader.load_triplets.call_args[0][0] + predicates = [t.predicate for t in triplets] + SKOS = self._SKOS + + self.assertIn(f"{SKOS}altLabel", predicates) + self.assertEqual(predicates.count(f"{SKOS}altLabel"), 2) + self.assertIn(f"{SKOS}broader", predicates) + self.assertIn(f"{SKOS}definition", predicates) + self.assertIn(f"{SKOS}notation", predicates) + + @patch('semantica.triplet_store.blazegraph_store.BlazegraphStore') + def test_add_skos_concept_scheme_triple_always_included(self, mock_bg): + """ConceptScheme rdf:type triple is always included even without optional args.""" + store, _, mock_loader = self._make_store(mock_bg) + + store.add_skos_concept( + concept_uri="http://example.org/concept/blue", + scheme_uri="http://example.org/vocab/colours", + pref_label="Blue", + ) + + triplets = mock_loader.load_triplets.call_args[0][0] + scheme_types = [ + t for t in triplets + if t.subject == "http://example.org/vocab/colours" + and t.predicate == self._RDF_TYPE + and t.object == f"{self._SKOS}ConceptScheme" + ] + self.assertEqual(len(scheme_types), 1) + + # --- get_skos_concepts --- + + @patch('semantica.triplet_store.blazegraph_store.BlazegraphStore') + def test_get_skos_concepts_all(self, mock_bg): + """get_skos_concepts returns all concepts when no scheme_uri given.""" + store, mock_backend, _ = self._make_store(mock_bg) + + from semantica.triplet_store.query_engine import QueryResult + mock_result = QueryResult( + bindings=[ + {"concept": {"value": "http://example.org/concept/red"}, + "prefLabel": {"value": "Red"}, + "altLabel": {"value": "Crimson"}, + "broader": None, "narrower": None, "related": None}, + {"concept": {"value": "http://example.org/concept/red"}, + "prefLabel": {"value": "Red"}, + "altLabel": {"value": "Rouge"}, + "broader": None, "narrower": None, "related": None}, + {"concept": {"value": "http://example.org/concept/blue"}, + "prefLabel": {"value": "Blue"}, + "altLabel": None, + "broader": None, "narrower": None, "related": None}, + ], + variables=["concept", "prefLabel", "altLabel"], + ) + mock_backend.execute_sparql.return_value = { + "bindings": mock_result.bindings, + "variables": mock_result.variables, + "metadata": {}, + } + + # Patch query_engine.execute_query to return mock_result directly + store.query_engine.execute_query = MagicMock(return_value=mock_result) + + concepts = store.get_skos_concepts() + self.assertEqual(len(concepts), 2) + + red = next(c for c in concepts if "red" in c["uri"]) + self.assertEqual(red["pref_label"], "Red") + self.assertIn("Crimson", red["alt_labels"]) + self.assertIn("Rouge", red["alt_labels"]) + + blue = next(c for c in concepts if "blue" in c["uri"]) + self.assertEqual(blue["alt_labels"], []) + + @patch('semantica.triplet_store.blazegraph_store.BlazegraphStore') + def test_get_skos_concepts_scheme_filter_in_query(self, mock_bg): + """When scheme_uri is given the scheme URI appears in the issued SPARQL.""" + store, _, _ = self._make_store(mock_bg) + + from semantica.triplet_store.query_engine import QueryResult + empty_result = QueryResult(bindings=[], variables=[]) + store.query_engine.execute_query = MagicMock(return_value=empty_result) + + store.get_skos_concepts(scheme_uri="http://example.org/vocab/colours") + + issued_sparql = store.query_engine.execute_query.call_args[0][0] + self.assertIn("http://example.org/vocab/colours", issued_sparql) + + @patch('semantica.triplet_store.blazegraph_store.BlazegraphStore') + def test_get_skos_concepts_empty_store(self, mock_bg): + """Returns empty list when no concepts exist.""" + store, _, _ = self._make_store(mock_bg) + from semantica.triplet_store.query_engine import QueryResult + store.query_engine.execute_query = MagicMock( + return_value=QueryResult(bindings=[], variables=[]) + ) + self.assertEqual(store.get_skos_concepts(), [])