diff --git a/.gitignore b/.gitignore
index ed5349ee..0e5128d7 100644
Binary files a/.gitignore and b/.gitignore differ
diff --git a/docs/reference/pipeline.md b/docs/reference/pipeline.md
index 56218391..de450ed2 100644
--- a/docs/reference/pipeline.md
+++ b/docs/reference/pipeline.md
@@ -495,6 +495,40 @@ result = engine.execute_pipeline(
Delta detection uses SHA-256 checksums on source content. Only sources whose checksum differs from `base_version_id` are passed to downstream steps. For pipelines that run hourly or daily against a growing corpus, delta mode eliminates redundant re-embedding and re-extraction.
+## SPARQL CONSTRUCT Template Steps
+
+Use the `"construct_template"` step type to render and execute a [SPARQL CONSTRUCT template](triplet_store#sparql-construct-templates) as part of a pipeline. `store_backend` and `construct_template_registry` are execution-time resources, not step config — pass them to `execute_pipeline()`, the same way `delta_mode` steps receive `version_manager` and `triplet_store`:
+
+```python
+from semantica.pipeline import PipelineBuilder, ExecutionEngine
+from semantica.triplet_store.construct_templates import construct_template_step_handler
+
+builder = PipelineBuilder()
+builder.add_step(
+ "apply_person_template",
+ "construct_template",
+ handler=construct_template_step_handler,
+ template_name="person_to_foaf",
+ params={"subject": "http://ex.org/p1", "name": "Alice", "age": 30},
+ target_graph="http://ex.org/graphs/people",
+)
+pipeline = builder.build("person_pipeline")
+
+engine = ExecutionEngine()
+result = engine.execute_pipeline(
+ pipeline,
+ data=None,
+ store_backend=store, # required: a BlazegraphStore instance
+ construct_template_registry=registry, # required: holds the registered template
+)
+
+triplets = result.output # List[Triplet], already persisted via store.add_triplets
+```
+
+
+ `construct_template` steps raise `ProcessingError` if `store_backend` or `construct_template_registry` is missing from `execute_pipeline()`'s options, and `ValidationError` if `template_name` isn't registered.
+
+
## Schemas
diff --git a/docs/reference/triplet_store.md b/docs/reference/triplet_store.md
index cb74bda0..97a5336e 100644
--- a/docs/reference/triplet_store.md
+++ b/docs/reference/triplet_store.md
@@ -277,6 +277,65 @@ store.execute_query("""
**`execute_query()` returns `QueryResult`, not a list.** Iterate `result.bindings`, not `result` directly. Each binding is a dict mapping variable name → `{"value": ..., "type": ...}`.
+## SPARQL CONSTRUCT Templates
+
+`semantica.triplet_store.construct_templates` provides parameterized SPARQL `CONSTRUCT` query templates: define a reusable query once, substitute typed parameters safely, and persist the resulting triples in one call. This is available for the **Blazegraph backend only** (see [Backends](#backends) above) — `BlazegraphStore.execute_sparql()` is the only backend with CONSTRUCT-aware RDF parsing.
+
+```python
+from semantica.triplet_store.construct_templates import (
+ ConstructTemplate,
+ ParameterDescriptor,
+ ConstructTemplateRegistry,
+ render_construct_template,
+ execute_construct_template,
+)
+from semantica.triplet_store import BlazegraphStore
+
+# Define and register a template
+template = ConstructTemplate(
+ name="person_to_foaf",
+ description="Maps a person record subject to a foaf:name triple",
+ construct_query="""
+ PREFIX foaf:
+ CONSTRUCT { <{{subject}}> foaf:name {{name}} ; foaf:age {{age}} }
+ WHERE { <{{subject}}> a }
+ """,
+ parameters=[
+ ParameterDescriptor(name="subject", type="uri", required=True),
+ ParameterDescriptor(name="name", type="literal", required=True),
+ ParameterDescriptor(
+ name="age", type="typed-literal", required=False, default=0,
+ datatype="xsd:integer",
+ ),
+ ],
+)
+
+registry = ConstructTemplateRegistry()
+registry.register(template)
+
+# Render only: inspect the substituted SPARQL string, no network call
+sparql = render_construct_template(
+ registry.get("person_to_foaf"),
+ params={"subject": "http://ex.org/p1", "name": "Alice", "age": 30},
+)
+
+# Render + execute + persist in one call
+store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph", namespace="kb")
+triplets = execute_construct_template(
+ template=registry.get("person_to_foaf"),
+ params={"subject": "http://ex.org/p1", "name": "Alice", "age": 30},
+ store_backend=store,
+ target_graph="http://ex.org/graphs/people",
+)
+# triplets: List[Triplet], already persisted via store.add_triplets
+```
+
+Each `ParameterDescriptor.type` controls how its value is rendered: `"uri"` values are validated against an allowlist and wrapped in `<...>`, `"literal"` values are escaped and quoted, and `"typed-literal"` values require a `datatype` (e.g. `"xsd:integer"`) and render unquoted for numeric/boolean XSD types. Placeholders use `{{param}}` rather than SPARQL's own `?param` syntax so template placeholders are never confused with real SPARQL variables in the query body.
+
+
+ CONSTRUCT templates are Blazegraph-only. `execute_construct_template()` raises `ProcessingError` if `store_backend` does not implement both `execute_sparql()` and `add_triplets()`.
+
+
## SPARQL Result Pagination
For large result sets, paginate with LIMIT and OFFSET:
diff --git a/semantica/triplet_store/blazegraph_store.py b/semantica/triplet_store/blazegraph_store.py
index 0b5cf70e..a2ad94aa 100644
--- a/semantica/triplet_store/blazegraph_store.py
+++ b/semantica/triplet_store/blazegraph_store.py
@@ -32,11 +32,15 @@ from typing import Any, Dict, List, Optional
from urllib.parse import urljoin, urlparse
import requests
+from rdflib import Graph
from ..semantic_extract.triplet_extractor import Triplet
from ..utils.exceptions import ProcessingError, ValidationError
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
+from . import sparql_escaping
+
+_CONSTRUCT_QUERY_RE = re.compile(r"\bCONSTRUCT\b", re.IGNORECASE)
class BlazegraphStore:
@@ -111,16 +115,42 @@ class BlazegraphStore:
"""Get SPARQL Update endpoint URL."""
return urljoin(self.endpoint, f"/blazegraph/namespace/{self.namespace}/sparql")
+ def _is_construct_query(self, query: str) -> bool:
+ """
+ Detect whether `query` is a SPARQL CONSTRUCT query.
+
+ This is a dispatch helper local to BlazegraphStore, distinct from
+ QueryEngine._validate_query (which already treats CONSTRUCT as one of
+ its valid_keywords and therefore requires no change — CONSTRUCT
+ queries already pass validation today). This helper only decides
+ which HTTP Accept header and response parser execute_sparql uses; it
+ does not gate query validity.
+ """
+ return _CONSTRUCT_QUERY_RE.search(query) is not None
+
def execute_sparql(self, query: str, **options) -> Dict[str, Any]:
"""
Execute SPARQL query.
Args:
query: SPARQL query string
- **options: Additional options
+ **options: Additional options:
+ - result_format: Optional[Literal["bindings", "construct"]].
+ If omitted, auto-detected via _is_construct_query(query).
Returns:
- Query results
+ Query results. For non-CONSTRUCT queries (or when result_format
+ resolves to "bindings"), the existing shape is unchanged:
+ {"success": bool, "bindings": [...], "variables": [...], "metadata": {...}}
+ For CONSTRUCT queries (or result_format="construct"), the shape is:
+ {"success": bool, "bindings": [], "variables": [], "triples": [...],
+ "metadata": {...}}
+ where "triples" is a list of (subject, predicate, object) string
+ 3-tuples parsed from the Turtle response via rdflib.
+
+ Raises:
+ ProcessingError: if not connected, the HTTP request fails, or (for
+ CONSTRUCT queries) the response body fails to parse as Turtle.
"""
tracking_id = self.progress_tracker.start_tracking(
module="triplet_store",
@@ -137,6 +167,62 @@ class BlazegraphStore:
sparql_endpoint = self._get_sparql_endpoint()
+ result_format = options.get("result_format")
+ if result_format is None:
+ result_format = "construct" if self._is_construct_query(query) else "bindings"
+
+ if result_format == "construct":
+ self.progress_tracker.update_tracking(
+ tracking_id, message="Sending CONSTRUCT query to Blazegraph endpoint..."
+ )
+ response = requests.post(
+ sparql_endpoint,
+ data={"query": query},
+ headers={
+ "Content-Type": "application/x-www-form-urlencoded",
+ "Accept": "text/turtle",
+ },
+ timeout=self.timeout,
+ auth=(self.username, self.password)
+ if self.username and self.password
+ else None,
+ )
+
+ response.raise_for_status()
+
+ self.progress_tracker.update_tracking(
+ tracking_id, message="Parsing CONSTRUCT response as Turtle..."
+ )
+ graph = Graph()
+ try:
+ graph.parse(data=response.content, format="turtle")
+ except Exception as parse_error:
+ raise ProcessingError(
+ f"Failed to parse CONSTRUCT response as Turtle: {parse_error}"
+ ) from parse_error
+
+ triples = [(str(s), str(p), str(o)) for s, p, o in graph]
+
+ result = {
+ "success": True,
+ "bindings": [],
+ "variables": [],
+ "triples": triples,
+ "metadata": {
+ "query": query,
+ "endpoint": sparql_endpoint,
+ "result_format": "construct",
+ },
+ }
+
+ self.progress_tracker.stop_tracking(
+ tracking_id,
+ status="completed",
+ message=f"CONSTRUCT query executed: {len(triples)} triples",
+ )
+ return result
+
+ # Non-CONSTRUCT path — unchanged from prior behavior.
self.progress_tracker.update_tracking(
tracking_id, message="Sending query to Blazegraph endpoint..."
)
@@ -307,31 +393,12 @@ class BlazegraphStore:
- Known prefixed names: ``xsd:integer``, ``rdf:langString``, etc.
Raises ValueError for anything else.
+
+ Delegates to the shared sparql_escaping.resolve_datatype_iri so this
+ logic has one canonical implementation shared with the CONSTRUCT
+ template renderer (semantica/triplet_store/construct_templates.py).
"""
- datatype = str(datatype)
-
- # Already angle-bracketed — validate the inner IRI contains no whitespace
- if datatype.startswith("<") and datatype.endswith(">"):
- inner = datatype[1:-1]
- if not inner or re.search(r"[\s<>\"{}|\\^`]", inner):
- raise ValueError(f"Invalid datatype IRI: {datatype!r}")
- return datatype
-
- # Full absolute IRI without brackets
- parsed = urlparse(datatype)
- if parsed.scheme in {"http", "https", "urn"} and not re.search(r"[\s<>\"{}|\\^`]", datatype):
- return f"<{datatype}>"
-
- # Prefixed form — expand known prefixes only
- if ":" in datatype:
- prefix, local = datatype.split(":", 1)
- if prefix in self._KNOWN_PREFIXES and re.match(r"^[A-Za-z0-9_\-\.]+$", local):
- return f"<{self._KNOWN_PREFIXES[prefix]}{local}>"
-
- raise ValueError(
- f"Unsupported datatype {datatype!r}: use a full IRI (http/https/urn), "
- f"an angle-bracketed IRI, or a known prefix (xsd/rdf/rdfs/owl/skos)."
- )
+ return sparql_escaping.resolve_datatype_iri(datatype)
def _is_uri_value(self, value: str) -> bool:
"""Detect if a value should be serialized as an IRI."""
@@ -346,15 +413,13 @@ class BlazegraphStore:
return not re.search(r"\s", value)
def _escape_literal(self, value: str) -> str:
- """Escape string literal for SPARQL."""
- return (
- str(value)
- .replace("\\", "\\\\")
- .replace("\"", "\\\"")
- .replace("\n", "\\n")
- .replace("\r", "\\r")
- .replace("\t", "\\t")
- )
+ """Escape string literal for SPARQL.
+
+ Delegates to the shared sparql_escaping.escape_literal so this logic
+ has one canonical implementation shared with the CONSTRUCT template
+ renderer (semantica/triplet_store/construct_templates.py).
+ """
+ return sparql_escaping.escape_literal(value)
def add_triplet(self, triplet: Triplet, **options) -> Dict[str, Any]:
"""Add single triplet."""
diff --git a/semantica/triplet_store/construct_templates.py b/semantica/triplet_store/construct_templates.py
new file mode 100644
index 00000000..dd30ec14
--- /dev/null
+++ b/semantica/triplet_store/construct_templates.py
@@ -0,0 +1,723 @@
+"""
+SPARQL CONSTRUCT Query Templates (Blazegraph-only)
+
+This module provides parameterized SPARQL CONSTRUCT query templates: a
+`ConstructTemplate` defines a reusable CONSTRUCT query body with `{{param}}`
+placeholders, `ConstructTemplateRegistry` stores/retrieves templates by name,
+and `render_construct_template` safely substitutes parameter values into a
+validated, injection-safe SPARQL string.
+
+This module is intentionally Blazegraph-only. `execute_construct_template`
+renders a template, executes it via `store_backend.execute_sparql(...,
+result_format="construct")` (the Blazegraph CONSTRUCT-aware extension),
+converts the parsed RDF triples into `Triplet` objects, and persists them via
+`store_backend.add_triplets`.
+
+All literal escaping, URI allowlist validation, and datatype-IRI resolution
+is delegated to the shared `semantica.triplet_store.sparql_escaping` module —
+no escaping/validation logic is reimplemented here.
+
+Main Classes:
+ - ParameterDescriptor: Declares one template parameter's name/type/validation.
+ - ConstructTemplate: Named, reusable CONSTRUCT query definition.
+ - ConstructTemplateRegistry: Stores and retrieves ConstructTemplate instances.
+
+Main Functions:
+ - render_construct_template: Safely render a ConstructTemplate into SPARQL.
+ - execute_construct_template: Render, execute, parse, and persist in one call.
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+import re
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Literal, Optional, Tuple
+
+from ..semantic_extract.triplet_extractor import Triplet
+from ..utils.exceptions import ProcessingError, ValidationError
+from ..utils.logging import get_logger
+from .sparql_escaping import escape_literal, resolve_datatype_iri, validate_uri
+
+ParameterKind = Literal["uri", "literal", "typed-literal"]
+
+# XSD local names (case-insensitive) that render as unquoted numeric/boolean
+# literals rather than quoted ""^^ literals. Matched against the
+# local name of the *resolved* datatype IRI so this works whether the
+# descriptor's datatype was given as a prefixed name (xsd:integer), a full
+# IRI, or a bracketed IRI.
+_INTEGER_LOCAL_NAMES = frozenset({"integer", "int", "long", "short"})
+_DECIMAL_LOCAL_NAMES = frozenset({"decimal", "double", "float"})
+_BOOLEAN_LOCAL_NAMES = frozenset({"boolean"})
+_NUMERIC_UNQUOTED_LOCAL_NAMES = _INTEGER_LOCAL_NAMES | _DECIMAL_LOCAL_NAMES | _BOOLEAN_LOCAL_NAMES
+
+_CONSTRUCT_KEYWORD_RE = re.compile(r"\bCONSTRUCT\b", re.IGNORECASE)
+_WHERE_KEYWORD_RE = re.compile(r"\bWHERE\b", re.IGNORECASE)
+_PLACEHOLDER_RE = re.compile(r"\{\{|\}\}")
+
+
+@dataclass
+class ParameterDescriptor:
+ """Describes one substitution parameter accepted by a ConstructTemplate."""
+
+ name: str
+ """Placeholder name as it appears in the query body, e.g. "subject" for {{subject}}."""
+
+ type: ParameterKind = "literal"
+ """One of "uri" | "literal" | "typed-literal"."""
+
+ required: bool = True
+ """If True and no value/default is supplied at render time, render raises ValidationError."""
+
+ default: Optional[Any] = None
+ """Used when the caller omits this parameter and required=False."""
+
+ datatype: Optional[str] = None
+ """Only meaningful when type == "typed-literal". An XSD datatype token accepted by
+ the shared resolve_datatype_iri, e.g. "xsd:integer", "xsd:dateTime", or a full IRI.
+ Required when type == "typed-literal"; render_construct_template (and
+ ConstructTemplateRegistry.register) raise ValidationError if type == "typed-literal"
+ and datatype is None."""
+
+ language: Optional[str] = None
+ """Only meaningful when type == "literal". RFC 5646 language tag, e.g. "en"."""
+
+
+@dataclass
+class ConstructTemplate:
+ """Parameterized SPARQL CONSTRUCT query template."""
+
+ name: str
+ """Unique registry key, e.g. "person_to_foaf"."""
+
+ description: str
+ """Human-readable summary, shown by list()/get_template_info()."""
+
+ construct_query: str
+ """Full CONSTRUCT query body containing {{param}} placeholders, e.g.:
+ "CONSTRUCT { <{{subject}}> foaf:name {{name}} } WHERE { ... }"
+ Must contain the CONSTRUCT keyword — enforced at register() time, not at
+ dataclass construction time."""
+
+ parameters: List[ParameterDescriptor] = field(default_factory=list)
+ """Ordered list of accepted parameters. Order has no runtime meaning, only used for
+ documentation / get_template_info() output."""
+
+ target_graph: Optional[str] = None
+ """Optional default named-graph IRI used when render_construct_template's/
+ execute_construct_template's target_graph argument is not supplied. This value,
+ like any caller-supplied target_graph, is ALWAYS passed through the same
+ validate_uri/escape path as a "uri"-typed parameter before being interpolated —
+ never through a raw f-string."""
+
+ metadata: dict = field(default_factory=dict)
+ """Free-form, mirrors PipelineTemplate.metadata (e.g. {"category": "rdf_mapping"})."""
+
+
+class ConstructTemplateRegistry:
+ """
+ CONSTRUCT template management system (Blazegraph-only).
+
+ Method shape mirrors PipelineTemplateManager:
+ PipelineTemplateManager.register_template(template) -> None
+ PipelineTemplateManager.get_template(name) -> Optional[PipelineTemplate]
+ PipelineTemplateManager.list_templates(category=None) -> List[str]
+
+ Unlike PipelineTemplateManager.register_template (which silently overwrites
+ by name), this registry rejects duplicate names with ValidationError —
+ CONSTRUCT queries execute against real triple stores, so silent overwrite
+ is a correctness hazard.
+ """
+
+ def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
+ """
+ Initialize template registry.
+
+ Args:
+ config: Configuration dictionary.
+ **kwargs: Additional configuration options.
+ """
+ self.logger = get_logger("construct_template_registry")
+ self.config = config or {}
+ self.config.update(kwargs)
+ self.templates: Dict[str, ConstructTemplate] = {}
+
+ def register(self, template: ConstructTemplate) -> None:
+ """
+ Register a CONSTRUCT template.
+
+ Args:
+ template: ConstructTemplate to register.
+
+ Raises:
+ ValidationError: if template.name is already registered,
+ construct_query does not contain the CONSTRUCT keyword, or any
+ ParameterDescriptor with type == "typed-literal" is missing
+ datatype. Validation happens before any mutation — a failed
+ register() call never partially overwrites the registry.
+ """
+ if template.name in self.templates:
+ raise ValidationError(
+ f"Template already registered: {template.name!r}. "
+ f"Remove it first via remove() if you intend to replace it."
+ )
+
+ if not _CONSTRUCT_KEYWORD_RE.search(template.construct_query):
+ raise ValidationError(
+ f"Template {template.name!r}: construct_query does not contain "
+ f"the CONSTRUCT keyword."
+ )
+
+ for descriptor in template.parameters:
+ if descriptor.type == "typed-literal" and descriptor.datatype is None:
+ raise ValidationError(
+ f"Template {template.name!r}: parameter {descriptor.name!r} has "
+ f"type='typed-literal' but no datatype declared."
+ )
+
+ self.templates[template.name] = template
+ self.logger.info(f"Registered CONSTRUCT template: {template.name}")
+
+ def get(self, name: str) -> Optional[ConstructTemplate]:
+ """
+ Get template by name.
+
+ Mirrors PipelineTemplateManager.get_template(template_name) -> Optional[PipelineTemplate].
+ """
+ return self.templates.get(name)
+
+ def list(self, category: Optional[str] = None) -> List[str]:
+ """
+ List registered template names, optionally filtered by metadata["category"].
+
+ Mirrors PipelineTemplateManager.list_templates(category=None) -> List[str].
+ """
+ if category:
+ return [
+ name
+ for name, template in self.templates.items()
+ if template.metadata.get("category") == category
+ ]
+ return list(self.templates.keys())
+
+ def remove(self, name: str) -> bool:
+ """
+ Remove a template by name.
+
+ Returns:
+ True if a template was removed, False if name was not registered.
+ """
+ if name in self.templates:
+ del self.templates[name]
+ self.logger.info(f"Removed CONSTRUCT template: {name}")
+ return True
+ return False
+
+ # --- PipelineTemplateManager-name aliases, for call-site consistency ---
+
+ def register_template(self, template: ConstructTemplate) -> None:
+ """Alias for register(), matching PipelineTemplateManager.register_template."""
+ self.register(template)
+
+ def get_template(self, template_name: str) -> Optional[ConstructTemplate]:
+ """Alias for get(), matching PipelineTemplateManager.get_template."""
+ return self.get(template_name)
+
+ def list_templates(self, category: Optional[str] = None) -> List[str]:
+ """Alias for list(), matching PipelineTemplateManager.list_templates."""
+ return self.list(category)
+
+ def get_template_info(self, template_name: str) -> Optional[Dict[str, Any]]:
+ """
+ Get template information.
+
+ Mirrors PipelineTemplateManager.get_template_info.
+
+ Returns:
+ {"name", "description", "parameter_count", "target_graph", "metadata"}
+ or None if template_name is not registered.
+ """
+ template = self.get(template_name)
+ if not template:
+ return None
+
+ return {
+ "name": template.name,
+ "description": template.description,
+ "parameter_count": len(template.parameters),
+ "target_graph": template.target_graph,
+ "metadata": template.metadata,
+ }
+
+
+# ---------------------------------------------------------------------------
+# render_construct_template and its internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _local_name_of_datatype_iri(datatype_iri: str) -> str:
+ """Extract the lower-cased local name from a resolved (bracketed) datatype IRI."""
+ inner = datatype_iri[1:-1] if datatype_iri.startswith("<") and datatype_iri.endswith(">") else datatype_iri
+ if "#" in inner:
+ return inner.rsplit("#", 1)[1].lower()
+ if "/" in inner:
+ return inner.rsplit("/", 1)[1].lower()
+ return inner.lower()
+
+
+def _render_numeric_literal(value: Any, local_name: str, descriptor_name: str) -> str:
+ """
+ Render an unquoted numeric/boolean literal for a "typed-literal" parameter
+ whose resolved datatype local name is in _NUMERIC_UNQUOTED_LOCAL_NAMES.
+
+ Raises:
+ ValidationError: if value cannot be coerced to the declared datatype.
+ """
+ if local_name in _INTEGER_LOCAL_NAMES:
+ if isinstance(value, bool):
+ raise ValidationError(
+ f"Parameter {descriptor_name!r}: expected an integer value for "
+ f"datatype local name {local_name!r}, got boolean {value!r}."
+ )
+ try:
+ return str(int(str(value)))
+ except (TypeError, ValueError):
+ raise ValidationError(
+ f"Parameter {descriptor_name!r}: value {value!r} is not a valid "
+ f"integer for its declared XSD datatype."
+ )
+
+ if local_name in _DECIMAL_LOCAL_NAMES:
+ try:
+ return str(float(str(value)))
+ except (TypeError, ValueError):
+ raise ValidationError(
+ f"Parameter {descriptor_name!r}: value {value!r} is not a valid "
+ f"decimal/double/float for its declared XSD datatype."
+ )
+
+ if local_name in _BOOLEAN_LOCAL_NAMES:
+ if isinstance(value, bool):
+ return "true" if value else "false"
+ normalized = str(value).strip().lower()
+ if normalized in ("true", "1"):
+ return "true"
+ if normalized in ("false", "0"):
+ return "false"
+ raise ValidationError(
+ f"Parameter {descriptor_name!r}: value {value!r} is not a valid "
+ f"boolean for its declared XSD datatype."
+ )
+
+ raise ValidationError(
+ f"Parameter {descriptor_name!r}: datatype local name {local_name!r} is not "
+ f"a recognized unquoted-numeric XSD datatype."
+ )
+
+
+def _find_matching_brace(text: str, open_index: int) -> int:
+ """
+ Given the index of an opening '{' in text, return the index of its matching '}'.
+
+ Skips over double-quoted string literal content while scanning, so a '{'
+ or '}' character that appears *inside* a rendered literal value (e.g. a
+ "typed-literal"/"literal" parameter whose value contains a brace
+ character — braces are not among the characters escape_literal escapes,
+ since SPARQL/Turtle string literals don't require it) does not corrupt
+ the brace-depth count. Quote-escaping is assumed to follow
+ escape_literal's convention (\\\\ and \\" are the only two-character
+ escapes that can produce a literal backslash or double-quote), so an
+ unescaped '"' reliably toggles in/out of string-literal content.
+ """
+ depth = 0
+ in_string = False
+ i = open_index
+ while i < len(text):
+ ch = text[i]
+ if in_string:
+ if ch == "\\":
+ # Skip the escaped character (e.g. \" or \\) without
+ # inspecting it, matching escape_literal's escaping scheme.
+ i += 2
+ continue
+ if ch == '"':
+ in_string = False
+ else:
+ if ch == '"':
+ in_string = True
+ elif ch == "{":
+ depth += 1
+ elif ch == "}":
+ depth -= 1
+ if depth == 0:
+ return i
+ i += 1
+ raise ValidationError("Unbalanced braces in construct_query: no matching '}' found.")
+
+
+def _split_construct_query(query_body: str) -> Tuple[str, str, str]:
+ """
+ Split a fully-substituted CONSTRUCT query into (preamble, construct_clause,
+ where_body), so the WHERE body can be re-wrapped in a GRAPH clause for
+ target_graph support.
+
+ Args:
+ query_body: Fully-substituted query string (all {{name}} tokens
+ already replaced).
+
+ Returns:
+ preamble: Everything before the CONSTRUCT keyword (e.g. PREFIX
+ declarations), stripped of leading/trailing whitespace. Preserved
+ verbatim so PREFIX declarations are not silently dropped when
+ target_graph wrapping rewrites the CONSTRUCT/WHERE structure.
+ construct_clause: The raw "{ ... }" template graph pattern immediately
+ following CONSTRUCT, braces included.
+ where_body: The raw text *inside* the "{ ... }" following WHERE,
+ braces excluded.
+
+ Raises:
+ ValidationError: if the CONSTRUCT/WHERE structure cannot be located.
+ """
+ construct_match = _CONSTRUCT_KEYWORD_RE.search(query_body)
+ if not construct_match:
+ raise ValidationError("construct_query does not contain the CONSTRUCT keyword.")
+ preamble = query_body[: construct_match.start()].strip()
+
+ construct_open = query_body.find("{", construct_match.end())
+ if construct_open == -1:
+ raise ValidationError("construct_query is missing '{' after the CONSTRUCT keyword.")
+ construct_close = _find_matching_brace(query_body, construct_open)
+ construct_clause = query_body[construct_open : construct_close + 1]
+
+ where_match = _WHERE_KEYWORD_RE.search(query_body, construct_close + 1)
+ if not where_match:
+ raise ValidationError("construct_query is missing a WHERE clause.")
+ where_open = query_body.find("{", where_match.end())
+ if where_open == -1:
+ raise ValidationError("construct_query is missing '{' after the WHERE keyword.")
+ where_close = _find_matching_brace(query_body, where_open)
+ where_body = query_body[where_open + 1 : where_close]
+
+ return preamble, construct_clause, where_body
+
+
+def render_construct_template(
+ template: ConstructTemplate,
+ params: Dict[str, Any],
+ target_graph: Optional[str] = None,
+) -> str:
+ """
+ Render a ConstructTemplate's construct_query into a safe, executable SPARQL
+ CONSTRUCT query string.
+
+ Args:
+ template: The ConstructTemplate to render.
+ params: Values for each {{name}} placeholder, keyed by ParameterDescriptor.name.
+ Values for missing optional parameters fall back to ParameterDescriptor.default.
+ target_graph: Named graph IRI to wrap the CONSTRUCT query in (via a GRAPH
+ clause around the WHERE body). If None, falls back to
+ template.target_graph. If both are None, no graph wrapping is applied.
+
+ Returns:
+ Fully-substituted SPARQL CONSTRUCT query string, safe to pass directly to
+ BlazegraphStore.execute_sparql.
+
+ Raises:
+ ValidationError: on any of:
+ - a required ParameterDescriptor has no value in params and no default
+ - a "uri"-typed parameter value fails validate_uri
+ - a "typed-literal"-typed parameter is missing template-declared datatype
+ - a "typed-literal"-typed parameter value cannot be coerced to that datatype
+ - target_graph (explicit arg or template.target_graph) fails validate_uri
+ - construct_query references a {{placeholder}} with no matching
+ ParameterDescriptor (detected as an unresolved placeholder after
+ substitution)
+ """
+ # Step 1: Resolve effective parameter values (params override, then default).
+ resolved: Dict[str, Any] = {}
+ for descriptor in template.parameters:
+ if descriptor.name in params:
+ resolved[descriptor.name] = params[descriptor.name]
+ elif not descriptor.required:
+ resolved[descriptor.name] = descriptor.default
+ else:
+ raise ValidationError(
+ f"Missing required parameter: {descriptor.name!r} "
+ f"(template {template.name!r})"
+ )
+
+ # Step 2: Render each parameter value according to its declared type.
+ rendered_values: Dict[str, str] = {}
+ for descriptor in template.parameters:
+ value = resolved[descriptor.name]
+
+ if descriptor.type == "uri":
+ safe_uri = validate_uri(value)
+ rendered_values[descriptor.name] = f"<{safe_uri}>"
+
+ elif descriptor.type == "literal":
+ escaped = escape_literal(value)
+ if descriptor.language is not None:
+ rendered_values[descriptor.name] = f'"{escaped}"@{descriptor.language}'
+ else:
+ rendered_values[descriptor.name] = f'"{escaped}"'
+
+ elif descriptor.type == "typed-literal":
+ if descriptor.datatype is None:
+ raise ValidationError(
+ f"typed-literal parameter requires datatype: {descriptor.name!r} "
+ f"(template {template.name!r})"
+ )
+ try:
+ datatype_iri = resolve_datatype_iri(descriptor.datatype)
+ except ValueError as exc:
+ raise ValidationError(
+ f"Parameter {descriptor.name!r}: {exc}"
+ ) from exc
+
+ local_name = _local_name_of_datatype_iri(datatype_iri)
+ if local_name in _NUMERIC_UNQUOTED_LOCAL_NAMES:
+ rendered_values[descriptor.name] = _render_numeric_literal(
+ value, local_name, descriptor.name
+ )
+ else:
+ escaped = escape_literal(value)
+ rendered_values[descriptor.name] = f'"{escaped}"^^{datatype_iri}'
+
+ else:
+ raise ValidationError(
+ f"Parameter {descriptor.name!r}: unknown parameter type {descriptor.type!r} "
+ f"(template {template.name!r})"
+ )
+
+ # Step 3: Substitute {{name}} tokens in construct_query.
+ query_body = template.construct_query
+ for name, rendered in rendered_values.items():
+ query_body = query_body.replace("{{" + name + "}}", rendered)
+
+ if _PLACEHOLDER_RE.search(query_body):
+ raise ValidationError(
+ f"Unresolved placeholder(s) in construct_query for template "
+ f"{template.name!r}: query still contains '{{{{' / '}}}}' tokens "
+ f"with no matching ParameterDescriptor."
+ )
+
+ # Step 4: Resolve effective target_graph — SAME validate_uri path as any
+ # "uri" parameter. No raw f-string interpolation is used here.
+ effective_graph = target_graph if target_graph is not None else template.target_graph
+ if effective_graph is not None:
+ safe_graph_uri = validate_uri(effective_graph)
+ wrapped_graph_token = f"<{safe_graph_uri}>"
+
+ preamble, construct_clause, where_body = _split_construct_query(query_body)
+ graph_wrapped_query = (
+ f"CONSTRUCT {construct_clause} "
+ f"WHERE {{ GRAPH {wrapped_graph_token} {{ {where_body} }} }}"
+ )
+ rendered_query = f"{preamble}\n{graph_wrapped_query}" if preamble else graph_wrapped_query
+ else:
+ rendered_query = query_body
+
+ return rendered_query
+
+
+def execute_construct_template(
+ template: ConstructTemplate,
+ params: Dict[str, Any],
+ store_backend: Any,
+ target_graph: Optional[str] = None,
+ **options: Any,
+) -> List[Triplet]:
+ """
+ Render, execute, parse, and persist a CONSTRUCT template in one call.
+
+ Args:
+ template: ConstructTemplate to execute.
+ params: Parameter values, forwarded to render_construct_template.
+ store_backend: A BlazegraphStore instance (Blazegraph-only; duck-typed
+ via hasattr(store_backend, "execute_sparql"), but this function
+ additionally requires store_backend to expose add_triplets — a
+ plain SPARQL-only backend without write support is rejected with
+ ProcessingError).
+ target_graph: Forwarded to render_construct_template; also used as
+ the `graph` option when persisting results via add_triplets so
+ constructed triples land in the same named graph they were
+ scoped to at query time.
+ **options: Forwarded to both store_backend.execute_sparql and
+ store_backend.add_triplets (e.g. timeout overrides).
+
+ Returns:
+ The List[Triplet] that were constructed AND successfully persisted
+ via add_triplets. Order matches the order the store backend's
+ execute_sparql yielded triples in its "triples" key.
+
+ Raises:
+ ValidationError: propagated from render_construct_template.
+ ProcessingError: if store_backend lacks execute_sparql/add_triplets,
+ or if persistence via add_triplets does not report success.
+
+ Exception-propagation convention:
+ This function does not wrap or catch exceptions raised by
+ render_construct_template or store_backend.execute_sparql — each
+ sub-layer is responsible for raising its own correctly-typed
+ exception (ValidationError for rendering failures; ProcessingError
+ for execution failures, as BlazegraphStore.execute_sparql already
+ does internally for connection/request/Turtle-parse errors). Adding
+ a second wrapping layer here would only obscure the original error
+ with no new information. The one exception this function DOES raise
+ itself is the add_triplets write-failure case immediately below,
+ because add_triplets signals failure via a returned dict rather than
+ an exception, so there is no pre-existing typed exception to let
+ propagate.
+ """
+ if not (hasattr(store_backend, "execute_sparql") and hasattr(store_backend, "add_triplets")):
+ raise ProcessingError(
+ "store_backend must support both execute_sparql and add_triplets "
+ "to use execute_construct_template."
+ )
+
+ # Step 1: Render (raises ValidationError unchanged on any failure).
+ rendered_query = render_construct_template(template, params, target_graph)
+
+ # Step 2: Execute via Blazegraph's CONSTRUCT-aware path.
+ query_result = store_backend.execute_sparql(
+ rendered_query, result_format="construct", **options
+ )
+
+ # Step 3: Convert parsed RDF triples to Triplet objects. store_backend is
+ # a BlazegraphStore instance, whose execute_sparql returns Dict[str, Any]
+ # with a "triples" key for CONSTRUCT queries (see BlazegraphStore.execute_sparql).
+ raw_triples = query_result.get("triples", [])
+ triplets: List[Triplet] = []
+ for s, p, o in raw_triples:
+ triplets.append(
+ Triplet(
+ subject=str(s),
+ predicate=str(p),
+ object=str(o),
+ # confidence=1.0 (explicit, matching Triplet's own default):
+ # a CONSTRUCT query is a deterministic graph transformation
+ # over already-persisted RDF/SPARQL-computed data, not a
+ # probabilistic extraction (unlike NER/LLM-based Triplet
+ # extraction, where confidence reflects genuine estimation
+ # uncertainty). There is no meaningful uncertainty to encode
+ # here, so full confidence is the correct value, not an
+ # accidental default.
+ confidence=1.0,
+ metadata={"source": "construct_template", "template": template.name},
+ )
+ )
+
+ # Step 4: Persist via add_triplets (same write path as any other bulk load).
+ effective_graph = target_graph if target_graph is not None else template.target_graph
+ write_result = store_backend.add_triplets(triplets, graph=effective_graph, **options)
+
+ if not write_result.get("success", False):
+ raise ProcessingError(
+ f"Failed to persist constructed triples for template "
+ f"{template.name!r}: {write_result}"
+ )
+
+ return triplets
+
+
+def construct_template_step_handler(data: Any, **options: Any) -> List[Triplet]:
+ """
+ Pipeline step handler for the "construct_template" step type.
+
+ Resolves `store_backend` and `construct_template_registry` from execution
+ options, looks up the named template, and delegates to
+ execute_construct_template. Mirrors the exact resolution pattern used for
+ `triplet_store`/`version_manager` in
+ ExecutionEngine._execute_step's delta_mode handling: call-time options
+ first, engine config fallback second, ProcessingError if either is
+ missing.
+
+ Called by ExecutionEngine._execute_step as
+ `step.handler(data, **step.config, **options)` — no change to
+ PipelineStep or _execute_step is required, since step.config and
+ **options already flow through generically for any step type.
+
+ Args:
+ data: Pipeline data flowing in (unused by this step type — a
+ construct_template step's output is independent of upstream
+ step data, exactly like other non-delta step types).
+ **options: The merged step.config + execution-time options dict, as
+ passed by ExecutionEngine._execute_step's
+ `step.handler(data, **step.config, **options)` call. Expected
+ keys:
+ - template_name: str — required, key into
+ construct_template_registry.
+ - params: Dict[str, Any] — required, forwarded to
+ render_construct_template.
+ - target_graph: Optional[str] — optional.
+ - store_backend: Any — resolved from options directly, or
+ from an "engine_config" dict passed alongside it (see
+ below); required.
+ - construct_template_registry: ConstructTemplateRegistry —
+ resolved the same way; required.
+ - engine_config: Optional[Dict[str, Any]] — fallback source
+ for store_backend/construct_template_registry when not
+ present directly in options, mirroring the *shape* of
+ self.config.get(...) in ExecutionEngine._execute_step's
+ delta_mode handling. One real difference from delta_mode:
+ delta_mode's fallback runs inside _execute_step itself (a
+ bound ExecutionEngine method with direct access to
+ self.config), whereas construct_template_step_handler is a
+ plain function with no such access — by design, this step
+ type requires no pre-handler interception in
+ _execute_step (see Requirement 7.5), so self.config is
+ simply never threaded to any step.handler call today. In
+ practice this means store_backend/construct_template_registry
+ should normally be passed as call-time options via
+ ExecutionEngine.execute_pipeline(pipeline, data,
+ store_backend=..., construct_template_registry=...),
+ which flow through untouched to this handler. The
+ engine_config parameter exists for callers who explicitly
+ forward their own ExecutionEngine(config=...) dict into
+ execute_pipeline's options (e.g. engine_config=engine.config)
+ and want the same two-tier resolution shape as delta_mode.
+ - step_name: Optional[str] — used only to name the failing
+ step in error messages; defaults to "construct_template"
+ if not supplied (design.md's own pseudocode references
+ step.name inside this handler, but the actual
+ step.handler(data, **step.config, **options) invocation
+ never passes step.name to a standalone handler function —
+ this default covers that gap without requiring any
+ ExecutionEngine/_execute_step change).
+
+ Returns:
+ The List[Triplet] returned by execute_construct_template.
+
+ Raises:
+ ProcessingError: if store_backend or construct_template_registry
+ cannot be resolved from options (or its "engine_config" fallback).
+ ValidationError: if template_name is not registered in the resolved
+ construct_template_registry.
+ """
+ step_name = options.get("step_name", "construct_template")
+ engine_config = options.get("engine_config") or {}
+
+ store_backend = options.get("store_backend") or engine_config.get("store_backend")
+ construct_template_registry = options.get(
+ "construct_template_registry"
+ ) or engine_config.get("construct_template_registry")
+
+ if not store_backend or not construct_template_registry:
+ raise ProcessingError(
+ f"Step '{step_name}' requires 'store_backend' and "
+ f"'construct_template_registry' in execution options for "
+ f"construct_template processing."
+ )
+
+ template_name = options.get("template_name")
+ template = construct_template_registry.get(template_name)
+ if template is None:
+ raise ValidationError(f"Unknown construct template: {template_name!r}")
+
+ return execute_construct_template(
+ template=template,
+ params=options.get("params", {}),
+ store_backend=store_backend,
+ target_graph=options.get("target_graph"),
+ )
diff --git a/semantica/triplet_store/query_engine.py b/semantica/triplet_store/query_engine.py
index 11c8bac7..db84f4ce 100644
--- a/semantica/triplet_store/query_engine.py
+++ b/semantica/triplet_store/query_engine.py
@@ -49,6 +49,12 @@ class QueryResult:
variables: List[str]
execution_time: float = 0.0
metadata: Dict[str, Any] = field(default_factory=dict)
+ triples: List[tuple] = field(default_factory=list)
+ """Populated only for CONSTRUCT queries. Each element is a (subject,
+ predicate, object) string 3-tuple, taken directly from the store
+ backend's execute_sparql "triples" key (see BlazegraphStore.execute_sparql
+ CONSTRUCT path). Empty list for all SELECT/ASK/DESCRIBE queries and for
+ backends without CONSTRUCT support."""
@dataclass
@@ -183,6 +189,7 @@ class QueryEngine:
bindings=result_data.get("bindings", []),
variables=result_data.get("variables", []),
execution_time=execution_time,
+ triples=result_data.get("triples", []),
metadata={
**result_data.get("metadata", {}),
"optimized": optimized_query != prepared_query,
diff --git a/semantica/triplet_store/sparql_escaping.py b/semantica/triplet_store/sparql_escaping.py
new file mode 100644
index 00000000..1863168a
--- /dev/null
+++ b/semantica/triplet_store/sparql_escaping.py
@@ -0,0 +1,159 @@
+"""
+Shared SPARQL literal-escaping and URI-validation primitives.
+
+This module centralizes string-escaping, datatype-IRI resolution, and URI
+allowlist validation logic that was previously duplicated (or would have
+been duplicated) between ``BlazegraphStore`` and the CONSTRUCT template
+renderer. ``BlazegraphStore._escape_literal`` and
+``BlazegraphStore._resolve_datatype_iri`` delegate to ``escape_literal`` and
+``resolve_datatype_iri`` here without any change in behavior; ``validate_uri``
+is new and is used by ``render_construct_template`` for both ``"uri"``-typed
+parameters and ``target_graph`` values.
+
+Author: Semantica Contributors
+License: MIT
+"""
+
+import re
+from typing import FrozenSet
+from urllib.parse import urlparse
+
+from ..utils.exceptions import ValidationError
+
+# Allowed URI schemes for validate_uri's allowlist (Requirement 3.2/4.4).
+_ALLOWED_URI_SCHEMES: FrozenSet[str] = frozenset({"http", "https", "urn"})
+
+# Known prefix expansions for XSD and common RDF vocabularies.
+# Identical table to BlazegraphStore._KNOWN_PREFIXES.
+KNOWN_PREFIXES: dict = {
+ "xsd": "http://www.w3.org/2001/XMLSchema#",
+ "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
+ "rdfs": "http://www.w3.org/2000/01/rdf-schema#",
+ "owl": "http://www.w3.org/2002/07/owl#",
+ "skos": "http://www.w3.org/2004/02/skos/core#",
+}
+
+# RFC 5646 language tag: primary subtag optionally followed by '-' + subtags.
+# Identical pattern to BlazegraphStore._LANG_TAG_RE.
+LANG_TAG_RE = re.compile(r"^[a-zA-Z]{1,8}(-[a-zA-Z0-9]{1,8})*$")
+
+# Disallowed-character check shared by validate_uri and resolve_datatype_iri,
+# identical to the character class used throughout BlazegraphStore.
+_DISALLOWED_URI_CHARS_RE = re.compile(r"[\s<>\"{}|\\^`]")
+
+
+def escape_literal(value: str) -> str:
+ """
+ Escape a string literal for safe inclusion inside SPARQL double quotes.
+
+ This is a byte-for-byte copy of BlazegraphStore._escape_literal's
+ transformation:
+ \\ -> \\\\ , " -> \\" , \\n -> \\n , \\r -> \\r , \\t -> \\t
+
+ Args:
+ value: Raw literal value (converted via str() first, matching the
+ original method's behavior of accepting non-str inputs).
+
+ Returns:
+ Escaped string safe to place inside a SPARQL/Turtle double-quoted
+ literal.
+ """
+ return (
+ str(value)
+ .replace("\\", "\\\\")
+ .replace("\"", "\\\"")
+ .replace("\n", "\\n")
+ .replace("\r", "\\r")
+ .replace("\t", "\\t")
+ )
+
+
+def resolve_datatype_iri(datatype: str) -> str:
+ """
+ Expand a datatype string to a validated SPARQL IRI token.
+
+ This is a byte-for-byte copy of BlazegraphStore._resolve_datatype_iri's
+ logic and exception behavior (raises ValueError, not ValidationError, to
+ match the original method's existing contract).
+
+ Accepts:
+ - Already-wrapped IRIs: ````
+ - Full IRIs: ``http://...`` / ``https://...`` / ``urn:...``
+ - Known prefixed names: ``xsd:integer``, ``rdf:langString``, etc.
+
+ Raises:
+ ValueError: for anything else, or for malformed/unsafe IRIs.
+
+ Returns:
+ An angle-bracketed IRI token, e.g. ````.
+ """
+ datatype = str(datatype)
+
+ # Already angle-bracketed — validate the inner IRI contains no whitespace
+ if datatype.startswith("<") and datatype.endswith(">"):
+ inner = datatype[1:-1]
+ if not inner or _DISALLOWED_URI_CHARS_RE.search(inner):
+ raise ValueError(f"Invalid datatype IRI: {datatype!r}")
+ return datatype
+
+ # Full absolute IRI without brackets
+ parsed = urlparse(datatype)
+ if parsed.scheme in {"http", "https", "urn"} and not _DISALLOWED_URI_CHARS_RE.search(datatype):
+ return f"<{datatype}>"
+
+ # Prefixed form — expand known prefixes only
+ if ":" in datatype:
+ prefix, local = datatype.split(":", 1)
+ if prefix in KNOWN_PREFIXES and re.match(r"^[A-Za-z0-9_\-\.]+$", local):
+ return f"<{KNOWN_PREFIXES[prefix]}{local}>"
+
+ raise ValueError(
+ f"Unsupported datatype {datatype!r}: use a full IRI (http/https/urn), "
+ f"an angle-bracketed IRI, or a known prefix (xsd/rdf/rdfs/owl/skos)."
+ )
+
+
+def validate_uri(
+ value: str,
+ *,
+ allowed_schemes: FrozenSet[str] = _ALLOWED_URI_SCHEMES,
+) -> str:
+ """
+ Validate that `value` is a safe absolute IRI using urllib.parse, and
+ return it unchanged (no bracket-wrapping — callers wrap with <...> at
+ render time).
+
+ Uses urllib.parse.urlparse(value):
+ - scheme must be in allowed_schemes (default {"http", "https", "urn"})
+ - the value must not contain whitespace or any of
+ < > " { } | \\ ^ ` characters (same disallowed-character set used by
+ resolve_datatype_iri / BlazegraphStore's existing IRI checks)
+
+ Args:
+ value: Candidate URI/IRI string.
+ allowed_schemes: Set of permitted URI schemes.
+
+ Returns:
+ The validated `value`, unchanged.
+
+ Raises:
+ ValidationError: if value is empty, not a string, has a scheme not in
+ allowed_schemes, or contains a disallowed character.
+ """
+ if not isinstance(value, str) or not value:
+ raise ValidationError(f"Invalid URI: value must be a non-empty string, got {value!r}")
+
+ if _DISALLOWED_URI_CHARS_RE.search(value):
+ raise ValidationError(
+ f"Invalid URI {value!r}: contains disallowed character(s) "
+ f"(whitespace or one of < > \" {{ }} | \\ ^ `)"
+ )
+
+ parsed = urlparse(value)
+ if parsed.scheme not in allowed_schemes:
+ raise ValidationError(
+ f"Invalid URI {value!r}: scheme {parsed.scheme!r} is not allowed "
+ f"(allowed: {sorted(allowed_schemes)})"
+ )
+
+ return value
diff --git a/tests/triplet_store/test_blazegraph_store.py b/tests/triplet_store/test_blazegraph_store.py
index 67f50a40..7a0525bb 100644
--- a/tests/triplet_store/test_blazegraph_store.py
+++ b/tests/triplet_store/test_blazegraph_store.py
@@ -1,7 +1,7 @@
import unittest
import os
import sys
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
if PROJECT_ROOT not in sys.path:
@@ -9,6 +9,7 @@ if PROJECT_ROOT not in sys.path:
from semantica.semantic_extract.triplet_extractor import Triplet
from semantica.triplet_store.blazegraph_store import BlazegraphStore
+from semantica.utils.exceptions import ProcessingError
class TestBlazegraphStoreSerialization(unittest.TestCase):
@@ -221,5 +222,315 @@ class TestBlazegraphStoreSerialization(unittest.TestCase):
self.assertEqual(obj, "\"Colour\"@en-GB")
+from semantica.triplet_store import sparql_escaping
+
+
+class TestSparqlEscapingExtractionParity(unittest.TestCase):
+ """
+ Regression tests proving BlazegraphStore._escape_literal /
+ BlazegraphStore._resolve_datatype_iri are byte-for-byte identical in
+ behavior to the extracted sparql_escaping.escape_literal /
+ sparql_escaping.resolve_datatype_iri functions, across every branch of
+ both original methods.
+ """
+
+ # --- escape_literal: every special character + combinations + plain text ---
+
+ ESCAPE_LITERAL_CASES = [
+ "", # empty string
+ "plain text, no special chars",
+ "back\\slash", # backslash
+ 'embedded "double" quote', # double quote
+ "line1\nline2", # newline
+ "line1\rline2", # carriage return
+ "tab\there", # tab
+ "\\\"\n\r\t", # all five special characters combined
+ 'mix \\ and " and \n and \r and \t together',
+ 42, # non-str input (both methods call str(value) first)
+ None,
+ ]
+
+ def test_escape_literal_matches_blazegraph_store_for_every_branch(self):
+ with patch.object(BlazegraphStore, "_connect", autospec=True):
+ store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+ for value in self.ESCAPE_LITERAL_CASES:
+ with self.subTest(value=value):
+ self.assertEqual(
+ store._escape_literal(value),
+ sparql_escaping.escape_literal(value),
+ )
+
+ # --- resolve_datatype_iri: every branch (bracketed, full IRI, prefixed,
+ # invalid bracketed, invalid full IRI-with-whitespace, unknown prefix) ---
+
+ RESOLVE_DATATYPE_IRI_VALID_CASES = [
+ "", # already bracketed
+ "http://www.w3.org/2001/XMLSchema#integer", # full IRI, no brackets
+ "https://example.org/type", # https scheme
+ "urn:isbn:0451450523", # urn scheme
+ "xsd:integer", # known prefix
+ "rdf:langString", # known prefix
+ "rdfs:label", # known prefix
+ "owl:Thing", # known prefix
+ "skos:Concept", # known prefix
+ ]
+
+ RESOLVE_DATATYPE_IRI_ERROR_CASES = [
+ "<>", # empty bracketed IRI
+ "", # bracketed IRI with whitespace
+ "", # bracketed IRI with disallowed char
+ "http://example.org/type CLEAR ALL", # full IRI with whitespace
+ "myns:customType", # unknown prefix
+ "not_a_uri_no_colon", # no scheme, no colon-prefixed form matches
+ "javascript:alert(1)", # disallowed scheme, not http/https/urn, no known prefix match
+ ]
+
+ def test_resolve_datatype_iri_matches_blazegraph_store_for_valid_cases(self):
+ with patch.object(BlazegraphStore, "_connect", autospec=True):
+ store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+ for datatype in self.RESOLVE_DATATYPE_IRI_VALID_CASES:
+ with self.subTest(datatype=datatype):
+ self.assertEqual(
+ store._resolve_datatype_iri(datatype),
+ sparql_escaping.resolve_datatype_iri(datatype),
+ )
+
+ def test_resolve_datatype_iri_matches_blazegraph_store_for_error_cases(self):
+ with patch.object(BlazegraphStore, "_connect", autospec=True):
+ store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+ for datatype in self.RESOLVE_DATATYPE_IRI_ERROR_CASES:
+ with self.subTest(datatype=datatype):
+ store_exc = None
+ shared_exc = None
+ try:
+ store._resolve_datatype_iri(datatype)
+ except ValueError as exc:
+ store_exc = exc
+ try:
+ sparql_escaping.resolve_datatype_iri(datatype)
+ except ValueError as exc:
+ shared_exc = exc
+
+ self.assertIsNotNone(store_exc, f"Expected ValueError from store for {datatype!r}")
+ self.assertIsNotNone(shared_exc, f"Expected ValueError from shared module for {datatype!r}")
+ self.assertEqual(str(store_exc), str(shared_exc))
+
+
+def _make_connected_store() -> BlazegraphStore:
+ """Create a BlazegraphStore instance bypassing the real _connect() call,
+ with .connected forced True (mirrors the state execute_sparql requires)."""
+ with patch.object(BlazegraphStore, "_connect", autospec=True):
+ store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+ store.connected = True
+ return store
+
+
+class TestBlazegraphStoreConstructExtension(unittest.TestCase):
+ """
+ Tests for the Blazegraph CONSTRUCT extension: _is_construct_query
+ detection, execute_sparql's CONSTRUCT branch (Accept header, Turtle
+ parsing via rdflib, triples shape, ProcessingError on malformed Turtle),
+ and Property 9 (non-CONSTRUCT queries are byte-for-byte unaffected).
+ """
+
+ # --- _is_construct_query detection ---
+
+ def test_is_construct_query_detects_uppercase_keyword(self):
+ store = _make_connected_store()
+ self.assertTrue(store._is_construct_query("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"))
+
+ def test_is_construct_query_detects_lowercase_keyword(self):
+ store = _make_connected_store()
+ self.assertTrue(store._is_construct_query("construct { ?s ?p ?o } where { ?s ?p ?o }"))
+
+ def test_is_construct_query_detects_mixed_case_keyword(self):
+ store = _make_connected_store()
+ self.assertTrue(store._is_construct_query("Construct { ?s ?p ?o } Where { ?s ?p ?o }"))
+
+ def test_is_construct_query_false_for_select(self):
+ store = _make_connected_store()
+ self.assertFalse(store._is_construct_query("SELECT ?s WHERE { ?s ?p ?o }"))
+
+ def test_is_construct_query_false_for_ask(self):
+ store = _make_connected_store()
+ self.assertFalse(store._is_construct_query("ASK { ?s ?p ?o }"))
+
+ def test_is_construct_query_does_not_match_substring_inside_identifier(self):
+ # "CONSTRUCTOR" contains "CONSTRUCT" as a substring but must not
+ # match due to \b word-boundary anchoring.
+ store = _make_connected_store()
+ self.assertFalse(
+ store._is_construct_query("SELECT ?s WHERE { ?s \"CONSTRUCTOR\" }")
+ )
+
+ # --- execute_sparql CONSTRUCT path ---
+
+ def test_execute_sparql_construct_sends_turtle_accept_header(self):
+ store = _make_connected_store()
+ mock_response = MagicMock()
+ mock_response.content = (
+ b"@prefix ex: .\n"
+ b'ex:s1 ex:p1 "value1" .\n'
+ )
+ mock_response.raise_for_status = MagicMock()
+
+ with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+ store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+ _, kwargs = mock_post.call_args
+ self.assertEqual(kwargs["headers"]["Accept"], "text/turtle")
+ self.assertEqual(
+ kwargs["headers"]["Content-Type"], "application/x-www-form-urlencoded"
+ )
+
+ def test_execute_sparql_construct_parses_triples_from_fixed_turtle_fixture(self):
+ store = _make_connected_store()
+ turtle_fixture = (
+ b"@prefix ex: .\n"
+ b'ex:s1 ex:p1 "value1" .\n'
+ b"ex:s1 ex:p2 ex:o2 .\n"
+ )
+ mock_response = MagicMock()
+ mock_response.content = turtle_fixture
+ mock_response.raise_for_status = MagicMock()
+
+ with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response):
+ result = store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+ self.assertTrue(result["success"])
+ self.assertEqual(result["bindings"], [])
+ self.assertEqual(result["variables"], [])
+ self.assertEqual(result["metadata"]["result_format"], "construct")
+
+ triples = set(result["triples"])
+ self.assertIn(("http://ex.org/s1", "http://ex.org/p1", "value1"), triples)
+ self.assertIn(("http://ex.org/s1", "http://ex.org/p2", "http://ex.org/o2"), triples)
+ self.assertEqual(len(result["triples"]), 2)
+
+ def test_execute_sparql_construct_result_format_option_forces_construct_path(self):
+ # Even for a query that doesn't literally contain "CONSTRUCT",
+ # result_format="construct" should force the Turtle-parsing path.
+ store = _make_connected_store()
+ mock_response = MagicMock()
+ mock_response.content = b'@prefix ex: .\nex:s1 ex:p1 "v" .\n'
+ mock_response.raise_for_status = MagicMock()
+
+ with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+ result = store.execute_sparql("SELECT ?s WHERE { ?s ?p ?o }", result_format="construct")
+
+ _, kwargs = mock_post.call_args
+ self.assertEqual(kwargs["headers"]["Accept"], "text/turtle")
+ self.assertIn("triples", result)
+
+ def test_execute_sparql_construct_malformed_turtle_raises_processing_error(self):
+ store = _make_connected_store()
+ mock_response = MagicMock()
+ # Deliberately invalid Turtle syntax.
+ mock_response.content = b"this is { not [ valid turtle syntax at all !!!"
+ mock_response.raise_for_status = MagicMock()
+
+ with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response):
+ with self.assertRaises(ProcessingError) as ctx:
+ store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+ # Must be a ProcessingError, not a raw rdflib exception leaking out.
+ self.assertIsInstance(ctx.exception, ProcessingError)
+ self.assertNotIsInstance(ctx.exception, (SyntaxError, ValueError))
+
+ def test_execute_sparql_construct_handles_literal_with_braces_in_valid_turtle(self):
+ # Adversarial case implied by the brace-matching bug found in the
+ # template-string layer (construct_templates._find_matching_brace):
+ # confirm rdflib itself parses a *valid* Turtle literal containing
+ # brace characters correctly, since this is a different parsing
+ # layer (real Turtle syntax, not our {{param}} template string).
+ store = _make_connected_store()
+ turtle_fixture = (
+ b"@prefix ex: .\n"
+ b'ex:s1 ex:p1 "text with { and } braces inside" .\n'
+ )
+ mock_response = MagicMock()
+ mock_response.content = turtle_fixture
+ mock_response.raise_for_status = MagicMock()
+
+ with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response):
+ result = store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
+
+ self.assertEqual(len(result["triples"]), 1)
+ subject, predicate, obj = result["triples"][0]
+ self.assertEqual(subject, "http://ex.org/s1")
+ self.assertEqual(predicate, "http://ex.org/p1")
+ self.assertEqual(obj, "text with { and } braces inside")
+
+ # --- Property 9: non-CONSTRUCT queries are byte-for-byte unaffected ---
+
+ def test_execute_sparql_select_query_response_shape_unchanged(self):
+ store = _make_connected_store()
+ mock_response = MagicMock()
+ mock_response.json.return_value = {
+ "head": {"vars": ["s", "p", "o"]},
+ "results": {
+ "bindings": [
+ {
+ "s": {"type": "uri", "value": "http://ex.org/s1"},
+ "p": {"type": "uri", "value": "http://ex.org/p1"},
+ "o": {"type": "literal", "value": "v1"},
+ }
+ ]
+ },
+ }
+ mock_response.raise_for_status = MagicMock()
+
+ with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+ result = store.execute_sparql("SELECT ?s ?p ?o WHERE { ?s ?p ?o }")
+
+ # Accept header must NOT be sent for a non-CONSTRUCT query — request
+ # shape is byte-for-byte identical to pre-CONSTRUCT-extension behavior.
+ _, kwargs = mock_post.call_args
+ self.assertEqual(kwargs["headers"], {"Content-Type": "application/x-www-form-urlencoded"})
+ self.assertNotIn("Accept", kwargs["headers"])
+
+ # Response shape must be exactly the pre-existing shape: no "triples"
+ # key at all (not even an empty list) for a plain SELECT response.
+ self.assertEqual(
+ result,
+ {
+ "success": True,
+ "bindings": mock_response.json.return_value["results"]["bindings"],
+ "variables": ["s", "p", "o"],
+ "metadata": {
+ "query": "SELECT ?s ?p ?o WHERE { ?s ?p ?o }",
+ "endpoint": store._get_sparql_endpoint(),
+ },
+ },
+ )
+ self.assertNotIn("triples", result)
+
+ def test_execute_sparql_ask_query_uses_bindings_path_not_construct(self):
+ store = _make_connected_store()
+ mock_response = MagicMock()
+ mock_response.json.return_value = {"head": {}, "boolean": True}
+ mock_response.raise_for_status = MagicMock()
+
+ with patch("semantica.triplet_store.blazegraph_store.requests.post", return_value=mock_response) as mock_post:
+ store.execute_sparql("ASK { ?s ?p ?o }")
+
+ _, kwargs = mock_post.call_args
+ self.assertNotIn("Accept", kwargs["headers"])
+
+
+class TestQueryEngineConstructValidation(unittest.TestCase):
+ """
+ Confirms QueryEngine._validate_query requires zero changes for CONSTRUCT
+ support — CONSTRUCT was already a valid keyword before this feature.
+ """
+
+ def test_construct_query_passes_validation_unchanged(self):
+ from semantica.triplet_store.query_engine import QueryEngine
+
+ engine = QueryEngine()
+ query = "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"
+ self.assertTrue(engine._validate_query(query))
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/triplet_store/test_construct_templates.py b/tests/triplet_store/test_construct_templates.py
new file mode 100644
index 00000000..fbe9d1cd
--- /dev/null
+++ b/tests/triplet_store/test_construct_templates.py
@@ -0,0 +1,966 @@
+"""
+Tests for semantica.triplet_store.construct_templates.
+
+Covers:
+ - ConstructTemplateRegistry: register/get/list/remove + aliases,
+ duplicate-name rejection, registration-time validation errors.
+ - render_construct_template: parametrized coverage of Correctness
+ Properties 1 (literal escaping), 2/3 (URI allowlist + target_graph
+ parity), and 4 (typed-literal numeric rendering), per design.md's
+ "Parametrized coverage of Correctness Properties 1, 2, 3, 4" section.
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import MagicMock
+
+import pytest
+from rdflib import Graph
+
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+if PROJECT_ROOT not in sys.path:
+ sys.path.insert(0, PROJECT_ROOT)
+
+from semantica.triplet_store.construct_templates import (
+ ConstructTemplate,
+ ConstructTemplateRegistry,
+ ParameterDescriptor,
+ _find_matching_brace,
+ _split_construct_query,
+ construct_template_step_handler,
+ execute_construct_template,
+ render_construct_template,
+)
+from semantica.utils.exceptions import ProcessingError, ValidationError
+
+
+def _simple_template(**overrides) -> ConstructTemplate:
+ """A minimal valid CONSTRUCT template: one literal param, no target_graph."""
+ defaults = dict(
+ name="simple_template",
+ description="A minimal template for testing.",
+ construct_query=(
+ "CONSTRUCT { {{value}} } "
+ "WHERE { {{value}} }"
+ ),
+ parameters=[ParameterDescriptor(name="value", type="literal", required=True)],
+ )
+ defaults.update(overrides)
+ return ConstructTemplate(**defaults)
+
+
+def _uri_template(**overrides) -> ConstructTemplate:
+ """A minimal valid CONSTRUCT template with one "uri"-typed param."""
+ defaults = dict(
+ name="uri_template",
+ description="A minimal template with a uri parameter.",
+ construct_query=(
+ "CONSTRUCT { {{subject}} \"x\" } "
+ "WHERE { {{subject}} \"x\" }"
+ ),
+ parameters=[ParameterDescriptor(name="subject", type="uri", required=True)],
+ )
+ defaults.update(overrides)
+ return ConstructTemplate(**defaults)
+
+
+def _typed_literal_template(datatype: str, **overrides) -> ConstructTemplate:
+ """A minimal valid CONSTRUCT template with one "typed-literal" param."""
+ defaults = dict(
+ name="typed_literal_template",
+ description="A minimal template with a typed-literal parameter.",
+ construct_query=(
+ "CONSTRUCT { {{value}} } "
+ "WHERE { {{value}} }"
+ ),
+ parameters=[
+ ParameterDescriptor(
+ name="value", type="typed-literal", required=True, datatype=datatype
+ )
+ ],
+ )
+ defaults.update(overrides)
+ return ConstructTemplate(**defaults)
+
+
+# ---------------------------------------------------------------------------
+# ConstructTemplateRegistry
+# ---------------------------------------------------------------------------
+
+
+class TestConstructTemplateRegistry(unittest.TestCase):
+ def test_register_and_get(self):
+ registry = ConstructTemplateRegistry()
+ template = _simple_template()
+ registry.register(template)
+ self.assertIs(registry.get("simple_template"), template)
+
+ def test_register_duplicate_name_raises_and_does_not_overwrite(self):
+ registry = ConstructTemplateRegistry()
+ t1 = _simple_template(description="original")
+ t2 = _simple_template(description="replacement")
+ registry.register(t1)
+ with self.assertRaises(ValidationError):
+ registry.register(t2)
+ self.assertIs(registry.get("simple_template"), t1)
+ self.assertEqual(registry.get("simple_template").description, "original")
+
+ def test_register_rejects_missing_construct_keyword(self):
+ registry = ConstructTemplateRegistry()
+ bad_template = _simple_template(
+ construct_query="SELECT ?s WHERE { ?s ?p ?o }"
+ )
+ with self.assertRaises(ValidationError):
+ registry.register(bad_template)
+ self.assertIsNone(registry.get("simple_template"))
+
+ def test_register_rejects_typed_literal_without_datatype(self):
+ registry = ConstructTemplateRegistry()
+ bad_template = _simple_template(
+ parameters=[ParameterDescriptor(name="value", type="typed-literal", datatype=None)]
+ )
+ with self.assertRaises(ValidationError):
+ registry.register(bad_template)
+ self.assertIsNone(registry.get("simple_template"))
+
+ def test_get_miss_returns_none(self):
+ registry = ConstructTemplateRegistry()
+ self.assertIsNone(registry.get("nonexistent"))
+
+ def test_list_without_category_returns_all_names(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(_simple_template(name="t1"))
+ registry.register(_simple_template(name="t2"))
+ self.assertEqual(sorted(registry.list()), ["t1", "t2"])
+
+ def test_list_with_category_filters(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(_simple_template(name="t1", metadata={"category": "mapping"}))
+ registry.register(_simple_template(name="t2", metadata={"category": "other"}))
+ registry.register(_simple_template(name="t3", metadata={"category": "mapping"}))
+ self.assertEqual(sorted(registry.list(category="mapping")), ["t1", "t3"])
+ self.assertEqual(registry.list(category="other"), ["t2"])
+ self.assertEqual(registry.list(category="nonexistent_category"), [])
+
+ def test_remove_idempotence(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(_simple_template())
+ self.assertTrue(registry.remove("simple_template"))
+ self.assertIsNone(registry.get("simple_template"))
+ self.assertFalse(registry.remove("simple_template"))
+
+ def test_register_template_alias_delegates_to_register(self):
+ registry = ConstructTemplateRegistry()
+ template = _simple_template()
+ registry.register_template(template)
+ self.assertIs(registry.get("simple_template"), template)
+ # Duplicate via alias still raises, same as register().
+ with self.assertRaises(ValidationError):
+ registry.register_template(_simple_template())
+
+ def test_get_template_alias_delegates_to_get(self):
+ registry = ConstructTemplateRegistry()
+ template = _simple_template()
+ registry.register(template)
+ self.assertIs(registry.get_template("simple_template"), template)
+ self.assertIsNone(registry.get_template("nonexistent"))
+
+ def test_list_templates_alias_delegates_to_list(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(_simple_template(name="t1", metadata={"category": "mapping"}))
+ registry.register(_simple_template(name="t2"))
+ self.assertEqual(sorted(registry.list_templates()), ["t1", "t2"])
+ self.assertEqual(registry.list_templates(category="mapping"), ["t1"])
+
+ def test_get_template_info_returns_expected_shape(self):
+ registry = ConstructTemplateRegistry()
+ template = _simple_template(
+ target_graph="http://ex.org/graphs/g1",
+ metadata={"category": "mapping"},
+ )
+ registry.register(template)
+ info = registry.get_template_info("simple_template")
+ self.assertEqual(
+ info,
+ {
+ "name": "simple_template",
+ "description": "A minimal template for testing.",
+ "parameter_count": 1,
+ "target_graph": "http://ex.org/graphs/g1",
+ "metadata": {"category": "mapping"},
+ },
+ )
+
+ def test_get_template_info_miss_returns_none(self):
+ registry = ConstructTemplateRegistry()
+ self.assertIsNone(registry.get_template_info("nonexistent"))
+
+
+# ---------------------------------------------------------------------------
+# Property 1: literal escaping
+# ---------------------------------------------------------------------------
+
+
+LITERAL_ESCAPING_CASES = [
+ pytest.param("back\\slash", id="backslash"),
+ pytest.param('embedded "double" quote', id="double-quote"),
+ pytest.param("line1\nline2", id="newline"),
+ pytest.param("line1\rline2", id="carriage-return"),
+ pytest.param("tab\there", id="tab"),
+ pytest.param("\\\"\n\r\t", id="all-combined"),
+]
+
+
+class TestRenderConstructTemplateLiteralEscaping:
+ """Property 1: no unescaped injection via literal parameters."""
+
+ @pytest.mark.parametrize("raw_value", LITERAL_ESCAPING_CASES)
+ def test_literal_param_round_trips_through_rdflib_turtle_parser(self, raw_value):
+ template = _simple_template()
+ rendered = render_construct_template(template, params={"value": raw_value})
+
+ # rendered must not contain an unescaped '"' that would terminate the
+ # literal early — verified by successfully round-tripping it through
+ # rdflib's Turtle parser back to the original string.
+ turtle_doc = f"@prefix ex: .\nex:s1 ex:p1 {rendered.split('} WHERE')[0].split('')[1].strip().rstrip('}').strip()} ."
+ graph = Graph()
+ graph.parse(data=turtle_doc, format="turtle")
+ literal_values = [str(o) for _, _, o in graph]
+ assert raw_value in literal_values
+
+ def test_render_produces_no_unescaped_quote_break(self):
+ template = _simple_template()
+ rendered = render_construct_template(
+ template, params={"value": 'has "quotes" inside'}
+ )
+ assert '"has \\"quotes\\" inside"' in rendered
+
+
+# ---------------------------------------------------------------------------
+# Property 2/3: URI allowlist + target_graph parity
+# ---------------------------------------------------------------------------
+
+
+BAD_URI_CASES = [
+ pytest.param("javascript:alert(1)", id="javascript-scheme"),
+ pytest.param("file:///etc/passwd", id="file-scheme"),
+ pytest.param("http://example.com/has space", id="embedded-whitespace"),
+ pytest.param("http://example.com/", id="embedded-angle-brackets"),
+ pytest.param("http://example.com/{injected}", id="embedded-braces"),
+ pytest.param("http://example.com/#frag with space", id="fragment-with-whitespace"),
+ pytest.param("", id="empty-string"),
+]
+
+GOOD_URI_CASES = [
+ pytest.param("http://example.com/ok", id="http-scheme"),
+ pytest.param("https://example.com/ok", id="https-scheme"),
+ pytest.param("urn:isbn:0451450523", id="urn-scheme"),
+]
+
+
+class TestRenderConstructTemplateUriValidation:
+ """Property 2 (URI allowlist enforcement) and Property 3 (target_graph parity)."""
+
+ @pytest.mark.parametrize("bad_uri", BAD_URI_CASES)
+ def test_uri_param_rejects_unsafe_values(self, bad_uri):
+ template = _uri_template()
+ with pytest.raises(ValidationError):
+ render_construct_template(template, params={"subject": bad_uri})
+
+ @pytest.mark.parametrize("bad_uri", BAD_URI_CASES)
+ def test_target_graph_rejects_same_unsafe_values_as_uri_param(self, bad_uri):
+ # A template with a valid "uri" param, but an unsafe target_graph.
+ template = _uri_template()
+ valid_params = {"subject": "http://example.com/ok"}
+
+ with pytest.raises(ValidationError) as uri_param_exc_info:
+ render_construct_template(
+ template, params={"subject": bad_uri}
+ )
+ with pytest.raises(ValidationError) as target_graph_exc_info:
+ render_construct_template(
+ template, params=valid_params, target_graph=bad_uri
+ )
+
+ # Parity check: both failure modes are ValidationError (same type);
+ # for non-empty invalid values, the underlying message text is
+ # identical because both call validate_uri() on the same bad_uri.
+ if bad_uri:
+ assert str(uri_param_exc_info.value) == str(target_graph_exc_info.value)
+
+ @pytest.mark.parametrize("good_uri", GOOD_URI_CASES)
+ def test_uri_param_accepts_allowed_schemes(self, good_uri):
+ template = _uri_template()
+ rendered = render_construct_template(template, params={"subject": good_uri})
+ assert f"<{good_uri}>" in rendered
+
+ @pytest.mark.parametrize("good_uri", GOOD_URI_CASES)
+ def test_target_graph_accepts_same_allowed_schemes_as_uri_param(self, good_uri):
+ template = _uri_template()
+ rendered = render_construct_template(
+ template,
+ params={"subject": "http://example.com/ok"},
+ target_graph=good_uri,
+ )
+ assert f"GRAPH <{good_uri}>" in rendered
+
+ def test_template_target_graph_used_when_argument_omitted(self):
+ template = _uri_template(target_graph="http://ex.org/graphs/default")
+ rendered = render_construct_template(
+ template, params={"subject": "http://example.com/ok"}
+ )
+ assert "GRAPH " in rendered
+
+ def test_explicit_target_graph_overrides_template_default(self):
+ template = _uri_template(target_graph="http://ex.org/graphs/default")
+ rendered = render_construct_template(
+ template,
+ params={"subject": "http://example.com/ok"},
+ target_graph="http://ex.org/graphs/override",
+ )
+ assert "GRAPH " in rendered
+ assert "GRAPH " not in rendered
+
+ def test_no_target_graph_means_no_graph_wrapping(self):
+ template = _uri_template()
+ rendered = render_construct_template(
+ template, params={"subject": "http://example.com/ok"}
+ )
+ assert "GRAPH" not in rendered
+
+
+# ---------------------------------------------------------------------------
+# Property 4: typed-literal numeric rendering
+# ---------------------------------------------------------------------------
+
+
+TYPED_LITERAL_CASES = [
+ pytest.param("xsd:integer", 42, True, id="xsd-integer"),
+ pytest.param("xsd:int", -7, True, id="xsd-int"),
+ pytest.param("xsd:long", 9999999999, True, id="xsd-long"),
+ pytest.param("xsd:short", 12, True, id="xsd-short"),
+ pytest.param("xsd:decimal", 3.14, True, id="xsd-decimal"),
+ pytest.param("xsd:double", 2.5e10, True, id="xsd-double"),
+ pytest.param("xsd:float", 1.5, True, id="xsd-float"),
+ pytest.param("xsd:boolean", True, True, id="xsd-boolean-true"),
+ pytest.param("xsd:boolean", False, True, id="xsd-boolean-false"),
+ pytest.param("xsd:dateTime", "2024-01-01T00:00:00Z", False, id="xsd-datetime-contrast"),
+]
+
+
+class TestRenderConstructTemplateTypedLiteralRendering:
+ """Property 4: typed-literal numeric rendering is unquoted, non-numeric is quoted."""
+
+ @pytest.mark.parametrize("datatype,value,expect_unquoted", TYPED_LITERAL_CASES)
+ def test_typed_literal_rendering_by_datatype(self, datatype, value, expect_unquoted):
+ template = _typed_literal_template(datatype)
+ rendered = render_construct_template(template, params={"value": value})
+
+ # Extract the substituted token from the rendered query.
+ placeholder_start = rendered.index("") + len(" ")
+ placeholder_end = rendered.index(" }", placeholder_start)
+ substituted = rendered[placeholder_start:placeholder_end]
+
+ if expect_unquoted:
+ assert not substituted.startswith('"'), (
+ f"Expected unquoted rendering for {datatype}, got: {substituted!r}"
+ )
+ else:
+ assert substituted.startswith('"') and "^^<" in substituted, (
+ f"Expected quoted ^^ rendering for {datatype}, got: {substituted!r}"
+ )
+
+ # rdflib should parse the resulting triple with the expected Literal.
+ prefix = "@prefix xsd: .\n"
+ turtle_doc = (
+ prefix
+ + f" {substituted} ."
+ )
+ graph = Graph()
+ graph.parse(data=turtle_doc, format="turtle")
+ literals = [o for _, _, o in graph]
+ assert len(literals) == 1
+
+ def test_typed_literal_rejects_non_numeric_value_for_integer(self):
+ template = _typed_literal_template("xsd:integer")
+ with pytest.raises(ValidationError):
+ render_construct_template(template, params={"value": "not-a-number"})
+
+ def test_typed_literal_missing_datatype_raises(self):
+ template = _simple_template(
+ parameters=[ParameterDescriptor(name="value", type="typed-literal", datatype=None)]
+ )
+ with pytest.raises(ValidationError):
+ render_construct_template(template, params={"value": 42})
+
+
+# ---------------------------------------------------------------------------
+# Required-parameter completeness (Property 5, spot-checked alongside 1-4)
+# ---------------------------------------------------------------------------
+
+
+class TestRenderConstructTemplateRequiredParameters:
+ def test_missing_required_parameter_raises(self):
+ template = _simple_template()
+ with pytest.raises(ValidationError):
+ render_construct_template(template, params={})
+
+ def test_optional_parameter_falls_back_to_default(self):
+ template = _simple_template(
+ parameters=[
+ ParameterDescriptor(name="value", type="literal", required=False, default="fallback")
+ ]
+ )
+ rendered = render_construct_template(template, params={})
+ assert '"fallback"' in rendered
+
+ def test_unresolved_placeholder_with_no_descriptor_raises(self):
+ template = _simple_template(
+ construct_query=(
+ "CONSTRUCT { {{value}} ; "
+ " {{undeclared}} } "
+ "WHERE { {{value}} }"
+ )
+ )
+ with pytest.raises(ValidationError):
+ render_construct_template(template, params={"value": "x"})
+
+
+# ---------------------------------------------------------------------------
+# _split_construct_query / _find_matching_brace: CONSTRUCT/WHERE boundary
+# parsing. This logic was added to fill a gap left as unspecified stubs
+# (extract_construct_clause/extract_where_body) in design.md; it is
+# exercised whenever target_graph wrapping is requested.
+# ---------------------------------------------------------------------------
+
+
+class TestSplitConstructQuery(unittest.TestCase):
+ """
+ Direct coverage of _split_construct_query / _find_matching_brace against
+ the 5 scenarios called out for verification, plus a regression case for
+ a bug found during that verification (see test 5b).
+ """
+
+ def test_1_prefix_declarations_before_construct_are_preserved_as_preamble(self):
+ query = (
+ "PREFIX foaf: \n"
+ 'CONSTRUCT { foaf:name "Alice" } '
+ 'WHERE { foaf:name "Alice" }'
+ )
+ preamble, construct_clause, where_body = _split_construct_query(query)
+ self.assertEqual(preamble, "PREFIX foaf: ")
+ self.assertEqual(construct_clause, '{ foaf:name "Alice" }')
+ self.assertEqual(where_body.strip(), ' foaf:name "Alice"')
+
+ def test_1b_target_graph_wrapping_preserves_prefix_preamble(self):
+ # End-to-end: PREFIX preamble must survive graph-wrapping, not just
+ # the standalone splitter.
+ template = ConstructTemplate(
+ name="prefixed_template",
+ description="test",
+ construct_query=(
+ "PREFIX foaf: \n"
+ 'CONSTRUCT { foaf:name "Alice" } '
+ 'WHERE { foaf:name "Alice" }'
+ ),
+ )
+ rendered = render_construct_template(
+ template, params={}, target_graph="http://ex.org/graphs/g1"
+ )
+ self.assertTrue(rendered.startswith("PREFIX foaf: "))
+ self.assertIn("GRAPH ", rendered)
+
+ def test_2_nested_braces_in_construct_clause_blank_node_pattern(self):
+ query = (
+ "CONSTRUCT { ?s ?p [ ?o2 ] } "
+ "WHERE { ?s ?p ?o2 }"
+ )
+ preamble, construct_clause, where_body = _split_construct_query(query)
+ self.assertEqual(preamble, "")
+ self.assertEqual(construct_clause, "{ ?s ?p [ ?o2 ] }")
+ self.assertEqual(where_body.strip(), "?s ?p ?o2")
+
+ def test_2b_nested_braces_in_construct_clause_graph_pattern(self):
+ # A CONSTRUCT clause containing an actual nested { ... } (not just
+ # blank-node [ ... ] syntax), to directly exercise brace-depth
+ # tracking rather than square-bracket handling.
+ query = (
+ "CONSTRUCT { GRAPH { ?s ?p ?o } } "
+ "WHERE { ?s ?p ?o }"
+ )
+ preamble, construct_clause, where_body = _split_construct_query(query)
+ self.assertEqual(construct_clause, "{ GRAPH { ?s ?p ?o } }")
+ self.assertEqual(where_body.strip(), "?s ?p ?o")
+
+ def test_3_nested_braces_in_where_clause_optional_block(self):
+ query = (
+ "CONSTRUCT { ?s ?o } "
+ "WHERE { ?s ?o . OPTIONAL { ?s ?o2 } }"
+ )
+ preamble, construct_clause, where_body = _split_construct_query(query)
+ self.assertEqual(construct_clause, "{ ?s ?o }")
+ self.assertEqual(
+ where_body.strip(),
+ "?s ?o . OPTIONAL { ?s ?o2 }",
+ )
+
+ def test_3b_nested_braces_in_where_clause_filter_not_exists_block(self):
+ query = (
+ "CONSTRUCT { ?s ?o } "
+ "WHERE { ?s ?o . "
+ "FILTER NOT EXISTS { ?s ?o2 } }"
+ )
+ preamble, construct_clause, where_body = _split_construct_query(query)
+ self.assertEqual(construct_clause, "{ ?s ?o }")
+ self.assertEqual(
+ where_body.strip(),
+ "?s ?o . FILTER NOT EXISTS { ?s ?o2 }",
+ )
+
+ def test_4_where_keyword_like_substring_inside_construct_clause_literal(self):
+ # "WHERE" appearing inside a string literal in the CONSTRUCT clause
+ # must not be mistaken for the real WHERE keyword.
+ query = (
+ 'CONSTRUCT { "the WHERE clause matters" } '
+ "WHERE { ?o }"
+ )
+ preamble, construct_clause, where_body = _split_construct_query(query)
+ self.assertEqual(
+ construct_clause,
+ '{ "the WHERE clause matters" }',
+ )
+ self.assertEqual(where_body.strip(), " ?o")
+
+ def test_4b_brace_character_inside_where_clause_literal_does_not_break_matching(self):
+ # A '}' character embedded inside a string literal within the WHERE
+ # body must not be counted as a real closing brace by
+ # _find_matching_brace (regression test for a bug found during
+ # verification: braces inside literals were originally counted
+ # regardless of string-literal context, corrupting brace-depth
+ # tracking and truncating/duplicating query text).
+ query = (
+ "CONSTRUCT { ?o } "
+ 'WHERE { "text with } inside" }'
+ )
+ preamble, construct_clause, where_body = _split_construct_query(query)
+ self.assertEqual(construct_clause, "{ ?o }")
+ self.assertEqual(
+ where_body.strip(),
+ ' "text with } inside"',
+ )
+
+ def test_4c_end_to_end_literal_with_brace_survives_target_graph_wrapping(self):
+ # End-to-end regression test: a rendered "literal" parameter value
+ # containing a '}' character must survive target_graph wrapping
+ # intact, not get truncated by brace-depth confusion.
+ template = ConstructTemplate(
+ name="brace_in_literal",
+ description="test",
+ construct_query=(
+ "CONSTRUCT { {{value}} } "
+ "WHERE { {{value}} }"
+ ),
+ parameters=[ParameterDescriptor(name="value", type="literal", required=True)],
+ )
+ rendered = render_construct_template(
+ template,
+ params={"value": "text with } inside"},
+ target_graph="http://ex.org/graphs/g1",
+ )
+ self.assertIn('"text with } inside"', rendered)
+ # Must appear intact in both the CONSTRUCT clause and the GRAPH-wrapped
+ # WHERE body — not truncated at the embedded '}'.
+ self.assertEqual(rendered.count('"text with } inside"'), 2)
+
+ def test_5_mismatched_braces_more_open_than_close_raises_validation_error(self):
+ query = "CONSTRUCT { ?s ?p ?o WHERE { ?s ?p ?o }"
+ with self.assertRaises(ValidationError):
+ _split_construct_query(query)
+
+ def test_5b_mismatched_braces_end_to_end_via_target_graph_wrapping(self):
+ # End-to-end: a malformed construct_query only surfaces the brace
+ # mismatch when target_graph wrapping is requested (no wrapping
+ # needed => no splitting attempted => no error from this code path).
+ template = ConstructTemplate(
+ name="malformed_template",
+ description="test",
+ construct_query="CONSTRUCT { ?s ?p ?o WHERE { ?s ?p ?o }",
+ )
+ with self.assertRaises(ValidationError):
+ render_construct_template(
+ template, params={}, target_graph="http://ex.org/graphs/g1"
+ )
+
+ def test_find_matching_brace_raises_on_no_closing_brace(self):
+ with self.assertRaises(ValidationError):
+ _find_matching_brace("{ unterminated", 0)
+
+
+# ---------------------------------------------------------------------------
+# execute_construct_template: render -> execute -> convert -> persist.
+# Property 6 (round-trip persistence) coverage via a stub store_backend.
+# ---------------------------------------------------------------------------
+
+
+class _StubStoreBackend:
+ """
+ Minimal stub store_backend for execute_construct_template tests.
+
+ execute_sparql returns a fixed CONSTRUCT-shaped result; add_triplets
+ records exactly what it was called with and returns a caller-controlled
+ success/failure result.
+ """
+
+ def __init__(self, triples, add_triplets_result=None, execute_sparql_error=None):
+ self._triples = triples
+ self._add_triplets_result = add_triplets_result or {"success": True}
+ self._execute_sparql_error = execute_sparql_error
+ self.execute_sparql_calls = []
+ self.add_triplets_calls = []
+
+ def execute_sparql(self, query, **options):
+ self.execute_sparql_calls.append((query, options))
+ if self._execute_sparql_error is not None:
+ raise self._execute_sparql_error
+ return {
+ "success": True,
+ "bindings": [],
+ "variables": [],
+ "triples": self._triples,
+ "metadata": {"query": query, "result_format": "construct"},
+ }
+
+ def add_triplets(self, triplets, **options):
+ self.add_triplets_calls.append((triplets, options))
+ return self._add_triplets_result
+
+
+class TestExecuteConstructTemplate(unittest.TestCase):
+ def _template(self, target_graph=None):
+ return ConstructTemplate(
+ name="exec_test_template",
+ description="test",
+ construct_query=(
+ "CONSTRUCT { {{value}} } "
+ "WHERE { {{value}} }"
+ ),
+ parameters=[ParameterDescriptor(name="value", type="literal", required=True)],
+ target_graph=target_graph,
+ )
+
+ # --- 1. Happy path ---
+
+ def test_happy_path_returns_triplets_and_calls_add_triplets_with_same_list(self):
+ fixed_triples = [
+ ("http://ex.org/s1", "http://ex.org/p1", "v1"),
+ ("http://ex.org/s2", "http://ex.org/p2", "v2"),
+ ]
+ stub = _StubStoreBackend(triples=fixed_triples)
+ template = self._template()
+
+ result = execute_construct_template(template, {"value": "x"}, stub)
+
+ self.assertEqual(len(result), 2)
+ self.assertEqual(
+ [(t.subject, t.predicate, t.object) for t in result],
+ fixed_triples,
+ )
+ for triplet in result:
+ self.assertEqual(
+ triplet.metadata, {"source": "construct_template", "template": "exec_test_template"}
+ )
+ # confidence is explicitly 1.0 (documented, deliberate default —
+ # CONSTRUCT results are deterministic, not probabilistic extraction).
+ self.assertEqual(triplet.confidence, 1.0)
+
+ # add_triplets must have been called with exactly this list (same
+ # object, not a copy or a differently-ordered list).
+ self.assertEqual(len(stub.add_triplets_calls), 1)
+ called_triplets, called_options = stub.add_triplets_calls[0]
+ self.assertIs(called_triplets, result)
+
+ # --- 2. Missing add_triplets ---
+
+ def test_missing_add_triplets_raises_before_execute_sparql_is_called(self):
+ class NoAddTriplets:
+ def execute_sparql(self, query, **options):
+ raise AssertionError("execute_sparql must not be called")
+
+ stub = NoAddTriplets()
+ template = self._template()
+
+ with self.assertRaises(ProcessingError):
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ def test_missing_add_triplets_execute_sparql_not_called_verified_via_mock(self):
+ stub = MagicMock()
+ del stub.add_triplets # hasattr(stub, "add_triplets") -> False
+ template = self._template()
+
+ with self.assertRaises(ProcessingError):
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ stub.execute_sparql.assert_not_called()
+
+ # --- 3. Missing execute_sparql ---
+
+ def test_missing_execute_sparql_raises_before_any_query_attempted(self):
+ stub = MagicMock()
+ del stub.execute_sparql # hasattr(stub, "execute_sparql") -> False
+ template = self._template()
+
+ with self.assertRaises(ProcessingError):
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ stub.add_triplets.assert_not_called()
+
+ # --- 4. add_triplets reports failure ---
+
+ def test_add_triplets_failure_raises_processing_error_not_partial_list(self):
+ fixed_triples = [("http://ex.org/s1", "http://ex.org/p1", "v1")]
+ stub = _StubStoreBackend(
+ triples=fixed_triples,
+ add_triplets_result={"success": False, "error": "write failed"},
+ )
+ template = self._template()
+
+ with self.assertRaises(ProcessingError):
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ # add_triplets was still attempted (with the constructed triplets)...
+ self.assertEqual(len(stub.add_triplets_calls), 1)
+ # ...but the function must not have returned anything — verified by
+ # the exception itself; there is no return value to inspect because
+ # execute_construct_template raised instead of returning.
+
+ # --- 5. execute_sparql raises an exception ---
+
+ def test_execute_sparql_exception_propagates_without_being_swallowed(self):
+ # Report exactly what happens: execute_construct_template does not
+ # wrap or catch exceptions raised by store_backend.execute_sparql —
+ # it calls it directly and lets whatever it raises propagate as-is.
+ network_error = ProcessingError("SPARQL query failed: connection refused")
+ stub = _StubStoreBackend(triples=[], execute_sparql_error=network_error)
+ template = self._template()
+
+ with self.assertRaises(ProcessingError) as ctx:
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ # The exact same exception instance propagates unchanged (not
+ # re-wrapped in a new ProcessingError with a different message).
+ self.assertIs(ctx.exception, network_error)
+
+ def test_execute_sparql_arbitrary_exception_type_also_propagates_unwrapped(self):
+ # Confirm this holds even for a non-ProcessingError exception type
+ # (e.g. a raw network library error) — execute_construct_template
+ # does not catch-and-wrap at all; add_triplets is simply never
+ # reached because execute_sparql raised first.
+ original_error = ConnectionError("simulated network failure")
+ stub = _StubStoreBackend(triples=[], execute_sparql_error=original_error)
+ template = self._template()
+
+ with self.assertRaises(ConnectionError) as ctx:
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ self.assertIs(ctx.exception, original_error)
+ self.assertEqual(stub.add_triplets_calls, [])
+
+ # --- 6. target_graph precedence ---
+
+ def test_target_graph_argument_overrides_template_target_graph(self):
+ stub = _StubStoreBackend(triples=[])
+ template = self._template(target_graph="http://ex.org/graphs/template_default")
+
+ execute_construct_template(
+ template, {"value": "x"}, stub, target_graph="http://ex.org/graphs/explicit"
+ )
+
+ _, add_triplets_options = stub.add_triplets_calls[0]
+ self.assertEqual(add_triplets_options["graph"], "http://ex.org/graphs/explicit")
+
+ def test_template_target_graph_used_when_argument_is_none(self):
+ stub = _StubStoreBackend(triples=[])
+ template = self._template(target_graph="http://ex.org/graphs/template_default")
+
+ execute_construct_template(template, {"value": "x"}, stub, target_graph=None)
+
+ _, add_triplets_options = stub.add_triplets_calls[0]
+ self.assertEqual(add_triplets_options["graph"], "http://ex.org/graphs/template_default")
+
+ def test_no_target_graph_anywhere_passes_none_to_add_triplets(self):
+ stub = _StubStoreBackend(triples=[])
+ template = self._template(target_graph=None)
+
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ _, add_triplets_options = stub.add_triplets_calls[0]
+ self.assertIsNone(add_triplets_options["graph"])
+
+ # --- 7. render_construct_template's ValidationError propagates unchanged ---
+
+ def test_missing_required_param_validation_error_propagates_unwrapped(self):
+ stub = _StubStoreBackend(triples=[])
+ template = self._template()
+
+ with self.assertRaises(ValidationError):
+ # Omit the required "value" parameter entirely.
+ execute_construct_template(template, {}, stub)
+
+ # Rendering must have failed before any query was attempted.
+ self.assertEqual(stub.execute_sparql_calls, [])
+ self.assertEqual(stub.add_triplets_calls, [])
+
+
+# ---------------------------------------------------------------------------
+# construct_template_step_handler: pipeline integration.
+# ---------------------------------------------------------------------------
+
+
+class TestConstructTemplateStepHandler(unittest.TestCase):
+ def _template(self):
+ return ConstructTemplate(
+ name="step_handler_template",
+ description="test",
+ construct_query=(
+ "CONSTRUCT { {{value}} } "
+ "WHERE { {{value}} }"
+ ),
+ parameters=[ParameterDescriptor(name="value", type="literal", required=True)],
+ )
+
+ def test_missing_store_backend_raises_processing_error(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(self._template())
+
+ with self.assertRaises(ProcessingError):
+ construct_template_step_handler(
+ None,
+ template_name="step_handler_template",
+ params={"value": "x"},
+ construct_template_registry=registry,
+ # store_backend deliberately omitted
+ )
+
+ def test_missing_registry_raises_processing_error(self):
+ stub = _StubStoreBackend(triples=[])
+
+ with self.assertRaises(ProcessingError):
+ construct_template_step_handler(
+ None,
+ template_name="step_handler_template",
+ params={"value": "x"},
+ store_backend=stub,
+ # construct_template_registry deliberately omitted
+ )
+
+ def test_unknown_template_name_raises_validation_error(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(self._template())
+ stub = _StubStoreBackend(triples=[])
+
+ with self.assertRaises(ValidationError):
+ construct_template_step_handler(
+ None,
+ template_name="does_not_exist",
+ params={"value": "x"},
+ store_backend=stub,
+ construct_template_registry=registry,
+ )
+
+ def test_happy_path_delegates_to_execute_construct_template(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(self._template())
+ fixed_triples = [("http://ex.org/s1", "http://ex.org/p1", "hello")]
+ stub = _StubStoreBackend(triples=fixed_triples)
+
+ result = construct_template_step_handler(
+ None,
+ template_name="step_handler_template",
+ params={"value": "hello"},
+ store_backend=stub,
+ construct_template_registry=registry,
+ )
+
+ self.assertEqual(
+ [(t.subject, t.predicate, t.object) for t in result],
+ fixed_triples,
+ )
+ self.assertEqual(len(stub.add_triplets_calls), 1)
+
+ def test_engine_config_fallback_used_when_options_omit_store_backend(self):
+ registry = ConstructTemplateRegistry()
+ registry.register(self._template())
+ stub = _StubStoreBackend(triples=[("http://ex.org/s1", "http://ex.org/p1", "v")])
+
+ result = construct_template_step_handler(
+ None,
+ template_name="step_handler_template",
+ params={"value": "v"},
+ engine_config={"store_backend": stub, "construct_template_registry": registry},
+ )
+
+ self.assertEqual(len(result), 1)
+
+
+class TestConstructTemplatePipelineIntegration(unittest.TestCase):
+ """
+ End-to-end: PipelineBuilder.add_step with the "construct_template" step
+ type -> ExecutionEngine.execute_pipeline with store_backend + registry
+ passed in -> triplets returned/persisted. Confirms Requirement 7.5 (no
+ PipelineStep/_execute_step changes needed) holds in practice.
+ """
+
+ def test_construct_template_step_executes_end_to_end_via_execution_engine(self):
+ from semantica.pipeline import ExecutionEngine, PipelineBuilder
+
+ registry = ConstructTemplateRegistry()
+ registry.register(
+ ConstructTemplate(
+ name="e2e_template",
+ description="test",
+ construct_query=(
+ "CONSTRUCT { {{value}} } "
+ "WHERE { {{value}} }"
+ ),
+ parameters=[ParameterDescriptor(name="value", type="literal", required=True)],
+ )
+ )
+ fixed_triples = [("http://ex.org/s1", "http://ex.org/p1", "Alice")]
+ stub = _StubStoreBackend(triples=fixed_triples)
+
+ builder = PipelineBuilder()
+ builder.add_step(
+ "apply_e2e_template",
+ "construct_template",
+ handler=construct_template_step_handler,
+ template_name="e2e_template",
+ params={"value": "Alice"},
+ )
+ pipeline = builder.build(name="e2e_pipeline")
+
+ engine = ExecutionEngine()
+ result = engine.execute_pipeline(
+ pipeline,
+ data=None,
+ store_backend=stub,
+ construct_template_registry=registry,
+ )
+
+ self.assertTrue(result.success, msg=f"Pipeline failed: {result.errors}")
+ returned_triplets = result.output
+ self.assertEqual(
+ [(t.subject, t.predicate, t.object) for t in returned_triplets],
+ fixed_triples,
+ )
+ # Confirm the triples were actually persisted via add_triplets.
+ self.assertEqual(len(stub.add_triplets_calls), 1)
+ persisted_triplets, _ = stub.add_triplets_calls[0]
+ self.assertEqual(
+ [(t.subject, t.predicate, t.object) for t in persisted_triplets],
+ fixed_triples,
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/triplet_store/test_query_engine.py b/tests/triplet_store/test_query_engine.py
new file mode 100644
index 00000000..babc2000
--- /dev/null
+++ b/tests/triplet_store/test_query_engine.py
@@ -0,0 +1,97 @@
+"""
+Tests for semantica.triplet_store.query_engine, focused on the
+QueryResult.triples field added for CONSTRUCT query support.
+"""
+
+import os
+import sys
+import unittest
+from unittest.mock import MagicMock
+
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+if PROJECT_ROOT not in sys.path:
+ sys.path.insert(0, PROJECT_ROOT)
+
+from semantica.triplet_store.query_engine import QueryEngine, QueryResult
+
+
+class TestQueryResultTriplesField(unittest.TestCase):
+ def test_query_result_triples_defaults_to_empty_list(self):
+ result = QueryResult(bindings=[], variables=[])
+ self.assertEqual(result.triples, [])
+
+ def test_query_result_triples_field_is_independent_per_instance(self):
+ # default_factory=list must produce a fresh list per instance, not a
+ # shared mutable default.
+ r1 = QueryResult(bindings=[], variables=[])
+ r2 = QueryResult(bindings=[], variables=[])
+ r1.triples.append(("s", "p", "o"))
+ self.assertEqual(r1.triples, [("s", "p", "o")])
+ self.assertEqual(r2.triples, [])
+
+ def test_query_result_accepts_explicit_triples(self):
+ triples = [("http://ex.org/s1", "http://ex.org/p1", "v1")]
+ result = QueryResult(bindings=[], variables=[], triples=triples)
+ self.assertEqual(result.triples, triples)
+
+
+class FakeConstructBackend:
+ """Fake store_backend whose execute_sparql returns a CONSTRUCT-shaped result."""
+
+ supports_named_graphs = True
+
+ def execute_sparql(self, query, **options):
+ return {
+ "success": True,
+ "bindings": [],
+ "variables": [],
+ "triples": [
+ ("http://ex.org/s1", "http://ex.org/p1", "v1"),
+ ("http://ex.org/s2", "http://ex.org/p2", "v2"),
+ ],
+ "metadata": {"query": query, "result_format": "construct"},
+ }
+
+
+class FakeBindingsBackend:
+ """Fake store_backend whose execute_sparql returns a plain bindings result
+ with no "triples" key at all, matching pre-CONSTRUCT-extension backends."""
+
+ supports_named_graphs = True
+
+ def execute_sparql(self, query, **options):
+ return {
+ "success": True,
+ "bindings": [{"s": {"value": "http://ex.org/s1"}}],
+ "variables": ["s"],
+ "metadata": {"query": query},
+ }
+
+
+class TestExecuteQueryPopulatesTriples(unittest.TestCase):
+ def test_execute_query_populates_triples_from_construct_backend(self):
+ engine = QueryEngine(enable_caching=False, enable_optimization=False)
+ result = engine.execute_query(
+ "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }", FakeConstructBackend()
+ )
+ self.assertEqual(
+ result.triples,
+ [
+ ("http://ex.org/s1", "http://ex.org/p1", "v1"),
+ ("http://ex.org/s2", "http://ex.org/p2", "v2"),
+ ],
+ )
+
+ def test_execute_query_defaults_triples_to_empty_list_when_backend_omits_key(self):
+ engine = QueryEngine(enable_caching=False, enable_optimization=False)
+ result = engine.execute_query(
+ "SELECT ?s WHERE { ?s ?p ?o }", FakeBindingsBackend()
+ )
+ self.assertEqual(result.triples, [])
+ # Non-CONSTRUCT behavior otherwise unaffected.
+ self.assertEqual(result.bindings, [{"s": {"value": "http://ex.org/s1"}}])
+ self.assertEqual(result.variables, ["s"])
+
+
+if __name__ == "__main__":
+ unittest.main()