Skip to content
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
303 changes: 302 additions & 1 deletion tests/test_the_schema_and_the_models_agree.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,39 @@
the shape, and ``spiffe://bernstein.run`` is one a real producer wrote. The schema is
tightened to the model in the same change; the four values stay as the guard that would have
caught it.

That fifth one is why the second half of this file exists. A hand-written matrix swept over
one base record answers "do the two disagree about any of these values", and the floors below
keep it from answering that about nothing. What it cannot answer is "is each constraint
*discriminated* at all" - whether the values reach the constraint's own boundary, or merely
fail so early that both validators reject for an unrelated reason and agree by accident. Two
gaps followed from that, and neither produced a failing test:

1. **Half the patterns were unreachable.** ``BASE`` carries five of the schema's ten pattern
constraints. ``model.weights_digest``, ``delegation.parent_record_hash``,
``references[].retention``, ``references[].digest`` and ``signature`` are all optional, all
absent from the fixture, and therefore never mutated. A sweep cannot disagree about a field
it never sets.
2. **The mirroring was claimed and not held.** ``models.py`` says its pattern constants are
"mirrored verbatim in schema/trace-claim.json and its copy, and held there by
tests/test_the_schema_and_the_models_agree.py". They were not held here. Two of the ten
were pinned in ``test_references_block.py``; the other eight were maintained by hand under
a comment that named this file.

The generator below closes both. It reads the ten patterns out of the schema rather than
listing them, so a new one cannot be added without appearing here; it probes each at its own
boundary with values derived from the pattern and from a valid instance of it; and for each
constraint it demands one of three outcomes - the two artifacts hold the same pattern string,
in which case no string can split them and that is a proof rather than an observation; or a
splitting value exists and is declared; or the constraint is neither, which fails.
"""
from __future__ import annotations

import copy
from typing import Any
import json
import re
from pathlib import Path
from typing import Any, get_args

import jsonschema
import pytest
Expand Down Expand Up @@ -208,3 +236,276 @@ def test_no_integer_field_accepts_a_boolean(path: tuple[str, ...], value: bool)
f"{'.'.join(path)} accepted {value!r}; before this guard it became "
f"{int(value)!r} and the record claimed it"
)


# ---------------------------------------------------------------------------
# Per-constraint discrimination
# ---------------------------------------------------------------------------

REPO_ROOT = Path(__file__).resolve().parents[1]
CANONICAL_SCHEMA: dict[str, Any] = json.loads(
(REPO_ROOT / "schema" / "trace-claim.json").read_text(encoding="utf-8")
)

#: A record valid to both validators that carries every pattern-constrained field.
#: ``BASE`` above is deliberately the minimum a producer must emit; five of the ten
#: pattern constraints sit on optional members it omits, and an absent field cannot
#: be mutated into a disagreement.
FULL: dict[str, Any] = {
**copy.deepcopy(BASE),
"model": {"provider": "anthropic", "model_id": "claude-sonnet-4-6",
"weights_digest": "sha256:" + "d" * 64},
"delegation": {"parent_record_hash": "sha256:" + "e" * 64,
"credential_id": "trace-spec-delegation-credential"},
"references": [{"rel": "behavior-trace", "id": "run-1",
"resolver": "https://agt.example.org",
"digest": "sha256:" + "f" * 64, "retention": "P30D"}],
"signature": "abcDEF-_123",
}

#: Pattern constraints the two artifacts do NOT hold as the same string, for which no
#: probe splits them either. Empty today, and it should stay that way: a constraint
#: that is neither mirrored nor split is one nothing is checking. An entry here says
#: a human looked and decided, which is the same contract as ``DECLARED_DIVERGENCES``.
UNMIRRORED_AND_UNSPLIT: dict[str, str] = {}

_LITERAL_RUN = re.compile(r"[A-Za-z0-9_.:/\-]{2,}")


def _pattern_constraints(schema: dict[str, Any]) -> dict[tuple[str, ...], str]:
"""Every ``pattern`` in *schema*, keyed by its path in a record.

Read out of the schema rather than listed, so a pattern added to the published
artifact cannot be added without this file noticing it.
"""
found: dict[tuple[str, ...], str] = {}

def walk(node: Any, path: tuple[str, ...]) -> None:
if not isinstance(node, dict):
return
if isinstance(node.get("pattern"), str):
found[path] = node["pattern"]
for key, value in node.items():
if key == "properties" and isinstance(value, dict):
for prop, sub in value.items():
walk(sub, path + (prop,))
elif key == "items":
walk(value, path + ("0",))
elif key in ("allOf", "anyOf", "oneOf") and isinstance(value, list):
for sub in value:
walk(sub, path)

walk(schema, ())
return found


def _get(obj: Any, path: tuple[str, ...]) -> Any:
for key in path:
obj = obj[int(key)] if isinstance(obj, list) else obj[key]
return obj


def _set_deep(obj: Any, path: tuple[str, ...], value: Any) -> None:
for key in path[:-1]:
obj = obj[int(key)] if isinstance(obj, list) else obj[key]
if isinstance(obj, list):
obj[int(path[-1])] = value
else:
obj[path[-1]] = value


def _probes(pattern: str, valid: str) -> list[str]:
"""Values that sit on *pattern*'s boundary, derived from it and from *valid*.

The classes, and what each is for:

* **structural prefixes of a valid value** - the #244 class. A pattern that tests a
prefix accepts ``spiffe://`` while a model requiring the full shape refuses it;
nothing that fails the prefix can tell the two apart, because both reject.
* **literal runs taken out of the pattern text** - the same class reached from the
other side, and the reason ``did:`` appears without anyone writing it here.
* **case flip, one character shorter, one longer, one outside the class** - the
length and alphabet boundaries of a fixed-width constraint.
* **empty, space, tab** - the values a producer sends when a field is unset.
"""
out: list[str] = []
for sep in ("://", ":", "/"):
start = valid.find(sep)
while start != -1:
out.append(valid[: start + len(sep)])
start = valid.find(sep, start + 1)
out += [valid.swapcase(), valid[:-1], valid + valid[-1:], valid[:-1] + "§", "", " ", "\t"]
out += _LITERAL_RUN.findall(pattern)
return list(dict.fromkeys(out))


def _splitters(path: tuple[str, ...], pattern: str) -> list[tuple[str, bool, bool]]:
"""Probes at *path* the two validators disagree about."""
valid = _get(FULL, path)
split: list[tuple[str, bool, bool]] = []
for probe in _probes(pattern, valid):
record = copy.deepcopy(FULL)
_set_deep(record, path, probe)
by_schema, by_model = _by_schema(record), _by_model(record)
if by_schema != by_model:
split.append((probe, by_schema, by_model))
return split


def _model_pattern_strings() -> set[str]:
"""Every pattern string the reference model constrains a field with.

Walked rather than listed, and walked through three indirections, because the
constraint is rarely where the obvious read looks for it. An optional field is
``Annotated[str, Field(pattern=...)] | None``, so the pattern sits on a union member
and ``field.metadata`` is empty; the union member's ``__metadata__`` holds a
``FieldInfo``, and the pattern is inside *its* metadata in turn. Reading only the
first of the three missed ``signature`` here, which is the same shape of miss the
rest of this file is about.
"""
from agentrust_trace import models

strings = {
value
for name, value in vars(models).items()
if name.endswith("_RE") and isinstance(value, str)
}

def constraints_of(obj: Any) -> None:
found = getattr(obj, "pattern", None)
if isinstance(found, str):
strings.add(found)
for nested in getattr(obj, "metadata", ()) or ():
constraints_of(nested)

seen: set[Any] = set()

def walk(annotation: Any) -> None:
try:
if annotation in seen:
return
seen.add(annotation)
except TypeError: # unhashable annotation; nothing below it to revisit
pass
for meta in getattr(annotation, "__metadata__", ()) or ():
constraints_of(meta)
fields = getattr(annotation, "model_fields", None)
if isinstance(fields, dict):
for field in fields.values():
for meta in field.metadata:
constraints_of(meta)
walk(field.annotation)
for arg in get_args(annotation):
walk(arg)

walk(TrustRecord)
return strings


def test_the_full_record_passes_both() -> None:
"""The control for everything below, same reason as the one above."""
assert _by_schema(FULL), "the schema rejects the full record"
assert _by_model(FULL), "the model rejects the full record"


def test_the_fixture_reaches_every_pattern_in_the_schema() -> None:
"""The gap this half of the file was written for.

``BASE`` reaches five of ten. A constraint on a field the fixture omits is never
mutated, never disagreed about, and reads in a green run exactly like a constraint
the two artifacts agree on.
"""
constraints = _pattern_constraints(CANONICAL_SCHEMA)
assert len(constraints) >= 10, "the schema lost pattern constraints; check before relaxing this"
missing = sorted(".".join(p) for p in constraints if not _reachable(FULL, p))
assert not missing, (
"these pattern constraints are not present in FULL, so nothing probes them:\n "
+ "\n ".join(missing)
)


def _reachable(record: dict[str, Any], path: tuple[str, ...]) -> bool:
try:
_get(record, path)
except (KeyError, IndexError, TypeError, ValueError):
return False
return True


def test_every_pattern_is_mirrored_or_split_or_declared() -> None:
"""Each of the ten constraints must land in one of three states, none of them silent.

*Mirrored*: the model constrains the field with the same pattern string the schema
publishes. Then no string can split the two, and saying so is a proof rather than an
observation about the values that happened to be tried.

*Split*: a probe the two disagree about, which belongs in ``DECLARED_DIVERGENCES``
with the reason it is not simply fixed.

*Neither*: the two carry different patterns and nothing here can tell them apart.
That is the state ``models.py`` already claims is impossible, and the one that
produced #244, so it fails unless a human has written down why.
"""
model_patterns = _model_pattern_strings()
unaccounted: list[str] = []
for path, pattern in sorted(_pattern_constraints(CANONICAL_SCHEMA).items()):
dotted = ".".join(path)
if pattern in model_patterns:
continue
split = _splitters(path, pattern)
declared = [s for s in split if (dotted, repr(s[0])[:26]) in DECLARED_DIVERGENCES]
if split and len(declared) == len(split):
continue
if dotted in UNMIRRORED_AND_UNSPLIT:
continue
undeclared = [repr(s[0]) for s in split if s not in declared]
unaccounted.append(
f"{dotted}: schema pattern {pattern!r} is not one the model carries; "
+ (
f"probes that split the two and are not declared: {undeclared}"
if undeclared
else "and no probe splits them, so nothing here is checking this constraint"
)
)
assert not unaccounted, "\n".join(unaccounted)


def test_no_unmirrored_declaration_has_quietly_been_resolved() -> None:
"""The other half of ``UNMIRRORED_AND_UNSPLIT``, same contract as the divergence set."""
model_patterns = _model_pattern_strings()
constraints = {".".join(p): pat for p, pat in _pattern_constraints(CANONICAL_SCHEMA).items()}
stale = sorted(
dotted
for dotted in UNMIRRORED_AND_UNSPLIT
if dotted not in constraints or constraints[dotted] in model_patterns
)
assert not stale, f"these are mirrored or gone and should leave the set: {stale}"


def test_the_generator_reaches_the_prefix_class_the_matrix_could_not() -> None:
"""The counterfactual. Without it the eight tests above prove only that today is fine.

#244 is replayed by putting the pre-fix prefix pattern back on ``subject`` in a copy
of the schema, and asserting the generated probes split the two validators against
it. They do, on ``spiffe://`` among others - a value nobody wrote into this file, and
the class the hand-written matrix could not reach, because every value in it fails a
prefix test as well and both validators agreed by rejecting.
"""
before = copy.deepcopy(CANONICAL_SCHEMA)
before["properties"]["subject"]["pattern"] = "^(spiffe://|did:)"
validator = jsonschema.Draft202012Validator(
before, format_checker=jsonschema.FormatChecker()
)

split = []
for probe in _probes(CANONICAL_SCHEMA["properties"]["subject"]["pattern"], FULL["subject"]):
record = copy.deepcopy(FULL)
record["subject"] = probe
if validator.is_valid(record) != _by_model(record):
split.append(probe)

assert "spiffe://" in split, (
"the generator no longer reaches the prefix class: against the pre-#244 schema it "
f"split the validators on {split!r}, which does not include the value a producer "
"actually sent"
)
Loading