feat: integrate SimpleCADAPI 2.0.2 CAD workflows

This commit is contained in:
Jerry
2026-08-03 11:17:05 +08:00
parent b5738e9109
commit c3a0f269b7
481 changed files with 110229 additions and 12826 deletions
@@ -0,0 +1,480 @@
"""Regenerate immutable cross-language Scene 1.0 contract vectors."""
from __future__ import annotations
import argparse
import json
import sys
from copy import deepcopy
from dataclasses import asdict
from pathlib import Path
from simplecadapi.scene import (
BASE_LIMITS,
canonical_json_bytes,
validate_connector_binding,
validate_entity_asset,
validate_normalized_product,
validate_presentation,
validate_scene_manifest,
validate_scene_package,
with_scene_revision,
)
from scene_contract_fixtures.authored_cases import (
build_connector_binding_matrix_cases,
build_normalized_product_matrix_cases,
build_presentation_matrix_cases,
)
from scene_contract_fixtures.base_scene import build_scene_package
from scene_contract_fixtures.binary_cases import build_binary_cases
from scene_contract_fixtures.common import b64, content_hash, first_issue, report_case
from scene_contract_fixtures.entity_cases import build_entity_matrix_cases
from scene_contract_fixtures.numeric_cases import build_numeric_vectors
from scene_contract_fixtures.package_cases import build_package_matrix_cases
from scene_contract_fixtures.resource_cases import build_resource_cases
from scene_contract_fixtures.schema_field_matrix import build_schema_field_matrices
from scene_contract_fixtures.scene_corpus import (
build_revision_vectors,
build_scene_field_cases,
build_scene_shape_cases,
)
ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "test" / "fixtures" / "scene-contract" / "corpus.json"
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--check", action="store_true", help="fail if output is stale")
args = parser.parse_args()
scene, scene_bytes, entity, entity_bytes, blobs = build_scene_package()
blob_pool = {uri: b64(payload) for uri, payload in blobs.items()}
scene_shape_cases, scene_packages = build_scene_shape_cases(
scene, entity, blobs, blob_pool
)
scene_field_cases = build_scene_field_cases(scene)
geometry_uri = next(uri for uri in blobs if uri.startswith("geometry/"))
edge_uri = next(uri for uri in blobs if uri.startswith("edges/"))
entity_uri = next(uri for uri in blobs if uri.startswith("entities/"))
triangle_glb = blobs[geometry_uri]
line_glb = blobs[edge_uri]
canonical_zip, deflate_zip, glb_cases, zip_cases = build_binary_cases(
triangle_glb,
line_glb,
{"scene.json": scene_bytes, **blobs},
)
missing_nodes = deepcopy(scene)
del missing_nodes["nodes"]
missing_nodes_bytes = canonical_json_bytes(missing_nodes)
empty_nodes = with_scene_revision({**scene, "nodes": []})
empty_nodes_bytes = canonical_json_bytes(empty_nodes)
revision_mismatch = deepcopy(scene)
revision_mismatch["revision"] = "sha256:" + "0" * 64
revision_mismatch_bytes = canonical_json_bytes(revision_mismatch)
transform_invalid = deepcopy(scene)
transform_invalid["nodes"][0]["transform"]["x_axis"] = [2, 0, 0]
transform_invalid = with_scene_revision(transform_invalid)
transform_invalid_bytes = canonical_json_bytes(transform_invalid)
nonunit_entity = deepcopy(entity)
next(item for item in nonunit_entity["entities"] if item["kind"] == "edge")[
"geometry"
]["direction"] = [2, 0, 0]
nonunit_entity_bytes = canonical_json_bytes(nonunit_entity)
unordered_entity = deepcopy(entity)
shared_edge = next(
item
for item in unordered_entity["entities"]
if item["kind"] == "edge" and len(item["parent_entity_ids"]) == 2
)
shared_edge["parent_entity_ids"].reverse()
unordered_entity_bytes = canonical_json_bytes(unordered_entity)
presentation = {
"appearances": [
{
"alpha_mode": "opaque",
"base_color": [0.2, 0.3, 0.4, 1],
"double_sided": False,
"edge_color": [0.05, 0.05, 0.05, 1],
"metallic": 0.1,
"name": "fixture",
"roughness": 0.7,
}
],
"cameras": [
{
"far": 10000,
"name": "overview",
"near": 1,
"parent_node_id": None,
"projection": "perspective",
"transform": {
"origin": [2000, -2000, 2000],
"x_axis": [1, 0, 0],
"y_axis": [0, 1, 0],
"z_axis": [0, 0, 1],
},
"vertical_fov_degrees": 45,
}
],
"node_overrides": [
{
"appearance_name": "fixture",
"node_id": "instance/root",
"visible": True,
}
],
"presentation_id": "fixture",
"schema_version": "1.0",
"source_scene_id": "fixture",
}
presentation_bytes = canonical_json_bytes(presentation)
missing_appearance = deepcopy(presentation)
missing_appearance["node_overrides"][0]["appearance_name"] = "missing"
missing_appearance_bytes = canonical_json_bytes(missing_appearance)
bad_camera = deepcopy(presentation)
bad_camera["cameras"][0]["far"] = 1
bad_camera_bytes = canonical_json_bytes(bad_camera)
connector_binding = {
"binding_id": "fixture",
"connector_id": "fixture_connector",
"name": None,
"owner_definition_id": "definition/root/part/fixture",
"schema_version": "1.0",
"selected_occurrence_node_id": "instance/root",
"source_model": {
"artifact_hash": "sha256:" + "1" * 64,
"graph_id": "fixture_graph",
"model_schema_version": "2.0",
},
"source_scene": {"revision": scene["revision"], "scene_id": "fixture"},
"target": {
"entity_asset_id": entity["geometry_asset_id"],
"entity_id": "entity/face/0",
"expected_source": {
"graph_id": "fixture_graph",
"kind": "model_output",
"node_id": "fixture_body",
"output_slot": 0,
},
"flip": True,
"kind": "topology_entity",
},
}
connector_binding_bytes = canonical_json_bytes(connector_binding)
wrong_binding_graph = deepcopy(connector_binding)
wrong_binding_graph["target"]["expected_source"]["graph_id"] = "other_graph"
wrong_binding_graph_bytes = canonical_json_bytes(wrong_binding_graph)
vertex_flip = deepcopy(connector_binding)
vertex_flip["target"]["entity_id"] = "entity/vertex/0"
vertex_flip_bytes = canonical_json_bytes(vertex_flip)
normalized_part = {
"body_source": {"kind": "manual", "source_id": "fixture_body"},
"connectors": [],
"kind": "part",
"material": None,
"metadata": {},
"name": None,
"part_id": "fixture_part",
}
normalized_part_bytes = canonical_json_bytes(normalized_part)
normalized_assembly = {
"assembly_id": "fixture_assembly",
"components": [
{
"component_id": "component_a",
"definition_ref": "definition/root/part/fixture",
"local_placement": {
"origin": [0, 0, 0],
"x_axis": [1, 0, 0],
"y_axis": [0, 1, 0],
"z_axis": [0, 0, 1],
},
"name": None,
}
],
"connectors": [],
"constraints": [],
"grounded_component_ids": ["component_a"],
"kind": "assembly",
"metadata": {},
"name": None,
}
normalized_assembly_bytes = canonical_json_bytes(normalized_assembly)
missing_grounded_component = deepcopy(normalized_assembly)
missing_grounded_component["grounded_component_ids"] = ["missing"]
missing_grounded_component_bytes = canonical_json_bytes(missing_grounded_component)
duplicate_components = deepcopy(normalized_assembly)
duplicate_components["components"].append(
deepcopy(duplicate_components["components"][0])
)
duplicate_components_bytes = canonical_json_bytes(duplicate_components)
bad_bounds = deepcopy(scene)
bad_bounds["geometry_assets"][0]["scene_local_bounds"]["max"][0] = 999
bad_bounds = with_scene_revision(bad_bounds)
bad_bounds_bytes = canonical_json_bytes(bad_bounds)
missing_blob = dict(blobs)
del missing_blob[edge_uri]
mutated_blobs = dict(blobs)
mutated_blobs[geometry_uri] = bytes([triangle_glb[0] ^ 1]) + triangle_glb[1:]
resource_cases = build_resource_cases(
scene,
normalized_part,
)
manifest_cases = [
report_case("valid_scene", scene_bytes, validate_scene_manifest(scene_bytes)),
report_case(
"missing_nodes",
missing_nodes_bytes,
validate_scene_manifest(missing_nodes_bytes),
),
report_case(
"empty_nodes",
empty_nodes_bytes,
validate_scene_manifest(empty_nodes_bytes),
),
report_case(
"revision_mismatch",
revision_mismatch_bytes,
validate_scene_manifest(revision_mismatch_bytes),
),
report_case(
"transform_invalid",
transform_invalid_bytes,
validate_scene_manifest(transform_invalid_bytes),
),
report_case(
"noncanonical_json",
scene_bytes + b"\n",
validate_scene_manifest(scene_bytes + b"\n"),
),
report_case(
"duplicate_json_key",
b'{"schema_version":"1.0","schema_version":"1.0"}',
validate_scene_manifest(
b'{"schema_version":"1.0","schema_version":"1.0"}'
),
),
report_case(
"bom",
b"\xef\xbb\xbf" + scene_bytes,
validate_scene_manifest(b"\xef\xbb\xbf" + scene_bytes),
),
]
entity_cases = [
report_case(
"valid_entities", entity_bytes, validate_entity_asset(entity_bytes)
),
report_case(
"nonunit_analytic_direction",
nonunit_entity_bytes,
validate_entity_asset(nonunit_entity_bytes),
),
report_case(
"unordered_adjacency",
unordered_entity_bytes,
validate_entity_asset(unordered_entity_bytes),
),
]
entity_cases.extend(build_entity_matrix_cases(entity))
presentation_cases = [
report_case(
"valid_presentation",
presentation_bytes,
validate_presentation(presentation_bytes),
),
report_case(
"missing_presentation_appearance",
missing_appearance_bytes,
validate_presentation(missing_appearance_bytes),
),
report_case(
"invalid_presentation_camera",
bad_camera_bytes,
validate_presentation(bad_camera_bytes),
),
]
presentation_cases.extend(build_presentation_matrix_cases(presentation))
connector_binding_cases = [
report_case(
"valid_connector_binding",
connector_binding_bytes,
validate_connector_binding(connector_binding_bytes),
),
report_case(
"connector_binding_graph_mismatch",
wrong_binding_graph_bytes,
validate_connector_binding(wrong_binding_graph_bytes),
),
report_case(
"connector_binding_vertex_flip",
vertex_flip_bytes,
validate_connector_binding(vertex_flip_bytes),
),
]
connector_binding_cases.extend(
build_connector_binding_matrix_cases(connector_binding)
)
normalized_product_cases = [
report_case(
"valid_normalized_part",
normalized_part_bytes,
validate_normalized_product(normalized_part_bytes),
),
report_case(
"valid_normalized_assembly",
normalized_assembly_bytes,
validate_normalized_product(normalized_assembly_bytes),
),
report_case(
"missing_grounded_component",
missing_grounded_component_bytes,
validate_normalized_product(missing_grounded_component_bytes),
),
report_case(
"duplicate_components",
duplicate_components_bytes,
validate_normalized_product(duplicate_components_bytes),
),
]
normalized_product_cases.extend(
build_normalized_product_matrix_cases(normalized_part, normalized_assembly)
)
package_cases = [
{
"blob_uris": sorted(blobs),
"expected": first_issue(validate_scene_package(scene_bytes, blobs)),
"manifest_base64": b64(scene_bytes),
"name": "valid_package",
"valid": True,
},
{
"blob_uris": sorted(blobs),
"expected": first_issue(validate_scene_package(bad_bounds_bytes, blobs)),
"manifest_base64": b64(bad_bounds_bytes),
"name": "glb_bounds_mismatch",
"valid": False,
},
{
"blob_uris": sorted(missing_blob),
"expected": first_issue(validate_scene_package(scene_bytes, missing_blob)),
"manifest_base64": b64(scene_bytes),
"name": "missing_blob",
"valid": False,
},
{
"blob_mutations": {geometry_uri: b64(mutated_blobs[geometry_uri])},
"blob_uris": sorted(mutated_blobs),
"expected": first_issue(validate_scene_package(scene_bytes, mutated_blobs)),
"manifest_base64": b64(scene_bytes),
"name": "blob_hash_mismatch",
"valid": False,
},
]
package_cases.extend(
build_package_matrix_cases(
scene, entity, blobs, scene_packages, blob_pool
)
)
jcs_input = {
"literals": [None, True, False],
"numbers": [333333333.33333329, 1e30, 4.50, 2e-3, 1e-27],
"string": "€$\u000f\nA'B\"\\\\\"/",
}
jcs_bytes = canonical_json_bytes(jcs_input)
corpus = {
"artifacts": {
"canonical_zip": {
"base64": b64(canonical_zip),
"sha256": content_hash(canonical_zip),
},
"deflate_zip": {
"base64": b64(deflate_zip),
"sha256": content_hash(deflate_zip),
},
"edge_glb": {
"base64": b64(line_glb),
"sha256": content_hash(line_glb),
"uri": edge_uri,
},
"entities": {
"base64": b64(entity_bytes),
"sha256": content_hash(entity_bytes),
"uri": entity_uri,
},
"geometry_glb": {
"base64": b64(triangle_glb),
"sha256": content_hash(triangle_glb),
"uri": geometry_uri,
},
"scene": {
"base64": b64(scene_bytes),
"revision": scene["revision"],
"sha256": content_hash(scene_bytes),
},
},
"blobs": dict(sorted(blob_pool.items())),
"entity_cases": entity_cases,
"connector_binding_cases": connector_binding_cases,
"format_version": "1.0",
"glb_cases": glb_cases,
"jcs_vectors": [
{
"canonical_base64": b64(jcs_bytes),
"input": jcs_input,
"name": "rfc8785_number_and_string_sample",
"sha256": content_hash(jcs_bytes),
}
],
"manifest_cases": manifest_cases,
"numeric_vectors": build_numeric_vectors(),
"normalized_product_cases": normalized_product_cases,
"package_cases": package_cases,
"presentation_cases": presentation_cases,
"resource_cases": resource_cases,
"resource_limits": asdict(BASE_LIMITS),
"revision_vectors": build_revision_vectors(scene_packages),
"scene_field_cases": scene_field_cases,
"scene_shape_cases": scene_shape_cases,
"schema_field_matrices": build_schema_field_matrices(
ROOT / "src/simplecadapi/scene/contracts/schemas"
),
"zip_cases": zip_cases,
}
for collection in (
manifest_cases,
entity_cases,
presentation_cases,
connector_binding_cases,
normalized_product_cases,
package_cases,
):
for case in collection:
assert case["valid"] == (case["expected"] is None), case["name"]
payload = (
json.dumps(corpus, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
).encode("utf-8")
relative_output = OUTPUT.relative_to(ROOT)
if args.check:
if not OUTPUT.exists() or OUTPUT.read_bytes() != payload:
print(f"stale generated fixture: {relative_output}", file=sys.stderr)
return 1
print(f"{relative_output} is current ({len(payload)} bytes)")
return 0
OUTPUT.write_bytes(payload)
print(f"wrote {relative_output} ({len(payload)} bytes)")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,267 @@
#!/usr/bin/env python3
"""Generate Python and TypeScript types from the five Scene contract schemas."""
from __future__ import annotations
import argparse
import json
import shutil
import subprocess
import sys
import tempfile
from importlib.metadata import version
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
SCHEMA_DIR = ROOT / "src/simplecadapi/scene/contracts/schemas"
PYTHON_OUTPUT = ROOT / "src/simplecadapi/scene/generated"
TYPESCRIPT_OUTPUT = ROOT / "scene-contract/src/generated"
PYTHON_GENERATOR_VERSION = "0.32.0"
TYPESCRIPT_GENERATOR_VERSION = "15.0.4"
SCHEMAS = (
("scene-1.0.schema.json", "scene", "SceneDocument"),
("entities-1.0.schema.json", "entities", "EntityDocument"),
("presentation-1.0.schema.json", "presentation", "PresentationDocument"),
(
"connector-binding-1.0.schema.json",
"connector_binding",
"ConnectorBindingDocument",
),
(
"normalized-product-1.schema.json",
"normalized_product",
"NormalizedProductDocument",
),
)
def _generator_schema(value: Any) -> Any:
"""Adapt 2020-12 vocabulary unsupported or weakened by the generators."""
if isinstance(value, list):
return [_generator_schema(item) for item in value]
if not isinstance(value, dict):
if isinstance(value, str) and value.startswith("#/$defs/"):
return "#/definitions/" + value.removeprefix("#/$defs/")
return value
result: dict[str, Any] = {}
for key, item in value.items():
if key in {"const", "prefixItems"}:
continue
if key == "$defs":
key = "definitions"
if key == "$schema":
item = "http://json-schema.org/draft-07/schema#"
if key == "items" and "prefixItems" in value:
continue
result[key] = _generator_schema(item)
if "const" in value:
constant = _generator_schema(value["const"])
if isinstance(constant, list):
result.update(
{
"type": "array",
"items": [{"enum": [item]} for item in constant],
"additionalItems": False,
"minItems": len(constant),
"maxItems": len(constant),
}
)
else:
result["enum"] = [constant]
if "prefixItems" in value:
result["items"] = _generator_schema(value["prefixItems"])
exact_length = value.get("maxItems") == len(value["prefixItems"])
result["additionalItems"] = (
False if exact_length else _generator_schema(value.get("items", True))
)
return result
def _run(command: list[str]) -> None:
subprocess.run(command, cwd=ROOT, check=True)
def _generate_python(schema_path: Path, output: Path, root_name: str) -> None:
_run(
[
sys.executable,
"-m",
"datamodel_code_generator",
"--input",
str(schema_path),
"--input-file-type",
"jsonschema",
"--output",
str(output),
"--output-model-type",
"typing.TypedDict",
"--target-python-version",
"3.10",
"--use-standard-collections",
"--use-union-operator",
"--enum-field-as-literal",
"all",
"--disable-timestamp",
"--use-double-quotes",
"--class-name",
root_name,
]
)
def _generate_typescript(schema_path: Path, output: Path) -> None:
executable = ROOT / "scene-contract/node_modules/.bin/json2ts"
if not executable.is_file():
raise RuntimeError("run `npm install` in scene-contract before generating types")
_run(
[
str(executable),
"--input",
str(schema_path),
"--output",
str(output),
"--no-enableConstEnums",
"--unreachableDefinitions",
"--maxItems=-1",
]
)
def _write_python_index(output: Path) -> None:
imports = []
names = []
for _, module_name, root_name in SCHEMAS:
imports.append(f"from .{module_name} import {root_name}\n")
names.append(f' "{root_name}",\n')
output.joinpath("__init__.py").write_text(
'"""Schema-generated Scene contract document types."""\n\n'
+ "".join(imports)
+ "\n__all__ = [\n"
+ "".join(names)
+ "]\n",
encoding="utf-8",
)
def _write_typescript_index(output: Path) -> None:
exports = [
f'export type {{ {root_name} }} from "./{module_name}.js";\n'
for _, module_name, root_name in SCHEMAS
]
output.joinpath("index.ts").write_text(
"// Generated by tools/generate_scene_contract_types.py. Do not edit.\n\n"
+ "".join(exports),
encoding="utf-8",
)
def _generate(destination: Path, language: str) -> tuple[Path | None, Path | None]:
schema_dir = destination / "schemas"
python_dir = destination / "python"
typescript_dir = destination / "typescript"
schema_dir.mkdir(parents=True)
if language in {"all", "python"}:
python_dir.mkdir()
if language in {"all", "typescript"}:
typescript_dir.mkdir()
for filename, module_name, root_name in SCHEMAS:
schema = json.loads(SCHEMA_DIR.joinpath(filename).read_text(encoding="utf-8"))
schema["title"] = root_name
adapted_path = schema_dir / filename
adapted_path.write_text(
json.dumps(_generator_schema(schema), indent=2, ensure_ascii=True) + "\n",
encoding="utf-8",
)
if language in {"all", "python"}:
_generate_python(adapted_path, python_dir / f"{module_name}.py", root_name)
if language in {"all", "typescript"}:
_generate_typescript(adapted_path, typescript_dir / f"{module_name}.ts")
if language in {"all", "python"}:
_write_python_index(python_dir)
if language in {"all", "typescript"}:
_write_typescript_index(typescript_dir)
return (
python_dir if language in {"all", "python"} else None,
typescript_dir if language in {"all", "typescript"} else None,
)
def _files(directory: Path) -> dict[str, bytes]:
if not directory.is_dir():
return {}
return {
path.relative_to(directory).as_posix(): path.read_bytes()
for path in sorted(directory.rglob("*"))
if path.is_file() and path.suffix != ".pyc" and "__pycache__" not in path.parts
}
def _sync(generated: Path, committed: Path, *, check: bool) -> bool:
generated_files = _files(generated)
committed_files = _files(committed)
if generated_files == committed_files:
return True
if check:
changed = sorted(generated_files.keys() ^ committed_files.keys())
changed.extend(
name
for name in sorted(generated_files.keys() & committed_files.keys())
if generated_files[name] != committed_files[name]
)
print(f"stale generated types in {committed.relative_to(ROOT)}:")
for name in changed:
print(f" {name}")
return False
if committed.exists():
shutil.rmtree(committed)
shutil.copytree(generated, committed)
return True
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--check", action="store_true", help="fail if output is stale")
parser.add_argument(
"--language",
choices=("all", "python", "typescript"),
default="all",
)
args = parser.parse_args()
if args.language in {"all", "python"}:
installed = version("datamodel-code-generator")
if installed != PYTHON_GENERATOR_VERSION:
raise RuntimeError(
f"expected datamodel-code-generator {PYTHON_GENERATOR_VERSION}, got {installed}"
)
if args.language in {"all", "typescript"}:
package_path = (
ROOT / "scene-contract/node_modules/json-schema-to-typescript/package.json"
)
if not package_path.is_file():
raise RuntimeError("run `npm install` in scene-contract before generating types")
installed = json.loads(package_path.read_text(encoding="utf-8"))["version"]
if installed != TYPESCRIPT_GENERATOR_VERSION:
raise RuntimeError(
f"expected json-schema-to-typescript {TYPESCRIPT_GENERATOR_VERSION}, "
f"got {installed}"
)
with tempfile.TemporaryDirectory(prefix="scene-contract-types-") as temporary:
python_dir, typescript_dir = _generate(Path(temporary), args.language)
current = True
if python_dir is not None:
current &= _sync(python_dir, PYTHON_OUTPUT, check=args.check)
if typescript_dir is not None:
current &= _sync(typescript_dir, TYPESCRIPT_OUTPUT, check=args.check)
return 0 if current else 1
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,383 @@
"""Run every current example model in an isolated process and export scenes.
This is intentionally a disposable harness for the example tagging migration.
It uses the model list from ``test_example_model_contract`` so the execution
set stays aligned with the repository's structural contract.
"""
from __future__ import annotations
import argparse
import ast
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
import importlib.util
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import time
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
EXAMPLES = ROOT / "examples"
DEFAULT_TIMEOUT_SECONDS = 900
DEFAULT_HEAVY_TIMEOUT_SECONDS = 1800
HEAVY_CASES = {
"examples/16_compact_two_stage_planetary_reducer/main.py",
"examples/20_integrated_bldc_joint_actuator/main.py",
}
def _is_model_decorator(node: ast.expr) -> bool:
if isinstance(node, ast.Call):
node = node.func
return (
isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id == "scad"
and node.attr == "model"
) or (isinstance(node, ast.Name) and node.id == "model")
def _builder_name(path: Path) -> str:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
model_functions = [
node.name
for node in tree.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and any(_is_model_decorator(decorator) for decorator in node.decorator_list)
]
if len(model_functions) != 1:
raise RuntimeError(f"expected one @model entry in {path}, got {model_functions}")
return model_functions[0]
def _model_cases() -> list[tuple[str, str]]:
contract_path = ROOT / "test" / "test_example_model_contract.py"
spec = importlib.util.spec_from_file_location(
"simplecadapi_example_model_contract", contract_path
)
if spec is None or spec.loader is None:
raise RuntimeError(f"could not load model contract: {contract_path}")
contract = importlib.util.module_from_spec(spec)
spec.loader.exec_module(contract)
cases = []
for path in contract._model_files():
relative_path = path.resolve().relative_to(ROOT).as_posix()
cases.append((relative_path, _builder_name(path)))
return cases
def _child_main(relative_path: str, builder_name: str, output_dir: Path) -> None:
path = (ROOT / relative_path).resolve()
sys.path.insert(0, str(path.parent))
spec = importlib.util.spec_from_file_location(
f"simplecadapi_scene_case_{os.getpid()}", path
)
if spec is None or spec.loader is None:
raise RuntimeError(f"could not load example module: {path}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
builder = getattr(module, builder_name)
result = builder()
exported = result.export_artifacts(output_dir=output_dir)
print(
"SCENE_RUN_RESULT="
+ json.dumps(
{
"graph_id": exported.session.graph.graph_id,
"artifact_paths": {
key: str(value) for key, value in exported.artifact_paths.items()
},
"result_node_count": len(exported.result_node_ids),
},
sort_keys=True,
),
flush=True,
)
def _kill_process_group(process: subprocess.Popen[str]) -> None:
try:
os.killpg(process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
def _run_case(
*,
relative_path: str,
builder_name: str,
run_dir: Path,
timeout_seconds: int,
heavy_timeout_seconds: int,
) -> dict[str, Any]:
case_name = relative_path.removesuffix(".py").replace("/", "__")
case_dir = run_dir / case_name
case_dir.mkdir(parents=True, exist_ok=True)
attempt = 1
while (case_dir / f"stdout.attempt-{attempt}.log").exists():
attempt += 1
for log_name in ("stdout.log", "stderr.log"):
log_path = case_dir / log_name
if log_path.exists():
log_path.replace(
case_dir / f"{log_path.stem}.attempt-{attempt}{log_path.suffix}"
)
timeout = (
heavy_timeout_seconds
if relative_path in HEAVY_CASES
else timeout_seconds
)
command = [
sys.executable,
str(Path(__file__).resolve()),
"--child",
relative_path,
builder_name,
str(case_dir),
]
environment = dict(os.environ)
environment["PYTHONHASHSEED"] = "0"
started_at = datetime.now(timezone.utc)
started = time.monotonic()
print(
f"[scene-export] starting {relative_path}::{builder_name} "
f"timeout={timeout}s",
flush=True,
)
stdout = ""
stderr = ""
returncode: int | None = None
timed_out = False
error: str | None = None
try:
process = subprocess.Popen(
command,
cwd=ROOT,
env=environment,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
errors="replace",
start_new_session=True,
)
try:
stdout, stderr = process.communicate(timeout=timeout)
returncode = process.returncode
except subprocess.TimeoutExpired as exc:
timed_out = True
_kill_process_group(process)
stdout, stderr = process.communicate()
if exc.stdout:
stdout = str(exc.stdout)
if exc.stderr:
stderr = str(exc.stderr)
except Exception as exc: # pragma: no cover - harness failure path
error = f"{exc.__class__.__name__}: {exc}"
elapsed = time.monotonic() - started
(case_dir / "stdout.log").write_text(stdout, encoding="utf-8")
(case_dir / "stderr.log").write_text(stderr, encoding="utf-8")
scene_paths = sorted(case_dir.glob("*.scene.zip"))
child_result: dict[str, Any] | None = None
for line in stdout.splitlines():
if line.startswith("SCENE_RUN_RESULT="):
try:
child_result = json.loads(line.removeprefix("SCENE_RUN_RESULT="))
except json.JSONDecodeError:
error = "invalid SCENE_RUN_RESULT payload"
status = "passed"
if timed_out:
status = "timeout"
elif error is not None or returncode != 0:
status = "failed"
elif len(scene_paths) != 1:
status = "failed"
error = f"expected one scene zip, found {len(scene_paths)}"
record = {
"case": relative_path,
"builder": builder_name,
"case_dir": str(case_dir.relative_to(ROOT)),
"status": status,
"returncode": returncode,
"timed_out": timed_out,
"timeout_seconds": timeout,
"elapsed_seconds": round(elapsed, 3),
"started_at": started_at.isoformat(),
"stdout_log": str((case_dir / "stdout.log").relative_to(ROOT)),
"stderr_log": str((case_dir / "stderr.log").relative_to(ROOT)),
"scene_paths": [str(path.relative_to(ROOT)) for path in scene_paths],
"child_result": child_result,
"error": error,
"attempt": attempt,
}
print(
f"[scene-export] {status} {relative_path} elapsed={elapsed:.1f}s",
flush=True,
)
return record
def _publish_scenes(
run_dir: Path,
records: list[dict[str, Any]],
) -> list[dict[str, Any]]:
scene_dir = run_dir / "scenes"
scene_dir.mkdir(parents=True, exist_ok=True)
for existing in scene_dir.glob("*.scene.zip"):
existing.unlink()
published: list[dict[str, Any]] = []
for record in records:
scene_paths = record.get("scene_paths", [])
if record.get("status") != "passed" or len(scene_paths) != 1:
continue
case_path = Path(record["case"])
case_name = (
case_path.parent.name if case_path.name == "main.py" else case_path.stem
)
source_path = ROOT / scene_paths[0]
destination = scene_dir / f"{case_name}.scene.zip"
shutil.copy2(source_path, destination)
published.append(
{
"case": record["case"],
"path": str(destination.relative_to(ROOT)),
}
)
return published
def _parent_main(args: argparse.Namespace) -> int:
all_cases = _model_cases()
previous_records: dict[str, dict[str, Any]] = {}
if args.resume is not None:
run_dir = args.resume.resolve()
report_path = run_dir / "execution_report.json"
previous_report = json.loads(report_path.read_text(encoding="utf-8"))
run_id = str(previous_report["run_id"])
started_at = datetime.fromisoformat(previous_report["started_at"])
previous_records = {
record["case"]: record for record in previous_report["cases"]
}
cases = [
case
for case in all_cases
if previous_records.get(case[0], {}).get("status") != "passed"
]
print(
f"[scene-export] resuming {run_id}; retrying {len(cases)} cases",
flush=True,
)
else:
cases = all_cases
run_id = datetime.now(timezone.utc).strftime("scene_export_%Y%m%dT%H%M%SZ")
run_dir = EXAMPLES / "out" / run_id
run_dir.mkdir(parents=True, exist_ok=False)
started_at = datetime.now(timezone.utc)
if args.case:
requested = set(args.case)
cases = [case for case in cases if case[0] in requested]
unknown = requested - {case[0] for case in all_cases}
if unknown:
raise ValueError(f"unknown example cases: {sorted(unknown)}")
if args.publish_only:
cases = []
records: list[dict[str, Any]] = []
with ThreadPoolExecutor(max_workers=args.workers) as executor:
futures = {
executor.submit(
_run_case,
relative_path=relative_path,
builder_name=builder_name,
run_dir=run_dir,
timeout_seconds=args.timeout,
heavy_timeout_seconds=args.heavy_timeout,
): (relative_path, builder_name)
for relative_path, builder_name in cases
}
for future in as_completed(futures):
relative_path, builder_name = futures[future]
try:
records.append(future.result())
except Exception as exc: # pragma: no cover - harness failure path
records.append(
{
"case": relative_path,
"builder": builder_name,
"status": "failed",
"error": f"runner worker {exc.__class__.__name__}: {exc}",
}
)
records_by_case = dict(previous_records)
records_by_case.update({record["case"]: record for record in records})
records = sorted(records_by_case.values(), key=lambda record: record["case"])
report = {
"run_id": run_id,
"started_at": started_at.isoformat(),
"finished_at": datetime.now(timezone.utc).isoformat(),
"root": str(ROOT),
"workers": args.workers,
"timeout_seconds": args.timeout,
"heavy_timeout_seconds": args.heavy_timeout,
"case_count": len(all_cases),
"passed": sum(record.get("status") == "passed" for record in records),
"failed": sum(record.get("status") == "failed" for record in records),
"timed_out": sum(record.get("status") == "timeout" for record in records),
"cases": records,
}
report_path = run_dir / "execution_report.json"
manifest_path = run_dir / "manifest.json"
report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
published_scenes = _publish_scenes(run_dir, records)
manifest = {
"run_id": run_id,
"case_count": len(all_cases),
"successful_scene_count": len(published_scenes),
"standalone_scene_packages": published_scenes,
}
manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
print(f"[scene-export] report={report_path}", flush=True)
print(f"[scene-export] scenes={run_dir / 'scenes'}", flush=True)
print(
f"[scene-export] passed={report['passed']} failed={report['failed']} "
f"timed_out={report['timed_out']}",
flush=True,
)
return 0 if report["failed"] == 0 and report["timed_out"] == 0 else 1
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--child", nargs=3, metavar=("PATH", "BUILDER", "OUTPUT"))
parser.add_argument("--workers", type=int, default=3)
parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SECONDS)
parser.add_argument("--resume", type=Path)
parser.add_argument("--case", action="append")
parser.add_argument("--publish-only", action="store_true")
parser.add_argument(
"--heavy-timeout", type=int, default=DEFAULT_HEAVY_TIMEOUT_SECONDS
)
args = parser.parse_args()
if args.child:
relative_path, builder_name, output = args.child
_child_main(relative_path, builder_name, Path(output).resolve())
return 0
if args.workers < 1:
parser.error("--workers must be at least 1")
if args.publish_only and args.resume is None:
parser.error("--publish-only requires --resume")
return _parent_main(args)
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1 @@
"""Builders for the immutable cross-language Scene 1.0 fixture corpus."""
@@ -0,0 +1,167 @@
"""Presentation, connector binding, and normalized product corpus cases."""
from __future__ import annotations
from copy import deepcopy
from simplecadapi.scene import (
validate_connector_binding,
validate_normalized_product,
validate_presentation,
)
from .common import JsonObject, validation_case
def build_presentation_matrix_cases(
presentation: JsonObject,
) -> list[dict[str, object]]:
orthographic = deepcopy(presentation)
orthographic["cameras"][0].pop("vertical_fov_degrees")
orthographic["cameras"][0].update(
{"projection": "orthographic", "vertical_span": 1000}
)
visible_only = deepcopy(presentation)
visible_only["node_overrides"][0].pop("appearance_name")
appearance_only = deepcopy(presentation)
appearance_only["node_overrides"][0].pop("visible")
neither = deepcopy(presentation)
neither["node_overrides"][0] = {"node_id": "instance/root"}
unknown_projection = deepcopy(presentation)
unknown_projection["cameras"][0]["projection"] = "unknown"
return [
validation_case("valid_orthographic_camera", orthographic, validate_presentation),
validation_case("valid_visible_only_override", visible_only, validate_presentation),
validation_case(
"valid_appearance_only_override", appearance_only, validate_presentation
),
validation_case("empty_node_override", neither, validate_presentation),
validation_case(
"unknown_camera_projection", unknown_projection, validate_presentation
),
]
def build_connector_binding_matrix_cases(
binding: JsonObject,
) -> list[dict[str, object]]:
cases: list[dict[str, object]] = []
for topology_kind, entity_id, flip in (
("FACE", "entity/face/0", True),
("EDGE", "entity/edge/0", True),
("VERTEX", "entity/vertex/0", False),
):
value = deepcopy(binding)
value["target"]["entity_id"] = entity_id
value["target"]["flip"] = flip
value["target"]["expected_source"] = {
"graph_id": "fixture_graph",
"kind": "model_topology",
"node_id": "fixture_body",
"output_slot": 0,
"topo_id": f"topo-{topology_kind.lower()}",
"topology_kind": topology_kind,
}
cases.append(
validation_case(
f"valid_binding_model_topology_{topology_kind.lower()}",
value,
validate_connector_binding,
)
)
named = deepcopy(binding)
named["name"] = "Fixture connector"
cases.append(validation_case("valid_binding_named", named, validate_connector_binding))
missing_name = deepcopy(binding)
del missing_name["name"]
cases.append(
validation_case("missing_binding_name", missing_name, validate_connector_binding)
)
unsupported_version = deepcopy(binding)
unsupported_version["source_model"]["model_schema_version"] = "3.0"
cases.append(
validation_case(
"unsupported_binding_model_schema_version",
unsupported_version,
validate_connector_binding,
)
)
return cases
def build_normalized_product_matrix_cases(
part: JsonObject, assembly: JsonObject
) -> list[dict[str, object]]:
material_part = deepcopy(part)
material_part["material"] = {
"color": [0.2, 0.3, 0.4],
"density": 7.85,
"density_unit": "g/cm3",
"material_id": "steel",
"metadata": {},
"name": "Steel",
}
model_part = deepcopy(part)
model_part["body_source"] = {
"graph_id": "graph",
"kind": "model_output",
"node_id": "body",
"output_slot": 0,
}
placement = deepcopy(part)
placement["connectors"] = [
{
"anchor": {
"anchor_kind": "placement",
"placement": {
"origin": [0, 0, 0],
"x_axis": [1, 0, 0],
"y_axis": [0, 1, 0],
"z_axis": [0, 0, 1],
},
},
"connector_id": "placement",
"name": None,
}
]
forwarded = deepcopy(assembly)
forwarded["connectors"] = [
{
"anchor": {
"anchor_kind": "forwarded",
"offset": None,
"source_component_id": "component_a",
"source_connector_id": "placement",
},
"connector_id": "forwarded",
"name": None,
}
]
forwarded_offset = deepcopy(forwarded)
forwarded_offset["connectors"][0]["anchor"]["offset"] = {
"origin": [0, 0, 5],
"x_axis": [1, 0, 0],
"y_axis": [0, 1, 0],
"z_axis": [0, 0, 1],
}
return [
validation_case(
"valid_complete_material", material_part, validate_normalized_product
),
validation_case(
"valid_model_body_source", model_part, validate_normalized_product
),
validation_case(
"valid_placement_connector", placement, validate_normalized_product
),
validation_case(
"valid_forwarded_connector_null_offset",
forwarded,
validate_normalized_product,
),
validation_case(
"valid_forwarded_connector_offset",
forwarded_offset,
validate_normalized_product,
),
]
@@ -0,0 +1,471 @@
"""Base Scene package and canonical tetrahedron asset construction."""
from __future__ import annotations
import hashlib
import math
import struct
from simplecadapi.scene import canonical_json_bytes, with_scene_revision
from .binary_cases import glb
from .common import content_hash, sha256_hex
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 _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 _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 = content_hash(triangle_glb)
edge_hash = content_hash(line_glb)
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 = content_hash(entity_bytes)
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/" + sha256_hex(
canonical_json_bytes(appearance_draft)
)
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
@@ -0,0 +1,703 @@
"""GLB and ZIP fixture construction, mutation, and case records."""
from __future__ import annotations
import binascii
import struct
import zlib
from copy import deepcopy
from typing import Callable
from simplecadapi.scene import (
canonical_json_bytes,
canonical_zip_bytes,
parse_strict_json,
preflight_glb,
preflight_zip_bytes,
)
from .common import JsonObject, b64
LOCAL = struct.Struct("<IHHHHHIIIHH")
CENTRAL = struct.Struct("<IHHHHHHIIIHHHHHII")
EOCD = struct.Struct("<IHHHHIIH")
def _f32(value: float) -> float:
result = struct.unpack("<f", struct.pack("<f", value))[0]
return 0.0 if result == 0 else result
def glb(
kind: str,
positions: list[tuple[float, float, float]],
indices: list[int],
normals: list[tuple[float, float, float]] | None = None,
) -> bytes:
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 replace_glb_json(data: bytes, mutate: Callable[[JsonObject], None]) -> bytes:
json_length = struct.unpack_from("<I", data, 12)[0]
document = parse_strict_json(data[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 = data[bin_header:]
return b"".join(
(
struct.pack(
"<III", 0x46546C67, 2, 12 + 8 + len(replacement) + len(bin_chunk)
),
struct.pack("<II", len(replacement), 0x4E4F534A),
replacement,
bin_chunk,
)
)
def _deflate_zip(members: dict[str, bytes]) -> bytes:
local_parts: list[bytes] = []
central_parts: list[bytes] = []
local_offset = 0
for name in sorted(members, key=lambda value: value.encode("ascii")):
name_bytes = name.encode("ascii")
payload = members[name]
compressor = zlib.compressobj(level=9, wbits=-zlib.MAX_WBITS)
compressed = compressor.compress(payload) + compressor.flush()
crc = binascii.crc32(payload) & 0xFFFFFFFF
local_header = LOCAL.pack(
0x04034B50,
20,
0x0800,
8,
0,
0x0021,
crc,
len(compressed),
len(payload),
len(name_bytes),
0,
)
local_parts.extend((local_header, name_bytes, compressed))
central_parts.extend(
(
CENTRAL.pack(
0x02014B50,
0x0314,
20,
0x0800,
8,
0,
0x0021,
crc,
len(compressed),
len(payload),
len(name_bytes),
0,
0,
0,
0,
0x81A40000,
local_offset,
),
name_bytes,
)
)
local_offset += len(local_header) + len(name_bytes) + len(compressed)
central = b"".join(central_parts)
return b"".join(local_parts) + central + EOCD.pack(
0x06054B50,
0,
0,
len(members),
len(members),
len(central),
local_offset,
0,
)
def _replace_u16(data: bytes, offset: int, value: int) -> bytes:
result = bytearray(data)
struct.pack_into("<H", result, offset, value)
return bytes(result)
def _replace_u32(data: bytes, offset: int, value: int) -> bytes:
result = bytearray(data)
struct.pack_into("<I", result, offset, value)
return bytes(result)
def _glb_bin_start(data: bytes) -> int:
return 28 + struct.unpack_from("<I", data, 12)[0]
def _replace_glb_bin_u32(data: bytes, relative_offset: int, value: int) -> bytes:
return _replace_u32(data, _glb_bin_start(data) + relative_offset, value)
def _set_path(value: object, path: tuple[object, ...], replacement: object) -> None:
current = value
for part in path[:-1]:
current = current[part] # type: ignore[index]
current[path[-1]] = replacement # type: ignore[index]
def _add_path(value: object, path: tuple[object, ...], key: str, item: object) -> None:
current = value
for part in path:
current = current[part] # type: ignore[index]
current[key] = item # type: ignore[index]
def _append_path(value: object, path: tuple[object, ...], item: object) -> None:
current = value
for part in path:
current = current[part] # type: ignore[index]
current.append(item) # type: ignore[attr-defined]
def _nonminimal_glb_json_padding(data: bytes) -> bytes:
json_length = struct.unpack_from("<I", data, 12)[0]
json_end = 20 + json_length
result = data[:json_end] + b" " + data[json_end:]
result = _replace_u32(result, 8, len(result))
return _replace_u32(result, 12, json_length + 4)
def _append_glb_bin_word(data: bytes, value: int = 0) -> bytes:
json_length = struct.unpack_from("<I", data, 12)[0]
bin_header = 20 + json_length
bin_length = struct.unpack_from("<I", data, bin_header)[0]
result = data + struct.pack("<I", value)
result = _replace_u32(result, 8, len(result))
return _replace_u32(result, bin_header, bin_length + 4)
def _f32_from_bits(value: int) -> float:
return struct.unpack("<f", struct.pack("<I", value))[0]
def _glb_case(
name: str,
payload: bytes,
expected_kind: str,
valid: bool,
) -> dict[str, object]:
try:
info = preflight_glb(payload, expected_kind=expected_kind)
error = None
actual_valid = True
actual_kind = info.kind
except ValueError as exc:
error = str(exc)
actual_valid = False
actual_kind = None
assert actual_valid == valid, name
return {
"error": error,
"expected_kind": expected_kind,
"kind": actual_kind,
"name": name,
"payload_base64": b64(payload),
"valid": valid,
}
def _zip_case(
name: str,
payload: bytes,
valid: bool,
used_deflate: bool | None,
) -> dict[str, object]:
try:
info = preflight_zip_bytes(payload)
error = None
actual_valid = True
actual_deflate = info.used_deflate
except ValueError as exc:
error = str(exc)
actual_valid = False
actual_deflate = None
assert actual_valid == valid, name
assert actual_deflate == used_deflate, name
return {
"error": error,
"name": name,
"payload_base64": b64(payload),
"used_deflate": used_deflate,
"valid": valid,
}
def _zip_layout(data: bytes) -> tuple[list[tuple[int, int, int]], int, int]:
eocd = len(data) - EOCD.size
count = struct.unpack_from("<H", data, eocd + 10)[0]
central = struct.unpack_from("<I", data, eocd + 16)[0]
entries: list[tuple[int, int, int]] = []
cursor = central
for _index in range(count):
name_length = struct.unpack_from("<H", data, cursor + 28)[0]
extra_length = struct.unpack_from("<H", data, cursor + 30)[0]
comment_length = struct.unpack_from("<H", data, cursor + 32)[0]
local_offset = struct.unpack_from("<I", data, cursor + 42)[0]
entries.append((cursor, local_offset, name_length))
cursor += CENTRAL.size + name_length + extra_length + comment_length
return entries, central, eocd
def _replace_zip_name(data: bytes, index: int, name: bytes) -> bytes:
entries, _central, _eocd = _zip_layout(data)
central_offset, local_offset, name_length = entries[index]
assert len(name) == name_length
assert struct.unpack_from("<H", data, local_offset + 26)[0] == name_length
result = bytearray(data)
result[local_offset + LOCAL.size : local_offset + LOCAL.size + name_length] = name
result[central_offset + CENTRAL.size : central_offset + CENTRAL.size + name_length] = name
return bytes(result)
def _zip_with_prefix(data: bytes) -> bytes:
entries, central, eocd = _zip_layout(data)
result = bytearray(b"x" + data)
for central_offset, local_offset, _name_length in entries:
struct.pack_into("<I", result, central_offset + 1 + 42, local_offset + 1)
struct.pack_into("<I", result, eocd + 1 + 16, central + 1)
return bytes(result)
def _zip_with_local_gap(data: bytes) -> bytes:
_entries, central, eocd = _zip_layout(data)
result = bytearray(data[:central] + b"x" + data[central:])
struct.pack_into("<I", result, eocd + 1 + 16, central + 1)
return bytes(result)
def build_binary_cases(
triangle_glb: bytes,
line_glb: bytes,
members: dict[str, bytes],
) -> tuple[bytes, bytes, list[dict[str, object]], list[dict[str, object]]]:
canonical_zip = canonical_zip_bytes(members)
deflate_zip = _deflate_zip(members)
string_bounds_glb = replace_glb_json(
triangle_glb,
lambda document: document["accessors"][0]["min"].__setitem__(0, "0"),
)
boolean_scene_glb = replace_glb_json(
triangle_glb, lambda document: document.__setitem__("scene", False)
)
simple_triangle = glb(
"triangle",
[(0, 0, 0), (1, 0, 0), (0, 1, 0)],
[0, 1, 2],
[(0, 0, 1)] * 3,
)
float_extreme_line = glb(
"line",
[
(
_f32_from_bits(0xFF7FFFFF),
_f32_from_bits(0x80000001),
_f32_from_bits(0x007FFFFF),
),
(
_f32_from_bits(0x80800000),
0.0,
_f32_from_bits(0x00000001),
),
(
_f32_from_bits(0x00800000),
_f32_from_bits(0x007FFFFF),
_f32_from_bits(0x7F7FFFFF),
),
],
[0, 1, 1, 2],
)
glb_payloads: list[tuple[str, bytes, str, bool]] = [
("valid_triangle", triangle_glb, "triangle", True),
("valid_line", line_glb, "line", True),
("valid_float32_finite_extremes", float_extreme_line, "line", True),
("string_accessor_bounds", string_bounds_glb, "triangle", False),
("boolean_scene_index", boolean_scene_glb, "triangle", False),
]
for name, base, path, key, value in (
("root_extensions_used", triangle_glb, (), "extensionsUsed", []),
("asset_extras", triangle_glb, ("asset",), "extras", {}),
("node_name", triangle_glb, ("nodes", 0), "name", "node"),
("scene_name", triangle_glb, ("scenes", 0), "name", "scene"),
("mesh_name", triangle_glb, ("meshes", 0), "name", "mesh"),
(
"primitive_material",
triangle_glb,
("meshes", 0, "primitives", 0),
"material",
0,
),
("buffer_uri", triangle_glb, ("buffers", 0), "uri", "buffer.bin"),
("buffer_view_byte_stride", triangle_glb, ("bufferViews", 0), "byteStride", 12),
("position_accessor_byte_offset", triangle_glb, ("accessors", 0), "byteOffset", 0),
("normal_accessor_normalized", triangle_glb, ("accessors", 1), "normalized", False),
("index_accessor_sparse", triangle_glb, ("accessors", 2), "sparse", {}),
):
glb_payloads.append(
(
name,
replace_glb_json(
base,
lambda document, path=path, key=key, value=value: _add_path(
document, path, key, value
),
),
"triangle",
False,
)
)
for name, base, path, value, kind in (
("empty_position_accessor", line_glb, ("accessors", 0, "count"), 0, "line"),
("empty_index_accessor", line_glb, ("accessors", 1, "count"), 0, "line"),
(
"wrong_triangle_mode",
triangle_glb,
("meshes", 0, "primitives", 0, "mode"),
1,
"triangle",
),
(
"wrong_position_component_type",
triangle_glb,
("accessors", 0, "componentType"),
5125,
"triangle",
),
("wrong_normal_count", triangle_glb, ("accessors", 1, "count"), 1, "triangle"),
(
"wrong_index_component_type",
triangle_glb,
("accessors", 2, "componentType"),
5125,
"triangle",
),
("buffer_view_gap", triangle_glb, ("bufferViews", 1, "byteOffset"), 4, "triangle"),
(
"buffer_view_wrong_target",
triangle_glb,
("bufferViews", 0, "target"),
34963,
"triangle",
),
(
"buffer_view_wrong_length",
triangle_glb,
("bufferViews", 0, "byteLength"),
4,
"triangle",
),
):
glb_payloads.append(
(
name,
replace_glb_json(
base,
lambda document, path=path, value=value: _set_path(
document, path, value
),
),
kind,
False,
)
)
glb_payloads.extend(
(
("bad_glb_magic", _replace_u32(triangle_glb, 0, 0), "triangle", False),
("bad_glb_version", _replace_u32(triangle_glb, 4, 1), "triangle", False),
(
"bad_glb_total_length",
_replace_u32(triangle_glb, 8, len(triangle_glb) + 4),
"triangle",
False,
),
("bad_json_chunk_type", _replace_u32(triangle_glb, 16, 0), "triangle", False),
(
"nonminimal_json_padding",
_nonminimal_glb_json_padding(triangle_glb),
"triangle",
False,
),
(
"bad_bin_chunk_type",
_replace_u32(
triangle_glb,
24 + struct.unpack_from("<I", triangle_glb, 12)[0],
0,
),
"triangle",
False,
),
(
"extra_accessor",
replace_glb_json(
triangle_glb,
lambda document: _append_path(
document,
("accessors",),
deepcopy(document["accessors"][0]),
),
),
"triangle",
False,
),
(
"buffer_unpadded_length",
replace_glb_json(
triangle_glb,
lambda document: _set_path(
document,
("buffers", 0, "byteLength"),
document["buffers"][0]["byteLength"] + 4,
),
),
"triangle",
False,
),
("bin_extra_zero_word", _append_glb_bin_word(triangle_glb), "triangle", False),
(
"bin_nonzero_padding",
simple_triangle[:-1] + b"\x01",
"triangle",
False,
),
(
"position_negative_zero",
_replace_glb_bin_u32(simple_triangle, 0, 0x80000000),
"triangle",
False,
),
(
"position_positive_infinity",
_replace_glb_bin_u32(simple_triangle, 0, 0x7F800000),
"triangle",
False,
),
(
"position_negative_infinity",
_replace_glb_bin_u32(simple_triangle, 0, 0xFF800000),
"triangle",
False,
),
(
"position_quiet_nan",
_replace_glb_bin_u32(simple_triangle, 0, 0x7FC00000),
"triangle",
False,
),
(
"normal_negative_zero",
_replace_glb_bin_u32(simple_triangle, 36, 0x80000000),
"triangle",
False,
),
(
"normal_nonfinite",
_replace_glb_bin_u32(simple_triangle, 44, 0x7FC00000),
"triangle",
False,
),
(
"normal_norm_below_lower_limit",
_replace_glb_bin_u32(simple_triangle, 44, 0x3F7FFFEF),
"triangle",
False,
),
(
"normal_norm_above_upper_limit",
_replace_glb_bin_u32(simple_triangle, 44, 0x3F800009),
"triangle",
False,
),
)
)
valid_normal_boundaries = _replace_glb_bin_u32(simple_triangle, 44, 0x3F7FFFF0)
valid_normal_boundaries = _replace_glb_bin_u32(valid_normal_boundaries, 56, 0x3F800008)
glb_payloads.append(
("valid_normal_norm_inner_neighbors", valid_normal_boundaries, "triangle", True)
)
tiny_zip = canonical_zip_bytes({"other.json": b"x", "scene.json": b"{}"})
tiny_entries, tiny_central, tiny_eocd = _zip_layout(tiny_zip)
first_central, first_local, _first_name_length = tiny_entries[0]
first_payload = first_local + LOCAL.size + struct.unpack_from("<H", tiny_zip, first_local + 26)[0]
crc_zip = bytearray(tiny_zip)
crc_zip[first_payload] ^= 1
exact_ratio_zip = _deflate_zip({"scene.json": b"a" * 1200})
over_ratio_zip = _deflate_zip({"scene.json": b"a" * 1201})
zip_payloads: list[tuple[str, bytes, bool, bool | None]] = [
("canonical_stored", canonical_zip, True, False),
("allowlisted_deflate", deflate_zip, True, True),
("exact_member_compression_ratio", exact_ratio_zip, True, True),
("over_member_compression_ratio", over_ratio_zip, False, None),
("crc_mismatch", bytes(crc_zip), False, None),
("trailing_data", tiny_zip + b"x", False, None),
("leading_prefix", _zip_with_prefix(tiny_zip), False, None),
("local_record_gap", _zip_with_local_gap(tiny_zip), False, None),
("case_collision", _replace_zip_name(tiny_zip, 0, b"SCENE.JSON"), False, None),
("path_traversal", _replace_zip_name(tiny_zip, 0, b"a/../b.txt"), False, None),
("backslash_name", _replace_zip_name(tiny_zip, 0, b"other\\json"), False, None),
("nul_name", _replace_zip_name(tiny_zip, 0, b"other.\0son"), False, None),
("bad_eocd_signature", _replace_u32(tiny_zip, tiny_eocd, 0), False, None),
("eocd_comment", _replace_u16(tiny_zip, tiny_eocd + 20, 1), False, None),
("multi_disk", _replace_u16(tiny_zip, tiny_eocd + 4, 1), False, None),
(
"member_count_over_limit",
_replace_u16(
_replace_u16(tiny_zip, tiny_eocd + 8, 50_001),
tiny_eocd + 10,
50_001,
),
False,
None,
),
(
"central_offset_mismatch",
_replace_u32(tiny_zip, tiny_eocd + 16, tiny_central + 1),
False,
None,
),
("bad_central_signature", _replace_u32(tiny_zip, first_central, 0), False, None),
("central_made_by", _replace_u16(tiny_zip, first_central + 4, 20), False, None),
("central_needed", _replace_u16(tiny_zip, first_central + 6, 45), False, None),
("central_flags", _replace_u16(tiny_zip, first_central + 8, 0x0808), False, None),
("central_method", _replace_u16(tiny_zip, first_central + 10, 99), False, None),
("central_time", _replace_u16(tiny_zip, first_central + 12, 1), False, None),
("central_date", _replace_u16(tiny_zip, first_central + 14, 0), False, None),
("central_extra", _replace_u16(tiny_zip, first_central + 30, 1), False, None),
("central_comment", _replace_u16(tiny_zip, first_central + 32, 1), False, None),
("central_disk_start", _replace_u16(tiny_zip, first_central + 34, 1), False, None),
("central_internal_attr", _replace_u16(tiny_zip, first_central + 36, 1), False, None),
("central_external_attr", _replace_u32(tiny_zip, first_central + 38, 0), False, None),
("central_crc", _replace_u32(tiny_zip, first_central + 16, 0), False, None),
(
"central_compressed_size",
_replace_u32(tiny_zip, first_central + 20, 2),
False,
None,
),
(
"central_uncompressed_size",
_replace_u32(tiny_zip, first_central + 24, 2),
False,
None,
),
("central_local_offset", _replace_u32(tiny_zip, first_central + 42, 1), False, None),
("local_signature", _replace_u32(tiny_zip, first_local, 0), False, None),
("local_needed", _replace_u16(tiny_zip, first_local + 4, 45), False, None),
("local_flags", _replace_u16(tiny_zip, first_local + 6, 0), False, None),
("local_method", _replace_u16(tiny_zip, first_local + 8, 8), False, None),
("local_time", _replace_u16(tiny_zip, first_local + 10, 1), False, None),
("local_date", _replace_u16(tiny_zip, first_local + 12, 0), False, None),
("local_crc", _replace_u32(tiny_zip, first_local + 14, 0), False, None),
("local_compressed_size", _replace_u32(tiny_zip, first_local + 18, 2), False, None),
("local_uncompressed_size", _replace_u32(tiny_zip, first_local + 22, 2), False, None),
("local_extra", _replace_u16(tiny_zip, first_local + 28, 1), False, None),
]
return (
canonical_zip,
deflate_zip,
[_glb_case(*case) for case in glb_payloads],
[_zip_case(*case) for case in zip_payloads],
)
@@ -0,0 +1,145 @@
"""Shared fixture construction and package-record helpers."""
from __future__ import annotations
import base64
import hashlib
from copy import deepcopy
from typing import Any, Callable
from simplecadapi.scene import (
canonical_json_bytes,
validate_scene_package,
)
JsonObject = dict[str, Any]
ScenePackage = tuple[JsonObject, JsonObject, dict[str, bytes]]
Validator = Callable[[Any], Any]
IDENTITY = {
"origin": [0, 0, 0],
"x_axis": [1, 0, 0],
"y_axis": [0, 1, 0],
"z_axis": [0, 0, 1],
}
def b64(value: bytes) -> str:
return base64.b64encode(value).decode("ascii")
def sha256_hex(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def content_hash(value: bytes) -> str:
return "sha256:" + sha256_hex(value)
def first_issue(report: Any) -> dict[str, str] | None:
issue = report.first_error
return None if issue is None else {"code": issue.code, "path": issue.path}
def report_case(name: str, payload: bytes, report: Any) -> dict[str, object]:
return {
"expected": first_issue(report),
"name": name,
"payload_base64": b64(payload),
"valid": report.valid,
}
def transform(origin: list[float] | None = None) -> JsonObject:
value = deepcopy(IDENTITY)
if origin is not None:
value["origin"] = origin
return value
def validation_case(
name: str, value: Any, validator: Validator
) -> dict[str, object]:
payload = value if isinstance(value, bytes) else canonical_json_bytes(value)
return report_case(name, payload, validator(payload))
def replace_entity_sidecar(
scene: JsonObject,
entity: JsonObject,
blobs: dict[str, bytes],
) -> ScenePackage:
entity_bytes = canonical_json_bytes(entity)
entity_hash = content_hash(entity_bytes)
entity_uri = f"entities/sha256-{entity_hash.removeprefix('sha256:')}.json"
entity_record = scene["entity_assets"][0]
old_uri = entity_record["uri"]
entity_record.update(
{
"byte_length": len(entity_bytes),
"content_hash": entity_hash,
"entity_asset_id": entity_hash,
"uri": entity_uri,
}
)
for definition in scene["definitions"]:
if definition["kind"] in {"part", "shape"}:
definition["entity_asset_id"] = entity_hash
updated_blobs = {uri: payload for uri, payload in blobs.items() if uri != old_uri}
updated_blobs[entity_uri] = entity_bytes
return scene, entity, updated_blobs
def set_product_material_appearance(
scene: JsonObject, material_id: str = "fixture_material"
) -> None:
appearance = scene["appearances"][0]
appearance["name"] = "Fixture material"
appearance["source"] = {
"kind": "product_material",
"material_id": material_id,
"root_id": "root",
}
draft = dict(appearance)
draft.pop("appearance_id")
appearance_id = "appearance/evaluated/" + sha256_hex(canonical_json_bytes(draft))
appearance["appearance_id"] = appearance_id
for definition in scene["definitions"]:
if definition["kind"] in {"part", "shape"}:
definition["appearance_id"] = appearance_id
def inline_package_case(
name: str,
scene: JsonObject,
blobs: dict[str, bytes],
blob_pool: dict[str, str],
) -> dict[str, object]:
scene_bytes = canonical_json_bytes(scene)
report = validate_scene_package(scene_bytes, blobs)
for uri, payload in blobs.items():
encoded = b64(payload)
previous = blob_pool.setdefault(uri, encoded)
if previous != encoded:
raise AssertionError(f"one fixture URI resolved to different bytes: {uri}")
return {
"blob_uris": sorted(blobs),
"expected": first_issue(report),
"manifest_base64": b64(scene_bytes),
"name": name,
"valid": report.valid,
}
def scene_shape_case(
name: str,
scene: JsonObject,
blobs: dict[str, bytes],
expected: dict[str, object],
blob_pool: dict[str, str],
) -> dict[str, object]:
package = inline_package_case(name, scene, blobs, blob_pool)
if not package["valid"]:
raise AssertionError(f"invalid scene shape fixture {name}: {package['expected']}")
return {**package, "expected_shape": expected}
@@ -0,0 +1,103 @@
"""Entity provenance, topology adjacency, and range corpus cases."""
from __future__ import annotations
from copy import deepcopy
from simplecadapi.scene import validate_entity_asset
from .common import JsonObject, validation_case
def _entity_with_source(entity: JsonObject, source_kind: str) -> JsonObject:
result = deepcopy(entity)
topology_kind = {
"solid": "SOLID",
"face": "FACE",
"edge": "EDGE",
"vertex": "VERTEX",
}
for record in result["entities"]:
if source_kind == "model_output":
record["source"] = {
"graph_id": "graph",
"kind": "model_output",
"node_id": "body",
"output_slot": 0,
}
elif source_kind == "model_topology":
record["source"] = {
"graph_id": "graph",
"kind": "model_topology",
"node_id": "body",
"output_slot": 0,
"topo_id": record["entity_id"],
"topology_kind": topology_kind[record["kind"]],
}
else:
record["source"] = {
"kind": "imported_primitive",
"source_element_id": record["entity_id"],
}
return result
def build_entity_matrix_cases(entity: JsonObject) -> list[dict[str, object]]:
cases = [
validation_case(
f"valid_entity_source_{source_kind}",
_entity_with_source(entity, source_kind),
validate_entity_asset,
)
for source_kind in ("model_output", "model_topology", "imported_primitive")
]
topology_mismatch = _entity_with_source(entity, "model_topology")
face = next(
record for record in topology_mismatch["entities"] if record["kind"] == "face"
)
face["source"]["topology_kind"] = "EDGE"
cases.append(
validation_case(
"model_topology_kind_mismatch", topology_mismatch, validate_entity_asset
)
)
missing_reciprocal = deepcopy(entity)
face = next(
record for record in missing_reciprocal["entities"] if record["kind"] == "face"
)
edge_id = face["child_entity_ids"][0]
edge = next(
record
for record in missing_reciprocal["entities"]
if record["entity_id"] == edge_id
)
edge["parent_entity_ids"].remove(face["entity_id"])
cases.append(
validation_case(
"missing_reciprocal_adjacency", missing_reciprocal, validate_entity_asset
)
)
range_gap = deepcopy(entity)
range_gap["face_groups"][1]["first_index"] += 3
cases.append(
validation_case("face_group_range_gap", range_gap, validate_entity_asset)
)
wrong_cardinality = deepcopy(entity)
wrong_cardinality["edge_groups"][0]["index_count"] = 3
cases.append(
validation_case(
"edge_group_wrong_cardinality", wrong_cardinality, validate_entity_asset
)
)
unsupported_version = deepcopy(entity)
unsupported_version["schema_version"] = "2.0"
cases.append(
validation_case(
"unsupported_entity_schema_version",
unsupported_version,
validate_entity_asset,
)
)
return cases
@@ -0,0 +1,215 @@
"""Declarative required, nullable, and discriminated-union mutations."""
from __future__ import annotations
from copy import deepcopy
from typing import Any, Iterable
from simplecadapi.scene import canonical_json_bytes, with_scene_revision
from .common import sha256_hex
Path = tuple[str | int, ...]
def _case(
name: str,
operation: str,
path: Path,
value: Any = None,
*,
recompute_revision: bool = False,
) -> dict[str, Any]:
mutation = {"operation": operation, "path": list(path)}
if operation != "delete":
mutation["value"] = value
return {
"mutations": [mutation],
"name": name,
"recompute_revision": recompute_revision,
}
def _missing_fields(path: Path, fields: Iterable[str]) -> Iterable[dict[str, Any]]:
for field in fields:
yield _case(
"missing_" + "_".join((*map(str, path), field)),
"delete",
(*path, field),
)
def scene_field_matrix(scene: dict[str, Any]) -> list[dict[str, Any]]:
"""Return compact mutations covering the frozen structural field matrix."""
records: list[tuple[Path, tuple[str, ...]]] = [
(
(),
(
"schema_version", "extensions_used", "extensions_required",
"extensions", "scene_id", "revision", "generator", "source",
"coordinate_system", "compile_options", "definitions", "nodes",
"geometry_assets", "edge_assets", "appearances", "entity_assets",
"connectors", "cameras", "lights", "annotations", "diagnostics",
),
),
(
("compile_options",),
("linear_tolerance", "angular_tolerance", "embed_source", "embed_presentation"),
),
(
("definitions", 0),
("definition_id", "kind", "name", "source", "sdk_metadata"),
),
(
("nodes", 0),
(
"node_id", "parent_node_id", "order", "definition_id", "name",
"transform", "visible", "selectable", "appearance_override_id",
"source", "sdk_metadata",
),
),
(
("appearances", 0),
(
"appearance_id", "name", "source", "base_color", "metallic",
"roughness", "alpha_mode", "double_sided", "edge_color", "sdk_metadata",
),
),
(
("geometry_assets", 0),
(
"asset_id", "uri", "media_type", "byte_length", "content_hash",
"scene_local_bounds", "asset_to_scene", "tessellation",
),
),
(
("entity_assets", 0),
("entity_asset_id", "uri", "media_type", "byte_length", "content_hash"),
),
]
result = [
mutation
for path, fields in records
for mutation in _missing_fields(path, fields)
]
result.extend(
_case(name, "add", path, True)
for name, path in (
("unknown_scene_field", ("unknown",)),
("unknown_generator_field", ("generator", "unknown")),
("unknown_compile_options_field", ("compile_options", "unknown")),
("unknown_definition_field", ("definitions", 0, "unknown")),
("unknown_node_field", ("nodes", 0, "unknown")),
("unknown_appearance_field", ("appearances", 0, "unknown")),
)
)
result.extend(
(
_case("optional_presentation_source_null", "add", ("presentation_source",), None),
_case("unknown_scene_source_kind", "set", ("source", "kind"), "unknown"),
_case(
"unknown_definition_source_kind",
"set",
("definitions", 0, "source", "kind"),
"unknown",
),
_case("missing_node_source_kind", "delete", ("nodes", 0, "source", "kind")),
_case("unsupported_scene_schema_version", "set", ("schema_version",), "2.0"),
_case(
"unsupported_generator_profile",
"set",
("generator", "profile"),
"scene-2.0",
),
_case(
"no_presentation_visibility_override",
"set",
("nodes", 0, "visible"),
False,
recompute_revision=True,
),
_case(
"no_presentation_selectability_override",
"set",
("nodes", 0, "selectable"),
False,
recompute_revision=True,
),
_case(
"no_presentation_appearance_override",
"set",
("nodes", 0, "appearance_override_id"),
scene["appearances"][0]["appearance_id"],
recompute_revision=True,
),
_case(
"no_presentation_camera",
"set",
("cameras",),
[
{
"camera_id": "camera/fixture/overview",
"far": 1000,
"name": "overview",
"near": 1,
"parent_node_id": None,
"projection": "perspective",
"transform": {
"origin": [10, -10, 10],
"x_axis": [1, 0, 0],
"y_axis": [0, 1, 0],
"z_axis": [0, 0, 1],
},
"vertical_fov_degrees": 45,
}
],
recompute_revision=True,
),
)
)
return result
def apply_scene_field_case(scene: dict[str, Any], case: dict[str, Any]) -> dict[str, Any]:
"""Replay one mutation descriptor to produce its validator input."""
result = deepcopy(scene)
for mutation in case["mutations"]:
path = mutation["path"]
current: Any = result
for part in path[:-1]:
current = current[part]
if mutation["operation"] == "delete":
del current[path[-1]]
else:
current[path[-1]] = deepcopy(mutation.get("value"))
if case.get("recompute_revision"):
result = with_scene_revision(result)
return result
def nullable_fields_nonnull_case(scene: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
"""Return the multi-field valid nullable vector and its compact descriptor."""
value = deepcopy(scene)
value["nodes"][0]["name"] = "Named occurrence"
value["appearances"][0]["name"] = "Named appearance"
appearance = value["appearances"][0]
draft = dict(appearance)
draft.pop("appearance_id")
appearance_id = "appearance/evaluated/" + sha256_hex(canonical_json_bytes(draft))
appearance["appearance_id"] = appearance_id
value["definitions"][0]["appearance_id"] = appearance_id
descriptor = {
"mutations": [
{"operation": "set", "path": ["nodes", 0, "name"], "value": "Named occurrence"},
{"operation": "set", "path": ["appearances", 0, "name"], "value": "Named appearance"},
{"operation": "set", "path": ["appearances", 0, "appearance_id"], "value": appearance_id},
{"operation": "set", "path": ["definitions", 0, "appearance_id"], "value": appearance_id},
],
"name": "nullable_fields_nonnull",
"recompute_revision": True,
}
return with_scene_revision(value), descriptor
@@ -0,0 +1,128 @@
"""Numeric profile fixture vector construction."""
from __future__ import annotations
import struct
from simplecadapi.scene import (
profile_cross,
profile_f32_bits,
profile_normalize,
)
def _f64_from_bits(value: str) -> float:
return struct.unpack(">d", bytes.fromhex(value))[0]
def _f64_bits(value: float) -> str:
return struct.pack(">d", value).hex()
def build_numeric_vectors() -> list[dict[str, object]]:
vectors: list[dict[str, object]] = []
for name, input_bits in (
("positive_zero", "0000000000000000"),
("negative_zero", "8000000000000000"),
("smallest_binary64_subnormal", "0000000000000001"),
("negative_smallest_binary64_subnormal", "8000000000000001"),
("binary32_smallest_subnormal", "36a0000000000000"),
("halfway_tie_to_even", "3ff0000010000000"),
("just_above_halfway", "3ff0000010000001"),
("binary32_largest_finite", "47efffffe0000000"),
):
expected = profile_f32_bits(_f64_from_bits(input_bits))
vectors.append(
{
"expected_bits": f"{expected:08x}",
"input_bits": [input_bits],
"name": name,
"operation": "f32",
"valid": True,
}
)
for name, input_bits in (
("positive_overflow", "7fefffffffffffff"),
("negative_overflow", "ffefffffffffffff"),
("positive_infinity", "7ff0000000000000"),
("quiet_nan", "7ff8000000000000"),
):
try:
profile_f32_bits(_f64_from_bits(input_bits))
except ValueError as exc:
error = str(exc)
else:
raise AssertionError(name)
vectors.append(
{
"error": error,
"input_bits": [input_bits],
"name": name,
"operation": "f32",
"valid": False,
}
)
cross_inputs = (
("basis_cross", (1.0, 0.0, 0.0), (0.0, 1.0, 0.0)),
(
"cancellation_cross",
(134217729.0, 134217728.0, 0.0),
(134217728.0, 134217727.0, 0.0),
),
)
for name, left, right in cross_inputs:
result = profile_cross(left, right)
vectors.append(
{
"expected_bits": [_f64_bits(component) for component in result],
"input_bits": [
[_f64_bits(component) for component in left],
[_f64_bits(component) for component in right],
],
"name": name,
"operation": "cross",
"valid": True,
}
)
normalize_inputs = (
("normalize_3_4_0", (3.0, 4.0, -0.0)),
("normalize_equal_components", (1.0, 1.0, 1.0)),
)
for name, value in normalize_inputs:
result = profile_normalize(value)
vectors.append(
{
"expected_bits": [
f"{profile_f32_bits(component):08x}" for component in result
],
"input_bits": [[_f64_bits(component) for component in value]],
"name": name,
"operation": "normalize",
"valid": True,
}
)
for name, value in (
("normalize_zero", (0.0, 0.0, 0.0)),
(
"normalize_overflow",
(_f64_from_bits("7fefffffffffffff"), 1.0, 0.0),
),
):
try:
profile_normalize(value)
except ValueError as exc:
error = str(exc)
else:
raise AssertionError(name)
vectors.append(
{
"error": error,
"input_bits": [[_f64_bits(component) for component in value]],
"name": name,
"operation": "normalize",
"valid": False,
}
)
return vectors
@@ -0,0 +1,263 @@
"""Scene package provenance and connector-binding fixture builders."""
from __future__ import annotations
from copy import deepcopy
from typing import Callable
from simplecadapi.scene import with_scene_revision
from .common import (
JsonObject,
ScenePackage,
inline_package_case,
replace_entity_sidecar,
set_product_material_appearance,
)
def build_source_scene_package(
base_scene: JsonObject,
base_entity: JsonObject,
base_blobs: dict[str, bytes],
variant: str,
) -> ScenePackage:
scene = deepcopy(base_scene)
entity = deepcopy(base_entity)
definition = scene["definitions"][0]
source_for: Callable[[JsonObject], JsonObject]
status_for: Callable[[JsonObject], str]
if variant == "imported_shape":
definition_id = "definition/root/shape/imported/element"
scene["scene_id"] = "imported_fixture"
scene["source"] = {
"artifact_hash": "sha256:" + "1" * 64,
"format": "step",
"kind": "imported",
}
definition.update(
{
"definition_id": definition_id,
"kind": "shape",
"source": {
"kind": "imported",
"root_id": "root",
"source_element_id": "element",
},
}
)
source_for = lambda record: {
"kind": "imported_primitive",
"source_element_id": record["entity_id"],
}
status_for = lambda record: (
"not_applicable" if record["kind"] == "solid" else "owner_not_part"
)
else:
definition_kind = "part" if variant == "model_part" else "shape"
definition_id = (
"definition/root/part/model_part"
if definition_kind == "part"
else "definition/root/shape/model/graph/body/0"
)
scene["scene_id"] = f"{variant}_fixture"
scene["source"] = {
"artifact_hash": "sha256:" + "2" * 64,
"graph_id": "graph",
"kind": "model",
"model_schema_version": "2.0",
}
definition_source: JsonObject = {
"graph_id": "graph",
"kind": "product_model" if definition_kind == "part" else "model_output",
"node_id": "body",
"output_slot": 0,
"root_id": "root",
}
if definition_kind == "part":
definition_source.update(
{"semantic_id": "model_part", "semantic_type": "Part"}
)
definition.update(
{
"definition_id": definition_id,
"kind": definition_kind,
"source": definition_source,
}
)
topology_kind = {
"solid": "SOLID",
"face": "FACE",
"edge": "EDGE",
"vertex": "VERTEX",
}
if variant == "model_part":
source_for = lambda record: {
"graph_id": "graph",
"kind": "model_topology",
"node_id": "body",
"output_slot": 0,
"topo_id": record["entity_id"],
"topology_kind": topology_kind[record["kind"]],
}
statuses = iter(
["supported", "selector_ambiguous", "selector_unstable", "supported"]
)
def status_for(record: JsonObject) -> str:
return (
"not_applicable"
if record["kind"] == "solid"
else next(statuses, "supported")
)
else:
source_for = lambda _record: {
"graph_id": "graph",
"kind": "model_output",
"node_id": "body",
"output_slot": 0,
}
status_for = lambda record: (
"not_applicable"
if record["kind"] == "solid"
else "owner_not_part"
)
if definition["kind"] == "part":
scene["nodes"][0]["source"] = {
"component_path": [],
"kind": "product_occurrence",
"root_id": "root",
}
scene["nodes"][0]["definition_id"] = definition_id
entity["definition_id"] = definition_id
for record in entity["entities"]:
record["source"] = source_for(record)
record["connector_binding_status"] = status_for(record)
if variant == "model_part":
edge = next(record for record in entity["entities"] if record["kind"] == "edge")
edge["sdk_connector_frame"] = None
edge["connector_binding_status"] = "frame_undefined"
scene, entity, blobs = replace_entity_sidecar(
scene, entity, dict(base_blobs)
)
if definition["kind"] == "part":
set_product_material_appearance(scene)
return with_scene_revision(scene), entity, blobs
def build_package_matrix_cases(
scene: JsonObject,
entity: JsonObject,
blobs: dict[str, bytes],
scene_packages: dict[str, ScenePackage],
blob_pool: dict[str, str],
) -> list[dict[str, object]]:
cases: list[dict[str, object]] = []
for variant in ("imported_shape", "model_shape", "model_part"):
package = build_source_scene_package(scene, entity, blobs, variant)
cases.append(
inline_package_case(
f"valid_{variant}", package[0], package[2], blob_pool
)
)
nested = scene_packages["nested_assembly"]
model_part = build_source_scene_package(scene, entity, blobs, "model_part")
status_mutations = [
(
"solid_status_precedence",
(deepcopy(scene), deepcopy(entity), dict(blobs)),
"solid",
"supported",
),
(
"owner_not_part_status_precedence",
(deepcopy(scene), deepcopy(entity), dict(blobs)),
"face",
"source_not_model",
),
(
"source_not_model_status_precedence",
deepcopy(nested),
"face",
"supported",
),
(
"frame_undefined_status_precedence",
deepcopy(model_part),
"edge",
"owner_not_part",
),
]
for name, package, kind, status in status_mutations:
status_scene, status_entity, status_blobs = package
target = next(
record for record in status_entity["entities"] if record["kind"] == kind
)
target["connector_binding_status"] = status
status_scene, _status_entity, status_blobs = replace_entity_sidecar(
status_scene, status_entity, status_blobs
)
cases.append(
inline_package_case(
name, with_scene_revision(status_scene), status_blobs, blob_pool
)
)
forwarded_scene, _forwarded_entity, forwarded_blobs = deepcopy(nested)
root_connector = next(
record
for record in forwarded_scene["connectors"]
if record["connector_id"] == "root_mount"
)
root_connector["local_transform"]["origin"] = [0, 0, 6]
cases.append(
inline_package_case(
"forwarded_transform_mismatch",
with_scene_revision(forwarded_scene),
forwarded_blobs,
blob_pool,
)
)
missing_child_scene, _missing_child_entity, missing_child_blobs = deepcopy(nested)
root_connector = next(
record
for record in missing_child_scene["connectors"]
if record["connector_id"] == "root_mount"
)
root_connector["forwarded_from"]["source_component_id"] = "missing"
cases.append(
inline_package_case(
"forwarded_direct_child_missing",
with_scene_revision(missing_child_scene),
missing_child_blobs,
blob_pool,
)
)
part_owner_scene, _part_owner_entity, part_owner_blobs = deepcopy(nested)
root_connector = next(
record
for record in part_owner_scene["connectors"]
if record["connector_id"] == "root_mount"
)
root_connector["owner_definition_id"] = "definition/root/part/shared_part"
root_connector["connector_snapshot_id"] = (
"connector/root/part/shared_part/root_mount"
)
part_owner_scene["connectors"].sort(
key=lambda record: record["connector_snapshot_id"]
)
cases.append(
inline_package_case(
"forwarded_owner_not_assembly",
with_scene_revision(part_owner_scene),
part_owner_blobs,
blob_pool,
)
)
return cases
@@ -0,0 +1,710 @@
"""Resource-limit fixture case construction."""
from __future__ import annotations
from copy import deepcopy
from dataclasses import replace
from typing import Any, Callable
from simplecadapi.scene import (
BASE_LIMITS,
SceneResourceLimits,
canonical_archive_size,
json_resource_issues,
preflight_aggregate_compression_ratio,
preflight_archive_member_sizes,
preflight_glb_counts,
preflight_input_archive_size,
preflight_member_compression_ratio,
resource_count_issues,
validate_scene_manifest,
with_scene_revision,
)
from simplecadapi.scene.validation import (
_compute_package_budget_totals,
_package_budget_issues,
_report,
)
from .common import JsonObject, first_issue
def _limits(overrides: dict[str, int]) -> SceneResourceLimits:
return replace(BASE_LIMITS, **overrides)
def _preflight_case(
name: str,
operation: str,
parameters: dict[str, object],
callback: Callable[[], object],
) -> dict[str, object]:
try:
callback()
except ValueError as exc:
valid, error = False, str(exc)
else:
valid, error = True, None
return {
"error": error,
"name": name,
"operation": operation,
"parameters": parameters,
"valid": valid,
}
def _issues_case(
name: str,
operation: str,
parameters: dict[str, object],
issues: tuple[Any, ...] | list[Any],
) -> dict[str, object]:
first = issues[0] if issues else None
return {
"expected": (
None if first is None else {"code": first.code, "path": first.path}
),
"name": name,
"operation": operation,
"parameters": parameters,
"valid": not issues,
}
def _validation_case(
name: str,
operation: str,
parameters: dict[str, object],
report: Any,
) -> dict[str, object]:
return {
"expected": first_issue(report),
"name": name,
"operation": operation,
"parameters": parameters,
"valid": report.valid,
}
def _input_size_case(name: str, size: int, overrides: dict[str, int]) -> dict[str, object]:
parameters: dict[str, object] = {"limits": overrides, "size": size}
return _preflight_case(
name,
"input_archive_size",
parameters,
lambda: preflight_input_archive_size(size, limits=_limits(overrides)),
)
def _archive_sizes_case(
name: str, sizes: dict[str, int], overrides: dict[str, int]
) -> dict[str, object]:
parameters: dict[str, object] = {"limits": overrides, "sizes": sizes}
return _preflight_case(
name,
"archive_member_sizes",
parameters,
lambda: preflight_archive_member_sizes(sizes, limits=_limits(overrides)),
)
def _archive_count_case(
name: str, count: int, overrides: dict[str, int]
) -> dict[str, object]:
sizes = {"scene.json": 0}
sizes.update({f"x/{index}": 0 for index in range(count - 1)})
parameters: dict[str, object] = {"count": count, "limits": overrides}
return _preflight_case(
name,
"archive_member_count",
parameters,
lambda: preflight_archive_member_sizes(sizes, limits=_limits(overrides)),
)
def _compression_case(
name: str,
operation: str,
uncompressed_size: int,
compressed_size: int,
) -> dict[str, object]:
overrides = {"compression_ratio": 2}
parameters = {
"compressed_size": compressed_size,
"limits": overrides,
"uncompressed_size": uncompressed_size,
}
if operation == "aggregate_compression_ratio":
callback = lambda: preflight_aggregate_compression_ratio(
uncompressed_size, compressed_size, limits=_limits(overrides)
)
else:
callback = lambda: preflight_member_compression_ratio(
uncompressed_size, compressed_size, limits=_limits(overrides)
)
return _preflight_case(name, operation, parameters, callback)
def _json_value(kind: str, text: str, field: str | None = None) -> object:
if kind == "value":
return {"value": text}
if kind == "object_key":
return {text: 0}
if kind == "metadata_key":
return {"metadata": {text: 0}}
if kind == "sdk_metadata_key":
return {"sdk_metadata": {text: 0}}
if kind == "identifier":
return {field or "node_id": text}
if kind == "identifier_array":
return {field or "component_path": [text]}
if kind == "uri":
return {"uri": text}
raise AssertionError(kind)
def _json_domain_case(
name: str,
kind: str,
text: str,
overrides: dict[str, int],
*,
field: str | None = None,
) -> dict[str, object]:
parameters: dict[str, object] = {
"field": field,
"kind": kind,
"limits": overrides,
"text": text,
}
issues = json_resource_issues(
_json_value(kind, text, field), limits=_limits(overrides)
)
return _issues_case(name, "json_domain", parameters, issues)
def _nested_value(depth: int) -> object:
value: dict[str, object] = {}
current = value
for _index in range(depth):
current["x"] = {}
current = current["x"] # type: ignore[assignment]
return value
def _json_depth_case(name: str, depth: int) -> dict[str, object]:
overrides = {"json_depth": 2}
parameters: dict[str, object] = {"depth": depth, "limits": overrides}
issues = json_resource_issues(_nested_value(depth), limits=_limits(overrides))
return _issues_case(name, "json_depth", parameters, issues)
def _resource_count_value(
artifact: str, kind: str, count: int, field: str | None
) -> object:
if kind == "collection":
assert field is not None
return {field: [None] * count}
if kind == "hierarchy":
return {"nodes": [{"source": {"component_path": ["x"] * count}}]}
if kind == "forwarded":
return {
"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)
]
}
raise AssertionError((artifact, kind))
def _resource_count_case(
name: str,
artifact: str,
kind: str,
count: int,
overrides: dict[str, int],
*,
field: str | None = None,
) -> dict[str, object]:
parameters: dict[str, object] = {
"artifact": artifact,
"count": count,
"field": field,
"kind": kind,
"limits": overrides,
}
issues = resource_count_issues(
_resource_count_value(artifact, kind, count, field),
artifact,
limits=_limits(overrides),
)
return _issues_case(name, "resource_count", parameters, issues)
def _glb_count_case(
name: str,
kind: str,
vertex_count: int,
index_count: int,
overrides: dict[str, int],
) -> dict[str, object]:
parameters: dict[str, object] = {
"index_count": index_count,
"kind": kind,
"limits": overrides,
"vertex_count": vertex_count,
}
return _preflight_case(
name,
"glb_counts",
parameters,
lambda: preflight_glb_counts(
kind, # type: ignore[arg-type]
vertex_count,
index_count,
limits=_limits(overrides),
),
)
_BUDGET_FIELDS = {
"entities_total": "entity_count",
"line_segments_total": "line_segment_count",
"line_vertices_total": "line_vertex_count",
"static_decoded_buffer_bytes": "scene_json_bytes",
"triangle_vertices_total": "triangle_vertex_count",
"triangles_total": "triangle_count",
}
def _package_budget_case(
name: str, limit_name: str, value: int
) -> dict[str, object]:
contributions = {
"scene_json_bytes": 0,
"glb_decoded_buffer_bytes": 0,
"entity_json_bytes": 0,
"other_immutable_json_bytes": 0,
"entity_count": 0,
"entity_vertex_count": 0,
"triangle_vertex_count": 0,
"triangle_count": 0,
"line_vertex_count": 0,
"line_segment_count": 0,
}
contributions[_BUDGET_FIELDS[limit_name]] = value
overrides = {limit_name: 2}
totals = _compute_package_budget_totals(**contributions)
budget_report = _report(
_package_budget_issues(totals, limits=_limits(overrides)), artifact="package"
)
return _validation_case(
name,
"package_budget",
{"contributions": contributions, "limits": overrides},
budget_report,
)
def build_resource_cases(
scene: JsonObject,
normalized_part: JsonObject,
) -> list[dict[str, object]]:
input_limits = {"input_archive_bytes": 8}
role_limits = {
"canonical_archive_bytes": 1_000,
"one_member_bytes": 100,
"total_uncompressed_bytes": 100,
}
canonical_limits = {
"canonical_archive_bytes": 256,
"one_member_bytes": 100,
"total_uncompressed_bytes": 100,
}
canonical_envelope = canonical_archive_size(
{"geometry/a.glb": 0, "scene.json": 0}
)
canonical_payload = canonical_limits["canonical_archive_bytes"] - canonical_envelope
one_member_limits = {
"canonical_archive_bytes": 1_000,
"one_member_bytes": 8,
"total_uncompressed_bytes": 100,
}
total_limits = {
"canonical_archive_bytes": 1_000,
"one_member_bytes": 8,
"total_uncompressed_bytes": 16,
}
count_limits = {"canonical_archive_bytes": 10_000, "zip_members": 3}
cases = [
_input_size_case("input_archive_exact_limit", 8, input_limits),
_input_size_case("input_archive_over_limit", 9, input_limits),
_archive_sizes_case(
"scene_json_exact_limit",
{"scene.json": 8},
{**role_limits, "scene_json_bytes": 8},
),
_archive_sizes_case(
"scene_json_over_limit",
{"scene.json": 9},
{**role_limits, "scene_json_bytes": 8},
),
_archive_sizes_case(
"entity_json_exact_limit",
{"entities/a.json": 8, "scene.json": 0},
{**role_limits, "entity_json_bytes": 8},
),
_archive_sizes_case(
"entity_json_over_limit",
{"entities/a.json": 9, "scene.json": 0},
{**role_limits, "entity_json_bytes": 8},
),
_archive_sizes_case(
"model_json_exact_limit",
{"model/model.json": 8, "scene.json": 0},
{**role_limits, "model_json_bytes": 8},
),
_archive_sizes_case(
"model_json_over_limit",
{"model/model.json": 9, "scene.json": 0},
{**role_limits, "model_json_bytes": 8},
),
_archive_sizes_case(
"presentation_json_exact_limit",
{"presentation/presentation.json": 8, "scene.json": 0},
{**role_limits, "presentation_json_bytes": 8},
),
_archive_sizes_case(
"presentation_json_over_limit",
{"presentation/presentation.json": 9, "scene.json": 0},
{**role_limits, "presentation_json_bytes": 8},
),
_archive_sizes_case(
"canonical_archive_exact_limit",
{"geometry/a.glb": canonical_payload, "scene.json": 0},
canonical_limits,
),
_archive_sizes_case(
"canonical_archive_over_limit",
{"geometry/a.glb": canonical_payload + 1, "scene.json": 0},
canonical_limits,
),
_archive_sizes_case(
"one_member_exact_limit",
{"geometry/a.glb": 8, "scene.json": 0},
one_member_limits,
),
_archive_sizes_case(
"one_member_over_limit",
{"geometry/a.glb": 9, "scene.json": 0},
one_member_limits,
),
_archive_sizes_case(
"total_uncompressed_exact_limit",
{"geometry/a.glb": 8, "geometry/b.glb": 8, "scene.json": 0},
total_limits,
),
_archive_sizes_case(
"total_uncompressed_over_limit",
{
"geometry/a.glb": 8,
"geometry/b.glb": 8,
"geometry/c.glb": 1,
"scene.json": 0,
},
total_limits,
),
_archive_count_case("archive_member_count_exact_limit", 3, count_limits),
_archive_count_case("archive_member_count_over_limit", 4, count_limits),
_compression_case(
"aggregate_compression_ratio_exact_limit",
"aggregate_compression_ratio",
4,
2,
),
_compression_case(
"aggregate_compression_ratio_over_limit",
"aggregate_compression_ratio",
5,
2,
),
_compression_case(
"member_compression_ratio_exact_limit",
"member_compression_ratio",
4,
2,
),
_compression_case(
"member_compression_ratio_over_limit",
"member_compression_ratio",
5,
2,
),
_json_depth_case("json_depth_exact_limit", 2),
_json_depth_case("json_depth_over_limit", 3),
]
for kind, overrides in (
("value", {"json_string_bytes": 4}),
("object_key", {"json_string_bytes": 4}),
(
"metadata_key",
{"json_string_bytes": 100, "structural_id_bytes": 4},
),
(
"sdk_metadata_key",
{"json_string_bytes": 100, "structural_id_bytes": 4},
),
("uri", {"json_string_bytes": 100, "uri_bytes": 4}),
):
cases.append(
_json_domain_case(
f"utf8_{kind}_exact_limit", kind, "éé", overrides
)
)
cases.append(
_json_domain_case(
f"utf8_{kind}_over_limit", kind, "ééa", overrides
)
)
identifier_fields = (
"definition_ref",
"node_id",
"source_element_id",
"topo_id",
)
for field in identifier_fields:
overrides = {"json_string_bytes": 100, "structural_id_bytes": 4}
cases.append(
_json_domain_case(
f"utf8_{field}_exact_limit",
"identifier",
"éé",
overrides,
field=field,
)
)
cases.append(
_json_domain_case(
f"utf8_{field}_over_limit",
"identifier",
"ééa",
overrides,
field=field,
)
)
for field in (
"child_entity_ids",
"component_path",
"evaluated_tags",
"grounded_component_ids",
"parent_entity_ids",
"semantic_binding_ids",
):
overrides = {"json_string_bytes": 100, "structural_id_bytes": 4}
cases.append(
_json_domain_case(
f"utf8_{field}_item_exact_limit",
"identifier_array",
"éé",
overrides,
field=field,
)
)
cases.append(
_json_domain_case(
f"utf8_{field}_item_over_limit",
"identifier_array",
"ééa",
overrides,
field=field,
)
)
collection_limits = (
("scene", "definitions", "definitions"),
("scene", "nodes", "nodes"),
("scene", "geometry_assets", "assets_per_kind"),
("scene", "edge_assets", "assets_per_kind"),
("scene", "entity_assets", "assets_per_kind"),
("scene", "appearances", "appearances"),
("scene", "connectors", "connectors"),
("scene", "cameras", "cameras"),
("entities", "entities", "entities_per_sidecar"),
("entities", "face_groups", "entities_per_sidecar"),
("entities", "edge_groups", "entities_per_sidecar"),
("presentation", "node_overrides", "nodes"),
("presentation", "appearances", "appearances"),
("presentation", "cameras", "cameras"),
)
for artifact, field, limit_name in collection_limits:
overrides = {limit_name: 2}
cases.append(
_resource_count_case(
f"{artifact}_{field}_exact_limit",
artifact,
"collection",
2,
overrides,
field=field,
)
)
cases.append(
_resource_count_case(
f"{artifact}_{field}_over_limit",
artifact,
"collection",
3,
overrides,
field=field,
)
)
cases.extend(
[
_resource_count_case(
"hierarchy_depth_exact_limit",
"scene",
"hierarchy",
2,
{"hierarchy_depth": 2, "nodes": 10},
),
_resource_count_case(
"hierarchy_depth_over_limit",
"scene",
"hierarchy",
3,
{"hierarchy_depth": 2, "nodes": 10},
),
_resource_count_case(
"forwarded_connector_depth_exact_limit",
"scene",
"forwarded",
2,
{"connectors": 10, "forwarded_connector_depth": 2},
),
_resource_count_case(
"forwarded_connector_depth_over_limit",
"scene",
"forwarded",
3,
{"connectors": 10, "forwarded_connector_depth": 2},
),
_glb_count_case(
"triangle_vertices_per_asset_exact_limit",
"triangle",
2,
3,
{"triangle_vertices_per_asset": 2},
),
_glb_count_case(
"triangle_vertices_per_asset_over_limit",
"triangle",
3,
3,
{"triangle_vertices_per_asset": 2},
),
_glb_count_case(
"triangles_per_asset_exact_limit",
"triangle",
3,
6,
{"triangles_per_asset": 2},
),
_glb_count_case(
"triangles_per_asset_over_limit",
"triangle",
3,
9,
{"triangles_per_asset": 2},
),
_glb_count_case(
"line_vertices_per_asset_exact_limit",
"line",
2,
2,
{"line_vertices_per_asset": 2},
),
_glb_count_case(
"line_vertices_per_asset_over_limit",
"line",
3,
2,
{"line_vertices_per_asset": 2},
),
_glb_count_case(
"line_segments_per_asset_exact_limit",
"line",
2,
4,
{"line_segments_per_asset": 2},
),
_glb_count_case(
"line_segments_per_asset_over_limit",
"line",
2,
6,
{"line_segments_per_asset": 2},
),
]
)
for limit_name in _BUDGET_FIELDS:
cases.append(
_package_budget_case(f"{limit_name}_exact_limit", limit_name, 2)
)
cases.append(
_package_budget_case(f"{limit_name}_over_limit", limit_name, 3)
)
for name, value_number in (
("safe_integer_exact_limit", 9_007_199_254_740_991),
("safe_integer_over_limit", 9_007_199_254_740_992),
):
value = deepcopy(scene)
value["geometry_assets"][0]["byte_length"] = value_number
if value_number <= 9_007_199_254_740_991:
value = with_scene_revision(value)
cases.append(
_validation_case(
name,
"scene_geometry_byte_length",
{"value": str(value_number)},
validate_scene_manifest(value),
)
)
for name, field, value_number in (
("linear_tolerance_exact_max", "linear_tolerance", 1_000_000),
("linear_tolerance_over_max", "linear_tolerance", 1_000_001),
("angular_tolerance_exact_max", "angular_tolerance", 3.141592653589793),
("angular_tolerance_over_max", "angular_tolerance", 3.1415926535897936),
):
value = deepcopy(scene)
value["compile_options"][field] = value_number
value["geometry_assets"][0]["tessellation"][field] = value_number
if field == "linear_tolerance":
value["edge_assets"][0]["tessellation"][field] = value_number
value = with_scene_revision(value)
cases.append(
_validation_case(
name,
"scene_compile_option",
{"field": field, "value": value_number},
validate_scene_manifest(value),
)
)
# Keep the argument in active use so fixture construction catches drift in
# the valid normalized product even though resource probes are synthetic.
assert normalized_part
return cases
@@ -0,0 +1,218 @@
"""Minimal, nested, and repeated-instance Scene fixture builders."""
from __future__ import annotations
from copy import deepcopy
from simplecadapi.scene import with_scene_revision
from .common import (
IDENTITY,
JsonObject,
ScenePackage,
replace_entity_sidecar,
set_product_material_appearance,
transform,
)
def _product_definition(
definition_id: str,
kind: str,
semantic_id: str,
renderable: JsonObject | None = None,
) -> JsonObject:
definition: JsonObject = {
"definition_id": definition_id,
"kind": kind,
"name": semantic_id.replace("_", " ").title(),
"sdk_metadata": {},
"source": {
"kind": "product_manual",
"root_id": "root",
"semantic_id": semantic_id,
"semantic_type": "Part" if kind == "part" else "Assembly",
},
}
if renderable is not None:
definition.update(
{
field: renderable[field]
for field in (
"appearance_id",
"edge_asset_id",
"entity_asset_id",
"geometry_asset_id",
)
}
)
return definition
def _product_node(
component_path: list[str],
definition_id: str,
order: int,
local_transform: JsonObject | None = None,
) -> JsonObject:
node_id = "instance/root" + "".join(f"/{part}" for part in component_path)
parent_id = (
None
if not component_path
else "instance/root" + "".join(f"/{part}" for part in component_path[:-1])
)
return {
"appearance_override_id": None,
"definition_id": definition_id,
"name": component_path[-1] if component_path else "root",
"node_id": node_id,
"order": order,
"parent_node_id": parent_id,
"sdk_metadata": {},
"selectable": True,
"source": {
"component_path": component_path,
"kind": "product_occurrence",
"root_id": "root",
},
"transform": deepcopy(local_transform or IDENTITY),
"visible": True,
}
def _manual_connector(
owner_definition_id: str,
owner_kind: str,
owner_semantic_id: str,
connector_id: str,
anchor_kind: str,
local_transform: JsonObject,
*,
forwarded_from: JsonObject | None = None,
) -> JsonObject:
connector: JsonObject = {
"anchor_kind": anchor_kind,
"connector_id": connector_id,
"connector_snapshot_id": (
f"connector/root/{owner_kind}/{owner_semantic_id}/{connector_id}"
),
"local_transform": local_transform,
"name": connector_id.replace("_", " ").title(),
"owner_definition_id": owner_definition_id,
"sdk_metadata": {},
"source": {"kind": "manual", "source_id": "fixture"},
}
if forwarded_from is not None:
connector["forwarded_from"] = forwarded_from
return connector
def build_product_scene_package(
base_scene: JsonObject,
base_entity: JsonObject,
base_blobs: dict[str, bytes],
*,
nested: bool,
) -> ScenePackage:
scene = deepcopy(base_scene)
entity = deepcopy(base_entity)
root_definition_id = "definition/root/assembly/root_assembly"
part_definition_id = "definition/root/part/shared_part"
renderable = deepcopy(scene["definitions"][0])
definitions = [
_product_definition(root_definition_id, "assembly", "root_assembly"),
_product_definition(part_definition_id, "part", "shared_part", renderable),
]
nodes = [_product_node([], root_definition_id, 0)]
connectors: list[JsonObject] = []
if nested:
nested_definition_id = "definition/root/assembly/nested_assembly"
definitions.append(
_product_definition(
nested_definition_id, "assembly", "nested_assembly"
)
)
nodes.extend(
[
_product_node(["subassembly"], nested_definition_id, 0),
_product_node(["subassembly", "part"], part_definition_id, 0),
]
)
part_connector = _manual_connector(
part_definition_id,
"part",
"shared_part",
"mount",
"placement",
transform(),
)
nested_connector = _manual_connector(
nested_definition_id,
"assembly",
"nested_assembly",
"nested_mount",
"forwarded",
transform(),
forwarded_from={
"offset": None,
"source_component_id": "part",
"source_connector_id": "mount",
"source_connector_snapshot_id": part_connector[
"connector_snapshot_id"
],
"source_definition_id": part_definition_id,
},
)
root_offset = transform([0, 0, 5])
root_connector = _manual_connector(
root_definition_id,
"assembly",
"root_assembly",
"root_mount",
"forwarded",
deepcopy(root_offset),
forwarded_from={
"offset": root_offset,
"source_component_id": "subassembly",
"source_connector_id": "nested_mount",
"source_connector_snapshot_id": nested_connector[
"connector_snapshot_id"
],
"source_definition_id": nested_definition_id,
},
)
connectors = [part_connector, nested_connector, root_connector]
scene["scene_id"] = "nested_fixture"
else:
nodes.extend(
[
_product_node(
["left"], part_definition_id, 0, transform([-1000, 0, 0])
),
_product_node(
["right"], part_definition_id, 1, transform([1000, 0, 0])
),
]
)
scene["scene_id"] = "repeated_fixture"
scene["definitions"] = sorted(
definitions, key=lambda value: value["definition_id"].encode("utf-8")
)
scene["nodes"] = sorted(
nodes, key=lambda value: value["node_id"].encode("utf-8")
)
scene["connectors"] = sorted(
connectors,
key=lambda value: value["connector_snapshot_id"].encode("utf-8"),
)
entity["definition_id"] = part_definition_id
for record in entity["entities"]:
if record["kind"] != "solid":
record["connector_binding_status"] = "source_not_model"
scene, entity, blobs = replace_entity_sidecar(
scene, entity, dict(base_blobs)
)
set_product_material_appearance(scene)
return with_scene_revision(scene), entity, blobs
@@ -0,0 +1,140 @@
"""Scene shape, field-matrix, and two-pass revision corpus sections."""
from __future__ import annotations
from copy import deepcopy
from simplecadapi.scene import canonical_json_bytes, validate_scene_manifest
from .common import (
JsonObject,
ScenePackage,
b64,
content_hash,
first_issue,
scene_shape_case,
)
from .field_matrix import (
apply_scene_field_case,
nullable_fields_nonnull_case,
scene_field_matrix,
)
from .scene_cases import build_product_scene_package
def build_scene_shape_cases(
scene: JsonObject,
entity: JsonObject,
blobs: dict[str, bytes],
blob_pool: dict[str, str],
) -> tuple[list[dict[str, object]], dict[str, ScenePackage]]:
nested = build_product_scene_package(scene, entity, blobs, nested=True)
repeated = build_product_scene_package(scene, entity, blobs, nested=False)
packages = {
"minimal_standalone_shape": (deepcopy(scene), deepcopy(entity), dict(blobs)),
"nested_assembly": nested,
"repeated_part_instance": repeated,
}
cases = [
scene_shape_case(
"minimal_standalone_shape",
scene,
blobs,
{
"definition_count": 1,
"definition_occurrence_counts": {
"definition/root/shape/manual/fixture": 1
},
"edge_asset_count": 1,
"entity_asset_count": 1,
"geometry_asset_count": 1,
"maximum_depth": 0,
"node_count": 1,
"root_node_ids": ["instance/root"],
},
blob_pool,
),
scene_shape_case(
"nested_assembly",
nested[0],
nested[2],
{
"definition_count": 3,
"definition_occurrence_counts": {
"definition/root/assembly/nested_assembly": 1,
"definition/root/assembly/root_assembly": 1,
"definition/root/part/shared_part": 1,
},
"edge_asset_count": 1,
"entity_asset_count": 1,
"geometry_asset_count": 1,
"maximum_depth": 2,
"node_count": 3,
"root_node_ids": ["instance/root"],
},
blob_pool,
),
scene_shape_case(
"repeated_part_instance",
repeated[0],
repeated[2],
{
"definition_count": 2,
"definition_occurrence_counts": {
"definition/root/assembly/root_assembly": 1,
"definition/root/part/shared_part": 2,
},
"edge_asset_count": 1,
"entity_asset_count": 1,
"geometry_asset_count": 1,
"maximum_depth": 1,
"node_count": 3,
"root_node_ids": ["instance/root"],
},
blob_pool,
),
]
return cases, packages
def build_scene_field_cases(scene: JsonObject) -> list[dict[str, object]]:
descriptors = scene_field_matrix(scene)
nullable_value, nullable_descriptor = nullable_fields_nonnull_case(scene)
descriptors.append(nullable_descriptor)
result = []
for descriptor in descriptors:
value = (
nullable_value
if descriptor["name"] == "nullable_fields_nonnull"
else apply_scene_field_case(scene, descriptor)
)
report = validate_scene_manifest(canonical_json_bytes(value))
result.append(
{
**descriptor,
"expected": first_issue(report),
"valid": report.valid,
}
)
return result
def build_revision_vectors(
scene_packages: dict[str, ScenePackage],
) -> list[dict[str, object]]:
vectors = []
for name, (scene, _entity, _blobs) in scene_packages.items():
draft = deepcopy(scene)
revision = draft.pop("revision")
draft_bytes = canonical_json_bytes(draft)
scene_bytes = canonical_json_bytes(scene)
vectors.append(
{
"canonical_base64": b64(scene_bytes),
"draft_base64": b64(draft_bytes),
"name": name,
"revision": revision,
"sha256": content_hash(scene_bytes),
}
)
return vectors
@@ -0,0 +1,330 @@
"""Schema-grounded structural field matrices for all contract artifacts."""
from __future__ import annotations
import json
import re
from copy import deepcopy
from pathlib import Path
from typing import Any, Mapping
from jsonschema import Draft202012Validator
Json = Any
SCHEMAS = {
"scene": "scene-1.0.schema.json",
"entities": "entities-1.0.schema.json",
"presentation": "presentation-1.0.schema.json",
"connector_binding": "connector-binding-1.0.schema.json",
"normalized_product": "normalized-product-1.schema.json",
}
def _resolve(root: Mapping[str, Any], schema: Mapping[str, Any]) -> Mapping[str, Any]:
reference = schema.get("$ref")
if not isinstance(reference, str) or not reference.startswith("#/"):
return schema
current: Any = root
for part in reference[2:].split("/"):
current = current[part.replace("~1", "/").replace("~0", "~")]
return current
def _wrapper(root: Mapping[str, Any], pointer: str) -> dict[str, Any]:
if pointer == "#":
return deepcopy(dict(root))
return {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$ref": pointer,
"$defs": deepcopy(root.get("$defs", {})),
}
def _string_example(schema: Mapping[str, Any]) -> str:
pattern = str(schema.get("pattern", ""))
if pattern == "^sha256:[0-9a-f]{64}$":
return "sha256:" + "0" * 64
if pattern.startswith("^appearance/evaluated/"):
return "appearance/evaluated/" + "0" * 64
if pattern.startswith("^entity/"):
return "entity/face/0"
for candidate in ("a", "x", "root", "entity/face/0"):
if not pattern or re.fullmatch(pattern, candidate):
return candidate
raise AssertionError(f"no structural string example for pattern {pattern!r}")
def _accepts_null(root: Mapping[str, Any], schema: Mapping[str, Any]) -> bool:
wrapper = deepcopy(dict(schema))
wrapper["$defs"] = deepcopy(root.get("$defs", {}))
return Draft202012Validator(wrapper).is_valid(None)
def _nonnull_example(root: Mapping[str, Any], schema: Mapping[str, Any]) -> Json:
resolved = _resolve(root, schema)
if "oneOf" in resolved:
for branch in resolved["oneOf"]:
if not _accepts_null(root, branch):
return _example(root, branch, prefer_nonnull=True)
types = resolved.get("type")
if isinstance(types, list):
draft = dict(resolved)
draft["type"] = next(item for item in types if item != "null")
return _example(root, draft, prefer_nonnull=True)
return _example(root, resolved, prefer_nonnull=True)
def _condition_matches(
root: Mapping[str, Any], condition: Mapping[str, Any], value: Json
) -> bool:
schema = deepcopy(dict(condition))
schema["$defs"] = deepcopy(root.get("$defs", {}))
return Draft202012Validator(schema).is_valid(value)
def _example(
root: Mapping[str, Any],
schema: Mapping[str, Any],
*,
overrides: Mapping[str, Json] | None = None,
prefer_nonnull: bool = False,
) -> Json:
resolved = _resolve(root, schema)
if "const" in resolved:
return deepcopy(resolved["const"])
if "enum" in resolved:
return deepcopy(resolved["enum"][0])
for union_key in ("oneOf", "anyOf"):
if union_key in resolved and not (
resolved.get("type") == "object" or "properties" in resolved
):
branches = resolved[union_key]
if prefer_nonnull:
branches = sorted(
branches,
key=lambda branch: _accepts_null(root, branch),
)
return _example(root, branches[0], prefer_nonnull=prefer_nonnull)
types = resolved.get("type")
if isinstance(types, list):
if "null" in types and not prefer_nonnull:
return None
types = next(item for item in types if item != "null")
if types == "object" or "properties" in resolved:
properties = resolved.get("properties", {})
result = {
field: _example(root, properties.get(field, {}))
for field in resolved.get("required", [])
}
if overrides:
result.update(deepcopy(dict(overrides)))
for branch in resolved.get("anyOf", []):
branch_required = branch.get("required", [])
if branch_required:
for field in branch_required:
result.setdefault(field, _example(root, properties.get(field, {})))
break
for condition in resolved.get("allOf", []):
then = condition.get("then")
if isinstance(then, dict) and _condition_matches(
root, condition.get("if", {}), result
):
for field in then.get("required", []):
result.setdefault(field, _example(root, properties.get(field, {})))
return result
if types == "array" or "prefixItems" in resolved:
result = [_example(root, item) for item in resolved.get("prefixItems", [])]
minimum = resolved.get("minItems", 0)
while len(result) < minimum:
result.append(_example(root, resolved.get("items", {})))
return result
if types == "string":
return _string_example(resolved)
if types == "integer":
return int(resolved.get("minimum", 0))
if types == "number":
if "exclusiveMinimum" in resolved:
return resolved["exclusiveMinimum"] + 1
return resolved.get("minimum", 0)
if types == "boolean":
return False
if types == "null":
return None
return {}
def _closed_records(schema: Mapping[str, Any]) -> list[tuple[str, Mapping[str, Any]]]:
result: dict[str, Mapping[str, Any]] = {}
if schema.get("type") == "object" and schema.get("additionalProperties") is False:
result["#"] = schema
def visit(value: Any, pointer: str) -> None:
if not isinstance(value, dict):
return
if value.get("type") == "object" and value.get("additionalProperties") is False:
result[pointer] = value
for key in ("properties", "items", "oneOf", "anyOf", "allOf"):
child = value.get(key)
if isinstance(child, dict):
for name, nested in child.items():
visit(nested, f"{pointer}/{key}/{name}")
elif isinstance(child, list):
for index, nested in enumerate(child):
visit(nested, f"{pointer}/{key}/{index}")
for name, definition in schema.get("$defs", {}).items():
visit(definition, f"#/$defs/{name}")
return list(result.items())
def _mutate(base: Json, operation: str, field: str, value: Json = None) -> Json:
result = deepcopy(base)
if operation == "delete":
del result[field]
else:
result[field] = deepcopy(value)
return result
def _record_matrix(
artifact: str,
root: Mapping[str, Any],
pointer: str,
schema: Mapping[str, Any],
*,
overrides: Mapping[str, Json] | None = None,
variant: str = "base",
) -> dict[str, Any]:
validator = Draft202012Validator(_wrapper(root, pointer))
base = _example(root, schema, overrides=overrides)
if not validator.is_valid(base):
errors = [error.message for error in validator.iter_errors(base)]
raise AssertionError(
f"invalid generated {artifact} {pointer} {variant}: {errors}"
)
properties = schema.get("properties", {})
cases: list[dict[str, Any]] = [
{"mutations": [], "name": "base_valid", "valid": True}
]
for field in base:
cases.append(
{
"mutations": [{"operation": "delete", "path": [field]}],
"name": f"missing_{field}",
"valid": validator.is_valid(_mutate(base, "delete", field)),
}
)
cases.append(
{
"mutations": [
{"operation": "set", "path": ["unknown"], "value": True}
],
"name": "unknown_field",
"valid": validator.is_valid(_mutate(base, "set", "unknown", True)),
}
)
optional = [field for field in properties if field not in base]
for field in optional:
present = _example(root, properties[field])
cases.append(
{
"mutations": [
{"operation": "set", "path": [field], "value": present}
],
"name": f"optional_{field}_present",
"valid": validator.is_valid(_mutate(base, "set", field, present)),
}
)
if optional:
all_present = deepcopy(base)
mutations = []
for field in optional:
present = _example(root, properties[field])
all_present[field] = present
mutations.append(
{"operation": "set", "path": [field], "value": present}
)
cases.append(
{
"mutations": mutations,
"name": "optional_fields_all_present",
"valid": validator.is_valid(all_present),
}
)
for field, field_schema in properties.items():
if field in base and base[field] is None and _accepts_null(root, field_schema):
nonnull = _nonnull_example(root, field_schema)
cases.append(
{
"mutations": [
{"operation": "set", "path": [field], "value": nonnull}
],
"name": f"nullable_{field}_nonnull",
"valid": validator.is_valid(
_mutate(base, "set", field, nonnull)
),
}
)
return {
"artifact": artifact,
"base": base,
"cases": cases,
"schema_pointer": pointer,
"variant": variant,
}
def build_schema_field_matrices(schema_dir: Path) -> list[dict[str, Any]]:
"""Build declarative matrices for every closed object and discriminator."""
matrices: list[dict[str, Any]] = []
for artifact, filename in SCHEMAS.items():
root = json.loads(schema_dir.joinpath(filename).read_text(encoding="utf-8"))
for pointer, schema in _closed_records(root):
discriminator = next(
(
(field, field_schema["enum"])
for field, field_schema in schema.get("properties", {}).items()
if isinstance(field_schema, dict)
and len(field_schema.get("enum", [])) > 1
),
None,
)
variants: list[tuple[str, Mapping[str, Json] | None]] = [("base", None)]
if discriminator is not None:
field, values = discriminator
variants = [
(f"{field}={value}", {field: value}) for value in values
]
if artifact == "normalized_product" and pointer == "#/$defs/connector":
variants = []
for anchor_name in (
"geometryAnchor",
"placementAnchor",
"forwardedAnchor",
):
anchor = _example(root, root["$defs"][anchor_name])
variants.append(
(f"anchor={anchor['anchor_kind']}", {"anchor": anchor})
)
for variant, overrides in variants:
matrices.append(
_record_matrix(
artifact,
root,
pointer,
schema,
overrides=overrides,
variant=variant,
)
)
identities = [
(matrix["artifact"], matrix["schema_pointer"], matrix["variant"])
for matrix in matrices
]
if len(identities) != len(set(identities)):
raise AssertionError("duplicate schema field matrix identity")
return matrices