Merge pull request #805 from Sameer6305/fix/773-sparql-test-coverage

test(explorer): add coverage for SPARQL route (#773)

CHANGELOG.md

@@ -11,6 +11,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Comprehensive unit and security test suite for the `/api/sparql` Explorer route** (#773) by @Sameer6305
+  - Added `tests/explorer/test_sparql_route.py` (34 tests) covering the SPARQL Explorer route (`semantica/explorer/routes/sparql.py`), which executes arbitrary SPARQL queries against an in-memory rdflib projection of the live graph and previously had zero test coverage
+  - Verified read-only allowlist enforcement against write and mutation queries (`INSERT DATA`, `DELETE DATA`, `DELETE WHERE`, `DROP ALL`, `CLEAR ALL`, `LOAD`, `CREATE GRAPH`, `MODIFY`, comments, and multi-statement injections like `SELECT ... ; DROP ALL`), confirming rejected queries short-circuit before any graph is built or queried
+  - Verified resource-limiting behavior, confirming row capping (`_SPARQL_MAX_ROWS`) truncates results and sets `truncated: true`, query timeout (`_SPARQL_TIMEOUT_S`) returns a clean error message without crashing, and concurrency semaphore (`_SPARQL_MAX_CONCURRENT`) prevents thread starvation under load
+  - Verified RDF projection fidelity for node properties and edge relationships, and error formatting for malformed SPARQL syntax with line and column extraction
+  - Follow-up review fixes (#805): extracted the duplicated row-cap-and-truncate loop (previously copy-pasted between the `CONSTRUCT`/`DESCRIBE` and `SELECT` branches) into a shared `_cap_rows()` helper so the `_SPARQL_MAX_ROWS` cap is enforced identically by both; added `test_row_cap_truncates_construct_results`, since the truncation path for `CONSTRUCT`/`DESCRIBE` results had no direct test coverage even though `SELECT` truncation did
+
 - **Global default persistent storage for `ProvenanceManager`, plus a working `provenance` CLI** (#795, #802) by @Sameer6305 and @KaifAhmad1
   - Every ingestion/processing module (`kg_provenance.py`, `pipeline_provenance.py`, and 20+ other call sites) instantiated its own `ProvenanceManager()` with no `storage_path`, so all of them silently fell back to `InMemoryStorage` and the SQLite audit trail was never actually written. `ProvenanceManager.set_default_storage_path(path)` now sets a class-level default that every no-arg instantiation picks up, and `Semantica.__init__` wires `config.provenance.storage_path` into it automatically during orchestrator init
   - Added the thread-safe `default_storage_path(path)` context manager (`semantica.provenance.default_storage_path`) for test isolation — it stacks nested overrides and guarantees restoration of the previous default on exit, even on exception, so tests can't leak global state into each other

docs/reference/explorer.md

@@ -262,7 +262,7 @@ Full interactive docs at `http://localhost:8000/docs`. All endpoints accept and
 
     | Endpoint | Method | Description |
     | :-------- | :------ | :----------- |
-    | `/api/sparql` | `POST` | Execute a SPARQL SELECT or ASK query |
+    | `/api/sparql` | `POST` | Execute a read-only SPARQL query (`SELECT`, `ASK`, `CONSTRUCT`, or `DESCRIBE`); `CONSTRUCT`/`DESCRIBE` return triples as `subject`, `predicate`, `object` columns, and `ASK` returns a `result` boolean column |
 
   </Accordion>
   <Accordion title="Decisions, Provenance, Annotations & Export">

semantica/explorer/routes/sparql.py

@@ -1,10 +1,20 @@
-"""
+"""
 SPARQL routes backed by an in-memory rdflib projection of the current graph.
+
+Security contract
+-----------------
+* Only SELECT, ASK, CONSTRUCT, and DESCRIBE are accepted (allowlist enforced
+  before graph construction so rejected queries never touch the session).
+* Multi-statement injections that start with an allowed keyword (e.g.
+  ``SELECT ... ; DROP ALL``) pass the prefix check and reach rdflib, which
+  rejects non-SELECT/ASK/CONSTRUCT/DESCRIBE update syntax in the parser.
+* The in-memory rdflib graph is a read-only projection — the live
+  ``GraphSession`` is never mutated by this route.
 """
 
 import asyncio
 import re
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import rdflib
 from fastapi import APIRouter, Depends
@@ -75,6 +85,9 @@ def _build_rdflib_graph(session: GraphSession) -> rdflib.Graph:
     return graph
 
 
+# ---------------------------------------------------------------------------
+# Resource limits (override in tests via patch.object)
+# ---------------------------------------------------------------------------
 _SPARQL_MAX_ROWS = 5_000     # hard cap on returned rows
 _SPARQL_TIMEOUT_S = 30       # seconds before abandoning the await
 _SPARQL_MAX_CONCURRENT = 4   # semaphore: max simultaneous executions
@@ -85,6 +98,22 @@ _SPARQL_MAX_CONCURRENT = 4   # semaphore: max simultaneous executions
 _sparql_semaphore = asyncio.Semaphore(_SPARQL_MAX_CONCURRENT)
 
 
+def _cap_rows(items: Any, row_builder) -> Tuple[List[Dict[str, Any]], bool]:
+    """Materialize up to ``_SPARQL_MAX_ROWS`` items via ``row_builder``, reporting truncation.
+
+    Shared by the SELECT and CONSTRUCT/DESCRIBE branches below so the row cap
+    is enforced identically regardless of query type.
+    """
+    rows: List[Dict[str, Any]] = []
+    truncated = False
+    for item in items:
+        if len(rows) >= _SPARQL_MAX_ROWS:
+            truncated = True
+            break
+        rows.append(row_builder(item))
+    return rows, truncated
+
+
 @router.post("", response_model=SparqlResponse)
 async def execute_sparql(
     req: SparqlRequest,
@@ -126,16 +155,38 @@ async def execute_sparql(
                 error_column=int(column_match.group(1)) if column_match else None,
             )
 
-    columns = [str(var) for var in query_results.vars] if query_results.vars else []
-    rows: List[Dict[str, Any]] = []
-    for row in query_results:
-        if len(rows) >= _SPARQL_MAX_ROWS:
-            break
-        row_data = {}
-        for index, column in enumerate(columns):
-            value = row[index]
-            row_data[column] = str(value) if value is not None else None
-        rows.append(row_data)
+    # ---------------------------------------------------------------------------
+    # Serialize results into a type-aware tabular representation.
+    # ASK      → single row: {"result": "true"|"false"}
+    # CONSTRUCT/DESCRIBE → rows of {"subject", "predicate", "object"} triples
+    # SELECT   → rows keyed by projected variable names
+    # ---------------------------------------------------------------------------
+    query_type: str = query_results.type  # always set by rdflib
+
+    if query_type == "ASK":
+        columns = ["result"]
+        rows = [{"result": "true" if query_results.askAnswer else "false"}]
+        truncated = False
+
+    elif query_type in ("CONSTRUCT", "DESCRIBE"):
+        columns = ["subject", "predicate", "object"]
+        rows, truncated = _cap_rows(
+            query_results,  # rdflib always yields (s, p, o) 3-tuples
+            lambda triple: {
+                "subject": str(triple[0]),
+                "predicate": str(triple[1]),
+                "object": str(triple[2]),
+            },
+        )
+
+    else:  # SELECT
+        columns = [str(var) for var in (query_results.vars or [])]
+        rows, truncated = _cap_rows(
+            query_results,
+            lambda row: {
+                column: (str(row[index]) if row[index] is not None else None)
+                for index, column in enumerate(columns)
+            },
+        )
 
-    truncated = len(rows) == _SPARQL_MAX_ROWS
     return SparqlResponse(columns=columns, rows=rows, total=len(rows), truncated=truncated)

tests/explorer/test_sparql_route.py

@@ -0,0 +1,346 @@
+"""Tests for the SPARQL Explorer route (semantica/explorer/routes/sparql.py).
+
+Filed as #773: this route executes arbitrary SPARQL against an in-memory
+rdflib projection of the live graph and had zero test coverage anywhere in
+the repo. Coverage here focuses on the two things that matter most for a
+query-execution surface: (1) the read-only allowlist can't be bypassed, and
+(2) the resource-limiting behavior (row cap, timeout) actually engages.
+"""
+
+import asyncio
+import concurrent.futures
+from unittest.mock import patch
+
+import pytest
+
+from semantica.context.context_graph import ContextGraph
+from semantica.explorer.app import create_app
+from semantica.explorer.session import GraphSession
+
+try:
+    from starlette.testclient import TestClient
+except ImportError:
+    pytest.skip(
+        "starlette TestClient is required for explorer tests. Install semantica[explorer].",
+        allow_module_level=True,
+    )
+
+import semantica.explorer.routes.sparql as sparql_mod
+
+
+def _build_sample_graph() -> ContextGraph:
+    graph = ContextGraph(advanced_analytics=False)
+    graph.add_node(
+        "python",
+        node_type="language",
+        content="Python programming language",
+        popularity="high",
+    )
+    graph.add_node("javascript", node_type="language", content="JavaScript programming language")
+    graph.add_node("web_dev", node_type="concept", content="Web Development")
+    graph.add_edge("python", "web_dev", edge_type="used_in", weight=0.5)
+    return graph
+
+
+@pytest.fixture(scope="module")
+def client():
+    session = GraphSession(_build_sample_graph())
+    app = create_app(session=session)
+    with TestClient(app) as test_client:
+        yield test_client
+
+
+def _post(client, query):
+    return client.post("/api/sparql", json={"query": query})
+
+
+# ---------------------------------------------------------------------------
+# Happy-path query types
+# ---------------------------------------------------------------------------
+
+def test_select_returns_expected_columns_and_rows(client):
+    resp = _post(client, "SELECT ?s ?label WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?label }")
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+    assert set(payload["columns"]) == {"s", "label"}
+    assert payload["total"] == len(payload["rows"])
+    labels = {row["label"] for row in payload["rows"]}
+    assert "Python programming language" in labels
+    assert "JavaScript programming language" in labels
+
+
+def test_ask_query_returns_boolean_like_result(client):
+    resp = _post(
+        client,
+        "ASK { ?s <http://www.w3.org/2000/01/rdf-schema#label> \"Python programming language\" }",
+    )
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+    assert payload["columns"] == ["result"]
+    assert payload["rows"] == [{"result": "true"}]
+    assert payload["total"] == 1
+
+
+def test_construct_query_succeeds(client):
+    resp = _post(
+        client,
+        "CONSTRUCT { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?label } "
+        "WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?label }",
+    )
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+    assert payload["columns"] == ["subject", "predicate", "object"]
+    assert payload["total"] > 0
+    assert any(
+        row["subject"] == "http://semantica.local/entity/python"
+        and row["predicate"] == "http://www.w3.org/2000/01/rdf-schema#label"
+        and row["object"] == "Python programming language"
+        for row in payload["rows"]
+    )
+
+
+def test_describe_query_succeeds(client):
+    resp = _post(client, "DESCRIBE <http://semantica.local/entity/python>")
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+    assert payload["columns"] == ["subject", "predicate", "object"]
+    assert payload["total"] > 0
+    assert any(
+        row["subject"] == "http://semantica.local/entity/python"
+        and row["predicate"] == "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"
+        and row["object"] == "http://semantica.local/entity/language"
+        for row in payload["rows"]
+    )
+
+
+def test_lowercase_query_keyword_is_accepted(client):
+    """The allowlist regex is case-insensitive; confirm lowercase keywords work too."""
+    resp = _post(client, "select ?s where { ?s a <http://semantica.local/entity/language> }")
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+
+
+def test_select_with_no_results_returns_empty_rows(client):
+    resp = _post(client, "SELECT ?s WHERE { ?s a <http://semantica.local/entity/nonexistent> }")
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+    assert payload["rows"] == []
+    assert payload["total"] == 0
+
+
+# ---------------------------------------------------------------------------
+# Read-only allowlist: this is the security-relevant surface (#773's core risk)
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize(
+    "query",
+    [
+        "INSERT DATA { <http://semantica.local/entity/x> a <http://semantica.local/entity/hacked> }",
+        "DELETE DATA { <http://semantica.local/entity/python> a <http://semantica.local/entity/language> }",
+        "DELETE WHERE { ?s ?p ?o }",
+        "DROP ALL",
+        "DROP GRAPH <http://example.org/g>",
+        "CLEAR ALL",
+        "CLEAR GRAPH <http://example.org/g>",
+        "LOAD <http://example.org/data.ttl>",
+        "CREATE GRAPH <http://example.org/g>",
+        "MODIFY <http://example.org/g> DELETE { ?s ?p ?o } WHERE { ?s ?p ?o }",
+        "",
+        "   ",
+        "not a sparql query at all",
+        # A write statement smuggled after a comment prefix fails prefix-matching
+        "# comment\nDROP ALL",
+    ],
+)
+def test_write_and_non_read_queries_are_rejected_by_allowlist(client, query):
+    resp = _post(client, query)
+    assert resp.status_code == 200, "rejection is a normal 200 response with an error field, not an HTTP error"
+    payload = resp.json()
+    assert payload["error"] is not None
+    assert payload["rows"] == []
+    assert payload["columns"] == []
+    assert payload["total"] == 0
+
+
+def test_multi_statement_injection_is_rejected_by_parser(client):
+    """A multi-statement injection starting with SELECT passes the prefix
+    allowlist check but is rejected by rdflib's SPARQL parser as invalid
+    query syntax, preventing any mutation or secondary execution."""
+    resp = _post(client, "SELECT ?s WHERE { ?s ?p ?o } ; DROP ALL")
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is not None
+    assert payload["rows"] == []
+    assert payload["columns"] == []
+    assert payload["total"] == 0
+
+
+@pytest.mark.parametrize(
+    "query",
+    [
+        "DROP ALL",
+        "INSERT DATA { <http://semantica.local/entity/x> a <http://semantica.local/entity/y> }",
+        "# comment\nDROP ALL",
+        "",
+    ],
+)
+def test_allowlist_rejected_query_never_touches_the_graph(client, query):
+    """An input failing _is_read_only_query must short-circuit before any graph is built/queried."""
+    with patch.object(sparql_mod, "_build_rdflib_graph") as mock_build:
+        resp = _post(client, query)
+    assert resp.status_code == 200
+    assert resp.json()["error"] is not None
+    mock_build.assert_not_called()
+
+
+def test_multi_statement_injection_reaches_graph_but_fails_in_parser(client):
+    """Confirms the distinction between allowlist rejection and parser rejection:
+    a string starting with SELECT passes _is_read_only_query and builds a graph,
+    but rdflib.Graph.query() rejects the trailing '; DROP ALL' syntax."""
+    with patch.object(
+        sparql_mod, "_build_rdflib_graph", wraps=sparql_mod._build_rdflib_graph
+    ) as spy_build:
+        resp = _post(client, "SELECT ?s WHERE { ?s ?p ?o } ; DROP ALL")
+    assert resp.status_code == 200
+    assert resp.json()["error"] is not None
+    spy_build.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Error handling for malformed queries
+# ---------------------------------------------------------------------------
+
+def test_malformed_query_returns_error_without_crashing(client):
+    resp = _post(client, "SELECT ?s WHERE { this is not valid sparql syntax")
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is not None
+    assert payload["rows"] == []
+    assert payload["total"] == 0
+
+
+def test_malformed_query_error_includes_line_and_column_when_present(client):
+    resp = _post(client, "SELECT ?s WHERE { $$$ invalid $$$ }")
+    payload = resp.json()
+    assert payload["error"] is not None
+    # error_line/error_column are best-effort extraction from the pyparsing
+    # error message; they may be None depending on rdflib's error text, but
+    # the fields must always be present and of the right type when set.
+    assert payload["error_line"] is None or isinstance(payload["error_line"], int)
+    assert payload["error_column"] is None or isinstance(payload["error_column"], int)
+
+
+# ---------------------------------------------------------------------------
+# Resource limits: row cap and timeout
+# ---------------------------------------------------------------------------
+
+def test_row_cap_truncates_results(client):
+    with patch.object(sparql_mod, "_SPARQL_MAX_ROWS", 1):
+        resp = _post(
+            client,
+            "SELECT ?s WHERE { ?s a <http://semantica.local/entity/language> }",
+        )
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+    assert len(payload["rows"]) == 1
+    assert payload["total"] == 1
+    assert payload["truncated"] is True
+
+
+def test_result_below_cap_is_not_marked_truncated(client):
+    resp = _post(client, "SELECT ?s WHERE { ?s a <http://semantica.local/entity/language> }")
+    payload = resp.json()
+    assert payload["truncated"] is False
+
+
+def test_row_cap_truncates_construct_results(client):
+    """CONSTRUCT/DESCRIBE share _cap_rows with SELECT; confirm the cap applies there too."""
+    with patch.object(sparql_mod, "_SPARQL_MAX_ROWS", 1):
+        resp = _post(
+            client,
+            "CONSTRUCT { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?label } "
+            "WHERE { ?s <http://www.w3.org/2000/01/rdf-schema#label> ?label }",
+        )
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is None
+    assert len(payload["rows"]) == 1
+    assert payload["total"] == 1
+    assert payload["truncated"] is True
+
+
+def test_query_timeout_returns_clean_error_not_a_crash(client):
+    async def _raise_timeout(coro, timeout=None):
+        coro.close()  # avoid a 'coroutine was never awaited' warning from the mock
+        raise asyncio.TimeoutError()
+
+    with patch.object(sparql_mod.asyncio, "wait_for", side_effect=_raise_timeout):
+        resp = _post(client, "SELECT ?s WHERE { ?s a <http://semantica.local/entity/language> }")
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["error"] is not None
+    assert "timed out" in payload["error"].lower()
+    assert payload["rows"] == []
+
+
+# ---------------------------------------------------------------------------
+# Data-mapping fidelity: does the graph->RDF projection reflect session state?
+# ---------------------------------------------------------------------------
+
+def test_node_properties_are_projected_as_literals_excluding_reserved_keys(client):
+    resp = _post(
+        client,
+        "SELECT ?p ?v WHERE { <http://semantica.local/entity/python> ?p ?v }",
+    )
+    payload = resp.json()
+    predicates = {row["p"] for row in payload["rows"]}
+    # 'popularity' is a real property and should be projected.
+    assert any("popularity" in p for p in predicates)
+    # content/valid_from/valid_until are excluded from prop: projection
+    # (content instead becomes rdfs:label, handled separately).
+    assert not any(p.endswith("/prop/content") for p in predicates)
+
+
+def test_edges_are_projected_with_their_relationship_type(client):
+    resp = _post(
+        client,
+        "SELECT ?o WHERE { <http://semantica.local/entity/python> "
+        "<http://semantica.local/prop/used_in> ?o }",
+    )
+    payload = resp.json()
+    assert payload["error"] is None
+    assert any("web_dev" in row["o"] for row in payload["rows"])
+
+
+def test_concurrent_requests_all_complete_successfully(client):
+    """Basic smoke test that the concurrency semaphore doesn't deadlock or
+    drop requests under light concurrent load."""
+    import concurrent.futures
+
+    def _run():
+        return _post(client, "SELECT ?s WHERE { ?s a <http://semantica.local/entity/language> }")
+
+    pool = concurrent.futures.ThreadPoolExecutor(max_workers=6)
+    try:
+        futures = [pool.submit(_run) for _ in range(6)]
+        results = []
+        for idx, fut in enumerate(futures):
+            try:
+                results.append(fut.result(timeout=10.0))
+            except concurrent.futures.TimeoutError:
+                pytest.fail(
+                    f"Concurrent SPARQL query #{idx} deadlocked or timed out after 10.0s"
+                )
+    finally:
+        pool.shutdown(wait=False, cancel_futures=True)
+
+    for resp in results:
+        assert resp.status_code == 200
+        assert resp.json()["error"] is None