tests/conftest.py

"""Shared test scaffolding — all zero-network.

Nothing here touches a real HTTP endpoint, LiDAR tile, or the segmentation model.
Adapters take an injectable ``session`` (see ``FakeSession``); geometry/strategy
code is pure and driven with synthetic shapes and structured point arrays built
around a real Omaha UTM location so projections are faithful.
"""

from __future__ import annotations

import os

# The API module reads these at import time — set them before any test imports
# `lawn_estimator.api`. Generous rate limits keep the limiter from tripping
# across the API contract tests; the model warm-up is disabled so the suite
# stays offline and fast.
os.environ.setdefault("ALLOWED_API_KEYS", "test-key")
os.environ.setdefault("WARM_MODEL_ON_STARTUP", "0")
os.environ.setdefault("RATE_LIMIT_QUOTE", "10000/minute")
os.environ.setdefault("RATE_LIMIT_BATCH", "10000/minute")
# Durable run ledger in-memory for the suite → no stray data/app.db, no cross-run state.
os.environ.setdefault("LAWN_DB_PATH", ":memory:")
# The suite always exercises the SQLite backend; a stray DATABASE_URL must not divert it
# to Postgres (the postgres path is verified against a real Neon DB, not in unit tests).
os.environ.pop("DATABASE_URL", None)

import numpy as np
import pytest
import requests
from pyproj import Transformer
from shapely.geometry import Polygon
from shapely.ops import transform as shapely_transform


# ---------------------------------------------------------------------------
# Fake HTTP layer — a stand-in for requests.Session used by every adapter.
# ---------------------------------------------------------------------------
class FakeResponse:
    """Mimics the slice of requests.Response the adapters use."""

    def __init__(self, json_data=None, status_code=200, content=b"", headers=None, text="",
                 stream_chunks=None):
        self._json = {} if json_data is None else json_data
        self.status_code = status_code
        self.content = content
        self.headers = headers or {}
        self.text = text
        self._stream_chunks = stream_chunks or []

    def raise_for_status(self):
        if self.status_code >= 400:
            raise requests.HTTPError(f"HTTP {self.status_code}")

    def json(self):
        return self._json

    # Streaming-download support (used by lidar._stream_download).
    def iter_content(self, chunk_size=None):
        yield from self._stream_chunks

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


class FakeSession:
    """A requests.Session look-alike.

    Construct with either a fixed ``payload`` (returned as JSON for every GET) or
    a ``handler(url, params) -> dict | FakeResponse`` for URL/param-aware routing
    (e.g. distinguishing a geocoder's exact vs. LIKE query). Records every call
    on ``.calls`` for assertions.
    """

    def __init__(self, payload=None, handler=None):
        self._payload = payload
        self._handler = handler
        self.calls: list[tuple[str, dict]] = []

    def get(self, url, params=None, timeout=None, **kwargs):
        self.calls.append((url, dict(params or {})))
        if self._handler is not None:
            result = self._handler(url, params or {})
            return result if isinstance(result, FakeResponse) else FakeResponse(json_data=result)
        return FakeResponse(json_data=self._payload if self._payload is not None else {})


# ---------------------------------------------------------------------------
# Geometry factories — built in real EPSG:26914 (UTM 14N) around Omaha so the
# WGS84<->UTM round-trip inside build_estimation_geometry is faithful.
# ---------------------------------------------------------------------------
LOCAL_CRS = "EPSG:26914"
_TO_UTM = Transformer.from_crs("EPSG:4326", LOCAL_CRS, always_xy=True).transform
_TO_WGS = Transformer.from_crs(LOCAL_CRS, "EPSG:4326", always_xy=True).transform

# A real Omaha point; its UTM easting/northing anchor all synthetic geometry.
CENTER_LAT, CENTER_LON = 41.26, -96.0
OMAHA_CX, OMAHA_CY = _TO_UTM(CENTER_LON, CENTER_LAT)

SQFT_PER_SQM = 10.76391041671


def utm_to_wgs(geom):
    return shapely_transform(_TO_WGS, geom)


def square_utm(cx: float, cy: float, size_m: float) -> Polygon:
    h = size_m / 2.0
    return Polygon([(cx - h, cy - h), (cx + h, cy - h), (cx + h, cy + h), (cx - h, cy + h)])


def rect_utm(cx: float, cy: float, width_m: float, height_m: float) -> Polygon:
    hw, hh = width_m / 2.0, height_m / 2.0
    return Polygon([(cx - hw, cy - hh), (cx + hw, cy - hh), (cx + hw, cy + hh), (cx - hw, cy + hh)])


def ground_points_utm(cx: float, cy: float, n_side: int = 10, half_m: float = 14.0,
                      classification: int = 2) -> np.ndarray:
    """A grid of structured LiDAR points (X,Y,Z,Classification) inside a square."""
    xs = np.linspace(cx - half_m, cx + half_m, n_side)
    ys = np.linspace(cy - half_m, cy + half_m, n_side)
    gx, gy = np.meshgrid(xs, ys)
    gx, gy = gx.ravel(), gy.ravel()
    pts = np.zeros(gx.size, dtype=[("X", "f8"), ("Y", "f8"), ("Z", "f8"), ("Classification", "u1")])
    pts["X"], pts["Y"], pts["Z"] = gx, gy, 300.0
    pts["Classification"] = classification
    return pts


# ---------------------------------------------------------------------------
# Canned ArcGIS / Google payloads — mirror the real service schemas (Douglas
# PROPERTY_A/BLDG_YRBLT, Sarpy SITEADDRESS, address-point geometry, parcel
# rings, street paths). Kept as builders so tests can tweak a field inline.
# ---------------------------------------------------------------------------
def address_point_feature(fulladdr="17531 MADISON ST", zip_code="68135",
                          municipality="Omaha", lon=CENTER_LON, lat=CENTER_LAT):
    """An Esri Address_Points query feature (FGDC/NENA model)."""
    return {"attributes": {"FULLADDR": fulladdr, "ZIP": zip_code, "MUNICIPALITY": municipality},
            "geometry": {"x": lon, "y": lat}}


def address_points_payload(*features):
    return {"features": list(features)}


def _rings_from_utm(poly_utm: Polygon) -> list:
    """ArcGIS rings (WGS84 lon/lat) for a UTM polygon."""
    wgs = utm_to_wgs(poly_utm)
    return [[[x, y] for x, y in wgs.exterior.coords]]


def douglas_parcel_payload(object_id=101, property_a="17531 MADISON ST", year_built=1999,
                           bldg_sf=2200, poly_utm=None):
    poly_utm = poly_utm if poly_utm is not None else square_utm(OMAHA_CX, OMAHA_CY, 30.0)
    return {"features": [{
        "attributes": {"OBJECTID": object_id, "PROPERTY_A": property_a,
                       "BLDG_YRBLT": year_built, "BLDG_SF": bldg_sf,
                       "PIN": "1234567890", "PROP_ZIP": "68135", "ACRES": 0.21, "SQ_FEET": 9000},
        "geometry": {"rings": _rings_from_utm(poly_utm)},
    }]}


def sarpy_parcel_payload(object_id=202, site_address="708 KOUNTZE MEMORIAL DR", poly_utm=None):
    # Sarpy has no BLDG_YRBLT / BLDG_SF fields.
    poly_utm = poly_utm if poly_utm is not None else square_utm(OMAHA_CX, OMAHA_CY, 30.0)
    return {"features": [{
        "attributes": {"OBJECTID": object_id, "SITEADDRESS": site_address,
                       "PARCELID": "011-2233", "ACREAGE": 0.19, "PSTLZIP5": "68005"},
        "geometry": {"rings": _rings_from_utm(poly_utm)},
    }]}


def street_feature(name="MADISON ST", path_wgs=None):
    return {"attributes": {"FULLNAME": name}, "geometry": {"paths": [path_wgs]}}


def google_geocode_payload(formatted="17531 Madison St, Omaha, NE 68135",
                           lat=CENTER_LAT, lon=CENTER_LON, location_type="ROOFTOP", status="OK"):
    return {"status": status, "results": [{
        "formatted_address": formatted,
        "geometry": {"location": {"lat": lat, "lng": lon}, "location_type": location_type},
    }]}


@pytest.fixture
def fake_session_factory():
    """Returns FakeSession-building helpers to keep test bodies terse."""
    return {"payload": FakeSession, "response": FakeResponse}


@pytest.fixture(autouse=True)
def _block_real_network(monkeypatch):
    """Enforce the zero-network contract: any real outbound socket connect fails.

    The suite mocks every data source, so a socket connect means a test regressed
    into hitting a live endpoint. The in-process ASGI TestClient uses no sockets,
    so the API tests are unaffected; localhost is allowed just in case.
    """
    import socket

    real_connect = socket.socket.connect

    def guard(self, address, *args, **kwargs):
        host = address[0] if isinstance(address, tuple) else address
        if host not in ("127.0.0.1", "::1", "localhost"):
            raise RuntimeError(f"Blocked network access in test: {address!r}")
        return real_connect(self, address, *args, **kwargs)

    monkeypatch.setattr(socket.socket, "connect", guard)