"""Tests for the sensor bundle contract: descriptor, resolver, registry, fingerprints. Each test states the property it protects, because the point of this layer is that a product name stops deciding which pipeline runs. The bundles built here are small: a resolver test does not need a real camera configuration, only the specific files the rule under test reads. """ from __future__ import annotations import json import shutil from pathlib import Path import pytest from orisys.sensors import ( BACKEND_COMPOUND_EYE, BACKEND_SINGLE_CAMERA, BACKEND_UV_UNROLL, BUILT_IN_PROFILES, CalibrationSelection, DescriptorError, ProcessorCreationError, backend_from_config, create_processor, known_backends, parse_descriptor, resolve_bundle, spec_for, validate_profile, ) REPO_ROOT = Path(__file__).resolve().parents[1] SDK_SENSORS = REPO_ROOT / "sensors" MINIMAL_CONFIG = { "name": "prototype", "version": "1.0", "type": "compound_eye", "camera": {"resolution_stitch": [64, 64], "fps": 30}, "stitcher": {"index": [0]}, "tracker": {"algorithm": "dis"}, "contact": {}, } def _write_bundle( root: Path, *, descriptor: dict | None = None, config: dict | None = None, extra: dict[str, bytes] | None = None, legacy_manifest: dict | None = None, ) -> Path: """A bundle folder: ``sensor.json`` plus whichever manifest shape is asked for.""" root.mkdir(parents=True, exist_ok=True) (root / "sensor.json").write_text( json.dumps(config if config is not None else MINIMAL_CONFIG), encoding="utf-8" ) if descriptor is not None: (root / "manifest.json").write_text(json.dumps(descriptor), encoding="utf-8") elif legacy_manifest is not None: (root / "manifest.json").write_text( json.dumps(legacy_manifest), encoding="utf-8" ) for relative, payload in (extra or {}).items(): target = root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(payload) return root def _descriptor(**overrides) -> dict: payload = { "schema_version": 1, "profile_id": "test.profile", "display_name": "Test Profile", "product_family": "mini", "processing_backend": BACKEND_COMPOUND_EYE, "sensor_json": "sensor.json", } payload.update(overrides) return payload # ------------------------------------------------------------------ descriptor def test_descriptor_is_required_to_identify_an_unknown_folder(tmp_path: Path): """A folder name is not evidence of a backend.""" root = _write_bundle(tmp_path / "mystery") with pytest.raises(DescriptorError) as excinfo: resolve_bundle(root) assert "not a built-in bundle" in str(excinfo.value) def test_descriptor_rejects_unknown_backend_and_schema_version(tmp_path: Path): root = _write_bundle( tmp_path / "b", descriptor=_descriptor(processing_backend="nonsense") ) with pytest.raises(DescriptorError) as excinfo: parse_descriptor(root) assert "unsupported processing_backend" in str(excinfo.value) root2 = _write_bundle(tmp_path / "c", descriptor=_descriptor(schema_version=99)) with pytest.raises(DescriptorError) as excinfo2: parse_descriptor(root2) assert "unsupported descriptor schema_version" in str(excinfo2.value) def test_descriptor_rejects_paths_that_leave_the_bundle(tmp_path: Path): root = _write_bundle( tmp_path / "d", descriptor=_descriptor(computational_assets={"map": "../outside.npz"}), ) with pytest.raises(DescriptorError) as excinfo: parse_descriptor(root) assert "escapes the bundle folder" in str(excinfo.value) root2 = _write_bundle( tmp_path / "e", descriptor=_descriptor(computational_assets={"map": str(tmp_path / "abs.npz")}), ) with pytest.raises(DescriptorError) as excinfo2: parse_descriptor(root2) assert "absolute" in str(excinfo2.value) def test_descriptor_round_trips_with_relative_paths(tmp_path: Path): root = _write_bundle( tmp_path / "rt", descriptor=_descriptor( computational_assets={"map": "assets/map.npz"}, assets={"scene": "scene/x.oryscene"}, ), extra={"assets/map.npz": b"payload", "scene/x.oryscene": b"scene"}, ) descriptor = parse_descriptor(root) payload = descriptor.as_dict() assert payload["computational_assets"] == {"map": "assets/map.npz"} assert payload["assets"] == {"scene": "scene/x.oryscene"} assert not Path(payload["sensor_json"]).is_absolute() # ------------------------------------------------------------------ legacy adapter @pytest.mark.parametrize("folder", sorted(BUILT_IN_PROFILES)) def test_every_built_in_resolves_through_compatibility(folder: str): """The acceptance rule: existing supported bundle formats keep loading.""" root = SDK_SENSORS / folder if not root.is_dir(): pytest.skip(f"{folder} is not installed in this checkout") profile = resolve_bundle(root) assert profile.product_family == BUILT_IN_PROFILES[folder]["product_family"] assert profile.processing_backend in known_backends() assert profile.validation.state in ("ok", "degraded", "failed") assert profile.processing_fingerprint def test_backend_follows_the_configuration_not_the_folder_name(): """finger's sensor.json says single_camera; its backend comes from the SDK's own rule.""" assert backend_from_config({"type": "single_camera"}) == BACKEND_SINGLE_CAMERA assert backend_from_config({"stitcher": {}}) == BACKEND_COMPOUND_EYE assert backend_from_config({}) == BACKEND_SINGLE_CAMERA assert backend_from_config({"type": "uv_unroll"}) == BACKEND_UV_UNROLL with pytest.raises(DescriptorError): backend_from_config({"type": "made_up"}) def test_single_camera_is_never_read_as_a_product_family(): """``single_camera`` describes the image arrangement, not what the product is.""" profile = resolve_bundle(SDK_SENSORS / "finger") assert profile.product_family == "finger" assert profile.processing_backend == BACKEND_UV_UNROLL def test_legacy_manifest_is_read_by_the_adapter_not_the_strict_parser(tmp_path: Path): """finger ships an old-shape manifest; it must not be rejected for lacking schema_version.""" root = _write_bundle( tmp_path / "finger", legacy_manifest={ "name": "finger", "sensor_json": "sensor.json", "assets_dir": "assets/30fps-0703", "viz_pose": "assets/visualization/world_to_cam.npy", }, config={**MINIMAL_CONFIG, "type": "uv_unroll"}, extra={"assets/visualization/world_to_cam.npy": b"pose"}, ) profile = resolve_bundle(root) assert profile.product_family == "finger" # Declared uv_unroll without the map: an error, not a silent downgrade. assert profile.validation.state == "failed" assert any( issue.code == "assets.computational-undeclared" for issue in profile.validation.errors ) # The viz pose is picked up as a visualization asset, never as computational. assert "viz_pose" in profile.assets assert "viz_pose" not in profile.computational_assets # ------------------------------------------------------------------ backend registry def test_registry_refuses_an_unknown_identifier(): with pytest.raises(ProcessorCreationError) as excinfo: spec_for("nonsense") assert "unsupported processing backend" in str(excinfo.value) def test_mini_uses_its_supported_backend_without_pretending_to_be_box(): profile = resolve_bundle(SDK_SENSORS / "mini") assert profile.product_family == "mini" assert profile.processing_backend == BACKEND_COMPOUND_EYE assert profile.profile_id != resolve_bundle(SDK_SENSORS / "box").profile_id def test_finger_mems_is_assessed_from_its_configuration(tmp_path: Path): """Its name must not imply MEMS acquisition support. The studio-side ``finger-mems`` bundle copies ``box``'s configuration, so the folder name is the only thing that makes it look special. Its backend has to come from that configuration. """ config = {**MINIMAL_CONFIG, "type": "compound_eye", "stitcher": {"index": [0]}} root = _write_bundle(tmp_path / "finger-mems", config=config) profile = resolve_bundle(root) assert profile.product_family == "finger-mems" assert profile.processing_backend == BACKEND_COMPOUND_EYE # No calibration and no metadata: raw yes, calibrated no, per axis. assert profile.capabilities.raw_force.available assert not profile.capabilities.calibrated_force.any_available() assert profile.capabilities.calibrated_force.fn.reason # A UV backend is never inferred from the name. assert profile.processing_backend != BACKEND_UV_UNROLL def test_create_processor_names_what_is_missing_rather_than_faking_it(): """uv_unroll needs a UV domain assembled outside a config-driven Sensor.""" profile = resolve_bundle(SDK_SENSORS / "finger") with pytest.raises(ProcessorCreationError) as excinfo: create_processor(profile) assert "uv_unroll" in str(excinfo.value) # ------------------------------------------------------------------ capabilities def test_missing_optional_scene_does_not_block_processing(tmp_path: Path): root = _write_bundle( tmp_path / "sceneless", descriptor=_descriptor( computational_assets={"map": "assets/map.npz"}, assets={"scene": "scene/absent.oryscene"}, ), extra={"assets/map.npz": b"map"}, ) profile = resolve_bundle(root) assert profile.validation.usable assert profile.capabilities.video_processing.available assert profile.capabilities.optical_flow.available assert not profile.capabilities.visualization_3d.available assert "missing" in profile.capabilities.visualization_3d.reason def test_missing_computational_asset_fails_the_profile(tmp_path: Path): root = _write_bundle( tmp_path / "nogeom", descriptor=_descriptor(computational_assets={"map": "assets/absent.npz"}), ) profile = resolve_bundle(root) assert profile.validation.state == "failed" assert any( issue.code == "assets.computational-missing" for issue in profile.validation.errors ) def test_single_camera_reports_no_force_capability(tmp_path: Path): config = {key: value for key, value in MINIMAL_CONFIG.items() if key != "stitcher"} config["type"] = "single_camera" root = _write_bundle( tmp_path / "mono", descriptor=_descriptor( processing_backend=BACKEND_SINGLE_CAMERA, product_family="custom" ), config=config, ) profile = resolve_bundle(root) assert profile.capabilities.video_processing.available assert not profile.capabilities.raw_force.available assert "single_camera" in profile.capabilities.raw_force.reason # ------------------------------------------------------------------ calibration def test_explicit_none_is_a_choice_not_a_failure(tmp_path: Path): root = _write_bundle(tmp_path / "nocal", descriptor=_descriptor()) profile = resolve_bundle(root, calibration_selection=CalibrationSelection.NONE) assert profile.calibration.selection is CalibrationSelection.NONE assert profile.calibration_fingerprint == "none" assert not profile.capabilities.calibrated_force.any_available() assert all( "explicitly uncalibrated" in axis.reason for axis in ( profile.capabilities.calibrated_force.fn, profile.capabilities.calibrated_force.fx, profile.capabilities.calibrated_force.fy, ) ) def test_an_unreadable_selected_calibration_is_an_error_not_a_substitution(tmp_path: Path): root = _write_bundle(tmp_path / "badcal", descriptor=_descriptor()) profile = resolve_bundle( root, calibration_selection=str(tmp_path / "does-not-exist.json") ) assert any( issue.code == "calibration.missing" for issue in profile.validation.errors ) assert profile.calibration.document is None assert profile.validation.state == "failed" def test_legacy_calibration_without_metadata_reports_unknown_applicability(): """box ships a calibration with no identifying metadata; it applies, but is flagged.""" root = SDK_SENSORS / "box" if not (root / "force_calibration.json").is_file(): pytest.skip("box ships no calibration in this checkout") profile = resolve_bundle(root) assert profile.calibration.applied assert profile.calibration.applicability == "unknown" assert any( issue.code == "calibration.applicability-unknown" for issue in profile.validation.warnings ) def test_calibrated_force_is_reported_per_axis(tmp_path: Path): calibration = { "schema_version": 2, "unit": "N", "channels": { "normal": {"method": "piecewise_linear", "points": [[0.0, 0.0], [10.0, 1.0]]} }, } root = _write_bundle( tmp_path / "fnonly", descriptor=_descriptor(default_calibration="force_calibration.json"), extra={"force_calibration.json": json.dumps(calibration).encode()}, ) profile = resolve_bundle(root) axes = profile.capabilities.calibrated_force assert axes.fn.available assert not axes.fx.available assert not axes.fy.available assert "no fn curve" not in axes.fx.detail # the detail names the axis that is missing assert "no fx curve" in axes.fx.detail # ------------------------------------------------------------------ fingerprints def test_relocating_a_bundle_preserves_the_processing_fingerprint(tmp_path: Path): """Moving or renaming must not change what would be computed.""" first = _write_bundle( tmp_path / "one" / "mini", descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}), extra={"assets/map.npz": b"shared-payload"}, ) second = tmp_path / "elsewhere" / "renamed-profile" shutil.copytree(first, second) a = resolve_bundle(first) b = resolve_bundle(second) assert a.processing_fingerprint == b.processing_fingerprint assert a.bundle_root != b.bundle_root def test_changing_a_computational_asset_changes_the_fingerprint(tmp_path: Path): root = _write_bundle( tmp_path / "asset", descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}), extra={"assets/map.npz": b"first"}, ) before = resolve_bundle(root).processing_fingerprint (root / "assets" / "map.npz").write_bytes(b"second") after = resolve_bundle(root).processing_fingerprint assert before != after def test_display_name_and_visualization_assets_stay_out_of_the_fingerprint(tmp_path: Path): """A cosmetic change must not invalidate a comparable record.""" def build(name: str, **descriptor_overrides) -> Path: overrides = {"computational_assets": {"map": "assets/map.npz"}} overrides.update(descriptor_overrides) return _write_bundle( tmp_path / name, descriptor=_descriptor(**overrides), extra={"assets/map.npz": b"map"}, ) plain = build("plain", display_name="One") fancy = build( "fancy", display_name="A Much Nicer Name", assets={"scene": "scene/absent.oryscene"}, ) assert ( resolve_bundle(plain).processing_fingerprint == resolve_bundle(fancy).processing_fingerprint ) def test_changing_the_configuration_changes_the_fingerprint(tmp_path: Path): root = _write_bundle(tmp_path / "cfg", descriptor=_descriptor()) before = resolve_bundle(root).processing_fingerprint changed = json.loads((root / "sensor.json").read_text(encoding="utf-8")) changed["tracker"]["algorithm"] = "raft" (root / "sensor.json").write_text(json.dumps(changed), encoding="utf-8") after = resolve_bundle(root).processing_fingerprint assert before != after def test_renaming_the_config_does_not_change_the_fingerprint(tmp_path: Path): """``name``/``version`` identify the document; they do not change the numbers.""" root = _write_bundle(tmp_path / "named", descriptor=_descriptor()) before = resolve_bundle(root).processing_fingerprint renamed = json.loads((root / "sensor.json").read_text(encoding="utf-8")) renamed["name"] = "something-else" renamed["version"] = "9.9" (root / "sensor.json").write_text(json.dumps(renamed), encoding="utf-8") assert resolve_bundle(root).processing_fingerprint == before def test_validate_profile_flags_assets_that_changed_since_resolution(tmp_path: Path): root = _write_bundle( tmp_path / "stale", descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}), extra={"assets/map.npz": b"original"}, ) profile = resolve_bundle(root) assert validate_profile(profile).state in ("ok", "degraded") (root / "assets" / "map.npz").write_bytes(b"tampered") report = validate_profile(profile) assert any(issue.code == "fingerprint.stale" for issue in report.errors) def test_inventory_is_bundle_relative(tmp_path: Path): profile = resolve_bundle(SDK_SENSORS / "box") inventory = profile.inventory() assert inventory["profile_id"] == profile.profile_id assert inventory["processing_fingerprint"] == profile.processing_fingerprint assert not str(inventory["sensor_config"]).startswith(str(profile.bundle_root)) assert "/" in inventory["sensor_config"] or "\\" in inventory["sensor_config"] or inventory["sensor_config"]