perfect-postcode/pipeline/download/test_development_sites.py

import polars as pl

from pipeline.download.development_sites import (
    SCHEMA,
    _has_dwellings,
    _is_residential,
    _point_lonlat,
    _valid_lonlat,
    brownfield_to_frame,
    homes_england_to_frame,
)


def test_point_lonlat_parses_wkt():
    assert _point_lonlat({"point": "POINT(-0.1278 51.5074)"}) == (-0.1278, 51.5074)
    # Tolerates extra spacing and uppercase.
    assert _point_lonlat({"point": "POINT ( -1.5 53.8 )"}) == (-1.5, 53.8)


def test_point_lonlat_falls_back_to_columns():
    assert _point_lonlat({"longitude": "-2.0", "latitude": "52.4"}) == (-2.0, 52.4)


def test_point_lonlat_rejects_out_of_range():
    # A WKT written "lat lon" would place lat (51.5) into the lon slot -> rejected.
    assert _point_lonlat({"point": "POINT(51.5 -0.12)"}) is None
    assert _point_lonlat({"point": "garbage"}) is None
    assert _point_lonlat({}) is None


def test_valid_lonlat_bounds():
    assert _valid_lonlat(-0.1, 51.5)
    assert not _valid_lonlat(2.5, 51.5)  # lon east of GB
    assert not _valid_lonlat(-0.1, 40.0)  # lat south of GB


def test_has_dwellings():
    assert _has_dwellings(0, 5)
    assert _has_dwellings(3, None)
    assert not _has_dwellings(None, None)
    assert not _has_dwellings(0, 0)


def test_is_residential():
    assert _is_residential("Residential", None)
    assert _is_residential("Mixed Use", None)
    assert _is_residential(None, 12)
    assert not _is_residential("Employment", None)
    assert not _is_residential(None, None)
    assert not _is_residential("Industrial", 0)


def test_brownfield_to_frame_normalises_and_filters():
    raw = pl.DataFrame(
        {
            "point": [
                "POINT(-0.1 51.5)",  # kept
                "POINT(-1.2 52.6)",  # dropped: no dwellings
                "POINT(-2.0 53.0)",  # dropped: end-date set (built out)
                "POINT(-3.0 54.0)",  # kept
            ],
            "site-address": ["10 High St", "Old Yard", "Mill Site", "Dock Road"],
            "minimum-net-dwellings": ["5", "0", "10", None],
            "maximum-net-dwellings": ["15", "0", "10", "40"],
            "planning-permission-status": [
                "permissioned",
                "not-permissioned",
                "permissioned",
                "pending-decision",
            ],
            "planning-permission-type": ["full", "", "outline", None],
            "planning-permission-date": ["2023-01-01", None, "2022-05-01", None],
            "hectares": ["0.5", "0.2", "1.0", "2.0"],
            "end-date": [None, None, "2021-06-01", ""],
            "entity": [1700000, 1700001, 1700002, 1700003],
            "organisation": [
                "London Borough of Lewisham",
                "Org B",
                "Org C",
                "London Borough of Newham",
            ],
        }
    )
    frame = brownfield_to_frame(raw)

    assert frame.columns == list(SCHEMA.keys())
    assert frame.schema["min_dwellings"] == pl.Int32
    assert frame.schema["lat"] == pl.Float64
    # Only the first and last rows survive (dwellings present, not built out).
    assert frame.height == 2
    assert frame["source"].to_list() == ["brownfield", "brownfield"]
    first = frame.row(0, named=True)
    assert first["lat"] == 51.5
    assert first["lon"] == -0.1
    assert first["min_dwellings"] == 5
    assert first["max_dwellings"] == 15
    assert first["name"] == "10 High St"
    assert first["local_authority"] == "London Borough of Lewisham"
    # Per-site Planning Data entity page, not the unreliable site-plan-url.
    assert first["url"] == "https://www.planning.data.gov.uk/entity/1700000"
    last = frame.row(1, named=True)
    assert last["min_dwellings"] is None
    assert last["max_dwellings"] == 40
    assert last["url"] == "https://www.planning.data.gov.uk/entity/1700003"


def test_brownfield_empty_input_yields_typed_empty_frame():
    raw = pl.DataFrame(
        {
            "point": [],
            "minimum-net-dwellings": [],
            "maximum-net-dwellings": [],
            "end-date": [],
        }
    )
    frame = brownfield_to_frame(raw)
    assert frame.height == 0
    assert frame.columns == list(SCHEMA.keys())


def test_homes_england_to_frame_uses_centroid_and_capacity():
    features = [
        {
            "geometry": {
                "type": "Polygon",
                "coordinates": [[[-0.2, 51.4], [0.0, 51.4], [0.0, 51.6], [-0.2, 51.6], [-0.2, 51.4]]],
            },
            # Real Land Hub field names (Housing_Capacity arrives as a float string).
            "properties": {
                "Parcel_Name": "South West Of Old Park Mound, Telford, TF3 4AW",
                "Proposed_Use": "Residential",
                "Housing_Capacity": "67.0",
                "Planning_Status": "Detailed Application Submitted",
                "Local_Authority": "Telford and Wrekin",
                "Gross_Area__Acres_": "2.8602",
                "Site_Reference": "1494",
            },
        },
        {
            # Non-residential, no capacity -> dropped.
            "geometry": {
                "type": "Polygon",
                "coordinates": [[[-1.0, 52.0], [-0.9, 52.0], [-0.9, 52.1], [-1.0, 52.1], [-1.0, 52.0]]],
            },
            "properties": {"Proposed_Use": "Employment"},
        },
    ]
    frame = homes_england_to_frame(features)

    assert frame.height == 1
    row = frame.row(0, named=True)
    assert row["source"] == "homes-england"
    # Real name, not the bare Site_Reference fallback.
    assert row["name"] == "South West Of Old Park Mound, Telford, TF3 4AW"
    assert row["max_dwellings"] == 67
    assert row["min_dwellings"] is None
    assert row["local_authority"] == "Telford and Wrekin"
    # Acres converted to hectares: 2.8602 * 0.404686 ≈ 1.1575.
    assert row["hectares"] == 1.1575
    # No stable per-site URL on the Land Hub.
    assert row["url"] is None
    # Centroid of the unit-ish box.
    assert abs(row["lon"] - (-0.1)) < 1e-6
    assert abs(row["lat"] - 51.5) < 1e-6
    assert frame.columns == list(SCHEMA.keys())