orisys-app/shared/tests/test_scene3d_package.py

"""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"