Presa-Mano/tests/test_open3dmodel_client.py

"""Unit tests for Open3DModelClient."""
from __future__ import annotations

from unittest.mock import patch

import httpx
import numpy as np
import pytest

from presa_mano.clients.open3dmodel import (
    Open3DModelClient,
    Open3DModelError,
    Open3DModelUnavailable,
)


def _intrinsics():
    return {"fx": 600.0, "fy": 600.0, "cx": 320.0, "cy": 240.0,
            "width": 640, "height": 480}


def _ok_pose_response():
    body = {
        "success": True,
        "data": {
            "symmetry_type": "cylindrical",
            "translation_mm": {"x": 1.0, "y": 2.0, "z": 500.0},
            "transform_4x4": [
                [1.0, 0.0, 0.0, 1.0],
                [0.0, 1.0, 0.0, 2.0],
                [0.0, 0.0, 1.0, 500.0],
                [0.0, 0.0, 0.0, 1.0],
            ],
            "fitness": 0.87,
            "inlier_rmse_mm": 1.8,
        },
        "error": None,
    }
    return httpx.Response(200, json=body)


def _envelope_err(status: int, code: str, message: str):
    body = {"success": False, "data": None,
            "error": {"code": code, "message": message}}
    return httpx.Response(status, json=body)


def test_list_recipes_happy_path():
    body = [{"id": "r1", "name": "pestello", "symmetry_type": "cylindrical"}]
    response = httpx.Response(200, json=body)
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.get", return_value=response) as mock_get:
        recs = client.list_recipes()
    assert recs == body
    posted_url = mock_get.call_args.args[0]
    assert posted_url.endswith("/api/recipes")
    # API key header
    headers = mock_get.call_args.kwargs.get("headers", {})
    assert headers.get("X-Api-Key") == "K"


def test_list_recipes_envelope_shape():
    """Open3DModel wraps /api/recipes in {success, data, error}."""
    body = {"success": True,
            "data": [{"id": "abc", "name": "P2", "symmetry_type": "cylindrical"}],
            "error": None}
    response = httpx.Response(200, json=body)
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.get", return_value=response):
        recs = client.list_recipes()
    assert recs == body["data"]


def test_list_recipes_connect_error_raises_unavailable():
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.get", side_effect=httpx.ConnectError("nope")):
        with pytest.raises(Open3DModelUnavailable):
            client.list_recipes()


def test_estimate_pose_happy_path():
    pts_m = np.array([[0.01, 0.02, 0.5], [0.015, 0.022, 0.49]])
    response = _ok_pose_response()
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.post", return_value=response) as mock_post:
        data = client.estimate_pose("r1", pts_m, _intrinsics())
    assert data["fitness"] == 0.87
    assert data["inlier_rmse_mm"] == 1.8
    body = mock_post.call_args.kwargs["json"]
    assert body["recipe_id"] == "r1"
    assert body["unit"] == "m"
    assert body["point_cloud"][0] == [0.01, 0.02, 0.5]
    headers = mock_post.call_args.kwargs.get("headers", {})
    assert headers.get("X-Api-Key") == "K"


def test_estimate_pose_envelope_failure_recipe_not_found():
    pts_m = np.array([[0.0, 0.0, 0.5]])
    response = _envelope_err(200, "RECIPE_NOT_FOUND", "Ricetta ... non esiste")
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.post", return_value=response):
        with pytest.raises(Open3DModelError) as exc:
            client.estimate_pose("r-bogus", pts_m, _intrinsics())
    assert exc.value.code == "RECIPE_NOT_FOUND"


def test_estimate_pose_envelope_failure_registration_failed():
    pts_m = np.array([[0.0, 0.0, 0.5]])
    response = _envelope_err(200, "REGISTRATION_FAILED", "low fitness")
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.post", return_value=response):
        with pytest.raises(Open3DModelError) as exc:
            client.estimate_pose("r1", pts_m, _intrinsics())
    assert exc.value.code == "REGISTRATION_FAILED"


def test_estimate_pose_http_403_raises_open3dmodel_error_auth():
    """403 con detail testuale (HTTPException di FastAPI, NON envelope)."""
    pts_m = np.array([[0.0, 0.0, 0.5]])
    response = httpx.Response(403, json={"detail": "API key non valida"})
    client = Open3DModelClient("http://o3d:8878", api_key="WRONG", timeout_s=2.0)
    with patch("httpx.Client.post", return_value=response):
        with pytest.raises(Open3DModelError) as exc:
            client.estimate_pose("r1", pts_m, _intrinsics())
    assert exc.value.code == "HTTP_403"


def test_estimate_pose_connect_error_raises_unavailable():
    pts_m = np.array([[0.0, 0.0, 0.5]])
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.post", side_effect=httpx.ConnectError("down")):
        with pytest.raises(Open3DModelUnavailable):
            client.estimate_pose("r1", pts_m, _intrinsics())


def test_get_stl_bytes_happy_path():
    response = httpx.Response(200, content=b"BINARYSTL", headers={"ETag": "abc123"})
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.get", return_value=response):
        body, etag = client.get_stl_bytes("r1")
    assert body == b"BINARYSTL"
    assert etag == "abc123"


def test_get_stl_bytes_oversize_raises():
    big = b"x" * (25 * 1024 * 1024)  # 25 MB
    response = httpx.Response(200, content=big)
    client = Open3DModelClient("http://o3d:8878", api_key="K",
                               timeout_s=2.0, max_stl_bytes=20_000_000)
    with patch("httpx.Client.get", return_value=response):
        with pytest.raises(Open3DModelError) as exc:
            client.get_stl_bytes("r1")
    assert exc.value.code == "STL_TOO_LARGE"


def test_get_stl_bytes_missing_etag_falls_back_to_hash():
    response = httpx.Response(200, content=b"BINARYSTL")  # no ETag header
    client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
    with patch("httpx.Client.get", return_value=response):
        body, etag = client.get_stl_bytes("r1")
    assert body == b"BINARYSTL"
    assert len(etag) > 0  # qualunque hash deterministico