From 36f22bb3af2610ddfd0d3f289ea41c1ffdbaf6dc Mon Sep 17 00:00:00 2001 From: ecrum19 Date: Fri, 25 Sep 2026 18:44:12 +0200 Subject: [PATCH 1/3] Gate the shape layer on the graph, and let Node's heap be raised Two memory guards, each fixed after it failed on the same 170,935,101-triple graph. The shape-layer gate measured the wrong thing. pyshacl is in-memory, so the wrapper size-gates it -- but on the PACKAGED artifact's bytes, while what pyshacl pays for is the graph. One graph therefore landed on both sides of one 512 MiB gate purely by packaging: cottas 390,728,158 B under -> shapes attempted -> OOM nt.gz 756,594,166 B over -> skipped hdt 1,182,206,289 B over -> skipped The COTTAS run was SIGKILLed at 32.2 GB RSS on a 31 GB machine, after COTTAS decoding and rapper had both succeeded on every triple -- nothing was wrong with the artifact or the engine. The better a format compresses, the likelier it was to exhaust memory, so the guard inverted exactly where it was needed most. The authoritative gate now lives in the runner, which is the only place that knows what pyshacl will load, and counts triples, which no packaging can change. --shacl-max-triples defaults to 50M: above the largest graph the published campaign validated with shapes (17.1M) and below the one that exhausted 31 GB. A skip is recorded with its reason rather than being silent. Node chose a ceiling the machine did not. On the same graph the HDT endpoint aborted with "Reached heap limit Allocation failed - JavaScript heap out of memory" on the sample-level query while ~25 GB was free; the kernel OOM killer was never involved. Node does not size its old-space from the host, and nothing in the image, runner or wrapper set NODE_OPTIONS. --node-heap-mb now sets it for the three Comunica-backed engines. It is unset by default, so every published measurement keeps the conditions it was produced under. Co-Authored-By: Claude Opus 5 --- src/validation/validation_runner.py | 126 ++++++++++++++++++- test/test_shacl_gate_and_node_heap_unit.py | 138 +++++++++++++++++++++ vcf_rdfizer.py | 37 ++++++ 3 files changed, 300 insertions(+), 1 deletion(-) create mode 100644 test/test_shacl_gate_and_node_heap_unit.py diff --git a/src/validation/validation_runner.py b/src/validation/validation_runner.py index e830f84..b8f034d 100644 --- a/src/validation/validation_runner.py +++ b/src/validation/validation_runner.py @@ -2024,6 +2024,64 @@ def merge_shapes_graph(shapes: list[Path]): return graph +#: pyshacl loads the whole graph into memory, so the shape layer needs a size +#: gate. This one is in triples rather than bytes, because what pyshacl pays +#: for is the graph, not the packaging -- see the comment at its call site for +#: the failure that motivated it. +#: +#: 50M is chosen to sit above the largest graph the published campaign actually +#: validated with shapes (17.1M triples, the 100,000-record HG005 slice) and +#: below the one that exhausted 31 GB (170.9M). It is deliberately not derived +#: from measured bytes-per-triple: rdflib's footprint depends on term sharing +#: and IRI length, so a constant here is a conservative guard rather than a +#: prediction, and it is overridable for a machine that can afford more. +DEFAULT_SHACL_MAX_TRIPLES = 50_000_000 + +#: Node's V8 heap ceiling for the Comunica-backed endpoints (comunica, hdt, +#: cottas -- all three go through ComunicaHttpEndpointMixin). +#: +#: Node sizes its old-space from a default that does not track the machine, so +#: on a 31 GB host the endpoint still died with "Reached heap limit Allocation +#: failed - JavaScript heap out of memory" on the sample-level query of a +#: 170.9M-triple HDT graph, with 25 GB free. The kernel OOM killer was never +#: involved: the process aborted itself inside a ceiling it chose. +#: +#: None keeps Node's default, which is the behaviour every published result was +#: produced under. Set a value to raise it. +DEFAULT_NODE_HEAP_MB: int | None = None + + +def node_endpoint_env(heap_mb: int | None, base: dict[str, str] | None = None) -> dict[str, str]: + """Environment for a Comunica endpoint, with an optional heap ceiling. + + Appends rather than replaces NODE_OPTIONS, so a caller's own setting is + kept and only the heap is added. + """ + env = dict(base if base is not None else os.environ) + if not heap_mb: + return env + option = f"--max-old-space-size={heap_mb}" + existing = env.get("NODE_OPTIONS", "").strip() + env["NODE_OPTIONS"] = f"{existing} {option}".strip() if existing else option + return env + + + +def shacl_exceeds_limit(triple_count: int | None, limit: int | None) -> bool: + """Whether a decoded graph is too large to hand to pyshacl. + + An unknown count is treated as too large, for the same reason the wrapper + treats an unreadable size that way: skipping a check is recoverable and is + recorded, while exhausting memory mid-run loses the whole validation. + A limit of 0 disables the gate. + """ + if not limit: + return False + if triple_count is None: + return True + return triple_count > limit + + def validate_shacl( source: Path, shapes: Path | list[Path], @@ -2435,6 +2493,9 @@ def _init_endpoint(self, options: dict[str, Any]) -> None: options.get("comunica_warmup_timeout") or max(DEFAULT_COMUNICA_WARMUP_TIMEOUT, self.query_timeout) ) + # None leaves Node's own default, which is what every published result + # was produced under. + self.node_heap_mb = options.get("node_heap_mb") or DEFAULT_NODE_HEAP_MB self.server: subprocess.Popen | None = None self.executable: str | None = None self.endpoint: str | None = None @@ -2486,6 +2547,7 @@ def _start_endpoint(self) -> None: stdout=server_log.open("wb"), stderr=subprocess.STDOUT, start_new_session=True, + env=node_endpoint_env(getattr(self, "node_heap_mb", None)), ) self._await_bind(server_log) self._await_warm(server_log) @@ -3958,11 +4020,49 @@ def run_validation(args: argparse.Namespace) -> int: ) rdf_validation = validate_ntriples(decoded, results_dir) shacl_result = None - if args.shacl_shapes is not None: + # The authoritative size gate for pyshacl, because this is the only + # point that knows what pyshacl will actually load. + # + # The wrapper also gates, on the PACKAGED artifact's bytes, and that + # measure is wrong in a way that bites hardest on the best format. + # On a 170,935,101-triple graph the same graph landed on both sides + # of the wrapper's 512 MiB gate purely by packaging: + # + # cottas 390,728,158 B under -> shapes attempted -> OOM + # nt.gz 756,594,166 B over -> skipped + # hdt 1,182,206,289 B over -> skipped + # + # The COTTAS run was SIGKILLed at 32.2 GB RSS on a 31 GB machine + # after decoding and rapper had both succeeded on every triple. So + # the better a format compresses, the likelier it was to exhaust + # memory -- the guard inverted. Gating on the triple count fixes + # that, because a graph's cost to pyshacl does not depend on how it + # arrived. + decoded_triples = rdf_validation.get("tripleCount") + shacl_skipped: dict[str, Any] | None = None + if args.shacl_shapes is not None and shacl_exceeds_limit( + decoded_triples, args.shacl_max_triples + ): + shacl_skipped = { + "status": "SKIPPED_TOO_LARGE", + "tripleCount": decoded_triples, + "limitTriples": args.shacl_max_triples, + "reason": ( + f"the decoded graph holds {decoded_triples:,} triples, above the " + f"--shacl-max-triples limit of {args.shacl_max_triples:,}. " + f"pyshacl is in-memory, so attempting it risks exhausting memory " + f"mid-run; skipping is recoverable and is recorded here." + ), + } + eprint(f"[{args.dataset_id}] shapes skipped: {shacl_skipped['reason']}") + if args.shacl_shapes is not None and shacl_skipped is None: progress.emit("progress", completed=0, detail="validating SHACL shapes") shacl_result = validate_shacl( decoded, args.shacl_shapes, results_dir, args.shacl_ontology ) + if shacl_skipped is not None: + shacl_result = shacl_skipped + write_json(results_dir / "shacl.json", shacl_skipped) write_json(results_dir / "rdf-validation.json", rdf_validation) materialization["decodedTripleCount"] = rdf_validation.get("tripleCount") write_json(results_dir / "materialization.json", materialization) @@ -3976,6 +4076,7 @@ def run_validation(args: argparse.Namespace) -> int: "comunica_port": args.comunica_port, "hdt_port": args.hdt_port, "comunica_bind_timeout": args.comunica_bind_timeout, + "node_heap_mb": args.node_heap_mb, "comunica_warmup_timeout": args.comunica_warmup_timeout, } engine_options["artifact_path"] = str(args.rdf) @@ -4351,6 +4452,29 @@ def build_arg_parser() -> argparse.ArgumentParser: "checked." ), ) + parser.add_argument( + "--node-heap-mb", + type=int, + default=DEFAULT_NODE_HEAP_MB, + help=( + "Raise the V8 old-space ceiling (MB) for the Comunica-backed " + "endpoints (comunica, hdt, cottas). Node does not size its heap " + "from the machine, so an endpoint can abort with a JavaScript " + "heap-out-of-memory while the host still has free memory. Unset " + "keeps Node's default" + ), + ) + parser.add_argument( + "--shacl-max-triples", + type=int, + default=DEFAULT_SHACL_MAX_TRIPLES, + help=( + "Skip the shape layer when the decoded graph holds more triples " + "than this, recording the skip and its reason (0 disables the " + "gate). pyshacl is in-memory, and its cost tracks the graph rather " + f"than the artifact it arrived in (default: {DEFAULT_SHACL_MAX_TRIPLES:,})" + ), + ) parser.add_argument("--filter-oracle", choices=("auto", "bcftools", "cyvcf2"), default="auto") parser.add_argument("--scratch-dir", type=Path, default=Path("/work")) parser.add_argument( diff --git a/test/test_shacl_gate_and_node_heap_unit.py b/test/test_shacl_gate_and_node_heap_unit.py new file mode 100644 index 0000000..a07e812 --- /dev/null +++ b/test/test_shacl_gate_and_node_heap_unit.py @@ -0,0 +1,138 @@ +"""Two memory guards, each fixed after it failed on a 170.9M-triple graph. + +**The shape-layer gate measured the wrong thing.** pyshacl is in-memory, so the +wrapper size-gates it -- but on the PACKAGED artifact's bytes, while what +pyshacl pays for is the graph. One 170,935,101-triple graph therefore landed on +both sides of one 512 MiB gate purely by packaging: + + cottas 390,728,158 B under -> shapes attempted -> OOM + nt.gz 756,594,166 B over -> skipped + hdt 1,182,206,289 B over -> skipped + +The COTTAS run was SIGKILLed at 32.2 GB RSS on a 31 GB machine, after COTTAS +decoding and rapper had both succeeded on every triple. The better a format +compresses, the likelier it was to exhaust memory: the guard inverted. The fix +gates on the decoded triple count, which no packaging can change. + +**Node chose a ceiling the machine did not.** On the same graph the HDT +endpoint aborted with "Reached heap limit Allocation failed - JavaScript heap +out of memory" while ~25 GB was free; the kernel OOM killer was never involved. +Node does not size its old-space from the host, so the endpoints now accept an +explicit ceiling. +""" + +import importlib.util +import os +import unittest +from pathlib import Path + +from test.helpers import VerboseTestCase + +RUNNER_PATH = Path(__file__).resolve().parents[1] / "src" / "validation" / "validation_runner.py" +_spec = importlib.util.spec_from_file_location("validation_runner_guards", RUNNER_PATH) +V = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(V) + +# The real numbers from the run that motivated the fix. +GRAPH_TRIPLES = 170_935_101 +ARTIFACT_BYTES = { + "cottas": 390_728_158, + "nt.gz": 756_594_166, + "hdt": 1_182_206_289, +} +WRAPPER_BYTE_GATE = 512 * 1024 * 1024 + + +class ShaclGateTests(VerboseTestCase): + def test_the_packaging_no_longer_decides(self): + """The property the bug violated: same graph, same verdict. + + Every packaging of one graph must get one answer. This is the + regression test for the inversion itself. + """ + verdicts = { + kind: V.shacl_exceeds_limit(GRAPH_TRIPLES, V.DEFAULT_SHACL_MAX_TRIPLES) + for kind in ARTIFACT_BYTES + } + self.assertEqual(set(verdicts.values()), {True}, + "the decoded graph is the same for every artifact, so " + "every artifact must reach the same decision") + + def test_the_old_byte_gate_did_disagree_across_packagings(self): + """Pins why the fix was needed, so nobody reinstates the byte gate. + + Not a test of current behaviour -- a record of the defect, expressed in + the numbers that produced it. + """ + byte_verdicts = {k: b <= WRAPPER_BYTE_GATE for k, b in ARTIFACT_BYTES.items()} + self.assertTrue(byte_verdicts["cottas"], "cottas slipped under the gate") + self.assertFalse(byte_verdicts["nt.gz"]) + self.assertFalse(byte_verdicts["hdt"]) + self.assertEqual(len(set(byte_verdicts.values())), 2, + "the byte gate split one graph three ways; that was the bug") + + def test_the_graph_that_oomed_is_refused(self): + self.assertTrue(V.shacl_exceeds_limit(GRAPH_TRIPLES, V.DEFAULT_SHACL_MAX_TRIPLES)) + + def test_the_campaigns_largest_validated_graph_still_passes(self): + """17.1M triples is the 100,000-record HG005 slice the campaign ran + shapes on. The gate must not retroactively disable published behaviour.""" + self.assertFalse(V.shacl_exceeds_limit(17_098_746, V.DEFAULT_SHACL_MAX_TRIPLES)) + + def test_an_unknown_count_is_treated_as_too_large(self): + """Skipping is recoverable and recorded; exhausting memory is not.""" + self.assertTrue(V.shacl_exceeds_limit(None, V.DEFAULT_SHACL_MAX_TRIPLES)) + + def test_zero_disables_the_gate(self): + for count in (0, 1, GRAPH_TRIPLES): + with self.subTest(count=count): + self.assertFalse(V.shacl_exceeds_limit(count, 0)) + self.assertFalse(V.shacl_exceeds_limit(None, 0)) + + def test_the_boundary_is_inclusive(self): + limit = V.DEFAULT_SHACL_MAX_TRIPLES + self.assertFalse(V.shacl_exceeds_limit(limit, limit)) + self.assertTrue(V.shacl_exceeds_limit(limit + 1, limit)) + + def test_the_default_sits_between_the_two_observations(self): + """It must admit what worked and refuse what died, or it is arbitrary.""" + self.assertGreater(V.DEFAULT_SHACL_MAX_TRIPLES, 17_098_746) + self.assertLess(V.DEFAULT_SHACL_MAX_TRIPLES, GRAPH_TRIPLES) + + +class NodeHeapEnvTests(VerboseTestCase): + def test_unset_leaves_the_environment_alone(self): + """Published results were produced under Node's default; keep it.""" + env = V.node_endpoint_env(None, {"PATH": "/usr/bin"}) + self.assertNotIn("NODE_OPTIONS", env) + self.assertEqual(env["PATH"], "/usr/bin") + + def test_zero_is_also_unset(self): + self.assertNotIn("NODE_OPTIONS", V.node_endpoint_env(0, {})) + + def test_a_ceiling_is_applied(self): + env = V.node_endpoint_env(16384, {}) + self.assertEqual(env["NODE_OPTIONS"], "--max-old-space-size=16384") + + def test_an_existing_setting_is_kept_not_replaced(self): + """A caller's own NODE_OPTIONS must survive; only the heap is added.""" + env = V.node_endpoint_env(8192, {"NODE_OPTIONS": "--enable-source-maps"}) + self.assertIn("--enable-source-maps", env["NODE_OPTIONS"]) + self.assertIn("--max-old-space-size=8192", env["NODE_OPTIONS"]) + + def test_the_base_environment_is_not_mutated(self): + base = {"NODE_OPTIONS": "--enable-source-maps"} + V.node_endpoint_env(4096, base) + self.assertEqual(base["NODE_OPTIONS"], "--enable-source-maps") + + def test_it_defaults_to_the_process_environment(self): + env = V.node_endpoint_env(None) + self.assertEqual(env.get("PATH"), os.environ.get("PATH")) + + def test_the_default_is_node_s_own(self): + """Unset by default, so this change alters no existing measurement.""" + self.assertIsNone(V.DEFAULT_NODE_HEAP_MB) + + +if __name__ == "__main__": + unittest.main() diff --git a/vcf_rdfizer.py b/vcf_rdfizer.py index 817b7d7..78bb7b8 100644 --- a/vcf_rdfizer.py +++ b/vcf_rdfizer.py @@ -8972,6 +8972,11 @@ def run_validation_mode( # A subset selection. The runner turns this into a TIMING_ONLY result # rather than a validation verdict -- see its --queries help. ("--queries", "queries"), + # The authoritative shape-layer gate lives in the runner, which is the + # only place that knows the decoded graph's size. The size gate here + # sees the packaged artifact and cannot. + ("--shacl-max-triples", "shacl_max_triples"), + ("--node-heap-mb", "node_heap_mb"), ): value = options.get(key) if value is not None: @@ -9580,6 +9585,24 @@ def main(): default=None, help="Per-query timeout in seconds for validation (default: engine default)", ) + parser.add_argument( + "--shacl-max-triples", + default=None, + help=( + "Skip the shape layer when the decoded graph exceeds this many " + "triples, recording the skip (0 disables). pyshacl is in-memory " + "and its cost tracks the graph, not the artifact it arrived in" + ), + ) + parser.add_argument( + "--node-heap-mb", + default=None, + help=( + "Raise the V8 old-space ceiling (MB) for the Comunica-backed " + "validation engines (comunica, hdt, cottas). Node does not size " + "its heap from the machine" + ), + ) parser.add_argument( "--validation-queries", default=None, @@ -9742,6 +9765,20 @@ def main(): validation_engine_options["stop_after_query_timeout"] = True if args.validation_queries is not None: validation_engine_options["queries"] = args.validation_queries + if args.shacl_max_triples is not None: + # 0 is meaningful here: it disables the gate, the way + # --validation-time-budget 0 means no ceiling. + try: + limit = int(args.shacl_max_triples) + except (TypeError, ValueError): + raise ValueError("--shacl-max-triples must be an integer") + if limit < 0: + raise ValueError("--shacl-max-triples must be zero or a positive integer") + validation_engine_options["shacl_max_triples"] = limit + if args.node_heap_mb is not None: + validation_engine_options["node_heap_mb"] = parse_positive_int( + args.node_heap_mb, name="--node-heap-mb" + ) if args.qlever_index_arg: validation_engine_options["qlever_index_args"] = list(args.qlever_index_arg) if args.qlever_server_arg: From 5ed75c8902513a02ec4b16b50e1513d333e3b84c Mon Sep 17 00:00:00 2001 From: ecrum19 Date: Fri, 25 Sep 2026 19:13:52 +0200 Subject: [PATCH 2/3] Read the new validation options with getattr, and pin it behaviourally CI failed on five tests from the already-merged query-selection feature: 'Namespace' object has no attribute 'node_heap_mb' run_validation is driven by hand-built argparse.Namespace objects in several places -- that test suite and the mutation harness among them -- so reading a new option by plain attribute access breaks those callers at runtime rather than at import. The file already guarded progress_path and quiet with getattr for exactly this reason; --shacl-max-triples and --node-heap-mb now do the same, and the query-selection fixture carries both fields so it still exercises the real path. The regression test is behavioural rather than textual. The first version asserted on the source of run_validation and failed twice on formatting -- once on a wrapped getattr, once on the space in "getattr( args," -- which constrains how the guard is written rather than that it works. It now drives run_validation with a Namespace deliberately missing both fields and asserts the run never dies with "has no attribute". Verified by reintroducing the bug: the test fails, and passes again once restored. A regression test that has never been seen to fail has not been tested. Full suite: 958 tests, OK. Co-Authored-By: Claude Opus 5 --- src/validation/validation_runner.py | 11 ++- test/test_query_selection_unit.py | 5 ++ test/test_shacl_gate_and_node_heap_unit.py | 87 ++++++++++++++++++++++ 3 files changed, 99 insertions(+), 4 deletions(-) diff --git a/src/validation/validation_runner.py b/src/validation/validation_runner.py index b8f034d..3987504 100644 --- a/src/validation/validation_runner.py +++ b/src/validation/validation_runner.py @@ -4040,16 +4040,19 @@ def run_validation(args: argparse.Namespace) -> int: # arrived. decoded_triples = rdf_validation.get("tripleCount") shacl_skipped: dict[str, Any] | None = None + shacl_limit = getattr( + args, "shacl_max_triples", DEFAULT_SHACL_MAX_TRIPLES + ) if args.shacl_shapes is not None and shacl_exceeds_limit( - decoded_triples, args.shacl_max_triples + decoded_triples, shacl_limit ): shacl_skipped = { "status": "SKIPPED_TOO_LARGE", "tripleCount": decoded_triples, - "limitTriples": args.shacl_max_triples, + "limitTriples": shacl_limit, "reason": ( f"the decoded graph holds {decoded_triples:,} triples, above the " - f"--shacl-max-triples limit of {args.shacl_max_triples:,}. " + f"--shacl-max-triples limit of {shacl_limit:,}. " f"pyshacl is in-memory, so attempting it risks exhausting memory " f"mid-run; skipping is recoverable and is recorded here." ), @@ -4076,7 +4079,7 @@ def run_validation(args: argparse.Namespace) -> int: "comunica_port": args.comunica_port, "hdt_port": args.hdt_port, "comunica_bind_timeout": args.comunica_bind_timeout, - "node_heap_mb": args.node_heap_mb, + "node_heap_mb": getattr(args, "node_heap_mb", DEFAULT_NODE_HEAP_MB), "comunica_warmup_timeout": args.comunica_warmup_timeout, } engine_options["artifact_path"] = str(args.rdf) diff --git a/test/test_query_selection_unit.py b/test/test_query_selection_unit.py index 57d6aba..5544e6e 100644 --- a/test/test_query_selection_unit.py +++ b/test/test_query_selection_unit.py @@ -422,6 +422,11 @@ def _args(self, tmp_path, queries): filter_oracle="cyvcf2", dataset_id="sample", queries=queries, + # Mirror the real argument surface. run_validation tolerates these + # being absent (see RunValidationToleratesAMinimalNamespaceTests), + # but a fixture that omits them stops exercising the real path. + shacl_max_triples=V.DEFAULT_SHACL_MAX_TRIPLES, + node_heap_mb=V.DEFAULT_NODE_HEAP_MB, ) def _run(self, queries, *, corrupt=None): diff --git a/test/test_shacl_gate_and_node_heap_unit.py b/test/test_shacl_gate_and_node_heap_unit.py index a07e812..1aaa067 100644 --- a/test/test_shacl_gate_and_node_heap_unit.py +++ b/test/test_shacl_gate_and_node_heap_unit.py @@ -136,3 +136,90 @@ def test_the_default_is_node_s_own(self): if __name__ == "__main__": unittest.main() + + +class RunValidationToleratesAMinimalNamespaceTests(VerboseTestCase): + """A new option must not become a required attribute of run_validation. + + run_validation is driven by hand-built namespaces in several places -- this + suite and the mutation harness among them -- so reading a new option with + plain attribute access breaks those callers at runtime rather than at + import. That is exactly how adding --shacl-max-triples and --node-heap-mb + broke CI: five tests failed with "'Namespace' object has no attribute + 'node_heap_mb'", and the failure surfaced in a merged feature's tests + rather than in the change that caused it. + + The convention the file already used for progress_path and quiet is + getattr with the documented default. This pins it, so the next option + added cannot reintroduce the same break silently. + """ + + def test_run_validation_accepts_a_namespace_without_the_new_options(self): + """Behavioural, not textual: drive run_validation and look for the break. + + An earlier version of this test asserted on the source text and failed + twice on formatting, which is the wrong instrument -- it constrains how + the guard is written rather than that it works. + """ + import argparse + import json + import tempfile + from unittest import mock + + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + (tmp / "scratch").mkdir() + (tmp / "s.vcf").write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n") + (tmp / "s.nt").write_text("

.\n") + # Deliberately missing shacl_max_triples and node_heap_mb. + args = argparse.Namespace( + results_dir=tmp / "results", representation="expanded", + info_representation="structured", header_representation="structured", + progress_path=None, quiet=True, scratch_dir=tmp / "scratch", + rdf=tmp / "s.nt", rdf_format="nt", engine="comunica", + engines=["comunica"], mapping_policy="strict", + strict_conformance=False, shacl_shapes=None, + query_timeout=60, validation_time_budget=0, + stop_after_query_timeout=False, qlever_memory_gb=4, + qlever_port=7019, comunica_port=7020, hdt_port=7021, + comunica_bind_timeout=60, comunica_warmup_timeout=60, + qlever_startup_timeout=60, qlever_index_arg=[], + qlever_server_arg=[], vcf=tmp / "s.vcf", + filter_oracle="cyvcf2", dataset_id="sample", queries=None, + ) + engine = mock.MagicMock() + engine.describe.return_value = {"engine": "comunica"} + engine.execute.return_value = {"status": "FAILED"} + with mock.patch.object(V, "parse_vcf", return_value={ + "totalRecords": 1, "sampleCount": 0, "gtRecordCount": 0, + "sourceSha256": "0" * 64}), \ + mock.patch.object(V, "attach_census_expectations", + side_effect=lambda p, *a, **k: p), \ + mock.patch.object(V, "validate_ntriples", + return_value={"status": "PASS", "tripleCount": 1}), \ + mock.patch.object(V, "materialize_ntriples", + return_value=(tmp / "s.nt", {})), \ + mock.patch.object(V, "build_manifest", return_value={}), \ + mock.patch.object(V, "build_engine", return_value=engine): + V.run_validation(args) + summary = json.loads((args.results_dir / "summary.json").read_text()) + + # The run may fail for its own reasons -- the engine here is a stub -- + # but never because an option was read by attribute. + self.assertNotIn( + "has no attribute", str(summary.get("error") or ""), + "run_validation read a new option by attribute; use getattr with " + "its documented default so callers that build their own Namespace " + "keep working", + ) + + def test_a_namespace_without_them_still_resolves(self): + """The behaviour the getattr guard buys, checked rather than assumed.""" + import argparse + + args = argparse.Namespace() + self.assertEqual( + getattr(args, "shacl_max_triples", V.DEFAULT_SHACL_MAX_TRIPLES), + V.DEFAULT_SHACL_MAX_TRIPLES, + ) + self.assertIsNone(getattr(args, "node_heap_mb", V.DEFAULT_NODE_HEAP_MB)) From 3f71ef06a27cd1331504d46a22b37d1ebdaaf929 Mon Sep 17 00:00:00 2001 From: ecrum19 Date: Tue, 29 Sep 2026 08:45:14 +0200 Subject: [PATCH 3/3] Cover the shape gate and Node heap end to end, and fix the unknown-count crash The existing tests pinned the two pure helpers. Nothing exercised the code that acts on them: the gate inside run_validation, the heap reaching the spawned endpoint, or either option crossing the wrapper and the container boundary. Every line and branch the PR adds is now covered, and each of 12 deliberate mutations (env dropped from Popen, a falsy 0 dropped by the wrapper, an exclusive boundary, ...) fails at least one test. The new tests found a crash. shacl_exceeds_limit treats an unknown triple count as too large, and the skip reason then formatted it with f"{None:,}", raising TypeError. With shapes requested, that turned two outcomes into EXECUTION_FAILED: rapper passing without a parsable count line, and rapper missing -- which should have been BLOCKED_BY_PREFLIGHT, because the gate runs before the preflight check. The skip is now recorded with a reason that says the count is unknown. Also moves the test file's __main__ guard to the end, so running it directly no longer skips the class defined below it. Full unit suite: 994 tests, OK. Mutation suite: 116 tests, OK. Co-Authored-By: Claude Opus 5.5 --- src/validation/validation_runner.py | 20 +- test/test_shacl_gate_and_node_heap_unit.py | 503 ++++++++++++++++++++- 2 files changed, 517 insertions(+), 6 deletions(-) diff --git a/src/validation/validation_runner.py b/src/validation/validation_runner.py index 3987504..73d1005 100644 --- a/src/validation/validation_runner.py +++ b/src/validation/validation_runner.py @@ -4046,15 +4046,27 @@ def run_validation(args: argparse.Namespace) -> int: if args.shacl_shapes is not None and shacl_exceeds_limit( decoded_triples, shacl_limit ): + # An unknown count is refused too (see shacl_exceeds_limit), so + # the reason must not assume there is a number to format. + if decoded_triples is None: + size = ( + "the decoded graph's triple count is unknown, so it cannot " + f"be shown to be within the --shacl-max-triples limit of " + f"{shacl_limit:,}" + ) + else: + size = ( + f"the decoded graph holds {decoded_triples:,} triples, above " + f"the --shacl-max-triples limit of {shacl_limit:,}" + ) shacl_skipped = { "status": "SKIPPED_TOO_LARGE", "tripleCount": decoded_triples, "limitTriples": shacl_limit, "reason": ( - f"the decoded graph holds {decoded_triples:,} triples, above the " - f"--shacl-max-triples limit of {shacl_limit:,}. " - f"pyshacl is in-memory, so attempting it risks exhausting memory " - f"mid-run; skipping is recoverable and is recorded here." + f"{size}. pyshacl is in-memory, so attempting it risks " + f"exhausting memory mid-run; skipping is recoverable and is " + f"recorded here." ), } eprint(f"[{args.dataset_id}] shapes skipped: {shacl_skipped['reason']}") diff --git a/test/test_shacl_gate_and_node_heap_unit.py b/test/test_shacl_gate_and_node_heap_unit.py index 1aaa067..da7dc79 100644 --- a/test/test_shacl_gate_and_node_heap_unit.py +++ b/test/test_shacl_gate_and_node_heap_unit.py @@ -21,12 +21,23 @@ explicit ceiling. """ +import argparse +import copy import importlib.util +import json import os +import sys +import tempfile import unittest +from contextlib import redirect_stderr, redirect_stdout +from io import StringIO from pathlib import Path +from unittest import mock +import vcf_rdfizer +from test import validation_fixtures as fixtures from test.helpers import VerboseTestCase +from test.test_query_selection_unit import bindings_for RUNNER_PATH = Path(__file__).resolve().parents[1] / "src" / "validation" / "validation_runner.py" _spec = importlib.util.spec_from_file_location("validation_runner_guards", RUNNER_PATH) @@ -133,9 +144,23 @@ def test_the_default_is_node_s_own(self): """Unset by default, so this change alters no existing measurement.""" self.assertIsNone(V.DEFAULT_NODE_HEAP_MB) + def test_the_heap_is_appended_after_the_callers_options(self): + """Pinned exactly: one space, caller's options first, nothing dropped.""" + env = V.node_endpoint_env( + 8192, {"NODE_OPTIONS": "--enable-source-maps --trace-warnings"}) + self.assertEqual( + env["NODE_OPTIONS"], + "--enable-source-maps --trace-warnings --max-old-space-size=8192") -if __name__ == "__main__": - unittest.main() + def test_a_blank_existing_setting_is_not_kept_as_padding(self): + env = V.node_endpoint_env(2048, {"NODE_OPTIONS": " "}) + self.assertEqual(env["NODE_OPTIONS"], "--max-old-space-size=2048") + + def test_unset_returns_a_copy_not_the_process_environment(self): + """Callers may mutate what they get back without touching os.environ.""" + env = V.node_endpoint_env(None) + self.assertIsNot(env, os.environ) + self.assertIsInstance(env, dict) class RunValidationToleratesAMinimalNamespaceTests(VerboseTestCase): @@ -223,3 +248,477 @@ def test_a_namespace_without_them_still_resolves(self): V.DEFAULT_SHACL_MAX_TRIPLES, ) self.assertIsNone(getattr(args, "node_heap_mb", V.DEFAULT_NODE_HEAP_MB)) + + +#: Marks a field _runner_args must leave off the Namespace entirely. +_ABSENT = object() + + +def _runner_args(tmp: Path, **overrides) -> argparse.Namespace: + """The runner's full argument surface, with any field overridable or absent.""" + fields = dict( + results_dir=tmp / "results", representation="expanded", + info_representation="structured", header_representation="structured", + progress_path=None, quiet=True, scratch_dir=tmp / "scratch", + rdf=tmp / "s.nt", rdf_format="nt", engine="comunica", + engines=["comunica"], mapping_policy="strict", + strict_conformance=False, shacl_shapes=None, shacl_ontology=None, + query_timeout=60, validation_time_budget=0, + stop_after_query_timeout=False, qlever_memory_gb=4, + qlever_port=7019, comunica_port=7020, hdt_port=7021, + comunica_bind_timeout=60, comunica_warmup_timeout=60, + qlever_startup_timeout=60, qlever_index_arg=[], + qlever_server_arg=[], vcf=tmp / "s.vcf", + filter_oracle="cyvcf2", dataset_id="sample", queries=None, + shacl_max_triples=V.DEFAULT_SHACL_MAX_TRIPLES, + node_heap_mb=V.DEFAULT_NODE_HEAP_MB, + ) + fields.update(overrides) + return argparse.Namespace( + **{k: v for k, v in fields.items() if v is not _ABSENT}) + + +class RunValidationShaclGateTests(VerboseTestCase): + """The gate where it actually runs: inside run_validation. + + The helper tests above prove shacl_exceeds_limit decides correctly; these + prove run_validation acts on the decision -- that a refused graph never + reaches pyshacl, that the skip is recorded rather than silent, and that an + admitted graph still gets its shapes checked. + """ + + def _drive(self, *, rdf_validation=None, shapes=True, **overrides): + """Run the validator with every expensive stage stubbed out. + + Returns what a caller can observe afterwards: the exit code, the + summary, shacl.json (or None), the validate_shacl and build_engine + mocks, and stderr. + """ + if rdf_validation is None: + rdf_validation = {"status": "PASS", "tripleCount": 1} + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + (tmp / "scratch").mkdir() + (tmp / "s.vcf").write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n") + (tmp / "s.nt").write_text("

.\n") + shapes_path = tmp / "shapes.ttl" + shapes_path.write_text("") + if shapes: + overrides.setdefault("shacl_shapes", [shapes_path]) + args = _runner_args(tmp, **overrides) + # An engine that agrees with the oracle, so the run reaches the + # full summary rather than stopping on a query failure. + parser_summary = copy.deepcopy(fixtures.parser_summary("expanded")) + parser_summary.setdefault("sourceSha256", "0" * 64) + raw_dir = tmp / "raw" + raw_dir.mkdir() + query_file = tmp / "query.rq" + query_file.write_text("SELECT * WHERE { ?s ?p ?o }\n") + + def execute(query_id, _path): + raw = raw_dir / f"{query_id}.json" + answer = parser_summary.get(query_id) + raw.write_text(json.dumps( + bindings_for(answer) if answer is not None + else {"results": {"bindings": []}})) + return {"status": "PASS", "rawResult": str(raw), "wallSeconds": 0.01} + + engine = mock.MagicMock() + engine.describe.return_value = {"engine": "comunica", "setupSeconds": 0.1} + engine.execute.side_effect = execute + engine.query_timeout = 60 + stderr = StringIO() + with mock.patch.object(V, "query_path", return_value=query_file), \ + mock.patch.object(V, "parse_vcf", return_value=parser_summary), \ + mock.patch.object(V, "attach_census_expectations", + side_effect=lambda p, *a, **k: p), \ + mock.patch.object(V, "validate_ntriples", + return_value=dict(rdf_validation)), \ + mock.patch.object(V, "materialize_ntriples", + return_value=(tmp / "s.nt", {})), \ + mock.patch.object(V, "validate_shacl", return_value={ + "status": "PASS", "wallSeconds": 0.1}) as validate_shacl, \ + mock.patch.object(V, "build_manifest", return_value={}), \ + mock.patch.object(V, "build_engine", + return_value=engine) as build_engine, \ + redirect_stderr(stderr): + rc = V.run_validation(args) + results = args.results_dir + summary = json.loads((results / "summary.json").read_text()) + shacl_path = results / "shacl.json" + shacl = json.loads(shacl_path.read_text()) if shacl_path.exists() else None + return { + "rc": rc, "summary": summary, "shacl": shacl, + "validate_shacl": validate_shacl, "build_engine": build_engine, + "stderr": stderr.getvalue(), "decoded": tmp / "s.nt", + } + + def assertRunDidNotError(self, run): + self.assertIsNone( + run["summary"].get("error"), + f"run_validation raised instead of deciding: {run['summary'].get('error')}") + + def test_a_graph_above_the_limit_never_reaches_pyshacl(self): + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}) + self.assertRunDidNotError(run) + run["validate_shacl"].assert_not_called() + + def test_the_skip_is_recorded_with_its_reason(self): + """A skip nobody can see is the silent failure this replaces.""" + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}) + shacl = run["shacl"] + self.assertIsNotNone(shacl, "the skip must be written to shacl.json") + self.assertEqual(shacl["status"], "SKIPPED_TOO_LARGE") + self.assertEqual(shacl["tripleCount"], GRAPH_TRIPLES) + self.assertEqual(shacl["limitTriples"], V.DEFAULT_SHACL_MAX_TRIPLES) + # Both numbers, human-formatted, and the flag that changes the outcome. + self.assertIn(f"{GRAPH_TRIPLES:,}", shacl["reason"]) + self.assertIn(f"{V.DEFAULT_SHACL_MAX_TRIPLES:,}", shacl["reason"]) + self.assertIn("--shacl-max-triples", shacl["reason"]) + + def test_the_skip_is_announced_on_stderr(self): + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}) + self.assertIn("[sample] shapes skipped:", run["stderr"]) + + def test_the_skip_is_what_the_summary_reports(self): + """Downstream reads the summary, not shacl.json; it must agree.""" + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}) + self.assertEqual(run["summary"]["shacl"], run["shacl"]) + + def test_a_skip_is_not_a_shape_failure(self): + """SKIPPED must not be mistaken for FAIL and block the run on shapes.""" + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}) + self.assertNotEqual(run["summary"].get("status"), "BLOCKED_BY_PREFLIGHT") + + def test_a_graph_within_the_limit_still_gets_its_shapes_checked(self): + """The campaign's largest shape-validated graph, at the default gate.""" + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": 17_098_746}) + self.assertRunDidNotError(run) + run["validate_shacl"].assert_called_once() + self.assertEqual(run["validate_shacl"].call_args[0][0], run["decoded"]) + self.assertEqual(run["summary"]["shacl"]["status"], "PASS") + self.assertNotIn("shapes skipped", run["stderr"]) + + def test_the_limit_itself_is_admitted(self): + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": 1_000}, + shacl_max_triples=1_000) + run["validate_shacl"].assert_called_once() + + def test_one_over_the_limit_is_refused(self): + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": 1_001}, + shacl_max_triples=1_000) + run["validate_shacl"].assert_not_called() + self.assertEqual(run["shacl"]["limitTriples"], 1_000) + + def test_zero_disables_the_gate_in_the_run(self): + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}, + shacl_max_triples=0) + self.assertRunDidNotError(run) + run["validate_shacl"].assert_called_once() + + def test_an_unknown_count_is_skipped_and_recorded_not_crashed(self): + """rapper passed but its count line did not parse, so tripleCount is None. + + The helper treats that as too large, so the run must record a skip -- + not die while formatting the reason. + """ + run = self._drive(rdf_validation={"status": "PASS", "tripleCount": None}) + self.assertRunDidNotError(run) + run["validate_shacl"].assert_not_called() + self.assertEqual(run["shacl"]["status"], "SKIPPED_TOO_LARGE") + self.assertIsNone(run["shacl"]["tripleCount"]) + self.assertIn("unknown", run["shacl"]["reason"]) + + def test_a_missing_rapper_still_blocks_on_preflight_not_on_the_gate(self): + """No tripleCount key at all: the preflight verdict must survive.""" + run = self._drive(rdf_validation={ + "status": "EXECUTION_FAILED", "error": "rapper is not installed"}) + self.assertRunDidNotError(run) + self.assertEqual(run["summary"]["status"], "BLOCKED_BY_PREFLIGHT") + run["validate_shacl"].assert_not_called() + + def test_without_shapes_there_is_nothing_to_gate(self): + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}, + shapes=False) + self.assertRunDidNotError(run) + run["validate_shacl"].assert_not_called() + self.assertIsNone(run["shacl"], "no shapes asked for, so no shacl.json") + self.assertNotIn("shapes skipped", run["stderr"]) + + def test_a_namespace_without_the_limit_gets_the_default_gate(self): + """Absent means the documented default, not 'ungated'.""" + run = self._drive( + rdf_validation={"status": "PASS", "tripleCount": GRAPH_TRIPLES}, + shacl_max_triples=_ABSENT) + self.assertRunDidNotError(run) + run["validate_shacl"].assert_not_called() + self.assertEqual(run["shacl"]["limitTriples"], V.DEFAULT_SHACL_MAX_TRIPLES) + + def test_the_heap_ceiling_reaches_the_engine(self): + run = self._drive(shapes=False, node_heap_mb=16384) + options = run["build_engine"].call_args.kwargs["options"] + self.assertEqual(options["node_heap_mb"], 16384) + + def test_a_namespace_without_the_heap_gets_node_s_default(self): + run = self._drive(shapes=False, node_heap_mb=_ABSENT) + options = run["build_engine"].call_args.kwargs["options"] + self.assertIsNone(options["node_heap_mb"]) + + +class EndpointHeapTests(VerboseTestCase): + """The ceiling must reach the Node process, for all three endpoints. + + node_endpoint_env is only half the fix; the other half is that each + Comunica-backed engine reads the option and hands the environment to the + process it spawns. A ceiling that stops at engine construction is the same + wiring gap as a flag the wrapper never forwards. + """ + + ENGINES = ("comunica", "hdt", "cottas") + + def _engine(self, name, tmp, options): + return V.build_engine( + name, tmp / "graph.nt", raw_dir=tmp, scratch=tmp, options=options) + + def _spawn_env(self, name, options, environ): + """Start the endpoint with the process stubbed; return the env it got.""" + with tempfile.TemporaryDirectory() as td: + engine = self._engine(name, Path(td), options) + engine.executable = f"/usr/bin/{engine.endpoint_binary}" + with mock.patch.dict(os.environ, environ, clear=True), \ + mock.patch.object(type(engine), "_endpoint_source_argument", + return_value="graph"), \ + mock.patch.object(V.subprocess, "Popen") as popen, \ + mock.patch.object(V.ComunicaHttpEndpointMixin, "_await_bind"), \ + mock.patch.object(V.ComunicaHttpEndpointMixin, "_await_warm"): + engine._start_endpoint() + popen.return_value.stdout = None + return popen.call_args.kwargs["env"] + + def test_every_endpoint_engine_uses_the_shared_mixin(self): + """If one stops inheriting it, it silently stops honouring the ceiling.""" + for name in self.ENGINES: + with self.subTest(engine=name): + self.assertTrue(issubclass( + V.ENGINE_CLASSES[name], V.ComunicaHttpEndpointMixin)) + + def test_each_engine_reads_the_ceiling(self): + for name in self.ENGINES: + with self.subTest(engine=name), tempfile.TemporaryDirectory() as td: + engine = self._engine(name, Path(td), {"node_heap_mb": 12288}) + self.assertEqual(engine.node_heap_mb, 12288) + + def test_each_engine_defaults_to_node_s_own(self): + for options in ({}, {"node_heap_mb": None}, {"node_heap_mb": 0}): + for name in self.ENGINES: + with self.subTest(engine=name, options=options), \ + tempfile.TemporaryDirectory() as td: + engine = self._engine(name, Path(td), options) + self.assertIsNone(engine.node_heap_mb) + + def test_the_ceiling_reaches_the_spawned_process(self): + for name in self.ENGINES: + with self.subTest(engine=name): + env = self._spawn_env( + name, {"node_heap_mb": 12288}, + {"PATH": "/usr/bin", "NODE_OPTIONS": "--enable-source-maps"}) + self.assertEqual( + env["NODE_OPTIONS"], + "--enable-source-maps --max-old-space-size=12288") + self.assertEqual(env["PATH"], "/usr/bin") + + def test_unset_spawns_with_the_environment_unchanged(self): + for name in self.ENGINES: + with self.subTest(engine=name): + env = self._spawn_env(name, {}, {"PATH": "/usr/bin"}) + self.assertEqual(env, {"PATH": "/usr/bin"}) + + +class RunnerArgParsingTests(VerboseTestCase): + """Both options exist on the runner, with the documented defaults and types.""" + + def _parse(self, extra_argv): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + (tmp / "s.vcf").write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n") + (tmp / "s.nt").write_text("

.\n") + argv = [ + "validation_runner.py", "--vcf", str(tmp / "s.vcf"), + "--rdf", str(tmp / "s.nt"), "--representation", "expanded", + "--results-dir", str(tmp / "results"), "--dataset-id", "sample", + # The default is the container's /work, which does not exist + # here; without this the parse fails for the wrong reason. + "--scratch-dir", str(tmp), + *extra_argv, + ] + with mock.patch.object(sys, "argv", argv): + return V.parse_args() + + def test_the_defaults(self): + args = self._parse([]) + self.assertEqual(args.shacl_max_triples, V.DEFAULT_SHACL_MAX_TRIPLES) + self.assertIsNone(args.node_heap_mb) + + def test_both_parse_as_integers(self): + args = self._parse(["--shacl-max-triples", "0", "--node-heap-mb", "16384"]) + self.assertEqual(args.shacl_max_triples, 0) + self.assertEqual(args.node_heap_mb, 16384) + + def test_a_non_integer_exits_two(self): + for flag in ("--shacl-max-triples", "--node-heap-mb"): + with self.subTest(flag=flag): + stderr = StringIO() + with self.assertRaises(SystemExit) as caught, redirect_stderr(stderr): + self._parse([flag, "lots"]) + self.assertEqual(caught.exception.code, 2) + self.assertIn(flag, stderr.getvalue()) + + +class WrapperCliTests(VerboseTestCase): + """main() must validate both flags and put them into engine_options. + + A flag that parses but is never read is the wiring gap this project has + been bitten by before, so absence and presence are both pinned. + """ + + def _main(self, extra_argv): + """Run main() up to run_validation_mode; return (rc, engine_options, stderr).""" + captured = {} + + def capture(**kwargs): + captured.update(kwargs) + return 0 + + stderr = StringIO() + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + (tmp / "s.vcf").write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n") + (tmp / "s.nt").write_text("

.\n") + argv = [ + "vcf_rdfizer.py", "--mode", "validation", + "--input", str(tmp / "s.vcf"), "--rdf", str(tmp / "s.nt"), + "--out", str(tmp / "out"), + "--image", "example/vcf-rdfizer:test", "--no-build", + *extra_argv, + ] + with mock.patch.object(sys, "argv", argv), \ + mock.patch.object(vcf_rdfizer, "check_docker", return_value=True), \ + mock.patch.object(vcf_rdfizer, "docker_image_exists", + return_value=True), \ + mock.patch.object(vcf_rdfizer, "run_validation_mode", + side_effect=capture), \ + redirect_stderr(stderr), redirect_stdout(StringIO()): + rc = vcf_rdfizer.main() + return rc, captured.get("engine_options"), stderr.getvalue() + + def test_neither_flag_means_neither_option(self): + """Absent stays absent, so the runner applies its own defaults.""" + _rc, options, _err = self._main([]) + self.assertNotIn("shacl_max_triples", options) + self.assertNotIn("node_heap_mb", options) + + def test_the_triple_limit_is_forwarded_as_an_integer(self): + _rc, options, _err = self._main(["--shacl-max-triples", "100000000"]) + self.assertEqual(options["shacl_max_triples"], 100_000_000) + + def test_zero_is_kept_because_it_disables_the_gate(self): + _rc, options, _err = self._main(["--shacl-max-triples", "0"]) + self.assertEqual(options["shacl_max_triples"], 0) + + def test_a_bad_triple_limit_is_a_usage_error(self): + for value, message in ( + ("lots", "--shacl-max-triples must be an integer"), + ("1.5", "--shacl-max-triples must be an integer"), + ("-1", "--shacl-max-triples must be zero or a positive integer"), + ): + with self.subTest(value=value): + rc, options, err = self._main(["--shacl-max-triples", value]) + self.assertEqual(rc, 2) + self.assertIsNone(options, "a bad value must stop before the run") + self.assertIn(message, err) + + def test_the_heap_is_forwarded_as_an_integer(self): + _rc, options, _err = self._main(["--node-heap-mb", "16384"]) + self.assertEqual(options["node_heap_mb"], 16384) + + def test_a_bad_heap_is_a_usage_error(self): + """Unlike the triple limit, 0 is not meaningful here: unset is how you + keep Node's default.""" + for value in ("0", "-512", "lots"): + with self.subTest(value=value): + rc, options, err = self._main(["--node-heap-mb", value]) + self.assertEqual(rc, 2) + self.assertIsNone(options) + self.assertIn("--node-heap-mb must be a positive integer", err) + + +class WrapperForwardsTheGuardsTests(VerboseTestCase): + """Both options must cross the container boundary, and the runner must + accept what arrives.""" + + def _command_for(self, engine_options): + with tempfile.TemporaryDirectory() as td: + tmp = Path(td) + (tmp / "s.vcf").write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n") + (tmp / "s.nt").write_bytes(b"

.\n") + commands = [] + with mock.patch.object( + vcf_rdfizer, "run", + side_effect=lambda cmd, **kw: commands.append(cmd) or 0, + ), redirect_stdout(StringIO()): + vcf_rdfizer.run_validation_mode( + vcf_path=tmp / "s.vcf", rdf_path=tmp / "s.nt", + representation="expanded", + info_representation="structured", + header_representation="structured", + validation_id="s", results_dir=tmp / "results", + metrics_dir=tmp / "metrics", run_id="RID", timestamp="TS", + image_ref="example/vcf-rdfizer:latest", + filter_oracle="auto", engine="hdt", + engine_options=engine_options, + wrapper_log_path=tmp / "wrapper.log") + return commands[0] + + def _value(self, command, flag): + return command[command.index(flag) + 1] + + def test_both_are_forwarded(self): + command = self._command_for( + {"shacl_max_triples": 100_000_000, "node_heap_mb": 16384}) + self.assertEqual(self._value(command, "--shacl-max-triples"), "100000000") + self.assertEqual(self._value(command, "--node-heap-mb"), "16384") + + def test_a_zero_limit_is_forwarded_not_dropped_as_falsy(self): + """Dropping 0 would silently turn 'no gate' back into the default gate.""" + command = self._command_for({"shacl_max_triples": 0}) + self.assertEqual(self._value(command, "--shacl-max-triples"), "0") + + def test_absent_options_send_no_flags(self): + command = self._command_for({}) + self.assertNotIn("--shacl-max-triples", command) + self.assertNotIn("--node-heap-mb", command) + + def test_the_runner_accepts_what_the_wrapper_sends(self): + """The contract across the boundary: the flag names and value types + the wrapper emits are ones the runner parses.""" + command = self._command_for( + {"shacl_max_triples": 0, "node_heap_mb": 16384}) + actions = V.build_arg_parser()._option_string_actions + for flag, expected in (("--shacl-max-triples", 0), ("--node-heap-mb", 16384)): + with self.subTest(flag=flag): + self.assertIn(flag, actions) + self.assertEqual(actions[flag].type(self._value(command, flag)), expected) + + +if __name__ == "__main__": + unittest.main()