mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Address Qodo review: reject unknown params, preserve literal datatype/lang, check backend success, fix options collision, tighten CONSTRUCT detection, fix docs, add validator integration for construct_template steps

Browse Source
This commit is contained in:
Sameer6305
2026-07-18 14:44:33 +05:30
parent c4e971c91c
commit 4f2c6c8229
8 changed files with 852 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/reference/triplet_store.md
+2 -2
View File
@@ -297,8 +297,8 @@ template = ConstructTemplate(
description="Maps a person record subject to a foaf:name triple",
construct_query="""
PREFIX foaf: <http://xmlns.com/foaf/0.1/>
CONSTRUCT { <{{subject}}> foaf:name {{name}} ; foaf:age {{age}} }
WHERE { <{{subject}}> a <http://ex.org/Person> }
CONSTRUCT { {{subject}} foaf:name {{name}} ; foaf:age {{age}} }
WHERE { {{subject}} a <http://ex.org/Person> }
""",
parameters=[
ParameterDescriptor(name="subject", type="uri", required=True),
semantica/pipeline/pipeline_validator.py
+72 -3
View File
@@ -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
# '_<type>_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
)
semantica/triplet_store/blazegraph_store.py
+40 -5
View File
@@ -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,
semantica/triplet_store/construct_templates.py
+109 -8
View File
@@ -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(
semantica/triplet_store/query_engine.py
+6 -3
View File
@@ -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
tests/pipeline/test_pipeline_validator_construct.py
+374
View File
@@ -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()
tests/triplet_store/test_blazegraph_store.py
+21 -2
View File
@@ -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 <http://xmlns.com/foaf/0.1/>\nCONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }",
"empty_prefix_namespace": "PREFIX : <http://ex.org/> CONSTRUCT { ?s ?p ?o }",
"inline_comment": "PREFIX ex: <http://ex.org/>\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 ---
tests/triplet_store/test_construct_templates.py
+228 -6
View File
@@ -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: <http://ex.org/> .\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"^^<http://www.w3.org/2001/XMLSchema#integer>',
)
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):