diff --git a/semantica/explorer/routes/vocabulary.py b/semantica/explorer/routes/vocabulary.py index 867af5e9..64b60622 100644 --- a/semantica/explorer/routes/vocabulary.py +++ b/semantica/explorer/routes/vocabulary.py @@ -53,21 +53,20 @@ async def import_vocabulary( parse_format = "xml" if filename.endswith((".rdf", ".owl")) else "turtle" try: - nodes, edges = await asyncio.to_thread(parse_skos_file, content, parse_format) - - - added_nodes = await asyncio.to_thread(session.add_nodes, nodes) - added_edges = await asyncio.to_thread(session.add_edges, edges) - - return { - "status": "success", - "filename": filename, - "nodes_added": added_nodes, - "edges_added": added_edges - } - except Exception as exc: - return {"status": "error", "detail": str(exc)} + except ValueError as exc: + from fastapi import HTTPException + raise HTTPException(status_code=422, detail=str(exc)) + + added_nodes = await asyncio.to_thread(session.add_nodes, nodes) + added_edges = await asyncio.to_thread(session.add_edges, edges) + + return { + "status": "success", + "filename": filename, + "nodes_added": added_nodes, + "edges_added": added_edges, + } @router.get("/hierarchy", response_model=List[ConceptNode]) @@ -122,17 +121,21 @@ async def get_hierarchy( parent_to_children[src].append(tgt) has_parent.add(tgt) - # assemble final nested tree - roots = [] - for nid, node_obj in node_map.items(): - - child_ids = parent_to_children.get(nid, []) + # Assemble nested tree — cycle-safe via visited set. + def _attach_children(nid: str, visited: set) -> ConceptNode: + node_obj = node_map[nid] + child_ids = [c for c in parent_to_children.get(nid, []) if c not in visited] if child_ids: - node_obj.children = [node_map[cid] for cid in child_ids] + node_obj.children = [ + _attach_children(cid, visited | {nid}) for cid in child_ids + ] else: - node_obj.children = None # indicates a leaf node to the UI - - if nid not in has_parent: - roots.append(node_obj) + node_obj.children = None # leaf node signal for the UI + return node_obj + roots = [ + _attach_children(nid, {nid}) + for nid in node_map + if nid not in has_parent + ] return roots \ No newline at end of file diff --git a/semantica/explorer/utils/__init__.py b/semantica/explorer/utils/__init__.py new file mode 100644 index 00000000..8f9b9f56 --- /dev/null +++ b/semantica/explorer/utils/__init__.py @@ -0,0 +1 @@ +"""Utility helpers for the Semantica Knowledge Explorer.""" diff --git a/semantica/explorer/utils/rdf_parser.py b/semantica/explorer/utils/rdf_parser.py new file mode 100644 index 00000000..9ab45ea8 --- /dev/null +++ b/semantica/explorer/utils/rdf_parser.py @@ -0,0 +1,138 @@ +""" +RDF / SKOS parsing utility for the knowledge Explorer + +Parses `.ttl` and `.rdf` files, extracting skos:Concept and skos:ConceptScheme entities into flat dicts +compatible with ContextGraph. +""" + +from typing import Any, Dict, List, Tuple +import rdflib +from rdflib.namespace import RDF, RDFS, SKOS + +def _get_best_label(graph: rdflib.Graph, subject: rdflib.URIRef, predicate: rdflib.URIRef) -> str: + """ + Extracts the best available string label for a given predicate. + Prioritizes English tags ('en'), then untagged strings, then falls back to whatever + is available. Strips language tags in the process. + """ + + labels = list(graph.objects(subject, predicate)) + if not labels: + return "" + + # priority 1: English match exact + for lbl in labels: + if getattr(lbl, "language", None) == "en": + return str(lbl) + + # priority 2: English variants + for lbl in labels: + lang = getattr(lbl, "language", "") + if lang and lang.startswith("en"): + return str(lbl) + + # priority 3: No lang tag + for lbl in labels: + if getattr(lbl, "language", None) is None: + return str(lbl) + + # whatever is first if not any of the three above + return str(labels[0]) + +def _get_all_labels(graph: rdflib.Graph, subject: rdflib.URIRef, predicate: rdflib.URIRef) -> List[str]: + """ Returns a list of all string values for a predicate, stripping lang tags.""" + return list({str(lbl) for lbl in graph.objects(subject, predicate)}) + +def parse_skos_file(file_bytes: bytes, rdf_format: str = "turtle") -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: + """ + Parses RDF data and extracts SKOS concepts and relationships. + + Args: + file_bytes: The raw bytes of the uploaded file. + rdf_format: The rdflib parse format (e.g., "turtle" for .ttl, "xml" for .rdf). + + Returns: + A tuple of (nodes_list, edges_list) formatted for ContextGraph ingestion. + + Note: + Edges are only emitted when both endpoints exist in the parsed file. + Relationships pointing to external URIs not declared as skos:Concept or + skos:ConceptScheme (e.g. cross-vocabulary broader links) are silently dropped. + """ + + g = rdflib.Graph() + + try: + g.parse(data=file_bytes, format=rdf_format) + except Exception as e: + raise ValueError(f"Failed to parse RDF file as {rdf_format}. Ensure the file is valid. Details: {str(e)}") from e + + nodes_dict: Dict[str, Dict[str, Any]] = {} + edges: List[Dict[str, Any]] = [] + + # extract concept schemas + for scheme in g.subjects(RDF.type, SKOS.ConceptScheme): + uri = str(scheme) + + # if no prefLabel + + pref_label = _get_best_label(g, scheme, SKOS.prefLabel) + if not pref_label: + pref_label = uri.split("/")[-1].split("#")[-1] + + nodes_dict[uri] = { + "id": uri, + "type": "skos:ConceptScheme", + "properties": { + "content": pref_label, + "alt_labels": _get_all_labels(g, scheme, SKOS.altLabel), + "description": _get_best_label(g, scheme, SKOS.definition) + } + } + + # Extract concepts + for concept in g.subjects(RDF.type, SKOS.Concept): + uri = str(concept) + + pref_label = _get_best_label(g, concept, SKOS.prefLabel) + if not pref_label: + pref_label = uri.split("/")[-1].split("#")[-1] + + nodes_dict[uri] = { + "id": uri, + "type": "skos:Concept", + "properties": { + "content": pref_label, + "alt_labels": _get_all_labels(g, concept, SKOS.altLabel), + "description": _get_best_label(g, concept, SKOS.definition) + } + } + + + # Extract Relationships aka edges + + structural_preds = { + SKOS.broader: "skos:broader", + SKOS.narrower: "skos:narrower", + SKOS.inScheme: "skos:inScheme", + SKOS.related: "skos:related", + SKOS.topConceptOf: "skos:topConceptOf", + SKOS.hasTopConcept: "skos:hasTopConcept" + } + + for pred, edge_type in structural_preds.items(): + for source, target in g.subject_objects(pred): + # Only track edges where nodes were successfully extracted + if str(source) in nodes_dict and str(target) in nodes_dict: + edges.append({ + "source_id": str(source), + "target_id": str(target), + "type": edge_type, + "weight": 1.0, + "properties": {} + }) + + + return list(nodes_dict.values()), edges + + diff --git a/tests/explorer/test_rdf_parser.py b/tests/explorer/test_rdf_parser.py new file mode 100644 index 00000000..5483b02e --- /dev/null +++ b/tests/explorer/test_rdf_parser.py @@ -0,0 +1,424 @@ +""" +Tests for semantica/explorer/utils/rdf_parser.py + +Covers: +- parse_skos_file() with Turtle and RDF/XML formats +- ConceptScheme and Concept node extraction +- Label priority resolution (en > en-* > untagged > fallback) +- altLabel collection +- Structural edge extraction (broader/narrower/inScheme/related/topConceptOf/hasTopConcept) +- Edge filtering: edges with unknown endpoints are dropped +- Invalid bytes raises ValueError +- Empty graph returns empty lists +- _get_best_label and _get_all_labels helpers +""" + +import pytest +import rdflib +from rdflib.namespace import RDF, SKOS + +from semantica.explorer.utils.rdf_parser import ( + _get_all_labels, + _get_best_label, + parse_skos_file, +) + +# --------------------------------------------------------------------------- +# Sample TTL fixtures +# --------------------------------------------------------------------------- + +MINIMAL_TTL = b""" +@prefix skos: . +@prefix ex: . + +ex:Animals a skos:ConceptScheme ; + skos:prefLabel "Animals"@en . + +ex:Mammal a skos:Concept ; + skos:prefLabel "Mammal"@en ; + skos:inScheme ex:Animals . + +ex:Dog a skos:Concept ; + skos:prefLabel "Dog"@en ; + skos:broader ex:Mammal ; + skos:inScheme ex:Animals . +""" + +MULTILINGUAL_TTL = b""" +@prefix skos: . +@prefix ex: . + +ex:C1 a skos:Concept ; + skos:prefLabel "French Only"@fr ; + skos:prefLabel "English Label"@en ; + skos:prefLabel "British English"@en-GB ; + skos:altLabel "Alias One"@en ; + skos:altLabel "Alias Two"@en . +""" + +UNTAGGED_TTL = b""" +@prefix skos: . +@prefix ex: . + +ex:C2 a skos:Concept ; + skos:prefLabel "No Language Tag" ; + skos:altLabel "alt1" ; + skos:altLabel "alt2" . +""" + +FALLBACK_TTL = b""" +@prefix skos: . +@prefix ex: . + +ex:C3 a skos:Concept ; + skos:prefLabel "Nur Deutsch"@de . +""" + +ALL_EDGE_TYPES_TTL = b""" +@prefix skos: . +@prefix ex: . + +ex:S1 a skos:ConceptScheme ; + skos:prefLabel "Scheme One" . + +ex:A a skos:Concept ; + skos:prefLabel "A" ; + skos:inScheme ex:S1 ; + skos:topConceptOf ex:S1 . + +ex:B a skos:Concept ; + skos:prefLabel "B" ; + skos:broader ex:A ; + skos:inScheme ex:S1 . + +ex:C a skos:Concept ; + skos:prefLabel "C" ; + skos:related ex:B ; + skos:inScheme ex:S1 . + +ex:S1 skos:hasTopConcept ex:A . +""" + +# An edge pointing to an external URI not declared as a Concept/ConceptScheme +ORPHAN_EDGE_TTL = b""" +@prefix skos: . +@prefix ex: . + +ex:Known a skos:Concept ; + skos:prefLabel "Known" ; + skos:broader ex:ExternalConcept . +""" + +MINIMAL_RDF_XML = b""" + + + + Scheme X + + + + Concept Y + + + + +""" + + +# --------------------------------------------------------------------------- +# Helper: get node by URI +# --------------------------------------------------------------------------- + +def _node(nodes, uri): + return next((n for n in nodes if n["id"] == uri), None) + +def _edges_of_type(edges, edge_type): + return [e for e in edges if e["type"] == edge_type] + + +# --------------------------------------------------------------------------- +# parse_skos_file — basic extraction +# --------------------------------------------------------------------------- + +class TestParseSkosFileBasic: + def test_returns_tuple_of_two_lists(self): + nodes, edges = parse_skos_file(MINIMAL_TTL) + assert isinstance(nodes, list) + assert isinstance(edges, list) + + def test_extracts_concept_scheme(self): + nodes, _ = parse_skos_file(MINIMAL_TTL) + scheme = _node(nodes, "http://example.org/Animals") + assert scheme is not None + assert scheme["type"] == "skos:ConceptScheme" + assert scheme["properties"]["content"] == "Animals" + + def test_extracts_concepts(self): + nodes, _ = parse_skos_file(MINIMAL_TTL) + uris = {n["id"] for n in nodes} + assert "http://example.org/Mammal" in uris + assert "http://example.org/Dog" in uris + + def test_concept_type_tag(self): + nodes, _ = parse_skos_file(MINIMAL_TTL) + mammal = _node(nodes, "http://example.org/Mammal") + assert mammal["type"] == "skos:Concept" + + def test_node_has_required_keys(self): + nodes, _ = parse_skos_file(MINIMAL_TTL) + for n in nodes: + assert "id" in n + assert "type" in n + assert "properties" in n + assert "content" in n["properties"] + assert "alt_labels" in n["properties"] + assert "description" in n["properties"] + + def test_edge_has_required_keys(self): + _, edges = parse_skos_file(MINIMAL_TTL) + for e in edges: + assert "source_id" in e + assert "target_id" in e + assert "type" in e + assert "weight" in e + assert "properties" in e + + def test_edge_weight_default(self): + _, edges = parse_skos_file(MINIMAL_TTL) + assert all(e["weight"] == 1.0 for e in edges) + + +# --------------------------------------------------------------------------- +# parse_skos_file — label priority +# --------------------------------------------------------------------------- + +class TestLabelPriority: + def test_en_preferred_over_fr(self): + nodes, _ = parse_skos_file(MULTILINGUAL_TTL) + c1 = _node(nodes, "http://example.org/C1") + assert c1 is not None + assert c1["properties"]["content"] == "English Label" + + def test_untagged_used_when_no_en(self): + nodes, _ = parse_skos_file(UNTAGGED_TTL) + c2 = _node(nodes, "http://example.org/C2") + assert c2 is not None + assert c2["properties"]["content"] == "No Language Tag" + + def test_fallback_to_any_language(self): + nodes, _ = parse_skos_file(FALLBACK_TTL) + c3 = _node(nodes, "http://example.org/C3") + assert c3 is not None + assert c3["properties"]["content"] == "Nur Deutsch" + + def test_uri_fragment_used_when_no_pref_label(self): + ttl = b""" +@prefix skos: . +@prefix ex: . +ex:NoLabel a skos:Concept . +""" + nodes, _ = parse_skos_file(ttl) + n = _node(nodes, "http://example.org/NoLabel") + assert n is not None + assert n["properties"]["content"] == "NoLabel" + + +# --------------------------------------------------------------------------- +# parse_skos_file — altLabels +# --------------------------------------------------------------------------- + +class TestAltLabels: + def test_alt_labels_collected(self): + nodes, _ = parse_skos_file(MULTILINGUAL_TTL) + c1 = _node(nodes, "http://example.org/C1") + assert set(c1["properties"]["alt_labels"]) == {"Alias One", "Alias Two"} + + def test_alt_labels_empty_when_none(self): + nodes, _ = parse_skos_file(MINIMAL_TTL) + mammal = _node(nodes, "http://example.org/Mammal") + assert mammal["properties"]["alt_labels"] == [] + + def test_alt_labels_deduped(self): + ttl = b""" +@prefix skos: . +@prefix ex: . +ex:C a skos:Concept ; + skos:prefLabel "C" ; + skos:altLabel "same"@en ; + skos:altLabel "same"@en . +""" + nodes, _ = parse_skos_file(ttl) + c = _node(nodes, "http://example.org/C") + assert c["properties"]["alt_labels"].count("same") == 1 + + +# --------------------------------------------------------------------------- +# parse_skos_file — edge types +# --------------------------------------------------------------------------- + +class TestEdgeTypes: + def setup_method(self): + self.nodes, self.edges = parse_skos_file(ALL_EDGE_TYPES_TTL) + + def test_in_scheme_edges(self): + in_scheme = _edges_of_type(self.edges, "skos:inScheme") + assert len(in_scheme) >= 2 # A, B, C all inScheme S1 + + def test_broader_edge(self): + broader = _edges_of_type(self.edges, "skos:broader") + assert any( + e["source_id"] == "http://example.org/B" and + e["target_id"] == "http://example.org/A" + for e in broader + ) + + def test_related_edge(self): + related = _edges_of_type(self.edges, "skos:related") + assert any( + e["source_id"] == "http://example.org/C" and + e["target_id"] == "http://example.org/B" + for e in related + ) + + def test_top_concept_of_edge(self): + top_concept_of = _edges_of_type(self.edges, "skos:topConceptOf") + assert any( + e["source_id"] == "http://example.org/A" and + e["target_id"] == "http://example.org/S1" + for e in top_concept_of + ) + + def test_has_top_concept_edge(self): + has_top = _edges_of_type(self.edges, "skos:hasTopConcept") + assert any( + e["source_id"] == "http://example.org/S1" and + e["target_id"] == "http://example.org/A" + for e in has_top + ) + + +# --------------------------------------------------------------------------- +# parse_skos_file — edge filtering (orphan edges dropped) +# --------------------------------------------------------------------------- + +class TestOrphanEdgeFiltering: + def test_edge_to_external_uri_is_dropped(self): + nodes, edges = parse_skos_file(ORPHAN_EDGE_TTL) + # ex:ExternalConcept is not declared as a Concept/ConceptScheme + # so the broader edge should be dropped + assert len(edges) == 0 + + def test_known_node_is_still_extracted(self): + nodes, _ = parse_skos_file(ORPHAN_EDGE_TTL) + assert _node(nodes, "http://example.org/Known") is not None + + +# --------------------------------------------------------------------------- +# parse_skos_file — empty and error cases +# --------------------------------------------------------------------------- + +class TestEmptyAndErrors: + def test_empty_graph_returns_empty_lists(self): + empty_ttl = b"@prefix skos: .\n" + nodes, edges = parse_skos_file(empty_ttl) + assert nodes == [] + assert edges == [] + + def test_invalid_bytes_raises_value_error(self): + with pytest.raises(ValueError, match="Failed to parse RDF file"): + parse_skos_file(b"this is not valid turtle !!!!", rdf_format="turtle") + + def test_invalid_xml_raises_value_error(self): + with pytest.raises(ValueError, match="Failed to parse RDF file"): + parse_skos_file(b"", rdf_format="xml") + + +# --------------------------------------------------------------------------- +# parse_skos_file — RDF/XML format +# --------------------------------------------------------------------------- + +class TestRdfXmlFormat: + def test_parses_rdf_xml(self): + nodes, edges = parse_skos_file(MINIMAL_RDF_XML, rdf_format="xml") + uris = {n["id"] for n in nodes} + assert "http://example.org/SchemeX" in uris + assert "http://example.org/ConceptY" in uris + + def test_rdf_xml_scheme_type(self): + nodes, _ = parse_skos_file(MINIMAL_RDF_XML, rdf_format="xml") + scheme = _node(nodes, "http://example.org/SchemeX") + assert scheme["type"] == "skos:ConceptScheme" + assert scheme["properties"]["content"] == "Scheme X" + + def test_rdf_xml_in_scheme_edge(self): + _, edges = parse_skos_file(MINIMAL_RDF_XML, rdf_format="xml") + in_scheme = _edges_of_type(edges, "skos:inScheme") + assert any( + e["source_id"] == "http://example.org/ConceptY" and + e["target_id"] == "http://example.org/SchemeX" + for e in in_scheme + ) + + +# --------------------------------------------------------------------------- +# _get_best_label helper +# --------------------------------------------------------------------------- + +class TestGetBestLabel: + def _make_graph(self, triples_ttl: bytes) -> rdflib.Graph: + g = rdflib.Graph() + g.parse(data=triples_ttl, format="turtle") + return g + + def test_returns_en_when_available(self): + ttl = b""" +@prefix skos: . +@prefix ex: . +ex:X skos:prefLabel "English"@en ; + skos:prefLabel "Deutsch"@de . +""" + g = self._make_graph(ttl) + result = _get_best_label(g, rdflib.URIRef("http://example.org/X"), SKOS.prefLabel) + assert result == "English" + + def test_returns_empty_string_when_no_labels(self): + g = rdflib.Graph() + result = _get_best_label(g, rdflib.URIRef("http://example.org/X"), SKOS.prefLabel) + assert result == "" + + def test_en_variant_beats_untagged(self): + ttl = b""" +@prefix skos: . +@prefix ex: . +ex:X skos:prefLabel "No Tag" ; + skos:prefLabel "British"@en-GB . +""" + g = self._make_graph(ttl) + result = _get_best_label(g, rdflib.URIRef("http://example.org/X"), SKOS.prefLabel) + assert result == "British" + + +# --------------------------------------------------------------------------- +# _get_all_labels helper +# --------------------------------------------------------------------------- + +class TestGetAllLabels: + def test_returns_all_values(self): + ttl = b""" +@prefix skos: . +@prefix ex: . +ex:X skos:altLabel "A"@en ; + skos:altLabel "B"@fr ; + skos:altLabel "C" . +""" + g = rdflib.Graph() + g.parse(data=ttl, format="turtle") + result = _get_all_labels(g, rdflib.URIRef("http://example.org/X"), SKOS.altLabel) + assert set(result) == {"A", "B", "C"} + + def test_returns_empty_list_when_no_labels(self): + g = rdflib.Graph() + result = _get_all_labels(g, rdflib.URIRef("http://example.org/X"), SKOS.altLabel) + assert result == [] diff --git a/tests/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index 42d2f2d6..cf576767 100644 --- a/tests/explorer/test_vocabulary.py +++ b/tests/explorer/test_vocabulary.py @@ -7,12 +7,8 @@ Covers: - POST /api/vocabulary/import """ -import sys -from unittest.mock import MagicMock - -sys.modules['spacy'] = MagicMock() - import pytest +from unittest.mock import MagicMock, patch from fastapi import FastAPI from fastapi.testclient import TestClient @@ -20,22 +16,31 @@ from semantica.explorer.routes.vocabulary import router from semantica.explorer.dependencies import get_session +# --------------------------------------------------------------------------- +# App + dependency override setup +# --------------------------------------------------------------------------- + app = FastAPI() app.include_router(router) mock_session = MagicMock() -def override_get_session(): - return mock_session - -app.dependency_overrides[get_session] = override_get_session +app.dependency_overrides[get_session] = lambda: mock_session client = TestClient(app) -# Test cases -def test_list_schemes(): - """Test that /schemes correctly maps graph nodes to the Pydantic schema.""" +def setup_function(): + """Reset mock call history before each test to prevent state pollution.""" + mock_session.reset_mock() + + +# --------------------------------------------------------------------------- +# GET /api/vocabulary/schemes +# --------------------------------------------------------------------------- + +def test_list_schemes_returns_correct_shape(): + """Maps skos:ConceptScheme nodes to VocabularyScheme schema.""" mock_session.get_nodes.return_value = ([ { "id": "http://example.org/Scheme1", @@ -48,7 +53,7 @@ def test_list_schemes(): ], 1) response = client.get("/api/vocabulary/schemes") - + assert response.status_code == 200 data = response.json() assert len(data) == 1 @@ -57,65 +62,287 @@ def test_list_schemes(): assert data[0]["description"] == "A scheme for testing" -def test_get_hierarchy(): - """Test the O(V+E) in-memory tree building algorithm.""" +def test_list_schemes_empty_graph(): + """Returns empty list when no ConceptScheme nodes exist.""" + mock_session.get_nodes.return_value = ([], 0) + response = client.get("/api/vocabulary/schemes") + + assert response.status_code == 200 + assert response.json() == [] + + +def test_list_schemes_no_description(): + """Description field is optional — None when not present in properties.""" mock_session.get_nodes.return_value = ([ - {"id": "http://example.org/Parent", "type": "skos:Concept", "properties": {"content": "Parent Node"}}, - {"id": "http://example.org/Child", "type": "skos:Concept", "properties": {"content": "Child Node"}} + {"id": "http://example.org/S", "type": "skos:ConceptScheme", + "properties": {"content": "Minimal"}} + ], 1) + + response = client.get("/api/vocabulary/schemes") + + assert response.status_code == 200 + assert response.json()[0]["description"] is None + + +def test_list_schemes_metadata_envelope(): + """Label is read from 'metadata' envelope when 'properties' key absent.""" + mock_session.get_nodes.return_value = ([ + {"id": "http://example.org/S", "type": "skos:ConceptScheme", + "metadata": {"content": "Via Metadata"}} + ], 1) + + response = client.get("/api/vocabulary/schemes") + + assert response.status_code == 200 + assert response.json()[0]["label"] == "Via Metadata" + + +# --------------------------------------------------------------------------- +# GET /api/vocabulary/hierarchy +# --------------------------------------------------------------------------- + +def test_hierarchy_parent_child_via_broader(): + """broader edge: child → parent. Returns single root with one child.""" + mock_session.get_nodes.return_value = ([ + {"id": "http://example.org/Parent", "type": "skos:Concept", + "properties": {"content": "Parent Node"}}, + {"id": "http://example.org/Child", "type": "skos:Concept", + "properties": {"content": "Child Node"}} ], 2) - - mock_session.get_edges.return_value = ([ - {"source": "http://example.org/Parent", "target": "http://example.org/Scheme1", "type": "skos:inScheme"}, - - {"source": "http://example.org/Child", "target": "http://example.org/Scheme1", "type": "skos:inScheme"}, - - {"source": "http://example.org/Child", "target": "http://example.org/Parent", "type": "skos:broader"} - ], 3) + {"source": "http://example.org/Parent", "target": "http://example.org/Scheme1", + "type": "skos:inScheme"}, + {"source": "http://example.org/Child", "target": "http://example.org/Scheme1", + "type": "skos:inScheme"}, + {"source": "http://example.org/Child", "target": "http://example.org/Parent", + "type": "skos:broader"}, + ], 3) response = client.get("/api/vocabulary/hierarchy?scheme=http://example.org/Scheme1") - + assert response.status_code == 200 data = response.json() - - assert len(data) == 1 root = data[0] assert root["uri"] == "http://example.org/Parent" assert root["pref_label"] == "Parent Node" - assert len(root["children"]) == 1 child = root["children"][0] assert child["uri"] == "http://example.org/Child" assert child["pref_label"] == "Child Node" - assert child["children"] is None -def test_import_vocabulary(): - """Test the file upload endpoint safely parses and calls add_nodes/add_edges.""" +def test_hierarchy_parent_child_via_narrower(): + """narrower edge: parent → child. Same tree as broader, different edge direction.""" + mock_session.get_nodes.return_value = ([ + {"id": "http://example.org/P", "type": "skos:Concept", + "properties": {"content": "P"}}, + {"id": "http://example.org/C", "type": "skos:Concept", + "properties": {"content": "C"}} + ], 2) + mock_session.get_edges.return_value = ([ + {"source": "http://example.org/P", "target": "http://example.org/S", + "type": "skos:inScheme"}, + {"source": "http://example.org/C", "target": "http://example.org/S", + "type": "skos:inScheme"}, + # narrower: P → C means C is a child of P + {"source": "http://example.org/P", "target": "http://example.org/C", + "type": "skos:narrower"}, + ], 3) - minimal_ttl = b""" - @prefix skos: . - @prefix ex: . - ex:S a skos:ConceptScheme ; skos:prefLabel "S" . - """ - + response = client.get("/api/vocabulary/hierarchy?scheme=http://example.org/S") + + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + assert data[0]["uri"] == "http://example.org/P" + assert len(data[0]["children"]) == 1 + assert data[0]["children"][0]["uri"] == "http://example.org/C" + + +def test_hierarchy_membership_via_top_concept_of(): + """topConceptOf edge includes node in scheme without inScheme edge.""" + mock_session.get_nodes.return_value = ([ + {"id": "http://example.org/Top", "type": "skos:Concept", + "properties": {"content": "Top"}} + ], 1) + mock_session.get_edges.return_value = ([ + {"source": "http://example.org/Top", "target": "http://example.org/S", + "type": "skos:topConceptOf"}, + ], 1) + + response = client.get("/api/vocabulary/hierarchy?scheme=http://example.org/S") + + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + assert data[0]["uri"] == "http://example.org/Top" + + +def test_hierarchy_membership_via_has_top_concept(): + """hasTopConcept edge (scheme → concept) includes the target concept.""" + mock_session.get_nodes.return_value = ([ + {"id": "http://example.org/TC", "type": "skos:Concept", + "properties": {"content": "TopConcept"}} + ], 1) + mock_session.get_edges.return_value = ([ + {"source": "http://example.org/S", "target": "http://example.org/TC", + "type": "skos:hasTopConcept"}, + ], 1) + + response = client.get("/api/vocabulary/hierarchy?scheme=http://example.org/S") + + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + assert data[0]["uri"] == "http://example.org/TC" + + +def test_hierarchy_empty_scheme(): + """No concepts in scheme returns empty list.""" + mock_session.get_nodes.return_value = ([], 0) + mock_session.get_edges.return_value = ([], 0) + + response = client.get("/api/vocabulary/hierarchy?scheme=http://example.org/Empty") + + assert response.status_code == 200 + assert response.json() == [] + + +def test_hierarchy_flat_scheme_all_roots(): + """All concepts without parent relationships are returned as roots.""" + mock_session.get_nodes.return_value = ([ + {"id": "http://example.org/A", "type": "skos:Concept", + "properties": {"content": "A"}}, + {"id": "http://example.org/B", "type": "skos:Concept", + "properties": {"content": "B"}}, + ], 2) + mock_session.get_edges.return_value = ([ + {"source": "http://example.org/A", "target": "http://example.org/S", + "type": "skos:inScheme"}, + {"source": "http://example.org/B", "target": "http://example.org/S", + "type": "skos:inScheme"}, + ], 2) + + response = client.get("/api/vocabulary/hierarchy?scheme=http://example.org/S") + + assert response.status_code == 200 + data = response.json() + assert len(data) == 2 + uris = {n["uri"] for n in data} + assert uris == {"http://example.org/A", "http://example.org/B"} + + +def test_hierarchy_missing_scheme_param(): + """scheme query param is required — returns 422 when omitted.""" + response = client.get("/api/vocabulary/hierarchy") + assert response.status_code == 422 + + +def test_hierarchy_cycle_does_not_hang(): + """Cyclic broader edges must not cause infinite recursion during serialization.""" + mock_session.get_nodes.return_value = ([ + {"id": "http://example.org/A", "type": "skos:Concept", + "properties": {"content": "A"}}, + {"id": "http://example.org/B", "type": "skos:Concept", + "properties": {"content": "B"}}, + ], 2) + mock_session.get_edges.return_value = ([ + {"source": "http://example.org/A", "target": "http://example.org/S", + "type": "skos:inScheme"}, + {"source": "http://example.org/B", "target": "http://example.org/S", + "type": "skos:inScheme"}, + # Cycle: A broader B AND B broader A + {"source": "http://example.org/A", "target": "http://example.org/B", + "type": "skos:broader"}, + {"source": "http://example.org/B", "target": "http://example.org/A", + "type": "skos:broader"}, + ], 4) + + response = client.get("/api/vocabulary/hierarchy?scheme=http://example.org/S") + + # Must return 200 without hanging or raising a RecursionError + assert response.status_code == 200 + data = response.json() + assert isinstance(data, list) + + +# --------------------------------------------------------------------------- +# POST /api/vocabulary/import +# --------------------------------------------------------------------------- + +MINIMAL_TTL = b""" +@prefix skos: . +@prefix ex: . +ex:S a skos:ConceptScheme ; skos:prefLabel "S" . +""" + +MINIMAL_RDF_XML = b""" + + + Scheme X + + +""" + + +def test_import_ttl_success(): + """Valid .ttl upload returns success and calls add_nodes/add_edges.""" mock_session.add_nodes.return_value = 1 mock_session.add_edges.return_value = 0 response = client.post( "/api/vocabulary/import", - files={"file": ("test.ttl", minimal_ttl, "text/turtle")} + files={"file": ("vocab.ttl", MINIMAL_TTL, "text/turtle")}, ) - + assert response.status_code == 200 data = response.json() - assert data["status"] == "success" + assert data["filename"] == "vocab.ttl" assert data["nodes_added"] == 1 assert data["edges_added"] == 0 - mock_session.add_nodes.assert_called_once() - mock_session.add_edges.assert_called_once() \ No newline at end of file + mock_session.add_edges.assert_called_once() + + +def test_import_rdf_xml_success(): + """.rdf extension triggers XML format path.""" + mock_session.add_nodes.return_value = 1 + mock_session.add_edges.return_value = 0 + + response = client.post( + "/api/vocabulary/import", + files={"file": ("vocab.rdf", MINIMAL_RDF_XML, "application/rdf+xml")}, + ) + + assert response.status_code == 200 + assert response.json()["status"] == "success" + + +def test_import_invalid_file_returns_422(): + """Unparseable file content returns HTTP 422, not a silent 200 error dict.""" + response = client.post( + "/api/vocabulary/import", + files={"file": ("bad.ttl", b"this is not valid RDF!", "text/turtle")}, + ) + + assert response.status_code == 422 + + +def test_import_owl_extension_uses_xml_format(): + """.owl extension treated the same as .rdf — uses XML parser.""" + mock_session.add_nodes.return_value = 1 + mock_session.add_edges.return_value = 0 + + response = client.post( + "/api/vocabulary/import", + files={"file": ("onto.owl", MINIMAL_RDF_XML, "application/rdf+xml")}, + ) + + assert response.status_code == 200 + assert response.json()["status"] == "success"