"""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)