Spaces:
Running
Running
File size: 6,088 Bytes
9ca4757 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | 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]}
|