backend/tests/test_numpy_serialization.py

from __future__ import annotations

from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
from fastapi.testclient import TestClient
from pydantic import TypeAdapter

from app.core.visualizacao.map_payload import build_elaboracao_map_payload
from app.main import app
from app.models.session import SessionState
from app.services import auth_service, elaboracao_service
from app.services.session_store import session_store


def _assert_no_numpy_scalars(value: Any, path: str = "root") -> None:
    assert not isinstance(value, np.generic), f"{path} contém {type(value).__name__}"
    if isinstance(value, dict):
        for key, item in value.items():
            _assert_no_numpy_scalars(item, f"{path}.{key}")
    elif isinstance(value, (list, tuple)):
        for index, item in enumerate(value):
            _assert_no_numpy_scalars(item, f"{path}[{index}]")


def _dump_as_fastapi_response(value: dict[str, Any]) -> bytes:
    return TypeAdapter(dict[str, Any]).dump_json(value)


def _session(tmp_path: Path) -> SessionState:
    session = SessionState(session_id="teste-numpy", workdir=tmp_path)
    session.df_original = pd.DataFrame(
        {
            "Y": np.arange(10, dtype=float),
            "X": np.arange(10, dtype=float) + 1,
            "INDICE": pd.Series(np.arange(100, 110), dtype="int64"),
            "LATITUDE": np.linspace(-30.10, -30.19, 10),
            "LONGITUDE": np.linspace(-51.20, -51.29, 10),
            "FINALIDADE": ["TERRENO"] * 10,
        }
    )
    session.df_filtrado = session.df_original.copy()
    session.coluna_y = "Y"
    session.colunas_x = ["X"]
    session.transformacoes_x = {"X": "(x)"}
    return session


def test_map_payload_converts_numpy_indice_to_native_python() -> None:
    frame = pd.DataFrame(
        {
            "INDICE": pd.Series([101, 102, 103], dtype="int64"),
            "LATITUDE": [-30.10, -30.10, -30.11],
            "LONGITUDE": [-51.20, -51.20, -51.21],
            "FINALIDADE": ["TERRENO", "TERRENO", "TERRENO"],
        }
    )

    payload = build_elaboracao_map_payload(frame)

    assert payload is not None
    mercado = next(layer for layer in payload["overlay_layers"] if layer["id"] == "mercado")
    ponto_simples = next(point for point in mercado["points"] if not point.get("grouped"))
    ponto_agrupado = next(point for point in mercado["points"] if point.get("grouped"))
    assert ponto_simples["indice"] == 103
    assert type(ponto_simples["indice"]) is int
    assert [item["indice"] for item in ponto_agrupado["group_items"]] == [101, 102]
    assert all(type(item["indice"]) is int for item in ponto_agrupado["group_items"])
    _assert_no_numpy_scalars(payload)
    assert _dump_as_fastapi_response(payload)


def test_restart_outliers_normalizes_state_and_success_response(monkeypatch, tmp_path: Path) -> None:
    session = _session(tmp_path)
    session.outliers_anteriores = [np.int64(1)]  # type: ignore[list-item]
    session.iteracao = np.int64(2)  # type: ignore[assignment]

    monkeypatch.setattr(elaboracao_service, "_validar_outliers_micronumerosidade_geral", lambda *args, **kwargs: None)
    monkeypatch.setattr(elaboracao_service, "_atualizar_periodo_dados_mercado_filtrado", lambda *args, **kwargs: None)
    monkeypatch.setattr(elaboracao_service, "_montar_tabela_outliers_excluidos", lambda *args, **kwargs: None)
    monkeypatch.setattr(elaboracao_service, "ajustar_modelo", lambda *args, **kwargs: {})
    monkeypatch.setattr(
        elaboracao_service,
        "apply_selection",
        lambda *args, **kwargs: {"valor_numpy_aninhado": {"indice": np.int64(7)}},
    )

    result = elaboracao_service.reiniciar_iteracao(
        session,
        outliers_texto="2",
        reincluir_texto="1",
    )

    assert session.outliers_anteriores == [2]
    assert all(type(item) is int for item in session.outliers_anteriores)
    assert session.iteracao == 3
    assert type(session.iteracao) is int
    assert result["valor_numpy_aninhado"]["indice"] == 7
    _assert_no_numpy_scalars(result)
    assert _dump_as_fastapi_response(result)


def test_restart_outliers_sanitizes_regression_failure_response(monkeypatch, tmp_path: Path) -> None:
    session = _session(tmp_path)
    session.outliers_anteriores = [np.int64(1)]  # type: ignore[list-item]
    session.iteracao = np.int64(4)  # type: ignore[assignment]

    monkeypatch.setattr(elaboracao_service, "_validar_outliers_micronumerosidade_geral", lambda *args, **kwargs: None)
    monkeypatch.setattr(elaboracao_service, "_atualizar_periodo_dados_mercado_filtrado", lambda *args, **kwargs: None)
    monkeypatch.setattr(
        elaboracao_service,
        "_montar_tabela_outliers_excluidos",
        lambda *args, **kwargs: {"rows": [{"indice": np.int64(1)}]},
    )
    monkeypatch.setattr(elaboracao_service, "ajustar_modelo", lambda *args, **kwargs: None)

    result = elaboracao_service.reiniciar_iteracao(
        session,
        outliers_texto="2",
        reincluir_texto=None,
    )

    assert result["regressao_ok"] is False
    assert session.iteracao == 5
    assert type(session.iteracao) is int
    _assert_no_numpy_scalars(result)
    assert _dump_as_fastapi_response(result)


def test_outlier_restart_http_boundary_serializes_numpy(monkeypatch, tmp_path: Path) -> None:
    session = _session(tmp_path)
    monkeypatch.setattr(session_store, "get", lambda _session_id: session)
    monkeypatch.setattr(auth_service, "get_user_by_token", lambda _token: {"usuario": "teste"})
    monkeypatch.setattr(
        elaboracao_service,
        "reiniciar_iteracao",
        lambda *args, **kwargs: {
            "iteracao": np.int64(2),
            "outliers_anteriores": [np.int64(101)],
        },
    )

    with TestClient(app) as client:
        response = client.post(
            "/api/elaboracao/outliers/restart",
            headers={"X-Auth-Token": "token-de-teste"},
            json={"session_id": session.session_id},
        )

    assert response.status_code == 200
    assert response.json() == {"iteracao": 2, "outliers_anteriores": [101]}