Jerry
1860 lines
69 KiB
Python
1860 lines
69 KiB
Python
import base64
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Mapping
|
|
from copy import deepcopy
|
|
from dataclasses import FrozenInstanceError, asdict, is_dataclass, replace
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from referencing import Registry, Resource
|
|
from referencing.jsonschema import DRAFT202012
|
|
import simplecadapi.scene.archive as scene_archive
|
|
|
|
from simplecadapi.scene import (
|
|
BASE_LIMITS,
|
|
ConnectorBindingDocument,
|
|
EntityDocument,
|
|
NormalizedProductDocument,
|
|
PresentationDocument,
|
|
SCHEMA_FILES,
|
|
SceneContractError,
|
|
SceneDocument,
|
|
SceneResourceLimits,
|
|
canonical_json_bytes,
|
|
canonical_zip_bytes,
|
|
parse_canonical_json,
|
|
parse_strict_json,
|
|
preflight_aggregate_compression_ratio,
|
|
preflight_archive_member_sizes,
|
|
preflight_glb,
|
|
preflight_glb_counts,
|
|
preflight_input_archive_size,
|
|
preflight_member_compression_ratio,
|
|
preflight_zip_bytes,
|
|
profile_cross,
|
|
profile_f32_bits,
|
|
profile_normalize,
|
|
validate_entity_asset,
|
|
validate_connector_binding,
|
|
validate_normalized_product,
|
|
validate_presentation,
|
|
validate_scene_manifest,
|
|
validate_scene_package,
|
|
json_resource_issues,
|
|
resource_count_issues,
|
|
with_scene_revision,
|
|
)
|
|
from simplecadapi.scene.validation import (
|
|
_compute_package_budget_totals,
|
|
_issue,
|
|
_package_budget_issues,
|
|
_report,
|
|
_validate_rule_registry,
|
|
load_contract_artifact,
|
|
)
|
|
from scene_contract_corpus_support import (
|
|
apply_scene_field_case,
|
|
corpus_blobs,
|
|
scene_shape_facts,
|
|
)
|
|
|
|
|
|
IDENTITY = {
|
|
"origin": [0, 0, 0],
|
|
"x_axis": [1, 0, 0],
|
|
"y_axis": [0, 1, 0],
|
|
"z_axis": [0, 0, 1],
|
|
}
|
|
ASSET_TO_SCENE = [1000, 0, 0, 0, 0, 0, -1000, 0, 0, 1000, 0, 0, 0, 0, 0, 1]
|
|
|
|
|
|
def _f32(value):
|
|
result = struct.unpack("<f", struct.pack("<f", value))[0]
|
|
return 0.0 if result == 0 else result
|
|
|
|
|
|
def _normalize(vector):
|
|
length = math.sqrt(sum(component * component for component in vector))
|
|
return tuple(component / length for component in vector)
|
|
|
|
|
|
def _cross(left, right):
|
|
return (
|
|
left[1] * right[2] - left[2] * right[1],
|
|
left[2] * right[0] - left[0] * right[2],
|
|
left[0] * right[1] - left[1] * right[0],
|
|
)
|
|
|
|
|
|
def _frame(origin, x_axis):
|
|
x_axis = _normalize(x_axis)
|
|
seed = (0.0, 0.0, 1.0) if abs(x_axis[2]) < 0.9 else (0.0, 1.0, 0.0)
|
|
projection = sum(left * right for left, right in zip(seed, x_axis))
|
|
y_axis = _normalize(tuple(seed[index] - projection * x_axis[index] for index in range(3)))
|
|
z_axis = _normalize(_cross(x_axis, y_axis))
|
|
return {
|
|
"origin": list(origin),
|
|
"x_axis": list(x_axis),
|
|
"y_axis": list(y_axis),
|
|
"z_axis": list(z_axis),
|
|
}
|
|
|
|
|
|
def _glb(kind, positions, indices, normals=None):
|
|
positions = [tuple(_f32(component) for component in point) for point in positions]
|
|
position_bytes = b"".join(struct.pack("<fff", *point) for point in positions)
|
|
chunks = [position_bytes]
|
|
views = [
|
|
{"buffer": 0, "byteLength": len(position_bytes), "byteOffset": 0, "target": 34962}
|
|
]
|
|
accessors = [
|
|
{
|
|
"bufferView": 0,
|
|
"componentType": 5126,
|
|
"count": len(positions),
|
|
"max": [max(point[axis] for point in positions) for axis in range(3)],
|
|
"min": [min(point[axis] for point in positions) for axis in range(3)],
|
|
"type": "VEC3",
|
|
}
|
|
]
|
|
if kind == "triangle":
|
|
assert normals is not None
|
|
normals = [tuple(_f32(component) for component in vector) for vector in normals]
|
|
normal_bytes = b"".join(struct.pack("<fff", *vector) for vector in normals)
|
|
views.append(
|
|
{
|
|
"buffer": 0,
|
|
"byteLength": len(normal_bytes),
|
|
"byteOffset": len(position_bytes),
|
|
"target": 34962,
|
|
}
|
|
)
|
|
accessors.append(
|
|
{
|
|
"bufferView": 1,
|
|
"componentType": 5126,
|
|
"count": len(normals),
|
|
"type": "VEC3",
|
|
}
|
|
)
|
|
chunks.append(normal_bytes)
|
|
index_view = 2
|
|
attributes = {"NORMAL": 1, "POSITION": 0}
|
|
mode = 4
|
|
else:
|
|
index_view = 1
|
|
attributes = {"POSITION": 0}
|
|
mode = 1
|
|
prefix = b"".join(chunks)
|
|
prefix += b"\0" * ((-len(prefix)) % 4)
|
|
component_type = 5123 if len(positions) <= 65536 else 5125
|
|
index_format = "H" if component_type == 5123 else "I"
|
|
index_bytes = struct.pack("<" + index_format * len(indices), *indices)
|
|
views.append(
|
|
{
|
|
"buffer": 0,
|
|
"byteLength": len(index_bytes),
|
|
"byteOffset": len(prefix),
|
|
"target": 34963,
|
|
}
|
|
)
|
|
accessors.append(
|
|
{
|
|
"bufferView": index_view,
|
|
"componentType": component_type,
|
|
"count": len(indices),
|
|
"type": "SCALAR",
|
|
}
|
|
)
|
|
unpadded_bin = prefix + index_bytes
|
|
document = {
|
|
"accessors": accessors,
|
|
"asset": {"generator": "SimpleCAD Scene GLB Profile 1", "version": "2.0"},
|
|
"bufferViews": views,
|
|
"buffers": [{"byteLength": len(unpadded_bin)}],
|
|
"meshes": [
|
|
{"primitives": [{"attributes": attributes, "indices": index_view, "mode": mode}]}
|
|
],
|
|
"nodes": [{"mesh": 0}],
|
|
"scene": 0,
|
|
"scenes": [{"nodes": [0]}],
|
|
}
|
|
json_bytes = canonical_json_bytes(document)
|
|
json_chunk = json_bytes + b" " * ((-len(json_bytes)) % 4)
|
|
bin_chunk = unpadded_bin + b"\0" * ((-len(unpadded_bin)) % 4)
|
|
total_length = 12 + 8 + len(json_chunk) + 8 + len(bin_chunk)
|
|
return b"".join(
|
|
(
|
|
struct.pack("<III", 0x46546C67, 2, total_length),
|
|
struct.pack("<II", len(json_chunk), 0x4E4F534A),
|
|
json_chunk,
|
|
struct.pack("<II", len(bin_chunk), 0x004E4942),
|
|
bin_chunk,
|
|
)
|
|
)
|
|
|
|
|
|
def _canonical_triangle_asset():
|
|
points = {
|
|
"a": (0.0, 0.0, 0.0),
|
|
"b": (1.0, 0.0, 0.0),
|
|
"c": (0.0, 1.0, 0.0),
|
|
"d": (0.0, 0.0, 1.0),
|
|
}
|
|
oriented_faces = {
|
|
"abc": ("a", "c", "b"),
|
|
"abd": ("a", "b", "d"),
|
|
"acd": ("a", "d", "c"),
|
|
"bcd": ("b", "c", "d"),
|
|
}
|
|
blocks = []
|
|
for name, corners in oriented_faces.items():
|
|
a, b, c = (points[corner] for corner in corners)
|
|
normal = _normalize(_cross(tuple(b[i] - a[i] for i in range(3)), tuple(c[i] - a[i] for i in range(3))))
|
|
local_positions = sorted(
|
|
(points[corner] for corner in corners), key=lambda point: struct.pack("<fff", *point)
|
|
)
|
|
remap = {point: index for index, point in enumerate(local_positions)}
|
|
triple = tuple(remap[points[corner]] for corner in corners)
|
|
rotations = (triple, triple[1:] + triple[:1], triple[2:] + triple[:2])
|
|
triple = min(rotations)
|
|
block = b"".join(
|
|
struct.pack("<ffffff", *point, *normal) for point in local_positions
|
|
) + struct.pack("<HHH", *triple)
|
|
blocks.append((hashlib.sha256(b"\x01" + block).digest(), block, name, local_positions, normal, triple))
|
|
blocks.sort(key=lambda item: (item[0], item[1]))
|
|
positions = []
|
|
normals = []
|
|
indices = []
|
|
group_names = []
|
|
for _render_key, _block, name, local_positions, normal, triple in blocks:
|
|
offset = len(positions)
|
|
positions.extend(local_positions)
|
|
normals.extend([normal] * len(local_positions))
|
|
indices.extend(offset + index for index in triple)
|
|
group_names.append(name)
|
|
return _glb("triangle", positions, indices, normals), group_names
|
|
|
|
|
|
def _canonical_line_asset():
|
|
points = {
|
|
"a": (0.0, 0.0, 0.0),
|
|
"b": (1.0, 0.0, 0.0),
|
|
"c": (0.0, 1.0, 0.0),
|
|
"d": (0.0, 0.0, 1.0),
|
|
}
|
|
blocks = []
|
|
for name in ("ab", "ac", "ad", "bc", "bd", "cd"):
|
|
local_positions = sorted(
|
|
(points[name[0]], points[name[1]]), key=lambda point: struct.pack("<fff", *point)
|
|
)
|
|
block = b"".join(struct.pack("<fff", *point) for point in local_positions)
|
|
block += struct.pack("<HH", 0, 1)
|
|
blocks.append((hashlib.sha256(b"\x02" + block).digest(), block, name, local_positions))
|
|
blocks.sort(key=lambda item: (item[0], item[1]))
|
|
positions = []
|
|
indices = []
|
|
group_names = []
|
|
for _render_key, _block, name, local_positions in blocks:
|
|
offset = len(positions)
|
|
positions.extend(local_positions)
|
|
indices.extend((offset, offset + 1))
|
|
group_names.append(name)
|
|
return _glb("line", positions, indices), group_names
|
|
|
|
|
|
def _entity(kind, index, parents, children, geometry, properties, frame, status):
|
|
return {
|
|
"child_entity_ids": sorted(children),
|
|
"connector_binding_status": status,
|
|
"entity_id": f"entity/{kind}/{index}",
|
|
"evaluated_tags": [],
|
|
"geometry": geometry,
|
|
"kind": kind,
|
|
"parent_entity_ids": sorted(parents),
|
|
"properties": properties,
|
|
"render_status": "rendered",
|
|
"sdk_connector_frame": frame,
|
|
"sdk_metadata": {},
|
|
"semantic_binding_ids": [],
|
|
"source": {"kind": "unbound"},
|
|
}
|
|
|
|
|
|
def _build_scene_package():
|
|
triangle_glb, face_order = _canonical_triangle_asset()
|
|
line_glb, edge_order = _canonical_line_asset()
|
|
geometry_hash = "sha256:" + hashlib.sha256(triangle_glb).hexdigest()
|
|
edge_hash = "sha256:" + hashlib.sha256(line_glb).hexdigest()
|
|
face_ids = {name: f"entity/face/{index}" for index, name in enumerate(face_order)}
|
|
edge_ids = {name: f"entity/edge/{index}" for index, name in enumerate(edge_order)}
|
|
vertex_ids = {name: f"entity/vertex/{index}" for index, name in enumerate("abcd")}
|
|
cad_points = {
|
|
"a": (0.0, 0.0, 0.0),
|
|
"b": (1000.0, 0.0, 0.0),
|
|
"c": (0.0, 0.0, 1000.0),
|
|
"d": (0.0, -1000.0, 0.0),
|
|
}
|
|
face_vertices = {
|
|
"abc": "abc",
|
|
"abd": "abd",
|
|
"acd": "acd",
|
|
"bcd": "bcd",
|
|
}
|
|
face_edges = {
|
|
"abc": ("ab", "ac", "bc"),
|
|
"abd": ("ab", "ad", "bd"),
|
|
"acd": ("ac", "ad", "cd"),
|
|
"bcd": ("bc", "bd", "cd"),
|
|
}
|
|
edge_faces = {
|
|
edge: tuple(name for name, edges in face_edges.items() if edge in edges)
|
|
for edge in edge_order
|
|
}
|
|
vertex_edges = {
|
|
vertex: tuple(edge for edge in edge_order if vertex in edge)
|
|
for vertex in "abcd"
|
|
}
|
|
entities = []
|
|
entities.append(
|
|
_entity(
|
|
"solid",
|
|
0,
|
|
[],
|
|
list(face_ids.values()),
|
|
{"type": "brep_solid"},
|
|
{
|
|
"bounds": {"max": [1000, 0, 1000], "min": [0, -1000, 0]},
|
|
"centroid": [250, -250, 250],
|
|
"quality": "kernel_evaluated",
|
|
"surface_area": 1_500_000 + 500_000 * math.sqrt(3),
|
|
"volume": 1_000_000_000 / 6,
|
|
},
|
|
None,
|
|
"not_applicable",
|
|
)
|
|
)
|
|
face_normals = {
|
|
"abc": (0.0, 1.0, 0.0),
|
|
"abd": (0.0, 0.0, -1.0),
|
|
"acd": (-1.0, 0.0, 0.0),
|
|
"bcd": _normalize((1.0, -1.0, 1.0)),
|
|
}
|
|
for name in face_order:
|
|
vertices = [cad_points[vertex] for vertex in face_vertices[name]]
|
|
minimum = [min(point[axis] for point in vertices) for axis in range(3)]
|
|
maximum = [max(point[axis] for point in vertices) for axis in range(3)]
|
|
centroid = [sum(point[axis] for point in vertices) / 3 for axis in range(3)]
|
|
edge_vector = tuple(vertices[1][axis] - vertices[0][axis] for axis in range(3))
|
|
frame = _frame(centroid, edge_vector)
|
|
frame["z_axis"] = list(face_normals[name])
|
|
frame["y_axis"] = list(_normalize(_cross(face_normals[name], frame["x_axis"])))
|
|
entities.append(
|
|
_entity(
|
|
"face",
|
|
int(face_ids[name].rsplit("/", 1)[1]),
|
|
["entity/solid/0"],
|
|
[edge_ids[edge] for edge in face_edges[name]],
|
|
{
|
|
"normal": list(face_normals[name]),
|
|
"origin": centroid,
|
|
"type": "plane",
|
|
"x_direction": frame["x_axis"],
|
|
},
|
|
{
|
|
"area": 500_000 * (math.sqrt(3) if name == "bcd" else 1),
|
|
"bounds": {"max": maximum, "min": minimum},
|
|
"centroid": centroid,
|
|
"orientation": "forward",
|
|
"quality": "kernel_evaluated",
|
|
},
|
|
frame,
|
|
"owner_not_part",
|
|
)
|
|
)
|
|
for name in edge_order:
|
|
start, end = (cad_points[vertex] for vertex in name)
|
|
direction = tuple(end[axis] - start[axis] for axis in range(3))
|
|
midpoint = [(start[axis] + end[axis]) / 2 for axis in range(3)]
|
|
minimum = [min(start[axis], end[axis]) for axis in range(3)]
|
|
maximum = [max(start[axis], end[axis]) for axis in range(3)]
|
|
entities.append(
|
|
_entity(
|
|
"edge",
|
|
int(edge_ids[name].rsplit("/", 1)[1]),
|
|
[face_ids[face] for face in edge_faces[name]],
|
|
[vertex_ids[vertex] for vertex in name],
|
|
{"direction": list(_normalize(direction)), "origin": list(start), "type": "line"},
|
|
{
|
|
"bounds": {"max": maximum, "min": minimum},
|
|
"centroid": midpoint,
|
|
"length": math.sqrt(sum(component * component for component in direction)),
|
|
"quality": "kernel_evaluated",
|
|
},
|
|
_frame(midpoint, direction),
|
|
"owner_not_part",
|
|
)
|
|
)
|
|
for name in "abcd":
|
|
point = cad_points[name]
|
|
entities.append(
|
|
_entity(
|
|
"vertex",
|
|
int(vertex_ids[name].rsplit("/", 1)[1]),
|
|
[edge_ids[edge] for edge in vertex_edges[name]],
|
|
[],
|
|
{"position": list(point), "type": "point"},
|
|
{
|
|
"bounds": {"max": list(point), "min": list(point)},
|
|
"position": list(point),
|
|
"quality": "kernel_evaluated",
|
|
},
|
|
{**IDENTITY, "origin": list(point)},
|
|
"owner_not_part",
|
|
)
|
|
)
|
|
definition_id = "definition/root/shape/manual/fixture"
|
|
entity_asset = {
|
|
"definition_id": definition_id,
|
|
"edge_asset_id": edge_hash,
|
|
"edge_groups": [
|
|
{
|
|
"entity_id": edge_ids[name],
|
|
"first_index": 2 * index,
|
|
"group_id": index,
|
|
"index_count": 2,
|
|
"mesh_index": 0,
|
|
"primitive_index": 0,
|
|
}
|
|
for index, name in enumerate(edge_order)
|
|
],
|
|
"entities": entities,
|
|
"face_groups": [
|
|
{
|
|
"entity_id": face_ids[name],
|
|
"first_index": 3 * index,
|
|
"group_id": index,
|
|
"index_count": 3,
|
|
"mesh_index": 0,
|
|
"primitive_index": 0,
|
|
}
|
|
for index, name in enumerate(face_order)
|
|
],
|
|
"geometry_asset_id": geometry_hash,
|
|
"geometry_engine": {
|
|
"name": "OpenCascade",
|
|
"profile": "ocp-evaluated-properties-1",
|
|
"version": "fixture",
|
|
},
|
|
"schema_version": "1.0",
|
|
}
|
|
entity_asset["entities"].sort(key=lambda entity: entity["entity_id"].encode("utf-8"))
|
|
entity_bytes = canonical_json_bytes(entity_asset)
|
|
entity_hash = "sha256:" + hashlib.sha256(entity_bytes).hexdigest()
|
|
appearance_draft = {
|
|
"alpha_mode": "opaque",
|
|
"base_color": [0.72, 0.75, 0.78, 1],
|
|
"double_sided": False,
|
|
"edge_color": [0.08, 0.09, 0.1, 1],
|
|
"metallic": 0,
|
|
"name": None,
|
|
"roughness": 0.55,
|
|
"sdk_metadata": {},
|
|
"source": None,
|
|
}
|
|
appearance_id = "appearance/evaluated/" + hashlib.sha256(
|
|
canonical_json_bytes(appearance_draft)
|
|
).hexdigest()
|
|
appearance = {"appearance_id": appearance_id, **appearance_draft}
|
|
geometry_uri = f"geometry/sha256-{geometry_hash.removeprefix('sha256:')}.glb"
|
|
edge_uri = f"edges/sha256-{edge_hash.removeprefix('sha256:')}.glb"
|
|
entity_uri = f"entities/sha256-{entity_hash.removeprefix('sha256:')}.json"
|
|
bounds = {"max": [1000, 0, 1000], "min": [0, -1000, 0]}
|
|
scene = with_scene_revision(
|
|
{
|
|
"annotations": [],
|
|
"appearances": [appearance],
|
|
"cameras": [],
|
|
"compile_options": {
|
|
"angular_tolerance": 0.22,
|
|
"embed_presentation": False,
|
|
"embed_source": False,
|
|
"linear_tolerance": 0.35,
|
|
},
|
|
"connectors": [],
|
|
"coordinate_system": {"handedness": "right", "length_unit": "mm", "up_axis": "+Z"},
|
|
"definitions": [
|
|
{
|
|
"appearance_id": appearance_id,
|
|
"definition_id": definition_id,
|
|
"edge_asset_id": edge_hash,
|
|
"entity_asset_id": entity_hash,
|
|
"geometry_asset_id": geometry_hash,
|
|
"kind": "shape",
|
|
"name": None,
|
|
"sdk_metadata": {},
|
|
"source": {"kind": "manual", "root_id": "root", "source_id": "fixture"},
|
|
}
|
|
],
|
|
"diagnostics": [],
|
|
"edge_assets": [
|
|
{
|
|
"asset_id": edge_hash,
|
|
"asset_to_scene": ASSET_TO_SCENE,
|
|
"byte_length": len(line_glb),
|
|
"content_hash": edge_hash,
|
|
"media_type": "model/gltf-binary",
|
|
"scene_local_bounds": bounds,
|
|
"tessellation": {"linear_tolerance": 0.35},
|
|
"uri": edge_uri,
|
|
}
|
|
],
|
|
"entity_assets": [
|
|
{
|
|
"byte_length": len(entity_bytes),
|
|
"content_hash": entity_hash,
|
|
"entity_asset_id": entity_hash,
|
|
"media_type": "application/vnd.simplecad.entities+json",
|
|
"uri": entity_uri,
|
|
}
|
|
],
|
|
"extensions": {},
|
|
"extensions_required": [],
|
|
"extensions_used": [],
|
|
"generator": {
|
|
"name": "simplecadapi",
|
|
"ocp_bindings_version": "fixture",
|
|
"ocp_version": "fixture",
|
|
"platform_tag": "fixture",
|
|
"profile": "scene-1.0-ocp-glb-2",
|
|
"python_abi": "fixture",
|
|
"simplecadapi_version": "fixture",
|
|
"toolchain_hash": "sha256:" + "0" * 64,
|
|
},
|
|
"geometry_assets": [
|
|
{
|
|
"asset_id": geometry_hash,
|
|
"asset_to_scene": ASSET_TO_SCENE,
|
|
"byte_length": len(triangle_glb),
|
|
"content_hash": geometry_hash,
|
|
"media_type": "model/gltf-binary",
|
|
"scene_local_bounds": bounds,
|
|
"tessellation": {"angular_tolerance": 0.22, "linear_tolerance": 0.35},
|
|
"uri": geometry_uri,
|
|
}
|
|
],
|
|
"lights": [],
|
|
"nodes": [
|
|
{
|
|
"appearance_override_id": None,
|
|
"definition_id": definition_id,
|
|
"name": None,
|
|
"node_id": "instance/root",
|
|
"order": 0,
|
|
"parent_node_id": None,
|
|
"sdk_metadata": {},
|
|
"selectable": True,
|
|
"source": {"kind": "shape_root", "root_id": "root"},
|
|
"transform": IDENTITY,
|
|
"visible": True,
|
|
}
|
|
],
|
|
"scene_id": "fixture",
|
|
"schema_version": "1.0",
|
|
"source": {"kind": "manual", "source_id": "fixture"},
|
|
}
|
|
)
|
|
scene_bytes = canonical_json_bytes(scene)
|
|
blobs = {geometry_uri: triangle_glb, edge_uri: line_glb, entity_uri: entity_bytes}
|
|
return scene, scene_bytes, entity_asset, entity_bytes, blobs
|
|
|
|
|
|
def _attach_presentation(scene, blobs):
|
|
presentation = {
|
|
"appearances": [
|
|
{
|
|
"alpha_mode": "opaque",
|
|
"base_color": [1, 0.5, 0.1, 1],
|
|
"double_sided": False,
|
|
"edge_color": [0.1, 0.05, 0, 1],
|
|
"metallic": 0,
|
|
"name": "Highlight",
|
|
"roughness": 0.4,
|
|
}
|
|
],
|
|
"cameras": [
|
|
{
|
|
"far": 10000,
|
|
"name": "Overview",
|
|
"near": 1,
|
|
"parent_node_id": "instance/root",
|
|
"projection": "perspective",
|
|
"transform": IDENTITY,
|
|
"vertical_fov_degrees": 45,
|
|
}
|
|
],
|
|
"node_overrides": [
|
|
{
|
|
"appearance_name": "Highlight",
|
|
"node_id": "instance/root",
|
|
"visible": False,
|
|
}
|
|
],
|
|
"presentation_id": "fixture-presentation",
|
|
"schema_version": "1.0",
|
|
"source_scene_id": scene["scene_id"],
|
|
}
|
|
authored = presentation["appearances"][0]
|
|
evaluated_draft = {
|
|
"alpha_mode": authored["alpha_mode"],
|
|
"base_color": authored["base_color"],
|
|
"double_sided": authored["double_sided"],
|
|
"edge_color": authored["edge_color"],
|
|
"metallic": authored["metallic"],
|
|
"name": authored["name"],
|
|
"roughness": authored["roughness"],
|
|
"sdk_metadata": {},
|
|
"source": {
|
|
"appearance_name": authored["name"],
|
|
"kind": "presentation",
|
|
"presentation_id": presentation["presentation_id"],
|
|
},
|
|
}
|
|
appearance_id = "appearance/evaluated/" + hashlib.sha256(
|
|
canonical_json_bytes(evaluated_draft)
|
|
).hexdigest()
|
|
scene["appearances"].append({"appearance_id": appearance_id, **evaluated_draft})
|
|
scene["appearances"].sort(key=lambda appearance: appearance["appearance_id"].encode("utf-8"))
|
|
scene["nodes"][0]["visible"] = False
|
|
scene["nodes"][0]["appearance_override_id"] = appearance_id
|
|
scene["cameras"] = [
|
|
{"camera_id": "camera/fixture-presentation/Overview", **presentation["cameras"][0]}
|
|
]
|
|
presentation_bytes = canonical_json_bytes(presentation)
|
|
scene["compile_options"]["embed_presentation"] = True
|
|
scene["presentation_source"] = {
|
|
"artifact_hash": "sha256:" + hashlib.sha256(presentation_bytes).hexdigest(),
|
|
"embedded_artifact_byte_length": len(presentation_bytes),
|
|
"embedded_artifact_uri": "presentation/presentation.json",
|
|
"presentation_id": presentation["presentation_id"],
|
|
"schema_version": "1.0",
|
|
}
|
|
scene = with_scene_revision(scene)
|
|
blobs = dict(blobs)
|
|
blobs["presentation/presentation.json"] = presentation_bytes
|
|
return scene, presentation, blobs
|
|
|
|
|
|
def _replace_embedded_presentation(scene, blobs, payload):
|
|
scene["presentation_source"]["artifact_hash"] = (
|
|
"sha256:" + hashlib.sha256(payload).hexdigest()
|
|
)
|
|
scene["presentation_source"]["embedded_artifact_byte_length"] = len(payload)
|
|
scene = with_scene_revision(scene)
|
|
blobs = dict(blobs)
|
|
blobs["presentation/presentation.json"] = payload
|
|
return scene, blobs
|
|
|
|
|
|
def _codes(report):
|
|
return [issue.code for issue in report.issues]
|
|
|
|
|
|
def _load_shared_corpus():
|
|
path = Path(__file__).parent / "fixtures" / "scene-contract" / "corpus.json"
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_generated_python_scene_contract_types_are_current():
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(Path(__file__).parents[1] / "tools/generate_scene_contract_types.py"),
|
|
"--check",
|
|
"--language",
|
|
"python",
|
|
],
|
|
cwd=Path(__file__).parents[1],
|
|
check=True,
|
|
)
|
|
|
|
|
|
def _decode_base64(value):
|
|
return base64.b64decode(value, validate=True)
|
|
|
|
|
|
def _f64_from_hex(value):
|
|
return struct.unpack(">d", bytes.fromhex(value))[0]
|
|
|
|
|
|
def _f64_hex(value):
|
|
return struct.pack(">d", value).hex()
|
|
|
|
|
|
def _first_report_issue(report):
|
|
issue = report.first_error
|
|
return None if issue is None else {"code": issue.code, "path": issue.path}
|
|
|
|
|
|
def _replace_glb_json(glb, mutate):
|
|
json_length = struct.unpack_from("<I", glb, 12)[0]
|
|
document = parse_strict_json(glb[20 : 20 + json_length].rstrip(b" "))
|
|
mutate(document)
|
|
replacement = canonical_json_bytes(document)
|
|
replacement += b" " * ((-len(replacement)) % 4)
|
|
bin_header = 20 + json_length
|
|
bin_chunk = glb[bin_header:]
|
|
result = b"".join(
|
|
(
|
|
struct.pack("<III", 0x46546C67, 2, 12 + 8 + len(replacement) + len(bin_chunk)),
|
|
struct.pack("<II", len(replacement), 0x4E4F534A),
|
|
replacement,
|
|
bin_chunk,
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
def _declared_deflate_zip(compressed, uncompressed_size, *, local_uncompressed_size=None):
|
|
name = b"scene.json"
|
|
if local_uncompressed_size is None:
|
|
local_uncompressed_size = uncompressed_size
|
|
local = struct.pack(
|
|
"<IHHHHHIIIHH",
|
|
0x04034B50,
|
|
20,
|
|
0x0800,
|
|
8,
|
|
0,
|
|
0x0021,
|
|
0,
|
|
len(compressed),
|
|
local_uncompressed_size,
|
|
len(name),
|
|
0,
|
|
)
|
|
central_offset = len(local) + len(name) + len(compressed)
|
|
central = struct.pack(
|
|
"<IHHHHHHIIIHHHHHII",
|
|
0x02014B50,
|
|
0x0314,
|
|
20,
|
|
0x0800,
|
|
8,
|
|
0,
|
|
0x0021,
|
|
0,
|
|
len(compressed),
|
|
uncompressed_size,
|
|
len(name),
|
|
0,
|
|
0,
|
|
0,
|
|
0,
|
|
0x81A40000,
|
|
0,
|
|
) + name
|
|
eocd = struct.pack(
|
|
"<IHHHHIIH",
|
|
0x06054B50,
|
|
0,
|
|
0,
|
|
1,
|
|
1,
|
|
len(central),
|
|
central_offset,
|
|
0,
|
|
)
|
|
return local + name + compressed + central + eocd
|
|
|
|
|
|
def test_strict_and_canonical_json_reject_ambiguous_encodings():
|
|
with pytest.raises(ValueError, match="duplicate"):
|
|
parse_strict_json(b'{"a":1,"a":2}')
|
|
with pytest.raises(ValueError, match="BOM"):
|
|
parse_strict_json(b"\xef\xbb\xbf{}")
|
|
with pytest.raises(ValueError, match="canonical"):
|
|
parse_canonical_json(b'{"b":1, "a":2}')
|
|
with pytest.raises(ValueError, match="surrogate"):
|
|
parse_strict_json(b'{"value":"\\ud800"}')
|
|
with pytest.raises(ValueError, match="surrogate"):
|
|
parse_strict_json(b'{"\\ud800":0}')
|
|
|
|
|
|
def test_packaged_contract_artifacts_validate_and_link_exact_pseudocode_hashes():
|
|
schema_paths = [
|
|
"schemas/scene-1.0.schema.json",
|
|
"schemas/entities-1.0.schema.json",
|
|
"schemas/presentation-1.0.schema.json",
|
|
"schemas/connector-binding-1.0.schema.json",
|
|
"schemas/normalized-product-1.schema.json",
|
|
"schemas/rules-1.schema.json",
|
|
"schemas/profile-1.schema.json",
|
|
"schemas/render-profile-2.schema.json",
|
|
"schemas/evaluated-properties-profile-1.schema.json",
|
|
]
|
|
schemas = {
|
|
path: parse_strict_json(load_contract_artifact(path)) for path in schema_paths
|
|
}
|
|
for schema in schemas.values():
|
|
Draft202012Validator.check_schema(schema)
|
|
rules_schema = parse_strict_json(load_contract_artifact("schemas/rules-1.schema.json"))
|
|
Draft202012Validator(rules_schema).validate(
|
|
parse_strict_json(load_contract_artifact("rules/scene-1.0-rules.json"))
|
|
)
|
|
profile_schemas = {
|
|
"render_asset": schemas["schemas/render-profile-2.schema.json"],
|
|
"evaluated_properties": schemas[
|
|
"schemas/evaluated-properties-profile-1.schema.json"
|
|
],
|
|
}
|
|
registry = Registry().with_resources(
|
|
(
|
|
schema["$id"],
|
|
Resource.from_contents(schema, default_specification=DRAFT202012),
|
|
)
|
|
for schema in profile_schemas.values()
|
|
)
|
|
dispatcher = Draft202012Validator(
|
|
schemas["schemas/profile-1.schema.json"], registry=registry
|
|
)
|
|
profile_cases = (
|
|
(
|
|
"profiles/scene-1.0-ocp-glb-2.profile.json",
|
|
"scene-1.0-ocp-glb-2",
|
|
"render_asset",
|
|
),
|
|
(
|
|
"profiles/ocp-evaluated-properties-1.profile.json",
|
|
"ocp-evaluated-properties-1",
|
|
"evaluated_properties",
|
|
),
|
|
)
|
|
for path, profile_id, profile_kind in profile_cases:
|
|
profile = parse_strict_json(load_contract_artifact(path))
|
|
applicable = Draft202012Validator(profile_schemas[profile_kind])
|
|
other_kind = next(kind for kind in profile_schemas if kind != profile_kind)
|
|
|
|
assert profile["profile_id"] == profile_id
|
|
assert profile["profile_kind"] == profile_kind
|
|
applicable.validate(profile)
|
|
dispatcher.validate(profile)
|
|
assert not Draft202012Validator(profile_schemas[other_kind]).is_valid(profile)
|
|
|
|
unknown_nested = deepcopy(profile)
|
|
unknown_nested["rules"]["implementation_status"]["unknown"] = True
|
|
assert not applicable.is_valid(unknown_nested)
|
|
assert not dispatcher.is_valid(unknown_nested)
|
|
|
|
deeply_unknown = deepcopy(profile)
|
|
if profile_kind == "render_asset":
|
|
deeply_unknown["rules"]["default_appearance"]["unknown"] = True
|
|
else:
|
|
deeply_unknown["rules"]["canonical_labeling"]["unknown"] = True
|
|
assert not applicable.is_valid(deeply_unknown)
|
|
assert not dispatcher.is_valid(deeply_unknown)
|
|
|
|
wrong_id = deepcopy(profile)
|
|
wrong_id["profile_id"] = "wrong-profile"
|
|
assert not applicable.is_valid(wrong_id)
|
|
assert not dispatcher.is_valid(wrong_id)
|
|
|
|
wrong_kind = deepcopy(profile)
|
|
wrong_kind["profile_kind"] = other_kind
|
|
assert not applicable.is_valid(wrong_kind)
|
|
assert not dispatcher.is_valid(wrong_kind)
|
|
|
|
pseudocode = load_contract_artifact(profile["pseudocode_uri"])
|
|
assert profile["pseudocode_hash"] == "sha256:" + hashlib.sha256(pseudocode).hexdigest()
|
|
|
|
|
|
def test_rule_registry_rejects_schema_and_ordering_invariant_violations():
|
|
schema = parse_strict_json(load_contract_artifact("schemas/rules-1.schema.json"))
|
|
registry = parse_strict_json(load_contract_artifact("rules/scene-1.0-rules.json"))
|
|
assert set(_validate_rule_registry(registry, schema)) == {
|
|
rule["id"] for rule in registry["rules"]
|
|
}
|
|
|
|
missing_precedence = deepcopy(registry)
|
|
missing_precedence["rules"][0].pop("precedence")
|
|
with pytest.raises(ValueError, match="precedence"):
|
|
_validate_rule_registry(missing_precedence, schema)
|
|
|
|
duplicate_id = deepcopy(registry)
|
|
duplicate_id["rules"][1]["id"] = duplicate_id["rules"][0]["id"]
|
|
with pytest.raises(ValueError, match="duplicate rule ID"):
|
|
_validate_rule_registry(duplicate_id, schema)
|
|
|
|
duplicate_precedence = deepcopy(registry)
|
|
duplicate_precedence["rules"][1]["precedence"] = 0
|
|
with pytest.raises(ValueError, match="duplicate precedence"):
|
|
_validate_rule_registry(duplicate_precedence, schema)
|
|
|
|
wrong_order = deepcopy(registry)
|
|
wrong_order["rules"][0], wrong_order["rules"][1] = (
|
|
wrong_order["rules"][1],
|
|
wrong_order["rules"][0],
|
|
)
|
|
with pytest.raises(ValueError, match="ordered by phase and precedence"):
|
|
_validate_rule_registry(wrong_order, schema)
|
|
|
|
wrong_id_order = deepcopy(registry)
|
|
wrong_id_order["rules"][0]["id"], wrong_id_order["rules"][1]["id"] = (
|
|
wrong_id_order["rules"][1]["id"],
|
|
wrong_id_order["rules"][0]["id"],
|
|
)
|
|
with pytest.raises(ValueError, match="unsigned UTF-8 rule-ID order"):
|
|
_validate_rule_registry(wrong_id_order, schema)
|
|
|
|
|
|
def test_rule_registry_controls_reports_and_closes_issue_authority():
|
|
report = _report(
|
|
[
|
|
_issue("transform_invalid", "/z", "later registered rule"),
|
|
_issue("bounds_invalid", "/z", "z message"),
|
|
_issue("bounds_invalid", "/z", "a message"),
|
|
_issue("bounds_invalid", "/a", "earlier pointer"),
|
|
],
|
|
artifact="scene",
|
|
)
|
|
assert [(issue.code, issue.path, issue.message) for issue in report.issues] == [
|
|
("bounds_invalid", "/a", "earlier pointer"),
|
|
("bounds_invalid", "/z", "a message"),
|
|
("bounds_invalid", "/z", "z message"),
|
|
("transform_invalid", "/z", "later registered rule"),
|
|
]
|
|
|
|
with pytest.raises(ValueError, match="unregistered"):
|
|
_report([_issue("not_registered", "", "no authority")], artifact="scene")
|
|
with pytest.raises(ValueError, match="registered phase"):
|
|
_report(
|
|
[_issue("bounds_invalid", "", "wrong phase", "package")],
|
|
artifact="scene",
|
|
)
|
|
with pytest.raises(ValueError, match="does not apply"):
|
|
_report(
|
|
[_issue("presentation_reference_invalid", "", "wrong artifact")],
|
|
artifact="scene",
|
|
)
|
|
with pytest.raises(ValueError, match="root pointer"):
|
|
_report(
|
|
[_issue("duplicate_json_key", "/value", "wrong pointer", "parse")],
|
|
artifact="scene",
|
|
)
|
|
|
|
|
|
def test_valid_tetrahedron_scene_package_and_canonical_zip_round_trip():
|
|
scene, scene_bytes, entity_asset, entity_bytes, blobs = _build_scene_package()
|
|
|
|
assert validate_scene_manifest(scene).valid
|
|
assert validate_scene_manifest(scene_bytes).valid
|
|
assert validate_entity_asset(entity_asset).valid
|
|
assert validate_entity_asset(entity_bytes).valid
|
|
assert validate_scene_package(scene_bytes, blobs).valid
|
|
assert {preflight_glb(payload).kind for uri, payload in blobs.items() if uri.endswith(".glb")} == {
|
|
"triangle",
|
|
"line",
|
|
}
|
|
|
|
archive_bytes = canonical_zip_bytes({"scene.json": scene_bytes, **blobs})
|
|
archive = preflight_zip_bytes(archive_bytes)
|
|
assert archive.input_size == archive.canonical_size == len(archive_bytes)
|
|
assert not archive.used_deflate
|
|
assert dict(archive.members) == {"scene.json": scene_bytes, **blobs}
|
|
assert canonical_zip_bytes(archive.members) == archive_bytes
|
|
assert validate_scene_package(
|
|
archive.members["scene.json"],
|
|
{name: payload for name, payload in archive.members.items() if name != "scene.json"},
|
|
).valid
|
|
|
|
|
|
def test_embedded_presentation_is_canonical_and_resolves_exactly_against_scene():
|
|
scene, _scene_bytes, _entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
scene, presentation, blobs = _attach_presentation(scene, blobs)
|
|
|
|
assert validate_scene_package(canonical_json_bytes(scene), blobs).valid
|
|
|
|
noncanonical = json.dumps(presentation, indent=2, sort_keys=True).encode("utf-8")
|
|
scene, blobs = _replace_embedded_presentation(scene, blobs, noncanonical)
|
|
report = validate_scene_package(canonical_json_bytes(scene), blobs)
|
|
assert report.first_error is not None
|
|
assert (report.first_error.code, report.first_error.path) == (
|
|
"noncanonical_json",
|
|
"",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutate", "code", "path"),
|
|
[
|
|
(
|
|
lambda presentation: presentation.__setitem__("presentation_id", "other"),
|
|
"source_matrix_invalid",
|
|
"/presentation/presentation.json/presentation_id",
|
|
),
|
|
(
|
|
lambda presentation: presentation.__setitem__("source_scene_id", "other"),
|
|
"source_matrix_invalid",
|
|
"/presentation/presentation.json/source_scene_id",
|
|
),
|
|
(
|
|
lambda presentation: presentation["node_overrides"][0].__setitem__(
|
|
"node_id", "instance/missing"
|
|
),
|
|
"source_matrix_invalid",
|
|
"/presentation/presentation.json/node_overrides/0/node_id",
|
|
),
|
|
(
|
|
lambda presentation: presentation["cameras"][0].__setitem__(
|
|
"parent_node_id", "instance/missing"
|
|
),
|
|
"source_matrix_invalid",
|
|
"/presentation/presentation.json/cameras/0/parent_node_id",
|
|
),
|
|
],
|
|
)
|
|
def test_embedded_presentation_ids_and_scene_references_are_contextual(
|
|
mutate, code, path
|
|
):
|
|
scene, _scene_bytes, _entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
scene, presentation, blobs = _attach_presentation(scene, blobs)
|
|
mutate(presentation)
|
|
scene, blobs = _replace_embedded_presentation(
|
|
scene, blobs, canonical_json_bytes(presentation)
|
|
)
|
|
|
|
report = validate_scene_package(canonical_json_bytes(scene), blobs)
|
|
|
|
assert any(issue.code == code and issue.path == path for issue in report.issues)
|
|
|
|
|
|
def test_embedded_presentation_appearance_and_camera_resolution_is_exact():
|
|
scene, _scene_bytes, _entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
scene, _presentation, blobs = _attach_presentation(scene, blobs)
|
|
presentation_appearance = next(
|
|
appearance
|
|
for appearance in scene["appearances"]
|
|
if isinstance(appearance["source"], dict)
|
|
)
|
|
presentation_appearance["source"]["appearance_name"] = "Other"
|
|
draft = dict(presentation_appearance)
|
|
draft.pop("appearance_id")
|
|
new_id = "appearance/evaluated/" + hashlib.sha256(
|
|
canonical_json_bytes(draft)
|
|
).hexdigest()
|
|
presentation_appearance["appearance_id"] = new_id
|
|
scene["nodes"][0]["appearance_override_id"] = new_id
|
|
scene["appearances"].sort(key=lambda appearance: appearance["appearance_id"].encode("utf-8"))
|
|
scene["cameras"] = []
|
|
scene = with_scene_revision(scene)
|
|
|
|
report = validate_scene_package(canonical_json_bytes(scene), blobs)
|
|
|
|
appearance_index = scene["appearances"].index(presentation_appearance)
|
|
assert any(
|
|
issue.code == "source_matrix_invalid"
|
|
and issue.path == f"/appearances/{appearance_index}/source/appearance_name"
|
|
for issue in report.issues
|
|
)
|
|
assert any(
|
|
issue.code == "source_matrix_invalid"
|
|
and issue.path == "/presentation/presentation.json/cameras/0"
|
|
for issue in report.issues
|
|
)
|
|
|
|
|
|
def test_product_material_appearance_provenance_must_match_a_part_root():
|
|
scene, _scene_bytes, _entity_asset, _entity_bytes, _blobs = _build_scene_package()
|
|
appearance = scene["appearances"][0]
|
|
appearance["source"] = {
|
|
"kind": "product_material",
|
|
"material_id": "steel",
|
|
"root_id": "other",
|
|
}
|
|
draft = dict(appearance)
|
|
draft.pop("appearance_id")
|
|
appearance_id = "appearance/evaluated/" + hashlib.sha256(
|
|
canonical_json_bytes(draft)
|
|
).hexdigest()
|
|
appearance["appearance_id"] = appearance_id
|
|
scene["definitions"][0]["appearance_id"] = appearance_id
|
|
scene = with_scene_revision(scene)
|
|
|
|
report = validate_scene_manifest(scene)
|
|
|
|
assert any(
|
|
issue.code == "source_matrix_invalid"
|
|
and issue.path == "/appearances/0/source/root_id"
|
|
for issue in report.issues
|
|
)
|
|
|
|
|
|
def test_package_rejects_conflicting_records_for_one_uri_and_dispatches_by_role():
|
|
scene, _scene_bytes, _entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
edge_uri = scene["edge_assets"][0]["uri"]
|
|
scene["edge_assets"][0]["uri"] = scene["geometry_assets"][0]["uri"]
|
|
scene = with_scene_revision(scene)
|
|
blobs = dict(blobs)
|
|
blobs.pop(edge_uri)
|
|
|
|
report = validate_scene_package(canonical_json_bytes(scene), blobs)
|
|
|
|
assert any(issue.code == "package_member_set_invalid" for issue in report.issues)
|
|
assert any(
|
|
issue.code == "source_matrix_invalid" and issue.path == "/edge_assets/0/uri"
|
|
for issue in report.issues
|
|
)
|
|
|
|
|
|
def test_package_budget_formula_and_first_error_use_declarative_limits():
|
|
totals = _compute_package_budget_totals(
|
|
scene_json_bytes=7,
|
|
glb_decoded_buffer_bytes=11,
|
|
entity_json_bytes=13,
|
|
other_immutable_json_bytes=17,
|
|
entity_count=23,
|
|
entity_vertex_count=19,
|
|
triangle_vertex_count=29,
|
|
triangle_count=31,
|
|
line_vertex_count=37,
|
|
line_segment_count=41,
|
|
)
|
|
assert totals == {
|
|
"static_decoded_buffer_bytes": 656,
|
|
"entity_count": 23,
|
|
"triangle_vertex_total": 29,
|
|
"triangle_total": 31,
|
|
"line_vertex_total": 37,
|
|
"line_segment_total": 41,
|
|
}
|
|
exact = SimpleNamespace(
|
|
static_decoded_buffer_bytes=656,
|
|
entities_total=23,
|
|
triangle_vertices_total=29,
|
|
triangles_total=31,
|
|
line_vertices_total=37,
|
|
line_segments_total=41,
|
|
)
|
|
assert not _package_budget_issues(totals, limits=exact)
|
|
|
|
over_every_limit = SimpleNamespace(
|
|
static_decoded_buffer_bytes=655,
|
|
entities_total=22,
|
|
triangle_vertices_total=28,
|
|
triangles_total=30,
|
|
line_vertices_total=36,
|
|
line_segments_total=40,
|
|
)
|
|
report = _report(
|
|
_package_budget_issues(totals, limits=over_every_limit), artifact="package"
|
|
)
|
|
assert report.first_error is not None
|
|
assert (
|
|
report.first_error.code,
|
|
report.first_error.path,
|
|
report.first_error.message,
|
|
) == (
|
|
"resource_limit_exceeded",
|
|
"",
|
|
"total line GLB vertex count exceeds resource limit",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("value", [None, [], {"definitions": 1}, {"nodes": 1}, {"entities": 1}])
|
|
def test_structurally_invalid_values_return_reports_instead_of_raising(value):
|
|
assert not validate_scene_manifest(value).valid
|
|
assert not validate_entity_asset(value).valid
|
|
|
|
|
|
def test_asset_bearing_scene_requires_a_root_occurrence():
|
|
scene, _scene_bytes, _entity_asset, _entity_bytes, _blobs = _build_scene_package()
|
|
scene["nodes"] = []
|
|
scene = with_scene_revision(scene)
|
|
|
|
report = validate_scene_manifest(scene)
|
|
|
|
assert "hierarchy_invalid" in _codes(report)
|
|
|
|
|
|
def test_package_rejects_manifest_bounds_that_differ_from_glb():
|
|
scene, _scene_bytes, _entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
scene["geometry_assets"][0]["scene_local_bounds"]["max"][0] = 999
|
|
scene = with_scene_revision(scene)
|
|
|
|
report = validate_scene_package(canonical_json_bytes(scene), blobs)
|
|
|
|
assert "bounds_invalid" in _codes(report)
|
|
|
|
|
|
def test_glb_rejects_json_strings_and_booleans_in_numeric_fields():
|
|
_scene, _scene_bytes, _entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
triangle = next(payload for uri, payload in blobs.items() if uri.startswith("geometry/"))
|
|
string_bounds = _replace_glb_json(
|
|
triangle, lambda document: document["accessors"][0]["min"].__setitem__(0, "0")
|
|
)
|
|
boolean_scene = _replace_glb_json(triangle, lambda document: document.__setitem__("scene", False))
|
|
|
|
with pytest.raises(ValueError, match="three-number"):
|
|
preflight_glb(string_bounds)
|
|
with pytest.raises(ValueError, match="non-negative safe integer"):
|
|
preflight_glb(boolean_scene)
|
|
|
|
|
|
def test_entity_asset_rejects_sparse_ids_and_non_unit_analytic_directions():
|
|
_scene, _scene_bytes, entity_asset, _entity_bytes, _blobs = _build_scene_package()
|
|
sparse = deepcopy(entity_asset)
|
|
sparse["entities"][0]["entity_id"] = "entity/edge/99"
|
|
non_unit = deepcopy(entity_asset)
|
|
edge = next(entity for entity in non_unit["entities"] if entity["kind"] == "edge")
|
|
edge["geometry"]["direction"] = [2, 0, 0]
|
|
|
|
assert "entity_topology_invalid" in _codes(validate_entity_asset(sparse))
|
|
assert "analytic_geometry_invalid" in _codes(validate_entity_asset(non_unit))
|
|
|
|
|
|
def test_package_enforces_contextual_entity_binding_status():
|
|
scene, _scene_bytes, entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
face = next(entity for entity in entity_asset["entities"] if entity["kind"] == "face")
|
|
face["connector_binding_status"] = "supported"
|
|
entity_bytes = canonical_json_bytes(entity_asset)
|
|
old_uri = scene["entity_assets"][0]["uri"]
|
|
entity_hash = "sha256:" + hashlib.sha256(entity_bytes).hexdigest()
|
|
entity_uri = f"entities/sha256-{entity_hash.removeprefix('sha256:')}.json"
|
|
scene["entity_assets"][0].update(
|
|
{
|
|
"byte_length": len(entity_bytes),
|
|
"content_hash": entity_hash,
|
|
"entity_asset_id": entity_hash,
|
|
"uri": entity_uri,
|
|
}
|
|
)
|
|
scene["definitions"][0]["entity_asset_id"] = entity_hash
|
|
scene = with_scene_revision(scene)
|
|
blobs = {uri: payload for uri, payload in blobs.items() if uri != old_uri}
|
|
blobs[entity_uri] = entity_bytes
|
|
|
|
report = validate_scene_package(canonical_json_bytes(scene), blobs)
|
|
|
|
assert "connector_invalid" in _codes(report)
|
|
|
|
|
|
def test_deep_python_value_returns_a_budget_report_without_recursion_error():
|
|
value = {}
|
|
current = value
|
|
for _index in range(1000):
|
|
child = {}
|
|
current["x"] = child
|
|
current = child
|
|
|
|
report = validate_scene_manifest(value)
|
|
|
|
assert report.first_error is not None
|
|
assert report.first_error.code == "resource_limit_exceeded"
|
|
|
|
|
|
def test_public_validator_resource_overflows_remain_budget_issues():
|
|
scene, scene_bytes, _entity_asset, _entity_bytes, _blobs = _build_scene_package()
|
|
collection_report = validate_scene_manifest(
|
|
scene, limits=replace(BASE_LIMITS, definitions=0)
|
|
)
|
|
serialized_report = validate_scene_manifest(
|
|
scene, limits=replace(BASE_LIMITS, scene_json_bytes=len(scene_bytes) - 1)
|
|
)
|
|
|
|
assert collection_report.first_error is not None
|
|
assert (
|
|
collection_report.first_error.code,
|
|
collection_report.first_error.phase,
|
|
collection_report.first_error.path,
|
|
) == ("resource_limit_exceeded", "budget", "/definitions")
|
|
assert serialized_report.first_error is not None
|
|
assert (
|
|
serialized_report.first_error.code,
|
|
serialized_report.first_error.phase,
|
|
serialized_report.first_error.path,
|
|
) == ("resource_limit_exceeded", "budget", "")
|
|
|
|
|
|
def test_zip_preflight_rejects_crc_mutation_and_trailing_data():
|
|
_scene, scene_bytes, _entity_asset, _entity_bytes, blobs = _build_scene_package()
|
|
archive = canonical_zip_bytes({"scene.json": scene_bytes, **blobs})
|
|
corrupted = bytearray(archive)
|
|
first_payload = 30 + struct.unpack_from("<H", archive, 26)[0]
|
|
corrupted[first_payload] ^= 1
|
|
|
|
with pytest.raises(ValueError, match="CRC"):
|
|
preflight_zip_bytes(corrupted)
|
|
with pytest.raises(ValueError, match="EOCD"):
|
|
preflight_zip_bytes(archive + b"x")
|
|
|
|
|
|
def test_zip_preflight_checks_aggregate_ratio_boundary_before_inflate(monkeypatch):
|
|
compressed = b"\0"
|
|
template = _declared_deflate_zip(compressed, 0)
|
|
exact_size = len(template) * BASE_LIMITS.compression_ratio
|
|
|
|
def fail_inflate(*_args, **_kwargs):
|
|
raise AssertionError("inflate must not run before ratio preflight")
|
|
|
|
monkeypatch.setattr(scene_archive.zlib, "decompressobj", fail_inflate)
|
|
with pytest.raises(
|
|
ValueError,
|
|
match=r'^member compression ratio exceeds limit: "scene\.json"$',
|
|
):
|
|
preflight_zip_bytes(_declared_deflate_zip(compressed, exact_size))
|
|
with pytest.raises(ValueError, match="^aggregate compression ratio exceeds limit$"):
|
|
preflight_zip_bytes(_declared_deflate_zip(compressed, exact_size + 1))
|
|
with pytest.raises(ValueError, match="^ZIP central/local header mismatch$"):
|
|
preflight_zip_bytes(
|
|
_declared_deflate_zip(
|
|
compressed,
|
|
exact_size + 1,
|
|
local_uncompressed_size=exact_size,
|
|
)
|
|
)
|
|
|
|
|
|
def test_zip_preflight_zero_compressed_size_uses_one_for_member_ratio():
|
|
with pytest.raises(ValueError, match=r'^invalid deflate stream: "scene\.json"$'):
|
|
preflight_zip_bytes(
|
|
_declared_deflate_zip(b"", BASE_LIMITS.compression_ratio)
|
|
)
|
|
with pytest.raises(
|
|
ValueError,
|
|
match=r'^member compression ratio exceeds limit: "scene\.json"$',
|
|
):
|
|
preflight_zip_bytes(
|
|
_declared_deflate_zip(b"", BASE_LIMITS.compression_ratio + 1)
|
|
)
|
|
|
|
|
|
def test_python_matches_all_shared_manifest_and_entity_cases():
|
|
corpus = _load_shared_corpus()
|
|
validators = (
|
|
(corpus["manifest_cases"], validate_scene_manifest),
|
|
(corpus["entity_cases"], validate_entity_asset),
|
|
(corpus["presentation_cases"], validate_presentation),
|
|
(corpus["connector_binding_cases"], validate_connector_binding),
|
|
(corpus["normalized_product_cases"], validate_normalized_product),
|
|
)
|
|
for cases, validator in validators:
|
|
for case in cases:
|
|
report = validator(_decode_base64(case["payload_base64"]))
|
|
assert report.valid is case["valid"], case["name"]
|
|
if case["expected"] is not None:
|
|
assert report.first_error is not None
|
|
assert {
|
|
"code": report.first_error.code,
|
|
"path": report.first_error.path,
|
|
} == case["expected"], case["name"]
|
|
|
|
|
|
def test_shared_positive_case_names_cannot_drift_to_negative_expectations():
|
|
corpus = _load_shared_corpus()
|
|
positive_names = []
|
|
for section_name, cases in corpus.items():
|
|
if not isinstance(cases, list):
|
|
continue
|
|
for case in cases:
|
|
if not isinstance(case, dict):
|
|
continue
|
|
name = case.get("name")
|
|
if not isinstance(name, str) or not (
|
|
name.startswith("valid_") or name == "nullable_fields_nonnull"
|
|
):
|
|
continue
|
|
positive_names.append(f"{section_name}:{name}")
|
|
assert case["valid"] is True, positive_names[-1]
|
|
if "expected" in case:
|
|
assert case["expected"] is None, positive_names[-1]
|
|
if "error" in case:
|
|
assert case["error"] is None, positive_names[-1]
|
|
assert len(positive_names) == 30
|
|
|
|
|
|
def test_accepted_package_fixtures_cover_every_connector_binding_status():
|
|
corpus = _load_shared_corpus()
|
|
statuses = set()
|
|
for case in (*corpus["package_cases"], *corpus["scene_shape_cases"]):
|
|
if not case["valid"]:
|
|
continue
|
|
for uri, payload in corpus_blobs(corpus, case).items():
|
|
if not uri.startswith("entities/"):
|
|
continue
|
|
sidecar = parse_canonical_json(payload)
|
|
statuses.update(
|
|
entity["connector_binding_status"]
|
|
for entity in sidecar["entities"]
|
|
)
|
|
assert statuses == {
|
|
"not_applicable",
|
|
"owner_not_part",
|
|
"source_not_model",
|
|
"frame_undefined",
|
|
"selector_ambiguous",
|
|
"selector_unstable",
|
|
"supported",
|
|
}
|
|
|
|
|
|
def test_python_matches_all_shared_package_cases():
|
|
corpus = _load_shared_corpus()
|
|
precedence_statuses = {
|
|
"solid_status_precedence": "not_applicable",
|
|
"owner_not_part_status_precedence": "owner_not_part",
|
|
"source_not_model_status_precedence": "source_not_model",
|
|
"frame_undefined_status_precedence": "frame_undefined",
|
|
}
|
|
for case in corpus["package_cases"]:
|
|
blobs = corpus_blobs(corpus, case)
|
|
report = validate_scene_package(_decode_base64(case["manifest_base64"]), blobs)
|
|
assert report.valid is case["valid"], case["name"]
|
|
if case["expected"] is not None:
|
|
assert report.first_error is not None
|
|
assert {
|
|
"code": report.first_error.code,
|
|
"path": report.first_error.path,
|
|
} == case["expected"], case["name"]
|
|
expected_status = precedence_statuses.get(case["name"])
|
|
if expected_status is not None:
|
|
assert report.first_error is not None
|
|
assert report.first_error.message == (
|
|
f"connector binding status must be {expected_status}"
|
|
)
|
|
|
|
|
|
def test_python_replays_shared_scene_field_matrix():
|
|
corpus = _load_shared_corpus()
|
|
scene = parse_canonical_json(_decode_base64(corpus["artifacts"]["scene"]["base64"]))
|
|
for case in corpus["scene_field_cases"]:
|
|
value = apply_scene_field_case(scene, case)
|
|
report = validate_scene_manifest(canonical_json_bytes(value))
|
|
assert report.valid is case["valid"], case["name"]
|
|
assert _first_report_issue(report) == case["expected"], case["name"]
|
|
|
|
|
|
def test_python_characterizes_shared_scene_shapes_and_asset_reuse():
|
|
corpus = _load_shared_corpus()
|
|
for case in corpus["scene_shape_cases"]:
|
|
manifest = _decode_base64(case["manifest_base64"])
|
|
scene = parse_canonical_json(manifest)
|
|
report = validate_scene_package(manifest, corpus_blobs(corpus, case))
|
|
assert report.valid is case["valid"], case["name"]
|
|
assert _first_report_issue(report) == case["expected"], case["name"]
|
|
assert canonical_json_bytes(scene) == manifest, case["name"]
|
|
assert scene_shape_facts(scene) == case["expected_shape"], case["name"]
|
|
|
|
definitions = {item["definition_id"]: item for item in scene["definitions"]}
|
|
for definition_id, count in case["expected_shape"][
|
|
"definition_occurrence_counts"
|
|
].items():
|
|
if count > 1:
|
|
occurrences = [
|
|
node for node in scene["nodes"] if node["definition_id"] == definition_id
|
|
]
|
|
assert len(occurrences) == count
|
|
assert len({node["definition_id"] for node in occurrences}) == 1
|
|
assert definitions[definition_id]["geometry_asset_id"] is not None
|
|
|
|
|
|
def test_python_matches_shared_two_pass_revision_vectors():
|
|
for vector in _load_shared_corpus()["revision_vectors"]:
|
|
draft_bytes = _decode_base64(vector["draft_base64"])
|
|
draft = parse_canonical_json(draft_bytes)
|
|
scene_bytes = _decode_base64(vector["canonical_base64"])
|
|
scene = with_scene_revision(draft)
|
|
|
|
assert canonical_json_bytes(draft) == draft_bytes, vector["name"]
|
|
assert scene["revision"] == vector["revision"], vector["name"]
|
|
assert canonical_json_bytes(scene) == scene_bytes, vector["name"]
|
|
assert "sha256:" + hashlib.sha256(scene_bytes).hexdigest() == vector["sha256"]
|
|
|
|
|
|
def test_python_matches_all_shared_glb_and_zip_cases():
|
|
corpus = _load_shared_corpus()
|
|
for case in corpus["glb_cases"]:
|
|
payload = _decode_base64(case["payload_base64"])
|
|
if case["valid"]:
|
|
assert preflight_glb(payload, expected_kind=case["expected_kind"]).kind == case["kind"]
|
|
else:
|
|
with pytest.raises(ValueError, match="^" + re.escape(case["error"]) + "$"):
|
|
preflight_glb(payload, expected_kind=case["expected_kind"])
|
|
for case in corpus["zip_cases"]:
|
|
payload = _decode_base64(case["payload_base64"])
|
|
if case["valid"]:
|
|
assert preflight_zip_bytes(payload).used_deflate is case["used_deflate"]
|
|
else:
|
|
with pytest.raises(ValueError, match="^" + re.escape(case["error"]) + "$"):
|
|
preflight_zip_bytes(payload)
|
|
|
|
|
|
def test_python_matches_shared_numeric_profile_vectors():
|
|
for vector in _load_shared_corpus()["numeric_vectors"]:
|
|
try:
|
|
if vector["operation"] == "f32":
|
|
actual = f"{profile_f32_bits(_f64_from_hex(vector['input_bits'][0])):08x}"
|
|
elif vector["operation"] == "cross":
|
|
left, right = (
|
|
tuple(_f64_from_hex(component) for component in value)
|
|
for value in vector["input_bits"]
|
|
)
|
|
actual = [_f64_hex(component) for component in profile_cross(left, right)]
|
|
elif vector["operation"] == "normalize":
|
|
value = tuple(
|
|
_f64_from_hex(component) for component in vector["input_bits"][0]
|
|
)
|
|
actual = [
|
|
f"{profile_f32_bits(component):08x}"
|
|
for component in profile_normalize(value)
|
|
]
|
|
else:
|
|
raise AssertionError(vector["operation"])
|
|
except ValueError as exc:
|
|
assert not vector["valid"], vector["name"]
|
|
assert str(exc) == vector["error"], vector["name"]
|
|
else:
|
|
assert vector["valid"], vector["name"]
|
|
assert actual == vector["expected_bits"], vector["name"]
|
|
|
|
|
|
def test_python_matches_shared_resource_boundaries():
|
|
corpus = _load_shared_corpus()
|
|
assert asdict(BASE_LIMITS) == corpus["resource_limits"]
|
|
valid_scene = parse_canonical_json(
|
|
_decode_base64(corpus["artifacts"]["scene"]["base64"])
|
|
)
|
|
|
|
for case in corpus["resource_cases"]:
|
|
operation = case["operation"]
|
|
parameters = case["parameters"]
|
|
limits = replace(BASE_LIMITS, **parameters.get("limits", {}))
|
|
try:
|
|
if operation == "input_archive_size":
|
|
preflight_input_archive_size(parameters["size"], limits=limits)
|
|
valid, error, first = True, None, None
|
|
elif operation in {"archive_member_sizes", "archive_member_count"}:
|
|
if operation == "archive_member_sizes":
|
|
sizes = parameters["sizes"]
|
|
else:
|
|
sizes = {"scene.json": 0}
|
|
sizes.update(
|
|
{
|
|
f"x/{index:05d}": 0
|
|
for index in range(parameters["count"] - 1)
|
|
}
|
|
)
|
|
preflight_archive_member_sizes(sizes, limits=limits)
|
|
valid, error, first = True, None, None
|
|
elif operation in {
|
|
"aggregate_compression_ratio",
|
|
"member_compression_ratio",
|
|
}:
|
|
callback = (
|
|
preflight_aggregate_compression_ratio
|
|
if operation == "aggregate_compression_ratio"
|
|
else preflight_member_compression_ratio
|
|
)
|
|
callback(
|
|
parameters["uncompressed_size"],
|
|
parameters["compressed_size"],
|
|
limits=limits,
|
|
)
|
|
valid, error, first = True, None, None
|
|
elif operation == "json_depth":
|
|
value = {}
|
|
current = value
|
|
for _index in range(parameters["depth"]):
|
|
current["x"] = {}
|
|
current = current["x"]
|
|
issues = json_resource_issues(value, limits=limits)
|
|
valid = not issues
|
|
error = None
|
|
first = (
|
|
None
|
|
if not issues
|
|
else {"code": issues[0].code, "path": issues[0].path}
|
|
)
|
|
elif operation == "json_domain":
|
|
text = parameters["text"]
|
|
kind = parameters["kind"]
|
|
field = parameters.get("field")
|
|
if kind == "value":
|
|
value = {"value": text}
|
|
elif kind == "object_key":
|
|
value = {text: 0}
|
|
elif kind == "metadata_key":
|
|
value = {"metadata": {text: 0}}
|
|
elif kind == "sdk_metadata_key":
|
|
value = {"sdk_metadata": {text: 0}}
|
|
elif kind == "identifier":
|
|
value = {field or "node_id": text}
|
|
elif kind == "identifier_array":
|
|
value = {field or "component_path": [text]}
|
|
elif kind == "uri":
|
|
value = {"uri": text}
|
|
else:
|
|
raise AssertionError(kind)
|
|
issues = json_resource_issues(value, limits=limits)
|
|
valid = not issues
|
|
error = None
|
|
first = (
|
|
None
|
|
if not issues
|
|
else {"code": issues[0].code, "path": issues[0].path}
|
|
)
|
|
elif operation == "resource_count":
|
|
count = parameters["count"]
|
|
kind = parameters["kind"]
|
|
field = parameters.get("field")
|
|
if kind == "collection":
|
|
value = {field: [None] * count}
|
|
elif kind == "hierarchy":
|
|
value = {
|
|
"nodes": [
|
|
{"source": {"component_path": ["x"] * count}}
|
|
]
|
|
}
|
|
elif kind == "forwarded":
|
|
value = {
|
|
"connectors": [
|
|
{
|
|
"anchor_kind": "forwarded",
|
|
"connector_snapshot_id": f"c{index}",
|
|
"forwarded_from": {
|
|
"source_connector_snapshot_id": f"c{index + 1}"
|
|
},
|
|
}
|
|
for index in range(count)
|
|
]
|
|
}
|
|
else:
|
|
raise AssertionError(kind)
|
|
issues = resource_count_issues(
|
|
value, parameters["artifact"], limits=limits
|
|
)
|
|
valid = not issues
|
|
error = None
|
|
first = (
|
|
None
|
|
if not issues
|
|
else {"code": issues[0].code, "path": issues[0].path}
|
|
)
|
|
elif operation == "glb_counts":
|
|
preflight_glb_counts(
|
|
parameters["kind"],
|
|
parameters["vertex_count"],
|
|
parameters["index_count"],
|
|
limits=limits,
|
|
)
|
|
valid, error, first = True, None, None
|
|
elif operation == "package_budget":
|
|
contributions = parameters["contributions"]
|
|
totals = _compute_package_budget_totals(**contributions)
|
|
result = _report(
|
|
_package_budget_issues(totals, limits=limits),
|
|
artifact="package",
|
|
)
|
|
valid, error, first = result.valid, None, _first_report_issue(result)
|
|
elif operation == "scene_geometry_byte_length":
|
|
value = deepcopy(valid_scene)
|
|
byte_length = int(parameters["value"])
|
|
value["geometry_assets"][0]["byte_length"] = byte_length
|
|
if byte_length <= 9_007_199_254_740_991:
|
|
value = with_scene_revision(value)
|
|
result = validate_scene_manifest(value)
|
|
valid, error, first = result.valid, None, _first_report_issue(result)
|
|
elif operation == "scene_compile_option":
|
|
value = deepcopy(valid_scene)
|
|
field, number = parameters["field"], parameters["value"]
|
|
value["compile_options"][field] = number
|
|
value["geometry_assets"][0]["tessellation"][field] = number
|
|
if field == "linear_tolerance":
|
|
value["edge_assets"][0]["tessellation"][field] = number
|
|
result = validate_scene_manifest(with_scene_revision(value))
|
|
valid, error, first = result.valid, None, _first_report_issue(result)
|
|
else:
|
|
raise AssertionError(operation)
|
|
except ValueError as exc:
|
|
valid, error, first = False, str(exc), None
|
|
|
|
assert valid == case["valid"], case["name"]
|
|
if "error" in case:
|
|
assert error == case["error"], case["name"]
|
|
if case.get("expected") is not None:
|
|
assert first == case["expected"], case["name"]
|
|
|
|
|
|
def test_shared_exact_byte_artifacts_and_jcs_vectors():
|
|
corpus = _load_shared_corpus()
|
|
for artifact in corpus["artifacts"].values():
|
|
payload = _decode_base64(artifact["base64"])
|
|
assert "sha256:" + hashlib.sha256(payload).hexdigest() == artifact["sha256"]
|
|
for vector in corpus["jcs_vectors"]:
|
|
payload = canonical_json_bytes(vector["input"])
|
|
assert payload == _decode_base64(vector["canonical_base64"])
|
|
assert "sha256:" + hashlib.sha256(payload).hexdigest() == vector["sha256"]
|
|
archive = preflight_zip_bytes(_decode_base64(corpus["artifacts"]["canonical_zip"]["base64"]))
|
|
assert canonical_zip_bytes(archive.members) == _decode_base64(
|
|
corpus["artifacts"]["canonical_zip"]["base64"]
|
|
)
|
|
|
|
|
|
def _apply_schema_mutations(base, mutations):
|
|
value = deepcopy(base)
|
|
for mutation in mutations:
|
|
current = value
|
|
for part in mutation["path"][:-1]:
|
|
current = current[part]
|
|
field = mutation["path"][-1]
|
|
if mutation["operation"] == "delete":
|
|
del current[field]
|
|
else:
|
|
current[field] = deepcopy(mutation["value"])
|
|
return value
|
|
|
|
|
|
def test_python_replays_all_closed_schema_record_field_matrices():
|
|
corpus = _load_shared_corpus()
|
|
schemas = {
|
|
artifact: parse_strict_json(load_contract_artifact(path))
|
|
for artifact, path in SCHEMA_FILES.items()
|
|
}
|
|
for matrix in corpus["schema_field_matrices"]:
|
|
root = schemas[matrix["artifact"]]
|
|
pointer = matrix["schema_pointer"]
|
|
schema = (
|
|
root
|
|
if pointer == "#"
|
|
else {
|
|
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
|
"$ref": pointer,
|
|
"$defs": root["$defs"],
|
|
}
|
|
)
|
|
validator = Draft202012Validator(schema)
|
|
for case in matrix["cases"]:
|
|
value = _apply_schema_mutations(matrix["base"], case["mutations"])
|
|
assert validator.is_valid(value) is case["valid"], (
|
|
matrix["artifact"],
|
|
pointer,
|
|
matrix["variant"],
|
|
case["name"],
|
|
)
|
|
|
|
|
|
def _valid_document_payloads():
|
|
corpus = _load_shared_corpus()
|
|
return (
|
|
("scene", SceneDocument, _decode_base64(corpus["artifacts"]["scene"]["base64"])),
|
|
("entities", EntityDocument, _decode_base64(corpus["artifacts"]["entities"]["base64"])),
|
|
(
|
|
"presentation",
|
|
PresentationDocument,
|
|
_decode_base64(next(case for case in corpus["presentation_cases"] if case["valid"])["payload_base64"]),
|
|
),
|
|
(
|
|
"connector_binding",
|
|
ConnectorBindingDocument,
|
|
_decode_base64(next(case for case in corpus["connector_binding_cases"] if case["valid"])["payload_base64"]),
|
|
),
|
|
(
|
|
"normalized_product",
|
|
NormalizedProductDocument,
|
|
_decode_base64(next(case for case in corpus["normalized_product_cases"] if case["valid"])["payload_base64"]),
|
|
),
|
|
)
|
|
|
|
|
|
def _assert_deeply_immutable(value):
|
|
if isinstance(value, Mapping):
|
|
assert isinstance(value, MappingProxyType)
|
|
for child in value.values():
|
|
_assert_deeply_immutable(child)
|
|
elif isinstance(value, tuple):
|
|
for child in value:
|
|
_assert_deeply_immutable(child)
|
|
else:
|
|
assert not isinstance(value, (dict, list))
|
|
|
|
|
|
def _nested_container(value):
|
|
for child in value.values():
|
|
if isinstance(child, (Mapping, tuple, dict, list)):
|
|
return child
|
|
raise AssertionError("fixture has no nested container")
|
|
|
|
|
|
def test_immutable_documents_construct_parse_validate_and_round_trip_canonically():
|
|
for name, document_type, payload in _valid_document_payloads():
|
|
mutable = parse_canonical_json(payload)
|
|
constructed = document_type.from_value(mutable)
|
|
parsed = document_type.parse(payload)
|
|
|
|
assert is_dataclass(constructed), name
|
|
assert constructed.canonical_bytes == payload, name
|
|
assert parsed.canonical_bytes == payload, name
|
|
assert constructed.canonical_hash == "sha256:" + hashlib.sha256(payload).hexdigest(), name
|
|
assert constructed == parsed, name
|
|
assert bytes(constructed) == payload, name
|
|
assert document_type.from_value(constructed.value) == constructed, name
|
|
assert document_type.from_value(constructed.to_mutable()) == constructed, name
|
|
|
|
|
|
def test_immutable_documents_resist_nested_mutation_and_isolate_mutable_exports():
|
|
for name, document_type, payload in _valid_document_payloads():
|
|
source = parse_canonical_json(payload)
|
|
document = document_type.from_value(source)
|
|
expected_bytes = document.canonical_bytes
|
|
|
|
_assert_deeply_immutable(document.value)
|
|
with pytest.raises(TypeError):
|
|
document.value["mutation"] = True
|
|
frozen_nested = _nested_container(document.value)
|
|
with pytest.raises((TypeError, AttributeError)):
|
|
if isinstance(frozen_nested, Mapping):
|
|
frozen_nested["mutation"] = True
|
|
else:
|
|
frozen_nested.append(None)
|
|
with pytest.raises(FrozenInstanceError):
|
|
document._canonical_bytes = b"mutated"
|
|
|
|
source["mutation"] = True
|
|
first_copy = document.to_mutable()
|
|
second_copy = document.to_mutable()
|
|
mutable_nested = _nested_container(first_copy)
|
|
if isinstance(mutable_nested, dict):
|
|
mutable_nested["mutation"] = True
|
|
else:
|
|
mutable_nested.append(None)
|
|
first_copy["mutation"] = True
|
|
|
|
assert "mutation" not in document.value, name
|
|
assert "mutation" not in second_copy, name
|
|
assert first_copy != second_copy, name
|
|
assert document.canonical_bytes == expected_bytes, name
|
|
|
|
|
|
def test_immutable_document_constructors_reject_invalid_values_and_stale_revisions():
|
|
for _name, document_type, _payload in _valid_document_payloads():
|
|
with pytest.raises(SceneContractError):
|
|
document_type.from_value({})
|
|
|
|
with pytest.raises(ValueError, match="canonical"):
|
|
SceneDocument.parse(b'{ "schema_version": "1.0" }')
|
|
|
|
corpus = _load_shared_corpus()
|
|
scene = parse_canonical_json(_decode_base64(corpus["artifacts"]["scene"]["base64"]))
|
|
scene["revision"] = "sha256:" + "0" * 64
|
|
with pytest.raises(SceneContractError) as error:
|
|
SceneDocument.from_value(scene)
|
|
assert error.value.report.first_error.code == "revision_mismatch"
|