"""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, describe_package, read_package, scene_from_contents, write_package, ) from orisys_app_shared.scene3d.package import PackagedAssetLoader @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.contracts import Sealing from orisys_app_shared.scene3d import protect as protect_module 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"