dsh/dual_camera_system/tests/test_scan_point_store.py

import json
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import pytest

from core.scan_point_store import ScanPointStore


def test_add_and_list_points(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("group_1", {"ptz_name": "PTZ1", "panorama_name": "PAN1"})
    store.add_enabled_point("group_1", pan=30.0, tilt=0.0, zoom=1, dwell_time=3.0)
    points = store.list_enabled_points("group_1")
    assert len(points) == 1
    assert points[0]["pan"] == 30.0


def test_persistence(tmp_path):
    path = str(tmp_path / "scan_models.json")
    store = ScanPointStore(path)
    store.ensure_group("group_1", {"ptz_name": "PTZ1", "panorama_name": "PAN1"})
    store.add_enabled_point("group_1", pan=60.0, tilt=-5.0, zoom=2, dwell_time=2.0)
    del store

    store2 = ScanPointStore(path)
    points = store2.list_enabled_points("group_1")
    assert len(points) == 1
    assert points[0]["zoom"] == 2


def test_delete_point(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p1 = store.add_enabled_point("g1", pan=10.0, tilt=0.0)
    store.add_enabled_point("g1", pan=20.0, tilt=0.0)
    assert len(store.list_enabled_points("g1")) == 2
    ok = store.delete_enabled_point("g1", p1["id"])
    assert ok
    points = store.list_enabled_points("g1")
    assert len(points) == 1
    assert points[0]["order"] == 0


def test_returns_deep_copies(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {"ptz_name": "PTZ1"})
    point = store.add_enabled_point("g1", pan=10.0, tilt=0.0)
    point["pan"] = 999.0
    points = store.list_enabled_points("g1")
    assert points[0]["pan"] == 10.0

    group = store.get_group("g1")
    group["ptz_name"] = "CHANGED"
    assert store.get_group("g1")["ptz_name"] == "PTZ1"

    points[0]["tilt"] = 999.0
    assert store.list_enabled_points("g1")[0]["tilt"] == 0.0


def test_reorder_points_exact_ids(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p1 = store.add_enabled_point("g1", pan=10.0, tilt=0.0)
    p2 = store.add_enabled_point("g1", pan=20.0, tilt=0.0)
    p3 = store.add_enabled_point("g1", pan=30.0, tilt=0.0)

    original_order = [p["id"] for p in store.list_enabled_points("g1")]
    assert original_order == [p1["id"], p2["id"], p3["id"]]

    reversed_order = [p3["id"], p2["id"], p1["id"]]
    store.reorder_points("g1", reversed_order)
    reordered = store.list_enabled_points("g1")
    assert [p["id"] for p in reordered] == reversed_order
    for idx, p in enumerate(reordered):
        assert p["order"] == idx


def test_reorder_points_rejects_mismatched_ids(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p1 = store.add_enabled_point("g1", pan=10.0, tilt=0.0)
    p2 = store.add_enabled_point("g1", pan=20.0, tilt=0.0)

    with pytest.raises(ValueError, match="point_ids must contain exactly"):
        store.reorder_points("g1", [p1["id"]])

    with pytest.raises(ValueError, match="point_ids must contain exactly"):
        store.reorder_points("g1", [p1["id"], p2["id"], 999])

    with pytest.raises(ValueError, match="point_ids must contain exactly"):
        store.reorder_points("g1", [p1["id"], p1["id"]])


def test_update_enabled_point(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p = store.add_enabled_point("g1", pan=10.0, tilt=0.0, zoom=1, dwell_time=3.0)
    ok = store.update_enabled_point("g1", p["id"], {"pan": 45.0, "zoom": 5})
    assert ok
    updated = store.list_enabled_points("g1")[0]
    assert updated["pan"] == 45.0
    assert updated["zoom"] == 5
    assert updated["tilt"] == 0.0


def test_invalid_inputs_raise(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})

    with pytest.raises(ValueError):
        store.add_enabled_point("g1", pan="bad", tilt=0.0)

    with pytest.raises(ValueError):
        store.add_enabled_point("g1", pan=10.0, tilt=0.0, zoom=0)

    with pytest.raises(ValueError):
        store.add_enabled_point("g1", pan=10.0, tilt=0.0, dwell_time=0)

    p = store.add_enabled_point("g1", pan=10.0, tilt=0.0)

    with pytest.raises(ValueError):
        store.update_enabled_point("g1", p["id"], {"pan": "bad"})

    with pytest.raises(ValueError):
        store.update_enabled_point("g1", p["id"], {"zoom": -1})

    with pytest.raises(ValueError):
        store.update_enabled_point("g1", p["id"], {"dwell_time": -1.0})


def test_corrupted_json_starts_fresh(tmp_path, capsys):
    path = str(tmp_path / "scan_models.json")
    with open(path, "w", encoding="utf-8") as f:
        f.write("{not valid json")

    store = ScanPointStore(path)
    captured = capsys.readouterr()
    assert "corrupted JSON" in captured.err
    store.ensure_group("g1", {"ptz_name": "PTZ1"})
    assert store.get_group("g1")["ptz_name"] == "PTZ1"


def test_point_id_generation(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p1 = store.add_enabled_point("g1", pan=10.0, tilt=0.0)
    p2 = store.add_enabled_point("g1", pan=20.0, tilt=0.0)
    store.delete_enabled_point("g1", p1["id"])
    p3 = store.add_enabled_point("g1", pan=30.0, tilt=0.0)
    assert p1["id"] < p2["id"] < p3["id"]
    assert p3["id"] == p2["id"] + 1


def test_set_samples_deep_copies(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    samples = [{"x": 1, "nested": {"y": 2}}]
    store.set_samples("g1", samples)
    samples[0]["nested"]["y"] = 999
    assert store.get_group("g1")["samples"][0]["nested"]["y"] == 2


def test_set_scan_config_deep_copies(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    config = {"threshold": {"value": 0.5}}
    store.set_scan_config("g1", config)
    config["threshold"]["value"] = 0.9
    assert store.get_group("g1")["scan_config"]["threshold"]["value"] == 0.5


def test_set_panorama_deep_copies(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    panorama = {"image": {"width": 1920}}
    store.set_panorama("g1", panorama)
    panorama["image"]["width"] = 999
    assert store.get_group("g1")["panorama"]["image"]["width"] == 1920


def test_next_point_id_seeded_from_legacy_points(tmp_path):
    path = str(tmp_path / "scan_models.json")
    legacy_data = {
        "camera_groups": {
            "g1": {
                "ptz_name": "g1",
                "panorama_name": "g1",
                "scan_config": {},
                "samples": [],
                "enabled_points": [
                    {"id": 5, "pan": 10.0, "tilt": 0.0, "zoom": 1, "dwell_time": 3.0, "order": 0},
                    {"id": 12, "pan": 20.0, "tilt": 0.0, "zoom": 1, "dwell_time": 3.0, "order": 1},
                ],
                "panorama": {},
            }
        }
    }
    with open(path, "w", encoding="utf-8") as f:
        json.dump(legacy_data, f)

    store = ScanPointStore(path)
    store.ensure_group("g1", {})
    p = store.add_enabled_point("g1", pan=30.0, tilt=0.0)
    assert p["id"] == 13


def test_reject_boolean_values(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})

    with pytest.raises(ValueError):
        store.add_enabled_point("g1", pan=True, tilt=0.0)

    with pytest.raises(ValueError):
        store.add_enabled_point("g1", pan=10.0, tilt=False)

    with pytest.raises(ValueError):
        store.add_enabled_point("g1", pan=10.0, tilt=0.0, zoom=True)

    with pytest.raises(ValueError):
        store.add_enabled_point("g1", pan=10.0, tilt=0.0, dwell_time=True)

    p = store.add_enabled_point("g1", pan=10.0, tilt=0.0)

    with pytest.raises(ValueError):
        store.update_enabled_point("g1", p["id"], {"pan": True})

    with pytest.raises(ValueError):
        store.update_enabled_point("g1", p["id"], {"tilt": False})

    with pytest.raises(ValueError):
        store.update_enabled_point("g1", p["id"], {"zoom": True})

    with pytest.raises(ValueError):
        store.update_enabled_point("g1", p["id"], {"dwell_time": True})


def test_update_unknown_key_raises(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p = store.add_enabled_point("g1", pan=10.0, tilt=0.0)

    with pytest.raises(ValueError, match="Unknown field: bad_key"):
        store.update_enabled_point("g1", p["id"], {"bad_key": 123})


def test_update_rejects_boolean_order(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p = store.add_enabled_point("g1", pan=10.0, tilt=0.0)

    with pytest.raises(ValueError, match="order must be an integer"):
        store.update_enabled_point("g1", p["id"], {"order": True})


def test_update_rejects_non_finite_values(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p = store.add_enabled_point("g1", pan=10.0, tilt=0.0)

    with pytest.raises(ValueError, match="pan must be finite"):
        store.update_enabled_point("g1", p["id"], {"pan": float("nan")})

    with pytest.raises(ValueError, match="tilt must be finite"):
        store.update_enabled_point("g1", p["id"], {"tilt": float("inf")})

    with pytest.raises(ValueError, match="dwell_time must be finite"):
        store.update_enabled_point("g1", p["id"], {"dwell_time": float("-inf")})


def test_update_rejects_negative_order(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})
    p = store.add_enabled_point("g1", pan=10.0, tilt=0.0)

    with pytest.raises(ValueError, match="order must be non-negative"):
        store.update_enabled_point("g1", p["id"], {"order": -1})


def test_add_rejects_non_finite_values(tmp_path):
    store = ScanPointStore(str(tmp_path / "scan_models.json"))
    store.ensure_group("g1", {})

    with pytest.raises(ValueError, match="pan must be finite"):
        store.add_enabled_point("g1", pan=float("nan"), tilt=0.0)

    with pytest.raises(ValueError, match="dwell_time must be finite"):
        store.add_enabled_point("g1", pan=10.0, tilt=0.0, dwell_time=float("inf"))