from __future__ import annotations
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "checks"))
from schema_contract import ( ADOPTED_SCHEMA_KEYWORDS,
KEYWORD_APPLICABLE_TYPES,
TYPE_SPECIFIC_SCHEMA_KEYWORDS,
UNRESOLVED_REFERENCE_KEYWORDS,
uses_runtime_profile,
validate_schema_node,
)
PACK_DIRS = ("checks", "policies", "review", "schemas", "skills", "templates", "tools")
PACK_FILES = (
"AGENT_USAGE.md", "AGENTS.md", "labels.yaml", "skills-lock.json",
"states.yaml", "workflow.yaml",
)
RUNTIME_SCHEMA_SCRIPT = """
import json
import sys
sys.path.insert(0, sys.argv[1])
from specrail_lib import validate_instance
with open(sys.argv[2], encoding="utf-8") as fh:
schema = json.load(fh)
validate_instance(schema, json.loads(sys.argv[3]))
"""
RUNTIME_DATA = {
"issue": 1,
"collected_at": "now",
"open_prs_complete": True,
"open_pr_limit": 10,
"open_prs": [{"number": 2, "head_ref": "review-fix", "references_issue": True}],
"remote_branches": [],
}
def run(command: list[str], *, cwd: Path) -> subprocess.CompletedProcess[str]:
return subprocess.run(command, cwd=cwd, capture_output=True, text=True, check=False)
def assert_passed(completed: subprocess.CompletedProcess[str], label: str) -> None:
assert completed.returncode == 0, f"{label} failed:\n{completed.stdout}{completed.stderr}"
def copy_pack(repo: Path) -> None:
for relative_path in PACK_DIRS:
shutil.copytree(
ROOT / relative_path, repo / relative_path,
ignore=shutil.ignore_patterns("__pycache__", "*.pyc"),
)
for relative_path in PACK_FILES:
shutil.copy2(ROOT / relative_path, repo / relative_path)
target = repo / "scripts" / "sync-specrail-checks.sh"
target.parent.mkdir(parents=True)
shutil.copy2(ROOT / "scripts" / "sync-specrail-checks.sh", target)
helper = repo / "scripts" / "ci" / "specrail_sync_lock.py"
helper.parent.mkdir(parents=True)
shutil.copy2(ROOT / "scripts" / "ci" / "specrail_sync_lock.py", helper)
assert_passed(run(["git", "init", "-q"], cwd=repo), "initialize isolated git repo")
assert_passed(run(["git", "add", "-A"], cwd=repo), "track isolated pack")
def write_lock(path: Path, lock: dict[str, object]) -> None:
path.write_text(json.dumps(lock, indent=2) + "\n", encoding="utf-8")
def update_lock_hash(lock: dict[str, object], relative_path: str, path: Path) -> None:
entries = lock["files"]
assert isinstance(entries, list)
for entry in entries:
if isinstance(entry, dict) and entry.get("path") == relative_path:
entry["sha256"] = hashlib.sha256(path.read_bytes()).hexdigest()
return
raise AssertionError(f"missing lock entry for {relative_path}")
def set_nested(value: dict[str, object], path: tuple[str, ...], replacement: object) -> None:
target = value
for key in path[:-1]:
child = target[key]
assert isinstance(child, dict), f"fixture path {path!r} is not an object"
target = child
target[path[-1]] = replacement
def run_workflow(repo: Path) -> subprocess.CompletedProcess[str]:
return run(
[sys.executable, str(repo / "checks" / "check_workflow.py"), "--repo", str(repo)],
cwd=repo,
)
def run_runtime(
repo: Path, schema_path: Path, data: dict[str, object] = RUNTIME_DATA
) -> subprocess.CompletedProcess[str]:
return run(
[sys.executable, "-c", RUNTIME_SCHEMA_SCRIPT, str(repo / "checks"),
str(schema_path), json.dumps(data)],
cwd=repo,
)
def assert_contract_failure(repo: Path, expected: str, label: str) -> None:
workflow = run_workflow(repo)
assert workflow.returncode != 0, f"{label} must fail workflow validation"
assert expected in workflow.stdout, workflow.stdout
sync = run([str(repo / "scripts" / "sync-specrail-checks.sh"), "--verify"], cwd=repo)
assert sync.returncode != 0, f"{label} must fail sync validation"
assert "files match lock" in sync.stdout
assert expected in sync.stdout, sync.stdout
def collect_keywords(schema: dict[str, object]) -> set[str]:
found = set(schema)
for keyword in ("$defs", "dependentSchemas", "patternProperties", "properties"):
children = schema.get(keyword)
if isinstance(children, dict):
for child in children.values():
if isinstance(child, dict):
found.update(collect_keywords(child))
for keyword in ("allOf", "anyOf", "oneOf", "prefixItems"):
children = schema.get(keyword)
if isinstance(children, list):
for child in children:
if isinstance(child, dict):
found.update(collect_keywords(child))
for keyword in (
"additionalProperties", "contains", "contentSchema", "else", "if", "items",
"not", "propertyNames", "then", "unevaluatedItems", "unevaluatedProperties",
):
child = schema.get(keyword)
if isinstance(child, dict):
found.update(collect_keywords(child))
return found
def assert_vocabulary_and_baselines() -> None:
values = {keyword: "value" for keyword in (
"$anchor", "$comment", "$dynamicAnchor", "$id", "$schema",
"contentEncoding", "contentMediaType", "description", "format", "title",
)}
values.update({keyword: True for keyword in ("deprecated", "readOnly", "uniqueItems", "writeOnly")})
values.update({keyword: 0 for keyword in (
"maxContains", "maxItems", "maxLength", "maxProperties",
"minContains", "minItems", "minLength", "minProperties",
)})
values.update({keyword: 1 for keyword in (
"exclusiveMaximum", "exclusiveMinimum", "maximum", "minimum", "multipleOf",
)})
values.update({keyword: {} for keyword in (
"$defs", "dependentSchemas", "patternProperties", "properties",
"contains", "contentSchema", "else", "if", "items", "not", "propertyNames", "then",
)})
values.update({keyword: [{}] for keyword in ("allOf", "anyOf", "oneOf", "prefixItems")})
values.update({keyword: True for keyword in (
"additionalProperties", "unevaluatedItems", "unevaluatedProperties",
)})
values.update({
"$vocabulary": {"urn:example:vocabulary": True}, "const": None, "default": None,
"dependentRequired": {}, "enum": [None], "examples": [], "pattern": "^ok$",
"required": [], "type": "object",
})
assert set(values) == ADOPTED_SCHEMA_KEYWORDS - set(UNRESOLVED_REFERENCE_KEYWORDS)
for keyword, value in values.items():
assert validate_schema_node({keyword: value}, Path("vocabulary.schema.json")) == []
assert validate_schema_node(
{"$defs": {"value": {"type": "string"}}}, Path("defs.schema.json")
) == []
for keyword in UNRESOLVED_REFERENCE_KEYWORDS:
errors = validate_schema_node({keyword: "#/$defs/value"}, Path("ref.schema.json"))
assert any("reference resolution is implemented" in error for error in errors)
expected_typed = frozenset({
"additionalProperties", "contains", "contentEncoding", "contentMediaType",
"contentSchema", "dependentRequired", "dependentSchemas", "exclusiveMaximum",
"exclusiveMinimum", "format", "items", "maxContains", "maxItems", "maxLength",
"maxProperties", "maximum", "minContains", "minItems", "minLength", "minProperties",
"minimum", "multipleOf", "pattern", "patternProperties", "prefixItems", "properties",
"propertyNames", "required", "unevaluatedItems", "unevaluatedProperties", "uniqueItems",
})
assert TYPE_SPECIFIC_SCHEMA_KEYWORDS == expected_typed
assert frozenset(KEYWORD_APPLICABLE_TYPES) == expected_typed
observed: set[str] = set()
for path in sorted((ROOT / "schemas").glob("*.schema.json")):
schema = json.loads(path.read_text(encoding="utf-8"))
observed.update(collect_keywords(schema))
assert validate_schema_node(
schema, path.relative_to(ROOT), runtime_profile=uses_runtime_profile(path)
) == []
assert observed <= ADOPTED_SCHEMA_KEYWORDS
def assert_regex_contracts() -> None:
path = Path("regex.schema.json")
assert validate_schema_node({"pattern": "(?<name>a)"}, path) == []
errors = validate_schema_node({"pattern": "(?P<name>a)"}, path)
assert any("invalid ECMAScript regex" in error for error in errors)
assert validate_schema_node({"patternProperties": {"(?<name>a)": True}}, path) == []
errors = validate_schema_node({"patternProperties": {"(?P<name>a)": True}}, path)
assert any("invalid ECMAScript regex" in error for error in errors)
python_only = validate_schema_node(
{"type": "string", "pattern": "(?P<name>.+)"}, path, runtime_profile=True
)
assert any("invalid ECMAScript regex" in error for error in python_only)
assert not any("invalid Python runtime regex" in error for error in python_only)
ecmascript_only = validate_schema_node(
{"type": "string", "pattern": "(?<name>.+)"}, path, runtime_profile=True
)
assert not any("invalid ECMAScript regex" in error for error in ecmascript_only)
assert any("invalid Python runtime regex" in error for error in ecmascript_only)
with mock.patch.dict(os.environ, {"PATH": ""}):
missing = validate_schema_node({"pattern": "ok"}, path)
assert any("ECMAScript regex engine failed" in error for error in missing)
def assert_boolean_children_and_enum_uniqueness() -> None:
boolean_children = {
"$defs": {"a": True}, "dependentSchemas": {"a": False},
"patternProperties": {"^a$": True}, "properties": {"a": False},
"allOf": [True], "anyOf": [False], "oneOf": [True], "prefixItems": [False],
"contains": True, "contentSchema": False, "else": True, "if": False,
"items": True, "not": False, "propertyNames": True, "then": False,
"additionalProperties": True, "unevaluatedItems": False,
"unevaluatedProperties": True,
}
assert validate_schema_node(boolean_children, Path("boolean.schema.json")) == []
runtime_errors = validate_schema_node(
boolean_children, Path("runtime.schema.json"), runtime_profile=True
)
assert sum("boolean schema is not executable" in error for error in runtime_errors) == 18
duplicate = validate_schema_node(
{"enum": [{"a": [1, 2]}, {"a": [1, 2]}]}, Path("enum.schema.json")
)
assert any("enum must contain unique JSON values" in error for error in duplicate)
assert validate_schema_node({"enum": [True, 1]}, Path("enum.schema.json")) == []
numeric_duplicate = validate_schema_node({"enum": [1, 1.0]}, Path("enum.schema.json"))
assert any("enum must contain unique JSON values" in error for error in numeric_duplicate)
def assert_published_contracts() -> None:
base = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "Published contract fixture", "type": "object",
"properties": {
"missing_type": {"minLength": 1},
"mixed_union": {"type": ["string", "null"], "minLength": 1, "pattern": "(?<ok>.+)"},
"conditional": {
"if": {"properties": {"kind": {"const": "primary"}}},
"then": {"required": ["value"]},
"else": {"properties": {"fallback": {"type": "string"}}},
},
},
"additionalProperties": True,
}
negative_cases = (
("minLength", {"type": "integer", "minLength": 1}, "minLength is incompatible"),
("pattern", {"type": "integer", "pattern": "^x$"}, "pattern is incompatible"),
("unevaluatedProperties", {"type": "integer", "unevaluatedProperties": True}, "unevaluatedProperties is incompatible"),
("unevaluatedItems", {"type": "string", "unevaluatedItems": True}, "unevaluatedItems is incompatible"),
("contentEncoding", {"type": "integer", "contentEncoding": "base64"}, "contentEncoding is incompatible"),
("python regex", {"type": "string", "pattern": "(?P<name>a)"}, "invalid ECMAScript regex"),
("patternProperties", {"type": "object", "patternProperties": {"[": True}}, "invalid ECMAScript regex"),
("unresolved ref", {"$ref": "#/$defs/value"}, "reference resolution is implemented"),
("unresolved dynamic ref", {"$dynamicRef": "#value"}, "reference resolution is implemented"),
)
with tempfile.TemporaryDirectory(prefix="remem-published-schema-") as raw:
repo = Path(raw)
copy_pack(repo)
path = repo / "schemas" / "published_fixture.schema.json"
path.write_text(json.dumps(base, indent=2) + "\n", encoding="utf-8")
assert_passed(run_workflow(repo), "published positive workflow")
assert_passed(
run([str(repo / "scripts" / "sync-specrail-checks.sh"), "--verify"], cwd=repo),
"published positive sync",
)
for label, child, expected in negative_cases:
malformed = json.loads(json.dumps(base))
malformed["properties"] = {"value": child}
path.write_text(json.dumps(malformed, indent=2) + "\n", encoding="utf-8")
assert_contract_failure(repo, expected, label)
def assert_locked_schema_contracts() -> None:
cases = (
("schemas/duplicate_work_evidence.schema.json", ("minimun",), 1, "$: unsupported JSON Schema keyword 'minimun'"),
("schemas/pr_review_gate.schema.json", ("properties", "pr", "minimun"), 1, "$.properties.pr: unsupported JSON Schema keyword 'minimun'"),
("schemas/review_result.schema.json", ("properties", "prior_findings", "items", "minimun"), 1, "$.properties.prior_findings.items: unsupported JSON Schema keyword 'minimun'"),
("schemas/runtime_checkpoint.schema.json", ("additionalProperties",), {"type": "string", "minimun": 1}, "$.additionalProperties: unsupported JSON Schema keyword 'minimun'"),
("schemas/review_result.schema.json", ("properties", "body", "pattern"), "[", "$.properties.body.pattern invalid ECMAScript regex"),
("schemas/duplicate_work_evidence.schema.json", ("$ref",), "#/$defs/value", "reference resolution is implemented"),
("schemas/runtime_checkpoint.schema.json", ("$dynamicRef",), "#value", "reference resolution is implemented"),
)
locked = {case[0] for case in cases}
with tempfile.TemporaryDirectory(prefix="remem-locked-schema-") as raw:
repo = Path(raw)
copy_pack(repo)
lock_path = repo / "checks" / "specrail-sync.lock.json"
baseline_lock = json.loads(lock_path.read_text(encoding="utf-8"))
for relative, nested, replacement, expected in cases:
for locked_file in locked:
shutil.copy2(ROOT / locked_file, repo / locked_file)
schema_path = repo / relative
schema = json.loads(schema_path.read_text(encoding="utf-8"))
set_nested(schema, nested, replacement)
schema_path.write_text(json.dumps(schema, indent=2) + "\n", encoding="utf-8")
changed_lock = json.loads(json.dumps(baseline_lock))
update_lock_hash(changed_lock, relative, schema_path)
write_lock(lock_path, changed_lock)
assert_contract_failure(repo, expected, relative)
def runtime_cases() -> tuple[tuple[str, tuple[str, ...], object, str], ...]:
return (
("properties", ("properties",), [], "$.properties must be an object"),
("property schema", ("properties", "issue"), [], "$.properties.issue must be a boolean or object"),
("items", ("properties", "open_prs", "items"), [], "$.properties.open_prs.items must be a boolean or object"),
("additionalProperties", ("additionalProperties",), [], "$.additionalProperties must be a boolean or object"),
("required shape", ("required",), ["issue", 7], "$.required must be a unique array of strings"),
("unknown keyword", ("properties", "issue", "minimun"), 1, "$.properties.issue: unsupported JSON Schema keyword 'minimun'"),
("unknown type", ("properties", "issue", "type"), "uint64", "$.properties.issue.type must be a supported JSON type"),
("empty type array", ("properties", "issue", "type"), [], "$.properties.issue.type must be a supported JSON type"),
("duplicate type array", ("properties", "issue", "type"), ["integer", "integer"], "$.properties.issue.type must be a supported JSON type"),
("unknown union member", ("properties", "issue", "type"), ["integer", "uint64"], "$.properties.issue.type must be a supported JSON type"),
("enum shape", ("properties", "issue", "enum"), {"one": 1}, "$.properties.issue.enum must be a non-empty array"),
("empty enum", ("properties", "issue", "enum"), [], "$.properties.issue.enum must be a non-empty array"),
("enum duplicate", ("properties", "issue", "enum"), [1, 1.0], "$.properties.issue.enum must contain unique JSON values"),
("minLength bool", ("properties", "collected_at", "minLength"), True, "$.properties.collected_at.minLength must be a non-negative integer"),
("minItems negative", ("properties", "open_prs", "minItems"), -1, "$.properties.open_prs.minItems must be a non-negative integer"),
("minimum bool", ("properties", "issue", "minimum"), True, "$.properties.issue.minimum must be a JSON number"),
("minimum shape", ("properties", "issue", "minimum"), "one", "$.properties.issue.minimum must be a JSON number"),
("exclusiveMinimum shape", ("properties", "issue", "exclusiveMinimum"), "one", "$.properties.issue.exclusiveMinimum must be a JSON number"),
("exclusiveMaximum shape", ("properties", "issue", "exclusiveMaximum"), "ten", "$.properties.issue.exclusiveMaximum must be a JSON number"),
("minLength type compatibility", ("properties", "collected_at", "type"), "integer", "$.properties.collected_at.minLength requires only type string"),
("minLength missing type", ("properties", "collected_at"), {"minLength": 1}, "$.properties.collected_at.minLength requires explicit type string"),
("minLength unsafe union", ("properties", "collected_at", "type"), ["string", "null"], "$.properties.collected_at.minLength requires only type string"),
("items type compatibility", ("properties", "open_prs", "type"), "object", "$.properties.open_prs.items requires only type array"),
("minItems type compatibility", ("properties", "open_prs"), {"type": "object", "minItems": 1}, "$.properties.open_prs.minItems requires only type array"),
("minimum type compatibility", ("properties", "issue", "type"), "string", "$.properties.issue.minimum requires only type integer or number"),
("required type compatibility", ("properties", "open_prs", "items"), {"type": "array", "required": ["number"]}, "$.properties.open_prs.items.required requires only type object"),
("recursive item keyword", ("properties", "open_prs", "items", "minimun"), 1, "$.properties.open_prs.items: unsupported JSON Schema keyword 'minimun'"),
("recursive additional keyword", ("additionalProperties",), {"type": "string", "minimun": 1}, "$.additionalProperties: unsupported JSON Schema keyword 'minimun'"),
("boolean runtime child", ("properties", "issue"), True, "$.properties.issue boolean schema is not executable"),
("properties scalar bypass", ("properties", "open_prs"), {"properties": {"ignored": {"type": "string"}}}, "$.properties.open_prs.properties requires explicit type object"),
("properties union bypass", ("properties", "open_prs"), {"type": ["array", "object"], "properties": {"ignored": {"type": "string"}}}, "$.properties.open_prs.properties requires only type object"),
("additionalProperties scalar bypass", ("properties", "collected_at"), {"additionalProperties": False}, "$.properties.collected_at.additionalProperties requires explicit type object"),
("additionalProperties union bypass", ("properties", "collected_at"), {"type": ["object", "string"], "additionalProperties": False}, "$.properties.collected_at.additionalProperties requires only type object"),
("Python-only runtime regex", ("properties", "collected_at", "pattern"), "(?P<name>.+)", "invalid ECMAScript regex"),
("ECMAScript-only runtime regex", ("properties", "collected_at", "pattern"), "(?<name>.+)", "invalid Python runtime regex"),
)
def assert_runtime_profile_matrix() -> None:
runtime_accepts = {
"duplicate type array",
"minLength missing type", "minLength unsafe union",
"enum duplicate", "properties scalar bypass", "Python-only runtime regex",
"properties union bypass", "additionalProperties scalar bypass",
"additionalProperties union bypass",
}
runtime_definition_rejects = {
"minLength bool": (
"$schema.properties.collected_at: minLength must be a non-negative integer",
),
"minItems negative": (
"$schema.properties.open_prs: minItems must be a non-negative integer",
),
"minimum bool": (
"$schema.properties.issue.minimum: must be a number",
),
"unknown union member": (
"$schema.properties.issue.type: unsupported type 'uint64'",
),
"recursive additional keyword": (
"$schema.additionalProperties: unsupported JSON Schema keyword 'minimun'",
),
}
with tempfile.TemporaryDirectory(prefix="remem-runtime-schema-") as raw:
repo = Path(raw)
copy_pack(repo)
schema_path = repo / "schemas" / "duplicate_work_evidence.schema.json"
lock_path = repo / "checks" / "specrail-sync.lock.json"
baseline_schema = json.loads(schema_path.read_text(encoding="utf-8"))
baseline_lock = json.loads(lock_path.read_text(encoding="utf-8"))
open_path = repo / "schemas" / "open_required.schema.json"
open_path.write_text(json.dumps({
"$schema": "https://json-schema.org/draft/2020-12/schema", "title": "Open",
"type": "object", "required": ["opaque"], "additionalProperties": True,
}, indent=2) + "\n", encoding="utf-8")
assert_passed(run_workflow(repo), "open required workflow")
assert_passed(run_runtime(repo, open_path, {"opaque": 1}), "open required runtime")
safe = json.loads(json.dumps(baseline_schema))
safe["properties"]["issue"] = {"type": ["integer", "number"], "minimum": 1}
schema_path.write_text(json.dumps(safe, indent=2) + "\n", encoding="utf-8")
safe_lock = json.loads(json.dumps(baseline_lock))
update_lock_hash(safe_lock, "schemas/duplicate_work_evidence.schema.json", schema_path)
write_lock(lock_path, safe_lock)
assert_passed(run_runtime(repo, schema_path), "safe numeric union runtime")
assert_passed(run_workflow(repo), "safe numeric union workflow")
assert_passed(
run([str(repo / "scripts" / "sync-specrail-checks.sh"), "--verify"], cwd=repo),
"safe numeric union sync",
)
const_schema = json.loads(json.dumps(baseline_schema))
const_schema["properties"]["issue"]["const"] = 999
schema_path.write_text(json.dumps(const_schema, indent=2) + "\n", encoding="utf-8")
const_lock = json.loads(json.dumps(baseline_lock))
update_lock_hash(const_lock, "schemas/duplicate_work_evidence.schema.json", schema_path)
write_lock(lock_path, const_lock)
assert run_runtime(repo, schema_path).returncode != 0
assert_passed(run_workflow(repo), "ordinary const mismatch workflow")
assert_passed(
run([str(repo / "scripts" / "sync-specrail-checks.sh"), "--verify"], cwd=repo),
"ordinary const mismatch sync",
)
for label, nested, replacement, expected in runtime_cases():
malformed = json.loads(json.dumps(baseline_schema))
set_nested(malformed, nested, replacement)
schema_path.write_text(json.dumps(malformed, indent=2) + "\n", encoding="utf-8")
changed_lock = json.loads(json.dumps(baseline_lock))
update_lock_hash(changed_lock, "schemas/duplicate_work_evidence.schema.json", schema_path)
write_lock(lock_path, changed_lock)
runtime = run_runtime(repo, schema_path)
if label in runtime_accepts:
assert_passed(runtime, f"runtime contrast {label}")
elif label in runtime_definition_rejects:
assert runtime.returncode != 0, f"runtime definition must reject {label}"
assert "schema_validation.SchemaDefinitionError:" in runtime.stderr
assert "schema_validation.InstanceMismatch:" not in runtime.stderr
for fragment in runtime_definition_rejects[label]:
assert fragment in runtime.stderr, runtime.stderr
else:
assert runtime.returncode != 0, f"runtime must reject {label}"
assert_contract_failure(repo, expected, label)
def run_schema_contract_tests() -> None:
assert_vocabulary_and_baselines()
assert_regex_contracts()
assert_boolean_children_and_enum_uniqueness()
assert_published_contracts()
assert_locked_schema_contracts()
assert_runtime_profile_matrix()
def main() -> int:
run_schema_contract_tests()
print("Schema contract test passed")
return 0
if __name__ == "__main__":
sys.exit(main())