fn_registry/python/functions/datascience/detect_latlon_columns_test.py

"""Tests para detect_latlon_columns."""

import os
import sys

sys.path.insert(0, os.path.dirname(__file__))

from detect_latlon_columns import detect_latlon_columns

# Keys that every result dict (success or failure) must expose.
_EXPECTED_KEYS = {"lat_col", "lon_col", "confidence", "reason"}


def _col(name, mn=None, mx=None, inferred="numeric"):
    """Build a minimal ColumnProfile-like dict for the tests."""
    col = {"name": name, "inferred_type": inferred}
    if mn is not None or mx is not None:
        col["numeric"] = {"min": mn, "max": mx}
    return col


def test_par_latitude_longitude_fuerte():
    """Golden: nombres latitude/longitude con rango valido -> par con confianza alta."""
    columns = [
        _col("id", mn=1, mx=9999, inferred="numeric"),
        _col("latitude", mn=-45.0, mx=45.0),
        _col("longitude", mn=-120.0, mx=120.0),
    ]
    res = detect_latlon_columns(columns)

    assert set(res.keys()) == _EXPECTED_KEYS
    assert res["lat_col"] == "latitude"
    assert res["lon_col"] == "longitude"
    # Nombre fuerte (0.6) + rango (0.4) en ambos lados -> 1.0.
    assert abs(res["confidence"] - 1.0) < 1e-9
    assert "rango valido" in res["reason"]


def test_par_lat_lon_abreviado():
    """Golden: nombres abreviados lat/lon tambien se detectan como fuertes."""
    columns = [
        _col("lat", mn=40.0, mx=43.0),
        _col("lon", mn=-4.0, mx=-1.0),
        _col("precio", mn=0.0, mx=500.0),
    ]
    res = detect_latlon_columns(columns)
    assert res["lat_col"] == "lat"
    assert res["lon_col"] == "lon"
    assert abs(res["confidence"] - 1.0) < 1e-9


def test_par_x_y_debil_con_rango_valido():
    """Edge: x/y genericos solo cuentan como par debil cuando el rango encaja."""
    columns = [
        _col("y_coord", mn=-10.0, mx=10.0),   # debil latitud
        _col("x_coord", mn=-150.0, mx=150.0),  # debil longitud
    ]
    res = detect_latlon_columns(columns)
    assert res["lat_col"] == "y_coord"
    assert res["lon_col"] == "x_coord"
    # Nombre debil (0.3) + rango (0.4) -> 0.7 en ambos lados.
    assert abs(res["confidence"] - 0.7) < 1e-9


def test_nombre_lat_lon_pero_rango_fuera_no_detecta():
    """Edge: nombre lat/lon con rango fuera de bounds -> NO es coordenada."""
    columns = [
        _col("latitude", mn=-200.0, mx=200.0),   # fuera de [-90, 90]
        _col("longitude", mn=-120.0, mx=120.0),  # valido, pero sin par lat
    ]
    res = detect_latlon_columns(columns)
    assert res["lat_col"] is None
    assert res["lon_col"] is None
    assert res["confidence"] == 0.0
    assert isinstance(res["reason"], str) and res["reason"]


def test_par_fuerte_prevalece_sobre_debil():
    """Edge: con candidatos fuertes y debiles, gana el par de mayor confianza."""
    columns = [
        _col("latitude", mn=-45.0, mx=45.0),    # fuerte lat
        _col("y", mn=-30.0, mx=30.0),           # debil lat
        _col("longitude", mn=-120.0, mx=120.0),  # fuerte lon
        _col("x", mn=-100.0, mx=100.0),         # debil lon
    ]
    res = detect_latlon_columns(columns)
    assert res["lat_col"] == "latitude"
    assert res["lon_col"] == "longitude"
    assert abs(res["confidence"] - 1.0) < 1e-9


def test_entradas_vacias_o_invalidas_no_lanzan():
    """Edge: sin columnas / vacio / no-lista / entradas no-dict -> dict None sin lanzar."""
    for bad in ([], None, "no soy lista", 42, [1, 2, 3], [{}], [{"foo": "bar"}]):
        res = detect_latlon_columns(bad)
        assert set(res.keys()) == _EXPECTED_KEYS
        assert res["lat_col"] is None
        assert res["lon_col"] is None
        assert res["confidence"] == 0.0
        assert isinstance(res["reason"], str)


def test_solo_latitud_sin_longitud_no_detecta():
    """Edge: solo hay latitud valida, falta la longitud -> sin par."""
    columns = [
        _col("latitude", mn=-45.0, mx=45.0),
        _col("temperatura", mn=-5.0, mx=40.0),
    ]
    res = detect_latlon_columns(columns)
    assert res["lat_col"] is None
    assert res["lon_col"] is None
    assert res["confidence"] == 0.0


def test_deteccion_por_samples_cuando_falta_numeric():
    """Edge: sin bloque numeric, el rango se valida con samples."""
    columns = [
        {"name": "lat"},  # sin numeric ni inferred_type
        {"name": "lon"},
    ]
    samples = {
        "lat": [10.5, 20.0, None, 30.25],   # todos dentro de [-90, 90]
        "lon": [-40.0, 50.5, 60.0],          # todos dentro de [-180, 180]
    }
    res = detect_latlon_columns(columns, samples)
    assert res["lat_col"] == "lat"
    assert res["lon_col"] == "lon"
    assert abs(res["confidence"] - 1.0) < 1e-9


def test_samples_fuera_de_rango_descarta():
    """Edge: samples fuera de bounds invalidan la columna pese al nombre fuerte."""
    columns = [{"name": "lat"}, {"name": "lon"}]
    samples = {
        "lat": [10.0, 95.0],   # 95 > 90 -> latitud invalida
        "lon": [-40.0, 50.0],
    }
    res = detect_latlon_columns(columns, samples)
    assert res["lat_col"] is None
    assert res["lon_col"] is None
    assert res["confidence"] == 0.0