NODEDC_MISSION_CORE/tests/test_insta360_control.py

"""Camera command boundaries with a synthetic camera; never load a vendor SDK."""

import importlib.util
import json
import sys
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1] / "plugins/insta360-x4/runtime"
spec = importlib.util.spec_from_file_location("missioncore_insta360", ROOT / "__init__.py")
package = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = package
spec.loader.exec_module(package)
from missioncore_insta360 import identity, native, operations  # noqa: E402

ID_A = "instax4_" + "a" * 32
ID_B = "instax4_" + "b" * 32


def command(action="record.start", *, device=ID_A, session="session_test_a", params=None):
    now = datetime.now(UTC)
    op = "op_" + uuid.uuid4().hex
    return {
        "api_version": "missioncore.nodedc/plugin-sdk/v0alpha2",
        "kind": "OperationRequest",
        "operation_id": op,
        "idempotency_key": op,
        "session": {"session_id": session, "device_id": device},
        "action_id": action,
        "requested_at": now.isoformat(),
        "deadline_at": (now + timedelta(seconds=60)).isoformat(),
        "parameters": params or {},
    }


class Camera:
    def __init__(self):
        self.calls = []
        self.answer = {"state": "complete", "result": {"recording": True}}
        self.hook = None

    def call(self, action, params):
        self.calls.append((action, params))
        if self.hook:
            self.hook()
        return self.answer


def test_concurrent_remote_and_local_retry_dispatches_once(tmp_path):
    camera = Camera()
    executor = operations.Operations(ID_A, "session_test_a", tmp_path, camera)
    request = command()
    entered, release = threading.Event(), threading.Event()

    def inspect_receipt():
        receipt = json.loads((tmp_path / (request["operation_id"] + ".json")).read_text())
        assert receipt["result"]["state"] == "unknown"
        entered.set()
        assert release.wait(3)

    camera.hook = inspect_receipt
    with ThreadPoolExecutor(max_workers=2) as pool:
        first = pool.submit(executor.execute, request)
        assert entered.wait(3)
        second = pool.submit(executor.execute, request)
        release.set()
        assert first.result() == second.result() == camera.answer
    assert len(camera.calls) == 1


def test_crash_receipt_survives_restart_without_replaying_recording(tmp_path):
    camera = Camera()

    def crash():
        raise SystemExit("synthetic crash after dispatch")

    camera.hook = crash
    request = command()
    with pytest.raises(SystemExit):
        operations.Operations(ID_A, "session_test_a", tmp_path, camera).execute(request)
    fresh = Camera()
    recovered = operations.Operations(ID_A, "session_after_restart", tmp_path, fresh)
    assert recovered.execute(request)["state"] == "unknown"
    assert not fresh.calls


def test_camera_binding_session_deadline_and_parameters_precede_effects(tmp_path):
    camera = Camera()
    executor = operations.Operations(ID_A, "session_test_a", tmp_path, camera)
    wrong = [
        command(device=ID_B),
        command(session="session_old"),
        command(params={"force": True}),
        command("settings.apply", params={"mode": 7, "key": "library_path", "value": 1}),
        command("settings.apply", params={"mode": 7, "key": "iso", "value": True}),
    ]
    expired = command()
    expired["requested_at"] = (datetime.now(UTC) - timedelta(minutes=2)).isoformat()
    expired["deadline_at"] = (datetime.now(UTC) - timedelta(minutes=1)).isoformat()
    wrong.append(expired)
    for request in wrong:
        with pytest.raises(ValueError):
            executor.execute(request)
    assert not camera.calls
    assert not list(tmp_path.iterdir())


def test_operation_identity_cannot_be_reused_for_another_effect(tmp_path):
    camera = Camera()
    executor = operations.Operations(ID_A, "session_test_a", tmp_path, camera)
    request = command()
    executor.execute(request)
    request["action_id"] = "record.stop"
    with pytest.raises(ValueError, match="already in use"):
        executor.execute(request)
    assert len(camera.calls) == 1


def test_two_instances_keep_receipts_and_recording_independent(tmp_path):
    a, b = Camera(), Camera()
    first = operations.Operations(ID_A, "session_test_a", tmp_path / "a", a)
    second = operations.Operations(ID_B, "session_test_b", tmp_path / "b", b)
    first.execute(command())
    second.execute(command("preview.stop", device=ID_B, session="session_test_b"))
    assert [action for action, _ in a.calls] == ["record.start"]
    assert [action for action, _ in b.calls] == ["preview.stop"]


def test_setting_acknowledgement_requires_camera_readback(tmp_path):
    camera = Camera()
    camera.answer = {"state": "complete", "result": {"values": {"white_balance": 0}}}
    executor = operations.Operations(ID_A, "session_test_a", tmp_path, camera)
    request = command("settings.apply", params={"mode": 7, "key": "white_balance", "value": 5000})
    assert executor.execute(request)["state"] == "unknown"
    camera.answer = {"state": "complete", "result": {"values": {"white_balance": 5000}}}
    # The old command remains uncertain. Only a new explicit command may run.
    assert executor.execute(request)["state"] == "unknown"
    assert len(camera.calls) == 1
    assert (
        executor.execute(command("settings.apply", params=request["parameters"]))["state"]
        == "complete"
    )


def test_failed_journal_prevents_sdk_dispatch(tmp_path, monkeypatch):
    camera = Camera()

    def full(*_):
        raise OSError("synthetic disk full")

    monkeypatch.setattr(operations, "atomic", full)
    with pytest.raises(OSError):
        operations.Operations(ID_A, "session_test_a", tmp_path, camera).execute(command())
    assert not camera.calls


def test_model_binding_uses_product_and_tracks_usb_replug(tmp_path):
    usb = tmp_path / "1-2.3"
    usb.mkdir()
    values = {
        "idVendor": "2e1a",
        "idProduct": "0002",
        "product": "Insta360 X4",
        "serial": "SYNTHETIC-X4-A",
        "busnum": "1",
        "devnum": "7",
    }
    for key, value in values.items():
        (usb / key).write_text(value)
    first = identity.read_binding("1-2.3", tmp_path)
    (usb / "devnum").write_text("8")
    second = identity.read_binding("1-2.3", tmp_path)
    assert first.device_id == second.device_id and first != second
    assert second.device_path == "/dev/bus/usb/001/008"
    (usb / "product").write_text("Insta360 ONE R")
    with pytest.raises(ValueError, match="model"):
        identity.read_binding("1-2.3", tmp_path)
    with pytest.raises(ValueError, match="binding"):
        identity.read_binding("../1-2.3", tmp_path)


def test_sdk_library_is_not_loaded_before_isolation_check(monkeypatch):
    calls = []

    def reject(_):
        raise RuntimeError("synthetic namespace refusal")

    monkeypatch.setattr(native, "verify_isolation", reject)
    monkeypatch.setattr(native.ctypes, "CDLL", lambda value: calls.append(value))
    with pytest.raises(RuntimeError, match="namespace"):
        native.NativeCamera(object(), "/unused/library.so", "/unused/logs")
    assert not calls


def test_video_queue_can_drain_while_a_camera_command_is_blocked():
    # Construct only the Python transport with a fake C ABI. No isolation
    # override or vendor library is used by the real NativeCamera constructor.
    camera = native.NativeCamera.__new__(native.NativeCamera)
    camera.lock, camera.video_lock = threading.RLock(), threading.Lock()
    camera.handle, camera.buffer = 1, object()

    class API:
        def mc_x4_read_video(self, *_):
            return 0

    camera.api = API()
    with ThreadPoolExecutor(max_workers=1) as pool, camera.lock:
        assert pool.submit(camera.video).result(timeout=2) is None