mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
a1c60d2fba
Slice 3c.2. The OS Open-UPRN reference set is too large to host in Postgres, so it lives in S3 and is cached per-UPRN in the existing `property_details_spatial` table (ADR-0020). `PropertyDetailsSpatialRow` mirrors that table (uprn unique); `SpatialRepository` / `SpatialPostgresRepository` upsert one shared row per UPRN and read the planning protections back by UPRN (a null flag reads as unrestricted; absent UPRNs are omitted so the caller defaults them). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
"""SpatialRepo caches the OS spatial reference (coords + planning flags) by UPRN.
|
|
|
|
The OS Open-UPRN reference set is too large to host in Postgres, so Ingestion
|
|
resolves it from S3 and writes a per-UPRN cache row here; Modelling reads the
|
|
planning protections back off it (ADR-0020). A real ephemeral Postgres exercises
|
|
the upsert-by-UPRN semantics (one shared row per UPRN).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import Engine
|
|
from sqlmodel import Session
|
|
|
|
from domain.geospatial.coordinates import Coordinates
|
|
from domain.geospatial.planning_restrictions import PlanningRestrictions
|
|
from domain.geospatial.spatial_reference import SpatialReference
|
|
from repositories.spatial.spatial_postgres_repository import SpatialPostgresRepository
|
|
|
|
|
|
def test_planning_restrictions_round_trip_by_uprn(db_engine: Engine) -> None:
|
|
# Arrange
|
|
reference = SpatialReference(
|
|
coordinates=Coordinates(longitude=-0.1278, latitude=51.5074),
|
|
restrictions=PlanningRestrictions(
|
|
in_conservation_area=True, is_listed=False, is_heritage=False
|
|
),
|
|
)
|
|
|
|
# Act
|
|
with Session(db_engine) as session:
|
|
SpatialPostgresRepository(session).save(uprn=12345, reference=reference)
|
|
session.commit()
|
|
with Session(db_engine) as session:
|
|
reloaded = SpatialPostgresRepository(session).get_for_uprns([12345])
|
|
|
|
# Assert
|
|
assert reloaded == {12345: reference.restrictions}
|
|
|
|
|
|
def test_save_upserts_the_shared_uprn_row(db_engine: Engine) -> None:
|
|
# Arrange — the same UPRN re-ingested with corrected flags.
|
|
unprotected = SpatialReference(
|
|
coordinates=Coordinates(longitude=-0.1, latitude=51.5),
|
|
restrictions=PlanningRestrictions(),
|
|
)
|
|
listed = SpatialReference(
|
|
coordinates=Coordinates(longitude=-0.1, latitude=51.5),
|
|
restrictions=PlanningRestrictions(is_listed=True),
|
|
)
|
|
|
|
# Act
|
|
with Session(db_engine) as session:
|
|
repo = SpatialPostgresRepository(session)
|
|
repo.save(uprn=999, reference=unprotected)
|
|
repo.save(uprn=999, reference=listed)
|
|
session.commit()
|
|
with Session(db_engine) as session:
|
|
reloaded = SpatialPostgresRepository(session).get_for_uprns([999])
|
|
|
|
# Assert — one row per UPRN; the latest write wins.
|
|
assert reloaded == {999: PlanningRestrictions(is_listed=True)}
|
|
|
|
|
|
def test_get_for_uprns_omits_uncovered_uprns(db_engine: Engine) -> None:
|
|
# Arrange / Act — nothing stored for this UPRN.
|
|
with Session(db_engine) as session:
|
|
reloaded = SpatialPostgresRepository(session).get_for_uprns([404])
|
|
|
|
# Assert — absent UPRNs are simply not in the map (caller defaults them to
|
|
# unrestricted).
|
|
assert reloaded == {}
|