test(validation): check context equality over every schema, not thirteen

tests/test_validation/conftest.py

@@ -16,6 +16,18 @@
 RESOLVER_CACHE = DATA.parent / "resolver_cache"
 
 
+def embedded_schemas(node):
+    """Every schema inside a compliance fixture, whatever nesting that file happens to use."""
+    if isinstance(node, dict):
+        if isinstance(node.get("schema"), dict):
+            yield node["schema"]
+        for value in node.values():
+            yield from embedded_schemas(value)
+    elif isinstance(node, list):
+        for value in node:
+            yield from embedded_schemas(value)
+
+
 @pytest.fixture
 def data_dir() -> Path:
     """The committed slice of oold-schema's examples."""

tests/test_validation/test_jsonld.py

@@ -26,7 +26,7 @@
     roundtrip,
 )
 
-from .conftest import read
+from .conftest import embedded_schemas, read
 
 # ------------------------------------------------------------------ canonical / lost_keys
 
@@ -245,25 +245,36 @@ def test_the_walker_and_pyld_agree_on_the_effective_term_set(resolver, data_dir)
     # @context" - a failure of the harness that reads exactly like a failure of the subject.
     options = {"processingMode": "json-ld-1.1"}
 
-    compared = 0
-    for path in sorted(data_dir.glob("*.schema.json")):
-        loaded = resolver.load(path)
-        if "@context" not in loaded.schema:
-            continue
-        context = resolve_context(loaded.schema, loaded.base_uri, resolver)
+    def agree(schema, base_uri, base_url, label):
+        """True when both produce the same term set; None when the schema has nothing to compare."""
+        if "@context" not in schema:
+            return None
+        context = resolve_context(schema, base_uri, resolver)
         if context.errors or context.is_empty:
-            continue
-
+            return None
         active = processor.process_context(
             processor._get_initial_context(options),
             context.as_jsonld(),
-            {**options, "base": loader.url_for(path.name), "documentLoader": loader},
+            {**options, "base": base_url, "documentLoader": loader},
         )
         theirs = {term for term in active["mappings"] if not term.startswith("@")}
-        assert set(context.terms()) == theirs, path.name
-        compared += 1
+        assert set(context.terms()) == theirs, label
+        return True
+
+    compared = 0
+    for path in sorted(data_dir.glob("*.schema.json")):
+        loaded = resolver.load(path)
+        compared += bool(agree(loaded.schema, loaded.base_uri, loader.url_for(path.name), path.name))
+
+    # The compliance fixtures carry as many schemas again as the examples do, and they are where
+    # the unusual constructs live - that is what they are for - so leaving them out would test the
+    # equivalence on exactly the documents least likely to break it.
+    synthetic = loader.url_for("compliance-case.schema.json")
+    for path in sorted((data_dir / "compliance").glob("*.json")):
+        for index, schema in enumerate(embedded_schemas(json.loads(path.read_text(encoding="utf-8")))):
+            compared += bool(agree(schema, synthetic, synthetic, f"{path.name}[{index}]"))
 
-    assert compared >= 13, f"only {compared} schemas carried a resolvable @context"
+    assert compared >= 26, f"only {compared} schemas carried a resolvable @context"
 
 
 def test_pyld_keeps_a_scoped_context_unresolved(resolver, data_dir):

tests/test_validation/test_parity_live.py

@@ -13,13 +13,19 @@
 
 from __future__ import annotations
 
+import json
 import shutil
 import subprocess
 from pathlib import Path
 
 import pytest
 
 from oold.validation import Options, run_compliance, validate_directory
+from oold.validation.context_resolution import resolve_context
+from oold.validation.loader import DocumentLoader
+from oold.validation.resolve import Resolver
+
+from .conftest import embedded_schemas
 
 pytestmark = pytest.mark.parity
 
@@ -149,3 +155,49 @@ def test_every_mapped_rule_resolves_against_the_upstream_catalog(upstream):
     known = {r["id"] for r in json.loads(catalog.read_text(encoding="utf-8"))["rules"]}
     unknown = {check: rule for check, rule in rule_map().items() if rule not in known}
     assert not unknown, f"the registry cites ids absent from the upstream catalog: {unknown}"
+
+
+def test_the_walker_and_pyld_agree_across_the_upstream_corpus(upstream):
+    """The same equivalence the committed slice pins, over a corpus that is still moving.
+
+    `context_resolution`'s docstring justifies a hand-written walk by what pyld does not return.
+    The fixture slice is a snapshot and can only pin that against constructs upstream had when it
+    was taken; this runs it against whatever `main` has now, which is where a construct that
+    breaks the claim would appear first.
+    """
+    from pyld.jsonld import JsonLdProcessor
+
+    resolver = Resolver(offline=False)
+    processor = JsonLdProcessor()
+    options = {"processingMode": "json-ld-1.1"}
+    compared = 0
+
+    def agree(schema, base_uri, base_url, loader, label):
+        if "@context" not in schema:
+            return False
+        context = resolve_context(schema, base_uri, resolver)
+        if context.errors or context.is_empty:
+            return False
+        active = processor.process_context(
+            processor._get_initial_context(options),
+            context.as_jsonld(),
+            {**options, "base": base_url, "documentLoader": loader},
+        )
+        theirs = {term for term in active["mappings"] if not term.startswith("@")}
+        assert set(context.terms()) == theirs, label
+        return True
+
+    examples = upstream / "examples"
+    loader = DocumentLoader(resolver, directory=examples)
+    for path in sorted(examples.glob("*.schema.json")):
+        loaded = resolver.load(path)
+        compared += agree(loaded.schema, loaded.base_uri, loader.url_for(path.name), loader, path.name)
+
+    compliance = examples / "compliance"
+    comp_loader = DocumentLoader(resolver, directory=compliance)
+    synthetic = comp_loader.url_for("compliance-case.schema.json")
+    for path in sorted(compliance.glob("*.json")):
+        for index, schema in enumerate(embedded_schemas(json.loads(path.read_text(encoding="utf-8")))):
+            compared += agree(schema, synthetic, synthetic, comp_loader, f"{path.name}[{index}]")
+
+    assert compared >= 20, f"only {compared} upstream schemas carried a resolvable @context"