diff --git a/docs/reference/triplet_store.md b/docs/reference/triplet_store.md
index 97a5336e..3a291d8c 100644
--- a/docs/reference/triplet_store.md
+++ b/docs/reference/triplet_store.md
@@ -297,8 +297,8 @@ template = ConstructTemplate(
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 }
+ CONSTRUCT { {{subject}} foaf:name {{name}} ; foaf:age {{age}} }
+ WHERE { {{subject}} a }
""",
parameters=[
ParameterDescriptor(name="subject", type="uri", required=True),
diff --git a/semantica/pipeline/pipeline_validator.py b/semantica/pipeline/pipeline_validator.py
index ed8757a3..dd93685b 100644
--- a/semantica/pipeline/pipeline_validator.py
+++ b/semantica/pipeline/pipeline_validator.py
@@ -85,13 +85,26 @@ class PipelineValidator:
return self.validate_pipeline(pipeline, **options)
def validate_pipeline(
- self, pipeline: Union["Pipeline", "PipelineBuilder"], **options
+ self,
+ pipeline: Union["Pipeline", "PipelineBuilder"],
+ construct_template_registry: Optional[Any] = None,
+ **options,
) -> ValidationResult:
"""
Validate entire pipeline.
Args:
pipeline: Pipeline object or builder
+ construct_template_registry: Optional ConstructTemplateRegistry
+ instance. When provided, steps whose step_type is
+ "construct_template" are additionally validated: the
+ template_name is checked for existence in the registry and
+ step.config["params"] is checked for all required template
+ parameters. When None (the default), a WARNING-level issue
+ is added for each construct_template step noting that
+ template existence could not be checked — existing callers
+ that do not pass this argument see zero behavior change for
+ any other step type.
**options: Additional options
Returns:
@@ -134,7 +147,10 @@ class PipelineValidator:
message=f"Validating {len(pipeline.steps)} pipeline steps...",
)
for step in pipeline.steps:
- step_result = self.validate_step(step)
+ step_result = self.validate_step(
+ step,
+ _construct_template_registry=construct_template_registry,
+ )
if not step_result.valid:
errors.extend(step_result.errors)
warnings.extend(step_result.warnings)
@@ -205,7 +221,19 @@ class PipelineValidator:
Args:
step: Pipeline step
- **constraints: Validation constraints
+ **constraints: Validation constraints. The following keys are
+ understood by this method and consumed internally; all others
+ are available for future extension:
+
+ allow_no_handler (bool, default False): suppress the
+ "has no handler" warning.
+
+ _construct_template_registry (ConstructTemplateRegistry | None,
+ default None): registry forwarded by validate_pipeline
+ for construct_template step-type validation. Prefixed
+ with '_' to signal internal plumbing — callers invoking
+ validate_step directly should use validate_pipeline's
+ construct_template_registry keyword argument instead.
Returns:
Validation result
@@ -227,6 +255,47 @@ class PipelineValidator:
if not step.config:
warnings.append(f"Step '{step.name}' has no configuration")
+ # --- construct_template step-type-specific validation ---
+ # This is the first step-type-specific check in this validator.
+ # Future step-type-specific checks should follow the same pattern:
+ # extract a registry/context object from constraints via a
+ # '__registry' key forwarded by validate_pipeline.
+ if step.step_type == "construct_template":
+ registry = constraints.get("_construct_template_registry")
+ if registry is None:
+ warnings.append(
+ f"Step '{step.name}' (construct_template): no "
+ f"construct_template_registry provided — template "
+ f"existence and required parameters could not be checked."
+ )
+ else:
+ template_name = step.config.get("template_name")
+ template = registry.get(template_name) if template_name else None
+ if not template_name or template is None:
+ errors.append(
+ f"Step '{step.name}' (construct_template): "
+ f"template_name {template_name!r} is not registered "
+ f"in the provided construct_template_registry."
+ )
+ else:
+ # Static required-param check — mirrors render_construct_template's
+ # exact runtime logic: required=True means the param is mandatory
+ # regardless of whether a default is declared (render only uses
+ # default when required=False, so a required param with a default
+ # still raises ValidationError at execution time).
+ provided_params = step.config.get("params") or {}
+ missing = [
+ d.name
+ for d in template.parameters
+ if d.required and d.name not in provided_params
+ ]
+ if missing:
+ errors.append(
+ f"Step '{step.name}' (construct_template): missing "
+ f"required parameter(s) for template "
+ f"{template_name!r}: {missing}."
+ )
+
return ValidationResult(
valid=len(errors) == 0, errors=errors, warnings=warnings
)
diff --git a/semantica/triplet_store/blazegraph_store.py b/semantica/triplet_store/blazegraph_store.py
index a2ad94aa..b657b8e8 100644
--- a/semantica/triplet_store/blazegraph_store.py
+++ b/semantica/triplet_store/blazegraph_store.py
@@ -32,7 +32,7 @@ from typing import Any, Dict, List, Optional
from urllib.parse import urljoin, urlparse
import requests
-from rdflib import Graph
+from rdflib import Graph, Literal
from ..semantic_extract.triplet_extractor import Triplet
from ..utils.exceptions import ProcessingError, ValidationError
@@ -40,7 +40,27 @@ 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)
+# Matches CONSTRUCT only as the actual SPARQL query-form keyword: anchored
+# from the start of the string, optionally preceded by PREFIX/BASE
+# declarations, then requires CONSTRUCT as the first non-whitespace keyword.
+# This prevents false-positives from SELECT/ASK queries that merely contain
+# the word "CONSTRUCT" inside a string literal or comment (e.g. a literal
+# value of '"please CONSTRUCT this"' or a comment line).
+_CONSTRUCT_QUERY_RE = re.compile(
+ r"""
+ \A # anchor to start of string
+ (?: # skip zero or more of:
+ \s+ # whitespace
+ | \#[^\n]* # comments (until newline)
+ | PREFIX\s+[\w\-]*:\s*<[^>]*> # PREFIX declaration
+ | BASE\s+<[^>]*> # BASE declaration
+ )*
+ \s* # any remaining whitespace before the query form
+ CONSTRUCT # the actual query-form keyword
+ \b # must be followed by a non-word character
+ """,
+ re.IGNORECASE | re.VERBOSE,
+)
class BlazegraphStore:
@@ -145,8 +165,15 @@ class BlazegraphStore:
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.
+ where "triples" is a list of (subject, predicate, object, metadata)
+ 4-tuples parsed from the Turtle response via rdflib. subject and
+ predicate are always plain strings. object is the literal's
+ lexical value or the IRI string. metadata is a dict that is empty
+ ({}) for URIs and plain untyped/unlang-tagged literals, and
+ otherwise contains "datatype" (the datatype IRI as a string) and/
+ or "language" (the RFC 5646 language tag) for literals that carry
+ that information — preserving what would otherwise be lost by
+ collapsing every rdflib term down to str(term).
Raises:
ProcessingError: if not connected, the HTTP request fails, or (for
@@ -201,7 +228,15 @@ class BlazegraphStore:
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]
+ triples = []
+ for s, p, o in graph:
+ obj_metadata: Dict[str, Any] = {}
+ if isinstance(o, Literal):
+ if o.datatype is not None:
+ obj_metadata["datatype"] = str(o.datatype)
+ if o.language is not None:
+ obj_metadata["language"] = str(o.language)
+ triples.append((str(s), str(p), str(o), obj_metadata))
result = {
"success": True,
diff --git a/semantica/triplet_store/construct_templates.py b/semantica/triplet_store/construct_templates.py
index dd30ec14..4051b9cd 100644
--- a/semantica/triplet_store/construct_templates.py
+++ b/semantica/triplet_store/construct_templates.py
@@ -446,6 +446,18 @@ def render_construct_template(
f"(template {template.name!r})"
)
+ # Reject any keys in params that don't correspond to a declared
+ # ParameterDescriptor — silently ignoring unknown parameters would let
+ # a caller's typo (e.g. "subjcet" instead of "subject") go unnoticed
+ # while the mistyped value is simply dropped.
+ declared_names = {descriptor.name for descriptor in template.parameters}
+ unexpected_keys = sorted(set(params) - declared_names)
+ if unexpected_keys:
+ raise ValidationError(
+ f"Unexpected parameter(s) for template {template.name!r}: "
+ f"{unexpected_keys}. Declared parameters: {sorted(declared_names)}."
+ )
+
# Step 2: Render each parameter value according to its declared type.
rendered_values: Dict[str, str] = {}
for descriptor in template.parameters:
@@ -544,7 +556,14 @@ def execute_construct_template(
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).
+ store_backend.add_triplets (e.g. timeout overrides). If options
+ contains "result_format" and/or "graph", those keys are
+ overridden by this function's own required values
+ ("construct" and effective_graph respectively) rather than
+ raising a duplicate-keyword-argument error — both keys are
+ load-bearing internal details of what this function does, so a
+ caller-supplied value for either is silently superseded, not an
+ error condition.
Returns:
The List[Triplet] that were constructed AND successfully persisted
@@ -569,6 +588,46 @@ def execute_construct_template(
because add_triplets signals failure via a returned dict rather than
an exception, so there is no pre-existing typed exception to let
propagate.
+
+ Why store_backend.execute_sparql is called directly instead of
+ QueryEngine.execute_query (investigated for issue #322 item on reusing
+ "Generic SPARQL execution ... QueryEngine.execute_query" — this is a
+ deliberate choice, not an oversight):
+ Routing through QueryEngine.execute_query was evaluated and found to
+ introduce three concrete regressions against this function's already
+ -tested behavior:
+ 1. QueryEngine.optimize_query's whitespace-collapse
+ (" ".join(query.split())) corrupts literal content. Verified: a
+ literal parameter value of "value: three spaces " comes
+ back from optimize_query as "value: three spaces " — the
+ collapsing operates on the whole query string with no awareness
+ of quoted-string boundaries, silently altering the literal's
+ actual content. This breaks the escaping guarantees
+ render_construct_template exists to provide (Property 1).
+ 2. QueryEngine.execute_query caches results keyed only on
+ normalized query text (enable_caching=True by default). A
+ CONSTRUCT query's correct results depend on the live state of the
+ graph at query time; repeated execute_construct_template calls
+ with identical params (a normal usage pattern — same template
+ re-run periodically) would silently return a stale cached
+ QueryResult instead of re-querying, causing incorrect
+ persistence via add_triplets.
+ 3. QueryEngine.execute_query wraps its entire body in a blanket
+ `except Exception: raise ProcessingError(...)`, re-typing every
+ exception regardless of origin. This directly conflicts with the
+ exception-propagation convention documented and tested above
+ (e.g. a raw ConnectionError from store_backend.execute_sparql
+ must propagate as ConnectionError, not get silently re-wrapped
+ into a differently-worded ProcessingError).
+ Fixing this properly would require QueryEngine itself to support a
+ "do not touch this already-rendered, already-safe query" mode
+ (disabling optimize_query and caching for CONSTRUCT) and to stop
+ re-wrapping already-correctly-typed exceptions — changes to a
+ shared, backend-agnostic module used by other query paths, which is
+ out of scope for this Blazegraph-only feature per the issue's own
+ no-scope-creep guidance. Calling store_backend.execute_sparql
+ directly is therefore the correct choice today, not a gap to close
+ casually.
"""
if not (hasattr(store_backend, "execute_sparql") and hasattr(store_backend, "add_triplets")):
raise ProcessingError(
@@ -579,17 +638,50 @@ def 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.
+ # Step 2: Execute via Blazegraph's CONSTRUCT-aware path. result_format is
+ # a load-bearing internal detail of this function (CONSTRUCT parsing
+ # requires it); if a caller's own **options happens to contain
+ # "result_format", the explicit value here must win rather than raising
+ # "got multiple values for keyword argument" — so it is popped out of a
+ # local copy of options and re-applied explicitly.
+ execute_options = dict(options)
+ execute_options.pop("result_format", None)
+ # Deliberately calling store_backend.execute_sparql directly, NOT
+ # QueryEngine.execute_query — see "Why store_backend.execute_sparql is
+ # called directly instead of QueryEngine.execute_query" in this
+ # function's docstring before routing through QueryEngine here.
query_result = store_backend.execute_sparql(
- rendered_query, result_format="construct", **options
+ rendered_query, result_format="construct", **execute_options
)
+ if not query_result.get("success", False):
+ raise ProcessingError(
+ f"CONSTRUCT query execution failed for template {template.name!r}: "
+ f"{query_result}"
+ )
+
# 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).
+ # with a "triples" key for CONSTRUCT queries: a list of
+ # (subject, predicate, object, metadata) 4-tuples (see
+ # BlazegraphStore.execute_sparql). object_metadata carries "datatype"
+ # and/or "language" for literals that have that information — those are
+ # folded into the resulting Triplet's own metadata under the
+ # "datatype"/"lang" keys, which is exactly what
+ # BlazegraphStore._format_object_for_sparql reads
+ # (metadata.get("datatype") / metadata.get("lang")) when re-serializing
+ # a Triplet back to SPARQL, so a typed/lang-tagged literal round-trips
+ # correctly through add_triplets instead of being silently flattened to
+ # an untyped plain string.
raw_triples = query_result.get("triples", [])
triplets: List[Triplet] = []
- for s, p, o in raw_triples:
+ for s, p, o, object_metadata in raw_triples:
+ triplet_metadata = {"source": "construct_template", "template": template.name}
+ if object_metadata.get("datatype"):
+ triplet_metadata["datatype"] = object_metadata["datatype"]
+ if object_metadata.get("language"):
+ triplet_metadata["lang"] = object_metadata["language"]
+
triplets.append(
Triplet(
subject=str(s),
@@ -604,13 +696,22 @@ def execute_construct_template(
# here, so full confidence is the correct value, not an
# accidental default.
confidence=1.0,
- metadata={"source": "construct_template", "template": template.name},
+ metadata=triplet_metadata,
)
)
- # Step 4: Persist via add_triplets (same write path as any other bulk load).
+ # Step 4: Persist via add_triplets (same write path as any other bulk
+ # load). Same reasoning as the result_format pop above: "graph" is the
+ # explicit target_graph/template.target_graph resolution this function
+ # exists to enforce, so a caller-supplied "graph" in **options must not
+ # crash the call or silently bypass that resolution — pop it before
+ # forwarding and let the computed effective_graph win.
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)
+ add_triplets_options = dict(options)
+ add_triplets_options.pop("graph", None)
+ write_result = store_backend.add_triplets(
+ triplets, graph=effective_graph, **add_triplets_options
+ )
if not write_result.get("success", False):
raise ProcessingError(
diff --git a/semantica/triplet_store/query_engine.py b/semantica/triplet_store/query_engine.py
index db84f4ce..1acafc19 100644
--- a/semantica/triplet_store/query_engine.py
+++ b/semantica/triplet_store/query_engine.py
@@ -51,10 +51,13 @@ class QueryResult:
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
+ predicate, object, metadata) 4-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."""
+ CONSTRUCT path). subject/predicate/object are strings; metadata is a dict
+ that is empty ({}) for URIs and plain untyped/unlang-tagged literals, and
+ otherwise carries "datatype" and/or "language" keys for literals that
+ have that information, so it is not silently lost. Empty list for all
+ SELECT/ASK/DESCRIBE queries and for backends without CONSTRUCT support."""
@dataclass
diff --git a/tests/pipeline/test_pipeline_validator_construct.py b/tests/pipeline/test_pipeline_validator_construct.py
new file mode 100644
index 00000000..63505c0e
--- /dev/null
+++ b/tests/pipeline/test_pipeline_validator_construct.py
@@ -0,0 +1,374 @@
+"""
+Tests for issue #3: construct_template_registry support in PipelineValidator.
+
+Covers four new behaviours and one explicit regression guard:
+ 1. No registry provided → WARNING-level issue, not an error (result still valid).
+ 2. Registry provided, template_name absent → ERROR.
+ 3. Registry provided, required param absent from step.config["params"] → ERROR.
+ 4. Registry provided, everything valid → no errors/warnings for that step.
+ 5. Regression: pipelines containing only non-construct_template steps see
+ identical validation output whether or not construct_template_registry is passed.
+"""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from semantica.pipeline.pipeline_builder import Pipeline, PipelineStep
+from semantica.pipeline.pipeline_validator import PipelineValidator
+from semantica.triplet_store.construct_templates import (
+ ConstructTemplate,
+ ConstructTemplateRegistry,
+ ParameterDescriptor,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_MINIMAL_QUERY = "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"
+
+
+def _make_registry(*templates):
+ """Return a ConstructTemplateRegistry pre-loaded with the given templates."""
+ reg = ConstructTemplateRegistry()
+ for t in templates:
+ reg.register(t)
+ return reg
+
+
+def _make_template(name, *param_descriptors):
+ return ConstructTemplate(
+ name=name,
+ description="test template",
+ construct_query=_MINIMAL_QUERY,
+ parameters=list(param_descriptors),
+ )
+
+
+def _make_pipeline(*steps):
+ """Wrap PipelineStep objects in a minimal Pipeline."""
+ return Pipeline(name="test_pipeline", steps=list(steps))
+
+
+def _make_step(name, step_type, config=None):
+ return PipelineStep(name=name, step_type=step_type, config=config or {})
+
+
+# ---------------------------------------------------------------------------
+# Test class
+# ---------------------------------------------------------------------------
+
+
+class TestPipelineValidatorConstructRegistry(unittest.TestCase):
+
+ def setUp(self):
+ self.tracker_patcher = patch(
+ "semantica.utils.progress_tracker.get_progress_tracker"
+ )
+ mock_get_tracker = self.tracker_patcher.start()
+ mock_tracker = MagicMock()
+ mock_get_tracker.return_value = mock_tracker
+
+ def tearDown(self):
+ self.tracker_patcher.stop()
+
+ # ------------------------------------------------------------------
+ # Test 1: registry not provided -> warning, not error
+ # ------------------------------------------------------------------
+
+ def test_no_registry_yields_warning_not_error(self):
+ """
+ When construct_template_registry is None (the default), a construct_template
+ step should produce exactly one WARNING mentioning the inability to check
+ template existence, and the result should still be valid (no errors).
+ """
+ step = _make_step(
+ "build_foaf",
+ "construct_template",
+ config={"template_name": "person_to_foaf", "params": {}},
+ )
+ pipeline = _make_pipeline(step)
+ validator = PipelineValidator()
+
+ result = validator.validate_pipeline(pipeline) # no registry kwarg
+
+ self.assertTrue(result.valid, msg=f"Expected valid; errors={result.errors}")
+ self.assertEqual(result.errors, [], msg="Expected no errors")
+ # At least one warning about the missing registry
+ registry_warnings = [
+ w for w in result.warnings if "construct_template_registry" in w
+ ]
+ self.assertGreater(
+ len(registry_warnings),
+ 0,
+ msg=f"Expected a warning about missing registry; warnings={result.warnings}",
+ )
+
+ # ------------------------------------------------------------------
+ # Test 2: registry provided, template_name absent -> error
+ # ------------------------------------------------------------------
+
+ def test_missing_template_name_in_registry_yields_error(self):
+ """
+ When a registry is provided but step.config["template_name"] is not
+ registered in it, validate_pipeline must produce an ERROR and result.valid
+ must be False.
+ """
+ # Registry has "other_template", not "person_to_foaf"
+ registry = _make_registry(
+ _make_template("other_template")
+ )
+ step = _make_step(
+ "build_foaf",
+ "construct_template",
+ config={"template_name": "person_to_foaf", "params": {}},
+ )
+ pipeline = _make_pipeline(step)
+ validator = PipelineValidator()
+
+ result = validator.validate_pipeline(
+ pipeline, construct_template_registry=registry
+ )
+
+ self.assertFalse(result.valid, msg="Expected invalid result")
+ template_errors = [e for e in result.errors if "person_to_foaf" in e]
+ self.assertGreater(
+ len(template_errors),
+ 0,
+ msg=f"Expected error mentioning 'person_to_foaf'; errors={result.errors}",
+ )
+
+ def test_none_template_name_in_config_yields_error(self):
+ """
+ Edge case: step.config has no 'template_name' key at all.
+ Registry is provided; should still produce an error.
+ """
+ registry = _make_registry(_make_template("some_template"))
+ step = _make_step(
+ "bad_step",
+ "construct_template",
+ config={"params": {}}, # no template_name key
+ )
+ pipeline = _make_pipeline(step)
+ validator = PipelineValidator()
+
+ result = validator.validate_pipeline(
+ pipeline, construct_template_registry=registry
+ )
+
+ self.assertFalse(result.valid)
+ self.assertTrue(any("template_name" in e or "None" in e for e in result.errors),
+ msg=f"Expected error about missing/None template_name; errors={result.errors}")
+
+ # ------------------------------------------------------------------
+ # Test 3: registry provided, required param absent -> error
+ # ------------------------------------------------------------------
+
+ def test_missing_required_param_yields_error(self):
+ """
+ When the registry is provided, the template is found, but a required
+ parameter is absent from step.config["params"], validate_pipeline must
+ produce an ERROR listing the missing parameter.
+ """
+ template = _make_template(
+ "person_to_foaf",
+ ParameterDescriptor(name="subject", type="uri", required=True),
+ ParameterDescriptor(name="name", type="literal", required=True),
+ ParameterDescriptor(name="lang", type="literal", required=False, default="en"),
+ )
+ registry = _make_registry(template)
+ # Provides "subject" but omits the required "name"
+ step = _make_step(
+ "build_foaf",
+ "construct_template",
+ config={
+ "template_name": "person_to_foaf",
+ "params": {"subject": "http://example.org/alice"},
+ },
+ )
+ pipeline = _make_pipeline(step)
+ validator = PipelineValidator()
+
+ result = validator.validate_pipeline(
+ pipeline, construct_template_registry=registry
+ )
+
+ self.assertFalse(result.valid, msg="Expected invalid result")
+ missing_errors = [e for e in result.errors if "name" in e]
+ self.assertGreater(
+ len(missing_errors),
+ 0,
+ msg=f"Expected error about missing 'name' param; errors={result.errors}",
+ )
+
+ def test_required_param_with_default_still_flagged_when_absent(self):
+ """
+ Correctness of the corrected check: required=True AND d.name not in
+ provided_params -> error, even when d.default is not None.
+ This verifies the 'and d.default is None' condition was NOT included,
+ matching render_construct_template's actual runtime behavior.
+ """
+ template = _make_template(
+ "tricky_template",
+ ParameterDescriptor(
+ name="subject",
+ type="literal",
+ required=True,
+ default="fallback_value", # default exists but required=True
+ ),
+ )
+ registry = _make_registry(template)
+ step = _make_step(
+ "tricky_step",
+ "construct_template",
+ config={
+ "template_name": "tricky_template",
+ "params": {}, # subject not supplied
+ },
+ )
+ pipeline = _make_pipeline(step)
+ validator = PipelineValidator()
+
+ result = validator.validate_pipeline(
+ pipeline, construct_template_registry=registry
+ )
+
+ # Must be an error even though default="fallback_value" exists,
+ # because render_construct_template raises on required=True + no caller value.
+ self.assertFalse(result.valid, msg=(
+ "Expected invalid: required=True param with a default should still be "
+ "flagged if absent from step.config['params']"
+ ))
+ self.assertTrue(
+ any("subject" in e for e in result.errors),
+ msg=f"Expected error mentioning 'subject'; errors={result.errors}",
+ )
+
+ # ------------------------------------------------------------------
+ # Test 4: all valid -> no errors, no construct-related warnings
+ # ------------------------------------------------------------------
+
+ def test_all_valid_yields_no_issues(self):
+ """
+ Registry provided, template found, all required params supplied ->
+ no errors, no construct_template-related warnings.
+ """
+ template = _make_template(
+ "person_to_foaf",
+ ParameterDescriptor(name="subject", type="uri", required=True),
+ ParameterDescriptor(name="name", type="literal", required=True),
+ ParameterDescriptor(name="lang", type="literal", required=False, default="en"),
+ )
+ registry = _make_registry(template)
+ step = _make_step(
+ "build_foaf",
+ "construct_template",
+ config={
+ "template_name": "person_to_foaf",
+ "params": {
+ "subject": "http://example.org/alice",
+ "name": "Alice",
+ # "lang" intentionally omitted -- optional, should not trigger error
+ },
+ },
+ )
+ pipeline = _make_pipeline(step)
+ validator = PipelineValidator()
+
+ result = validator.validate_pipeline(
+ pipeline, construct_template_registry=registry
+ )
+
+ self.assertEqual(result.errors, [], msg=f"Expected no errors; got {result.errors}")
+ # No construct-specific warnings (handler/config warnings from generic check
+ # are fine -- the step has no handler and that's expected in this test fixture)
+ construct_warnings = [
+ w for w in result.warnings if "construct_template_registry" in w
+ ]
+ self.assertEqual(
+ construct_warnings, [],
+ msg=f"Expected no registry-related warnings; got {result.warnings}",
+ )
+
+ # ------------------------------------------------------------------
+ # Test 5: regression -- non-construct_template steps unaffected
+ # ------------------------------------------------------------------
+
+ def test_non_construct_steps_unchanged(self):
+ """
+ Regression guard: validate_pipeline's output must be identical for a
+ pipeline containing only non-construct_template steps, whether or not
+ construct_template_registry is passed.
+
+ Both calls (with and without the registry kwarg) must produce the same
+ valid/errors/warnings, confirming zero behavior change for existing
+ callers of any other step type.
+ """
+ steps = [
+ _make_step("ingest", "file_ingest", config={"path": "/data"}),
+ _make_step("parse", "document_parse", config={"format": "pdf"}),
+ _make_step("embed", "embedding", config={"model": "openai"}),
+ ]
+ pipeline = _make_pipeline(*steps)
+ validator = PipelineValidator()
+
+ # Call without registry (existing behavior)
+ result_without = validator.validate_pipeline(pipeline)
+ # Call with a registry (should not affect these steps at all)
+ dummy_registry = ConstructTemplateRegistry()
+ result_with = validator.validate_pipeline(
+ pipeline, construct_template_registry=dummy_registry
+ )
+
+ self.assertEqual(
+ result_without.valid,
+ result_with.valid,
+ msg="valid flag changed for non-construct_template pipeline",
+ )
+ self.assertEqual(
+ result_without.errors,
+ result_with.errors,
+ msg="errors changed for non-construct_template pipeline",
+ )
+ self.assertEqual(
+ result_without.warnings,
+ result_with.warnings,
+ msg="warnings changed for non-construct_template pipeline",
+ )
+
+ def test_mixed_pipeline_only_construct_step_gets_warning(self):
+ """
+ A pipeline with mixed step types: only the construct_template step should
+ receive the 'no registry' warning; other steps should be unaffected.
+ """
+ steps = [
+ _make_step("ingest", "file_ingest", config={"path": "/data"}),
+ _make_step(
+ "build_graph",
+ "construct_template",
+ config={"template_name": "some_template", "params": {}},
+ ),
+ _make_step("embed", "embedding", config={"model": "openai"}),
+ ]
+ pipeline = _make_pipeline(*steps)
+ validator = PipelineValidator()
+
+ result = validator.validate_pipeline(pipeline) # no registry
+
+ self.assertTrue(result.valid, msg=f"Expected valid; errors={result.errors}")
+ # Exactly one construct-registry warning (for "build_graph")
+ registry_warnings = [
+ w for w in result.warnings if "construct_template_registry" in w
+ ]
+ self.assertEqual(
+ len(registry_warnings),
+ 1,
+ msg=f"Expected exactly 1 registry warning; warnings={result.warnings}",
+ )
+ # The warning should mention the step name
+ self.assertIn("build_graph", registry_warnings[0])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/triplet_store/test_blazegraph_store.py b/tests/triplet_store/test_blazegraph_store.py
index 7a0525bb..dda94d28 100644
--- a/tests/triplet_store/test_blazegraph_store.py
+++ b/tests/triplet_store/test_blazegraph_store.py
@@ -347,6 +347,20 @@ class TestBlazegraphStoreConstructExtension(unittest.TestCase):
store = _make_connected_store()
self.assertTrue(store._is_construct_query("Construct { ?s ?p ?o } Where { ?s ?p ?o }"))
+ def test_is_construct_query_detects_complex_preambles(self):
+ # Permanent regression tests covering edge cases discovered during
+ # regex stress-testing (issue #7): multiline declarations, empty
+ # prefix namespaces, and inline comments embedded in the preamble.
+ store = _make_connected_store()
+ cases = {
+ "multiline_prefix": "PREFIX foaf:\n \nCONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }",
+ "empty_prefix_namespace": "PREFIX : CONSTRUCT { ?s ?p ?o }",
+ "inline_comment": "PREFIX ex: \n# inline comment\nCONSTRUCT { ?s ?p ?o }",
+ }
+ for name, query in cases.items():
+ with self.subTest(case=name):
+ self.assertTrue(store._is_construct_query(query))
+
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 }"))
@@ -402,10 +416,14 @@ class TestBlazegraphStoreConstructExtension(unittest.TestCase):
self.assertEqual(result["variables"], [])
self.assertEqual(result["metadata"]["result_format"], "construct")
- triples = set(result["triples"])
+ # "triples" is now a list of (s, p, o, metadata) 4-tuples. Both
+ # triples here are plain untyped literals/URIs, so metadata is {}.
+ triples = {(s, p, o) for s, p, o, _metadata in 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)
+ for _s, _p, _o, metadata in result["triples"]:
+ self.assertEqual(metadata, {})
def test_execute_sparql_construct_result_format_option_forces_construct_path(self):
# Even for a query that doesn't literally contain "CONSTRUCT",
@@ -456,10 +474,11 @@ class TestBlazegraphStoreConstructExtension(unittest.TestCase):
result = store.execute_sparql("CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }")
self.assertEqual(len(result["triples"]), 1)
- subject, predicate, obj = result["triples"][0]
+ subject, predicate, obj, metadata = 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")
+ self.assertEqual(metadata, {})
# --- Property 9: non-CONSTRUCT queries are byte-for-byte unaffected ---
diff --git a/tests/triplet_store/test_construct_templates.py b/tests/triplet_store/test_construct_templates.py
index fbe9d1cd..69fb5443 100644
--- a/tests/triplet_store/test_construct_templates.py
+++ b/tests/triplet_store/test_construct_templates.py
@@ -426,6 +426,28 @@ class TestRenderConstructTemplateRequiredParameters:
with pytest.raises(ValidationError):
render_construct_template(template, params={"value": "x"})
+ def test_unexpected_param_key_raises_and_names_the_key(self):
+ # Fix 1: params containing a key with no matching ParameterDescriptor
+ # must raise ValidationError naming the unexpected key(s), rather
+ # than silently ignoring the extra key.
+ template = _simple_template()
+ with pytest.raises(ValidationError) as exc_info:
+ render_construct_template(
+ template, params={"value": "x", "subjcet": "typo-of-subject"}
+ )
+ assert "subjcet" in str(exc_info.value)
+
+ def test_multiple_unexpected_param_keys_all_named_in_error(self):
+ template = _simple_template()
+ with pytest.raises(ValidationError) as exc_info:
+ render_construct_template(
+ template,
+ params={"value": "x", "extra_one": 1, "extra_two": 2},
+ )
+ message = str(exc_info.value)
+ assert "extra_one" in message
+ assert "extra_two" in message
+
# ---------------------------------------------------------------------------
# _split_construct_query / _find_matching_brace: CONSTRUCT/WHERE boundary
@@ -613,7 +635,14 @@ class _StubStoreBackend:
"""
def __init__(self, triples, add_triplets_result=None, execute_sparql_error=None):
- self._triples = triples
+ # Normalize plain (s, p, o) 3-tuples to the real (s, p, o, metadata)
+ # 4-tuple shape execute_sparql's CONSTRUCT path now returns, so
+ # existing call sites that only care about subject/predicate/object
+ # don't need to be rewritten; tests exercising metadata pass real
+ # 4-tuples directly and are left untouched here.
+ self._triples = [
+ t if len(t) == 4 else (t[0], t[1], t[2], {}) for t in triples
+ ]
self._add_triplets_result = add_triplets_result or {"success": True}
self._execute_sparql_error = execute_sparql_error
self.execute_sparql_calls = []
@@ -674,11 +703,95 @@ class TestExecuteConstructTemplate(unittest.TestCase):
# 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)
+ def test_datatype_and_language_metadata_round_trip_through_real_execute_sparql(self):
+ # End-to-end round-trip: a real BlazegraphStore.execute_sparql CONSTRUCT
+ # response (parsed via rdflib from a Turtle fixture containing an
+ # xsd:integer-typed literal and an @en-language-tagged literal) must
+ # carry that datatype/language info through execute_construct_template's
+ # Triplet conversion — not just matching lexical string values.
+ from unittest.mock import MagicMock, patch as mock_patch
+ from semantica.triplet_store.blazegraph_store import BlazegraphStore
+
+ turtle_fixture = (
+ b"@prefix ex: .\n"
+ b'ex:s1 ex:p_age 42 .\n'
+ b'ex:s2 ex:p_label "hello"@en .\n'
+ )
+ mock_response = MagicMock()
+ mock_response.content = turtle_fixture
+ mock_response.raise_for_status = MagicMock()
+
+ with mock_patch.object(BlazegraphStore, "_connect", autospec=True):
+ store = BlazegraphStore(endpoint="http://localhost:9999/blazegraph")
+ store.connected = True
+
+ with mock_patch(
+ "semantica.triplet_store.blazegraph_store.requests.post",
+ return_value=mock_response,
+ ):
+ real_query_result = store.execute_sparql(
+ "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"
+ )
+
+ # Sanity check on the raw execute_sparql output itself before going
+ # through execute_construct_template, so a failure here pinpoints
+ # whether the bug is in execute_sparql's metadata extraction or in
+ # execute_construct_template's consumption of it.
+ raw_by_subject = {s: (s, p, o, meta) for s, p, o, meta in real_query_result["triples"]}
+ self.assertEqual(
+ raw_by_subject["http://ex.org/s1"][3].get("datatype"),
+ "http://www.w3.org/2001/XMLSchema#integer",
+ )
+ self.assertEqual(raw_by_subject["http://ex.org/s2"][3].get("language"), "en")
+
+ class _RealResultStub:
+ def execute_sparql(self, query, **options):
+ return real_query_result
+
+ def add_triplets(self, triplets, **options):
+ self.persisted = triplets
+ return {"success": True}
+
+ stub = _RealResultStub()
+ template = self._template()
+
+ result = execute_construct_template(template, {"value": "x"}, stub)
+
+ by_subject = {t.subject: t for t in result}
+
+ age_triplet = by_subject["http://ex.org/s1"]
+ self.assertEqual(age_triplet.object, "42")
+ self.assertEqual(
+ age_triplet.metadata.get("datatype"),
+ "http://www.w3.org/2001/XMLSchema#integer",
+ )
+ self.assertNotIn("lang", age_triplet.metadata)
+
+ label_triplet = by_subject["http://ex.org/s2"]
+ self.assertEqual(label_triplet.object, "hello")
+ self.assertEqual(label_triplet.metadata.get("lang"), "en")
+ self.assertNotIn("datatype", label_triplet.metadata)
+
+ # Confirm the round-tripped metadata is exactly what
+ # BlazegraphStore._format_object_for_sparql reads when re-serializing
+ # (metadata.get("datatype") / metadata.get("lang")), proving this is
+ # not just a string match but a genuine type-preserving round trip
+ # usable for real re-persistence.
+ rendered_age = store._format_object_for_sparql(age_triplet)
+ self.assertEqual(
+ rendered_age,
+ '"42"^^',
+ )
+ rendered_label = store._format_object_for_sparql(label_triplet)
+ self.assertEqual(rendered_label, '"hello"@en')
+
+ # add_triplets was called with exactly the persisted triplets
+ # (same objects that were returned), confirming the round-tripped
+ # metadata made it all the way through the persistence step too.
+ self.assertEqual(
+ [(t.subject, t.predicate, t.object) for t in stub.persisted],
+ [(t.subject, t.predicate, t.object) for t in result],
+ )
# --- 2. Missing add_triplets ---
@@ -734,6 +847,60 @@ class TestExecuteConstructTemplate(unittest.TestCase):
# the exception itself; there is no return value to inspect because
# execute_construct_template raised instead of returning.
+ # --- Fix 3: execute_sparql success=False must be checked before triples
+ # conversion, not silently treated as an empty-triples success ---
+
+ def test_execute_sparql_success_false_raises_processing_error(self):
+ class _FailedExecuteSparqlBackend:
+ def __init__(self):
+ self.add_triplets_calls = []
+
+ def execute_sparql(self, query, **options):
+ return {
+ "success": False,
+ "error": "Blazegraph returned HTTP 500",
+ "bindings": [],
+ "variables": [],
+ }
+
+ def add_triplets(self, triplets, **options):
+ self.add_triplets_calls.append((triplets, options))
+ return {"success": True}
+
+ stub = _FailedExecuteSparqlBackend()
+ template = self._template()
+
+ with self.assertRaises(ProcessingError):
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ # The failure must be caught before ever reaching triples conversion
+ # or persistence — add_triplets must never be called.
+ self.assertEqual(stub.add_triplets_calls, [])
+
+ def test_execute_sparql_success_false_with_no_triples_key_still_raises(self):
+ # Even if a failed backend response happens to omit "triples"
+ # entirely (rather than including an empty list), success=False
+ # alone must be sufficient to raise — the bug being fixed is
+ # "success is never checked", not "triples key is missing".
+ class _FailedNoTriplesKeyBackend:
+ def __init__(self):
+ self.add_triplets_calls = []
+
+ def execute_sparql(self, query, **options):
+ return {"success": False}
+
+ def add_triplets(self, triplets, **options):
+ self.add_triplets_calls.append((triplets, options))
+ return {"success": True}
+
+ stub = _FailedNoTriplesKeyBackend()
+ template = self._template()
+
+ with self.assertRaises(ProcessingError):
+ execute_construct_template(template, {"value": "x"}, stub)
+
+ self.assertEqual(stub.add_triplets_calls, [])
+
# --- 5. execute_sparql raises an exception ---
def test_execute_sparql_exception_propagates_without_being_swallowed(self):
@@ -797,6 +964,61 @@ class TestExecuteConstructTemplate(unittest.TestCase):
_, add_triplets_options = stub.add_triplets_calls[0]
self.assertIsNone(add_triplets_options["graph"])
+ # --- Fix 4: caller-supplied "result_format"/"graph" in **options must
+ # not crash with a duplicate-keyword TypeError, and must be overridden
+ # by this function's own required values rather than silently honored ---
+
+ def test_options_result_format_collision_does_not_raise_and_is_overridden(self):
+ stub = _StubStoreBackend(triples=[("http://ex.org/s1", "http://ex.org/p1", "v1")])
+ template = self._template()
+
+ # Caller passes a conflicting result_format in **options; this must
+ # not raise "got multiple values for keyword argument 'result_format'".
+ execute_construct_template(
+ template, {"value": "x"}, stub, result_format="bindings"
+ )
+
+ # The explicit "construct" value must have won — not the caller's
+ # "bindings" override.
+ _, execute_sparql_options = stub.execute_sparql_calls[0]
+ self.assertEqual(execute_sparql_options["result_format"], "construct")
+
+ def test_options_graph_collision_does_not_raise_and_is_overridden(self):
+ stub = _StubStoreBackend(triples=[("http://ex.org/s1", "http://ex.org/p1", "v1")])
+ template = self._template(target_graph="http://ex.org/graphs/correct")
+
+ # Caller passes a conflicting graph in **options; this must not
+ # raise "got multiple values for keyword argument 'graph'".
+ execute_construct_template(
+ template, {"value": "x"}, stub, graph="http://ex.org/graphs/wrong"
+ )
+
+ # The correctly-resolved effective_graph must have won — not the
+ # caller's "wrong" override passed via **options.
+ _, add_triplets_options = stub.add_triplets_calls[0]
+ self.assertEqual(add_triplets_options["graph"], "http://ex.org/graphs/correct")
+
+ def test_options_result_format_and_graph_collision_both_overridden_together(self):
+ stub = _StubStoreBackend(triples=[("http://ex.org/s1", "http://ex.org/p1", "v1")])
+ template = self._template()
+
+ # Both keys colliding at once, combined with an explicit target_graph
+ # argument (which itself must win over template.target_graph too).
+ execute_construct_template(
+ template,
+ {"value": "x"},
+ stub,
+ target_graph="http://ex.org/graphs/explicit",
+ result_format="bindings",
+ graph="http://ex.org/graphs/wrong",
+ )
+
+ _, execute_sparql_options = stub.execute_sparql_calls[0]
+ self.assertEqual(execute_sparql_options["result_format"], "construct")
+
+ _, add_triplets_options = stub.add_triplets_calls[0]
+ self.assertEqual(add_triplets_options["graph"], "http://ex.org/graphs/explicit")
+
# --- 7. render_construct_template's ValidationError propagates unchanged ---
def test_missing_required_param_validation_error_propagates_unwrapped(self):