ykcheungab
446 lines
18 KiB
Python
446 lines
18 KiB
Python
"""Stage 2 acceptance for the ``.oryscene`` container.
|
|
|
|
The interesting half is the refusals: a package that is *nearly* right is the case a
|
|
loader has to catch, because the failure mode it prevents (a scene that renders with
|
|
someone else's geometry) is silent. Every test below mutates one thing about a
|
|
package that was just written correctly, and asserts that the loader says no — with
|
|
the code the design document assigns to that failure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import zipfile
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from orisys_app_shared.scene3d import synthetic
|
|
from orisys_app_shared.scene3d.contracts import ExpectedGeometry, ThreeDError
|
|
from orisys_app_shared.scene3d.digest import asset_hash, content_digest
|
|
from orisys_app_shared.scene3d.glb import with_housing
|
|
from orisys_app_shared.scene3d.package import (
|
|
MAGIC,
|
|
PackagedAssetLoader,
|
|
describe_package,
|
|
read_package,
|
|
scene_from_contents,
|
|
write_package,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def scene():
|
|
"""A complete little scene: one sensing plate and one housing box."""
|
|
return synthetic.poc_scene()
|
|
|
|
|
|
@pytest.fixture
|
|
def package(tmp_path, scene):
|
|
path = tmp_path / "poc.oryscene"
|
|
report = write_package(scene, path)
|
|
assert report.output == path
|
|
return path
|
|
|
|
|
|
def _rewrite(source: Path, destination: Path, mutate) -> Path:
|
|
"""Copy a package, letting ``mutate`` change the members on the way through."""
|
|
with zipfile.ZipFile(source) as archive:
|
|
members = {info.filename: archive.read(info.filename) for info in archive.infolist()}
|
|
members = mutate(members)
|
|
with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_DEFLATED) as archive:
|
|
for name in sorted(members):
|
|
archive.writestr(name, members[name])
|
|
return destination
|
|
|
|
|
|
def _manifest(members) -> dict:
|
|
return json.loads(members["manifest.json"])
|
|
|
|
|
|
def _store_manifest(members, manifest: dict) -> None:
|
|
members["manifest.json"] = json.dumps(manifest, indent=2, sort_keys=True).encode("utf-8")
|
|
|
|
|
|
def _rehash(members, manifest: dict) -> None:
|
|
"""Recompute the asset hash after a deliberate edit: the *content* is what lies."""
|
|
stripped = {k: v for k, v in manifest.items() if k not in ("asset_hash", "entry_digest")}
|
|
digest = asset_hash(
|
|
{name: str(record["sha256"]) for name, record in stripped["entries"].items()},
|
|
stripped,
|
|
)
|
|
manifest["asset_hash"] = digest
|
|
manifest["entry_digest"] = {
|
|
name: record["sha256"] for name, record in stripped["entries"].items()
|
|
}
|
|
_store_manifest(members, manifest)
|
|
|
|
|
|
# ------------------------------------------------------------------ round trip
|
|
def test_a_package_round_trips_into_the_same_scene(package, scene):
|
|
contents = read_package(package)
|
|
rebuilt = scene_from_contents(contents)
|
|
assert rebuilt.geometry_id == scene.geometry_id
|
|
assert rebuilt.field_domain_id == scene.field_domain_id
|
|
assert rebuilt.asset_id == scene.asset_id
|
|
assert rebuilt.shape_hw == scene.shape_hw
|
|
assert len(rebuilt.parts) == len(scene.parts)
|
|
for original in scene.parts:
|
|
copy = rebuilt.part(original.part_id)
|
|
assert copy is not None
|
|
assert np.array_equal(copy.positions_sensor_m, original.positions_sensor_m)
|
|
assert np.array_equal(copy.triangles, original.triangles)
|
|
assert np.array_equal(
|
|
np.asarray(rebuilt.binding.sample_indices),
|
|
np.asarray(scene.binding.sample_indices),
|
|
)
|
|
assert not np.asarray(rebuilt.binding.sample_weights).flags["WRITEABLE"]
|
|
assert [preset.name for preset in rebuilt.camera_presets] == [
|
|
preset.name for preset in scene.camera_presets
|
|
]
|
|
|
|
|
|
def test_the_loader_accepts_its_own_package_and_checks_the_device(package, scene):
|
|
loader = PackagedAssetLoader()
|
|
expected = ExpectedGeometry(
|
|
"finger", "rev-a", scene.geometry_id, scene.field_domain_id, scene.shape_hw
|
|
)
|
|
loaded = loader.load(package, expected)
|
|
assert loaded.geometry_id == scene.geometry_id
|
|
assert loader.validate(loaded, expected) == ()
|
|
wrong = ExpectedGeometry("finger", "rev-a", "other-geometry", scene.field_domain_id, scene.shape_hw)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
loader.load(package, wrong)
|
|
assert excinfo.value.diagnostic.code == "geometry_mismatch"
|
|
|
|
|
|
def test_the_package_report_describes_the_package(package):
|
|
report = describe_package(package)
|
|
assert report["asset_id"] == "poc-scene"
|
|
assert report["parts"] == {"sensing_surface": "sensing", "housing": "housing"}
|
|
assert report["shape_hw"] == [3, 3]
|
|
assert "sensing/positions.npy" in report["entries"]
|
|
assert report["vertices"]["sensing"] == 9
|
|
|
|
|
|
def test_a_housing_part_survives_packaging(tmp_path):
|
|
"""The housing has its own entries and keeps its role through the round trip."""
|
|
sensing_only = replace(synthetic.poc_scene(), parts=(synthetic.poc_scene().sensing_part(),))
|
|
housing = synthetic.poc_scene().part("housing")
|
|
scene = with_housing(sensing_only, housing)
|
|
path = tmp_path / "with-housing.oryscene"
|
|
write_package(scene, path)
|
|
rebuilt = scene_from_contents(read_package(path))
|
|
assert {part.role for part in rebuilt.parts} == {"sensing", "housing"}
|
|
assert np.array_equal(
|
|
rebuilt.part("housing").positions_sensor_m, housing.positions_sensor_m
|
|
)
|
|
assert rebuilt.part("housing").material.shading == "diffuse"
|
|
|
|
|
|
# --------------------------------------------------------------- refusals
|
|
def test_a_tampered_entry_is_refused(package, tmp_path):
|
|
def mutate(members):
|
|
members["sensing/positions.npy"] = b"not an array any more"
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "tampered.oryscene", mutate)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert excinfo.value.diagnostic.code == "asset_auth_failed"
|
|
|
|
|
|
def test_a_tampered_manifest_hash_is_refused(package, tmp_path):
|
|
def mutate(members):
|
|
manifest = _manifest(members)
|
|
manifest["geometry_id"] = "somebody-elses-geometry"
|
|
_store_manifest(members, manifest) # hash deliberately *not* recomputed
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "manifest.oryscene", mutate)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert "manifest hash does not match" in excinfo.value.diagnostic.message
|
|
|
|
|
|
def test_geometry_that_is_swapped_in_with_a_matching_hash_is_still_refused(package, tmp_path):
|
|
"""Consistency is not correctness: the binding must belong to the mesh."""
|
|
def mutate(members):
|
|
manifest = _manifest(members)
|
|
# Drop the sensing mesh to nine *different* positions but keep the binding,
|
|
# then fix every hash: the package is internally consistent and still wrong.
|
|
positions = np.zeros((9, 3), dtype=np.float32)
|
|
positions[:, 0] = np.arange(9, dtype=np.float32) * 0.01
|
|
import io
|
|
|
|
buffer = io.BytesIO()
|
|
np.save(buffer, positions, allow_pickle=False)
|
|
members["sensing/positions.npy"] = buffer.getvalue()
|
|
manifest["entries"]["sensing/positions.npy"]["sha256"] = content_digest(
|
|
{"data": positions}
|
|
)
|
|
_rehash(members, manifest)
|
|
return members
|
|
|
|
swapped = _rewrite(package, tmp_path / "swapped.oryscene", mutate)
|
|
contents = read_package(swapped) # hash-wise it is a valid package
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
scene_from_contents(contents)
|
|
assert excinfo.value.diagnostic.code == "geometry_mismatch"
|
|
assert "geometry" in excinfo.value.diagnostic.message.lower()
|
|
|
|
|
|
def test_non_finite_positions_are_refused(package, tmp_path):
|
|
def mutate(members):
|
|
import io
|
|
|
|
manifest = _manifest(members)
|
|
positions = np.zeros((9, 3), dtype=np.float32)
|
|
positions[0, 0] = np.nan
|
|
buffer = io.BytesIO()
|
|
np.save(buffer, positions, allow_pickle=False)
|
|
members["sensing/positions.npy"] = buffer.getvalue()
|
|
manifest["entries"]["sensing/positions.npy"]["sha256"] = content_digest(
|
|
{"data": positions}
|
|
)
|
|
_rehash(members, manifest)
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "nan.oryscene", mutate)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
scene_from_contents(read_package(broken))
|
|
assert "non-finite" in excinfo.value.diagnostic.message
|
|
|
|
|
|
def test_an_undeclared_entry_is_refused(package, tmp_path):
|
|
def mutate(members):
|
|
members["sensing/extra.npy"] = b"\x00"
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "extra.oryscene", mutate)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert "undeclared entries" in excinfo.value.diagnostic.message
|
|
|
|
|
|
def test_a_promised_entry_that_is_absent_is_refused(package, tmp_path):
|
|
def mutate(members):
|
|
members.pop("sensing/normals.npy")
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "absent.oryscene", mutate)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert "which is absent" in excinfo.value.diagnostic.message
|
|
|
|
|
|
def test_a_path_that_escapes_the_package_is_refused(package, tmp_path):
|
|
def mutate(members):
|
|
manifest = _manifest(members)
|
|
record = manifest["entries"].pop("sensing/normals.npy")
|
|
manifest["entries"]["../normals.npy"] = record
|
|
_rehash(members, manifest)
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "traversal.oryscene", mutate)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert "escapes the package" in excinfo.value.diagnostic.message
|
|
|
|
|
|
def test_a_dtype_or_shape_that_disagrees_with_the_manifest_is_refused(package, tmp_path):
|
|
for field, value, expected in (
|
|
("dtype", "i4", "field_domain_mismatch"),
|
|
("shape", [8, 3], "field_domain_mismatch"),
|
|
):
|
|
def mutate(members, field=field, value=value):
|
|
manifest = _manifest(members)
|
|
manifest["entries"]["sensing/positions.npy"][field] = value
|
|
_rehash(members, manifest)
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / f"bad-{field}.oryscene", mutate)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert excinfo.value.diagnostic.code == expected, field
|
|
|
|
|
|
def test_an_envelope_from_the_future_or_another_tool_is_refused(package, tmp_path):
|
|
def wrong_magic(members):
|
|
envelope = json.loads(members["envelope.json"])
|
|
envelope["magic"] = "not-our-magic"
|
|
members["envelope.json"] = json.dumps(envelope).encode()
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "magic.oryscene", wrong_magic)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert excinfo.value.diagnostic.code == "asset_auth_failed"
|
|
|
|
def wrong_version(members):
|
|
envelope = json.loads(members["envelope.json"])
|
|
envelope["envelope_version"] = 99
|
|
members["envelope.json"] = json.dumps(envelope).encode()
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "version.oryscene", wrong_version)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
assert "envelope version" in excinfo.value.diagnostic.message
|
|
|
|
|
|
def test_an_envelope_that_claims_an_unknown_payload_kind_is_refused(package, tmp_path):
|
|
"""An unknown payload is a refusal with its own reason, not a numpy error later."""
|
|
def encrypt(members):
|
|
envelope = json.loads(members["envelope.json"])
|
|
envelope["payload"] = "rot13"
|
|
envelope["key_id"] = "customer-1"
|
|
members["envelope.json"] = json.dumps(envelope).encode()
|
|
return members
|
|
|
|
broken = _rewrite(package, tmp_path / "encrypted.oryscene", encrypt)
|
|
with pytest.raises(ThreeDError) as excinfo:
|
|
read_package(broken)
|
|
item = excinfo.value.diagnostic
|
|
assert item.code == "asset_auth_failed"
|
|
assert "rot13" in item.message
|
|
|
|
|
|
def test_an_encrypted_payload_needs_a_key_provider_and_a_verification_key(scene, tmp_path):
|
|
"""The two ways a sealed package can be unreadable here, named separately.
|
|
|
|
Both are *licensing* states, not corruption: the message has to let an operator
|
|
tell "this machine has no key for it" from "these bytes were changed".
|
|
"""
|
|
crypto = pytest.importorskip(
|
|
"cryptography", reason="sealing needs the crypto backend"
|
|
)
|
|
del crypto
|
|
from orisys_app_shared.scene3d import protect as protect_module
|
|
from orisys_app_shared.scene3d.contracts import Sealing
|
|
|
|
private = protect_module.generate_signing_key()
|
|
public = protect_module.public_key_pem(private)
|
|
provider = protect_module.MappingKeyProvider({"customer-1": "11" * 32})
|
|
sealed = tmp_path / "sealed.oryscene"
|
|
write_package(
|
|
scene,
|
|
sealed,
|
|
sealing=Sealing(
|
|
key_id="customer-1",
|
|
provider=provider,
|
|
signing_key_pem=private,
|
|
signing_key_reference="build-key",
|
|
verify_key_pem=public,
|
|
),
|
|
)
|
|
|
|
with pytest.raises(ThreeDError) as no_key:
|
|
read_package(sealed, verify_key_pem=public)
|
|
assert "key provider" in no_key.value.diagnostic.message
|
|
|
|
with pytest.raises(ThreeDError) as no_verify:
|
|
read_package(sealed, keys=provider)
|
|
assert "verification key" in no_verify.value.diagnostic.message
|
|
|
|
contents = read_package(sealed, keys=provider, verify_key_pem=public)
|
|
assert contents.arrays["sensing/positions.npy"].shape[0] > 0
|
|
|
|
|
|
def test_a_key_provider_is_unused_by_a_plain_package(package, scene):
|
|
"""A provider is not an error on a plain package: it is simply not needed.
|
|
|
|
Refusing here would break the one call site that hands the same provider to every
|
|
load, which is how a client with a mix of sealed and plain packages is written.
|
|
"""
|
|
|
|
class Keys:
|
|
def get_content_key(self, key_id: str) -> bytes: # pragma: no cover - not used
|
|
raise AssertionError("a plain package must not ask for a key")
|
|
|
|
expected = ExpectedGeometry(
|
|
"finger", "rev-a", scene.geometry_id, scene.field_domain_id, scene.shape_hw
|
|
)
|
|
assets = PackagedAssetLoader().load(package, expected, Keys())
|
|
assert assets.geometry_id == scene.geometry_id
|
|
|
|
|
|
def test_the_manifest_carries_what_a_release_needs(package):
|
|
manifest = read_package(package).manifest
|
|
assert manifest["schema_version"] == "1.0"
|
|
assert manifest["coordinate_frame"] == "sensor"
|
|
assert manifest["length_unit"] == "m"
|
|
assert manifest["compatible"]["sensor_model"] == "finger"
|
|
assert manifest["field_domain"]["index_order"] == "row-major"
|
|
assert manifest["field_domain"]["origin"] == "top-left"
|
|
assert manifest["asset_hash"] and len(manifest["asset_hash"]) == 64
|
|
assert all(len(record["sha256"]) == 64 for record in manifest["entries"].values())
|
|
|
|
|
|
def test_the_same_scene_compiles_to_the_same_hash(tmp_path, scene):
|
|
first = write_package(scene, tmp_path / "a.oryscene")
|
|
second = write_package(scene, tmp_path / "b.oryscene")
|
|
assert first.asset_hash == second.asset_hash
|
|
assert write_package(scene, tmp_path / "a.oryscene", sensor_model="box").asset_hash != (
|
|
first.asset_hash
|
|
), "a different target device is a different package"
|
|
|
|
|
|
def test_writing_is_atomic_so_a_failed_build_leaves_nothing(tmp_path, scene, monkeypatch):
|
|
"""A half-written package must never be picked up by a later run."""
|
|
import orisys_app_shared.scene3d.package as package_module
|
|
|
|
target = tmp_path / "broken.oryscene"
|
|
|
|
def explode(*args, **kwargs):
|
|
raise RuntimeError("disk full")
|
|
|
|
monkeypatch.setattr(package_module.zipfile, "ZipFile", explode)
|
|
with pytest.raises(RuntimeError):
|
|
write_package(scene, target)
|
|
assert not target.exists()
|
|
assert not (tmp_path / "broken.oryscene.partial").exists(), (
|
|
"the partial file was left behind"
|
|
)
|
|
|
|
|
|
def test_the_envelope_names_the_magic_and_the_payload_state(package):
|
|
with zipfile.ZipFile(package) as archive:
|
|
envelope = json.loads(archive.read("envelope.json"))
|
|
assert envelope["magic"] == MAGIC
|
|
# An unsealed write stays at version 1: it is the package every release before
|
|
# stage 5 produced, and bumping it would make old readers refuse a plain file.
|
|
assert envelope["envelope_version"] == 1
|
|
assert envelope["payload"] == "plain"
|
|
assert envelope["entry_count"] == len(read_package(package).arrays)
|
|
assert "signature" not in envelope, "a plain package must not look signed"
|
|
|
|
|
|
# ------------------------------------------- the device-verifiable geometry record
|
|
def test_the_source_geometry_identity_round_trips(tmp_path, scene):
|
|
"""The record a device can reproduce without the operator's unit statement."""
|
|
from dataclasses import replace
|
|
|
|
marked = replace(scene, source_geometry_id="src-46a03a5f")
|
|
path = tmp_path / "marked.oryscene"
|
|
write_package(marked, path)
|
|
manifest = read_package(path).manifest
|
|
assert manifest["source_geometry"]["id"] == "src-46a03a5f"
|
|
assert manifest["source_geometry"]["unit"] == "source"
|
|
rebuilt = scene_from_contents(read_package(path))
|
|
assert rebuilt.source_geometry_id == "src-46a03a5f"
|
|
assert rebuilt.compatible_sensor_model == "finger"
|
|
assert rebuilt.compatible_hardware_revisions == ("*",)
|
|
|
|
|
|
def test_a_package_without_the_record_reads_as_unstated(tmp_path, scene):
|
|
"""An older package must not look as if its geometry had been checked."""
|
|
path = tmp_path / "plain.oryscene"
|
|
write_package(scene, path) # the synthetic scene states no source geometry
|
|
manifest = read_package(path).manifest
|
|
assert manifest["source_geometry"] is None
|
|
rebuilt = scene_from_contents(read_package(path))
|
|
assert rebuilt.source_geometry_id == ""
|
|
assert rebuilt.compatible_sensor_model == "finger"
|