junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

typing and renaming 🟪

Browse source
This commit is contained in:
Daniel Roth 2026-05-07 13:26:49 +00:00
parent 91a634e637
commit 6b29086a1e
13 changed files with 405 additions and 142 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

79
backend/app/db/functions/magic_plan_functions.py
View file

@ -6,25 +6,25 @@ from sqlmodel import Session, col
from datatypes.magicplan.domain.models import Floor, Plan
from backend.app.db.models.magic_plan import (
MagicPlanDoor,
MagicPlanFloor,
MagicPlanPlan,
MagicPlanRoom,
MagicPlanWindow,
MagicPlanDoorModel,
MagicPlanFloorModel,
MagicPlanPlanModel,
MagicPlanRoomModel,
MagicPlanWindowModel,
)
def save_plan(session: Session, plan: Plan) -> None:
plan_id = _upsert_plan(session, plan)
plan_id: int = _upsert_plan(session, plan)
_delete_children(session, plan_id)
floor_ids = _insert_floors(session, plan.floors, plan_id)
room_ids = _insert_rooms(session, plan.floors, floor_ids)
floor_ids: list[int] = _insert_floors(session, plan.floors, plan_id)
room_ids: list[int] = _insert_rooms(session, plan.floors, floor_ids)
_insert_windows_and_doors(session, plan.floors, room_ids)
def _upsert_plan(session: Session, plan: Plan) -> int:
stmt = (
pg_insert(MagicPlanPlan)
pg_insert(MagicPlanPlanModel)
.values(
magic_plan_uid=plan.uid,
name=plan.name,
@ -33,9 +33,13 @@ def _upsert_plan(session: Session, plan: Plan) -> int:
)
.on_conflict_do_update(
index_elements=["magic_plan_uid"],
set_={"name": plan.name, "address": plan.address, "postcode": plan.postcode},
set_={
"name": plan.name,
"address": plan.address,
"postcode": plan.postcode,
},
)
.returning(col(MagicPlanPlan.id))
.returning(col(MagicPlanPlanModel.id))
)
row_id: int = session.execute(stmt).scalar_one()
return row_id
@ -43,33 +47,52 @@ def _upsert_plan(session: Session, plan: Plan) -> int:
def _delete_children(session: Session, plan_id: int) -> None:
floor_subq = (
select(col(MagicPlanFloor.id))
.where(col(MagicPlanFloor.magic_plan_plan_id) == plan_id)
select(col(MagicPlanFloorModel.id))
.where(col(MagicPlanFloorModel.magic_plan_plan_id) == plan_id)
.scalar_subquery()
)
room_subq = (
select(col(MagicPlanRoom.id))
.where(col(MagicPlanRoom.magic_plan_floor_id).in_(floor_subq))
select(col(MagicPlanRoomModel.id))
.where(col(MagicPlanRoomModel.magic_plan_floor_id).in_(floor_subq))
.scalar_subquery()
)
session.execute(delete(MagicPlanWindow).where(col(MagicPlanWindow.magic_plan_room_id).in_(room_subq)))
session.execute(delete(MagicPlanDoor).where(col(MagicPlanDoor.magic_plan_room_id).in_(room_subq)))
session.execute(delete(MagicPlanRoom).where(col(MagicPlanRoom.magic_plan_floor_id).in_(floor_subq)))
session.execute(delete(MagicPlanFloor).where(col(MagicPlanFloor.magic_plan_plan_id) == plan_id))
session.execute(
delete(MagicPlanWindowModel).where(
col(MagicPlanWindowModel.magic_plan_room_id).in_(room_subq)
)
)
session.execute(
delete(MagicPlanDoorModel).where(
col(MagicPlanDoorModel.magic_plan_room_id).in_(room_subq)
)
)
session.execute(
delete(MagicPlanRoomModel).where(
col(MagicPlanRoomModel.magic_plan_floor_id).in_(floor_subq)
)
)
session.execute(
delete(MagicPlanFloorModel).where(
col(MagicPlanFloorModel.magic_plan_plan_id) == plan_id
)
)
def _insert_floors(session: Session, floors: list[Floor], plan_id: int) -> list[int]:
rows: list[dict[str, Any]] = [
{"magic_plan_plan_id": plan_id, "level": floor.level}
for floor in floors
{"magic_plan_plan_id": plan_id, "level": floor.level} for floor in floors
]
result = session.execute(
pg_insert(MagicPlanFloor).values(rows).returning(col(MagicPlanFloor.id))
pg_insert(MagicPlanFloorModel)
.values(rows)
.returning(col(MagicPlanFloorModel.id))
)
return cast(list[int], list(result.scalars().all()))
def _insert_rooms(session: Session, floors: list[Floor], floor_ids: list[int]) -> list[int]:
def _insert_rooms(
session: Session, floors: list[Floor], floor_ids: list[int]
) -> list[int]:
rows: list[dict[str, Any]] = [
{
"magic_plan_floor_id": floor_id,
@ -82,12 +105,14 @@ def _insert_rooms(session: Session, floors: list[Floor], floor_ids: list[int]) -
for room in floor.rooms
]
result = session.execute(
pg_insert(MagicPlanRoom).values(rows).returning(col(MagicPlanRoom.id))
pg_insert(MagicPlanRoomModel).values(rows).returning(col(MagicPlanRoomModel.id))
)
return cast(list[int], list(result.scalars().all()))
def _insert_windows_and_doors(session: Session, floors: list[Floor], room_ids: list[int]) -> None:
def _insert_windows_and_doors(
session: Session, floors: list[Floor], room_ids: list[int]
) -> None:
all_rooms = [room for floor in floors for room in floor.rooms]
window_rows: list[dict[str, Any]] = [
@ -111,6 +136,6 @@ def _insert_windows_and_doors(session: Session, floors: list[Floor], room_ids: l
]
if window_rows:
session.execute(pg_insert(MagicPlanWindow).values(window_rows))
session.execute(pg_insert(MagicPlanWindowModel).values(window_rows))
if door_rows:
session.execute(pg_insert(MagicPlanDoor).values(door_rows))
session.execute(pg_insert(MagicPlanDoorModel).values(door_rows))

86
backend/app/db/functions/tests/test_magic_plan_functions.py
View file

@ -5,7 +5,7 @@ from unittest.mock import MagicMock
import pytest
from sqlalchemy.dialects import postgresql
from datatypes.magicplan.api.response import MagicPlan
from datatypes.magicplan.api.response import MagicPlanPlan
from datatypes.magicplan.domain.mapper import map_plan
from datatypes.magicplan.domain.models import Plan
@ -19,15 +19,19 @@ PLAN_UID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
@pytest.fixture(scope="module")
def domain_plan() -> Plan:
data = json.loads((FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text())
return map_plan(MagicPlan.model_validate(data["data"]))
data = json.loads(
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
)
return map_plan(MagicPlanPlan.model_validate(data["data"]))
def _compiled(stmt: object) -> str:
return str(stmt.compile( # type: ignore[union-attr]
dialect=postgresql.dialect(),
compile_kwargs={"literal_binds": True},
))
return str(
stmt.compile( # type: ignore[union-attr]
dialect=postgresql.dialect(),
compile_kwargs={"literal_binds": True},
)
)
@pytest.fixture()
@ -44,15 +48,15 @@ def mock_session() -> MagicMock:
room_result.scalars.return_value.all.return_value = list(range(100, 114))
session.execute.side_effect = [
plan_result, # upsert plan
None, # delete windows
None, # delete doors
None, # delete rooms
None, # delete floors
plan_result, # upsert plan
None, # delete windows
None, # delete doors
None, # delete rooms
None, # delete floors
floor_result, # insert floors
room_result, # insert rooms
None, # insert windows
None, # insert doors
room_result, # insert rooms
None, # insert windows
None, # insert doors
]
return session
@ -65,7 +69,9 @@ def test_save_plan_does_not_raise(mock_session: MagicMock, domain_plan: Plan) ->
save_plan(mock_session, domain_plan)
def test_save_plan_upserts_plan_table_first(mock_session: MagicMock, domain_plan: Plan) -> None:
def test_save_plan_upserts_plan_table_first(
mock_session: MagicMock, domain_plan: Plan
) -> None:
# Act
save_plan(mock_session, domain_plan)
# Assert
@ -75,7 +81,9 @@ def test_save_plan_upserts_plan_table_first(mock_session: MagicMock, domain_plan
assert "INSERT" in sql.upper()
def test_save_plan_upsert_contains_plan_uid(mock_session: MagicMock, domain_plan: Plan) -> None:
def test_save_plan_upsert_contains_plan_uid(
mock_session: MagicMock, domain_plan: Plan
) -> None:
# Act
save_plan(mock_session, domain_plan)
# Assert
@ -83,7 +91,9 @@ def test_save_plan_upsert_contains_plan_uid(mock_session: MagicMock, domain_plan
assert PLAN_UID in _compiled(first_stmt)
def test_save_plan_upsert_contains_plan_name(mock_session: MagicMock, domain_plan: Plan) -> None:
def test_save_plan_upsert_contains_plan_name(
mock_session: MagicMock, domain_plan: Plan
) -> None:
# Act
save_plan(mock_session, domain_plan)
# Assert
@ -91,41 +101,63 @@ def test_save_plan_upsert_contains_plan_name(mock_session: MagicMock, domain_pla
assert domain_plan.name in _compiled(first_stmt)
def test_save_plan_deletes_floors_before_inserting(mock_session: MagicMock, domain_plan: Plan) -> None:
def test_save_plan_deletes_floors_before_inserting(
mock_session: MagicMock, domain_plan: Plan
) -> None:
# Act
save_plan(mock_session, domain_plan)
# Assert — find delete and insert stmts targeting magic_plan_floor
stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
floor_delete_idx = next(i for i, s in enumerate(stmts) if "DELETE" in s.upper() and "magic_plan_floor" in s)
floor_insert_idx = next(i for i, s in enumerate(stmts) if "INSERT" in s.upper() and "magic_plan_floor" in s)
floor_delete_idx = next(
i
for i, s in enumerate(stmts)
if "DELETE" in s.upper() and "magic_plan_floor" in s
)
floor_insert_idx = next(
i
for i, s in enumerate(stmts)
if "INSERT" in s.upper() and "magic_plan_floor" in s
)
assert floor_delete_idx < floor_insert_idx
def test_save_plan_floor_insert_contains_all_levels(mock_session: MagicMock, domain_plan: Plan) -> None:
def test_save_plan_floor_insert_contains_all_levels(
mock_session: MagicMock, domain_plan: Plan
) -> None:
# Act
save_plan(mock_session, domain_plan)
# Assert — each floor's level value appears in the INSERT
stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
floor_insert = next(s for s in stmts if "INSERT" in s.upper() and "magic_plan_floor" in s)
floor_insert = next(
s for s in stmts if "INSERT" in s.upper() and "magic_plan_floor" in s
)
for floor in domain_plan.floors:
if floor.level is not None:
assert str(floor.level) in floor_insert
def test_save_plan_room_insert_uses_all_floor_ids(mock_session: MagicMock, domain_plan: Plan) -> None:
def test_save_plan_room_insert_uses_all_floor_ids(
mock_session: MagicMock, domain_plan: Plan
) -> None:
# Act
save_plan(mock_session, domain_plan)
# Assert — both mocked floor ids (10, 20) appear in the room INSERT
stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
room_insert = next(s for s in stmts if "INSERT" in s.upper() and "magic_plan_room" in s)
room_insert = next(
s for s in stmts if "INSERT" in s.upper() and "magic_plan_room" in s
)
assert "10" in room_insert
assert "20" in room_insert
def test_save_plan_windows_use_room_ids_from_insert(mock_session: MagicMock, domain_plan: Plan) -> None:
def test_save_plan_windows_use_room_ids_from_insert(
mock_session: MagicMock, domain_plan: Plan
) -> None:
# Act
save_plan(mock_session, domain_plan)
# Assert — window INSERT references one of the mocked room ids (100–113)
stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
window_insert = next(s for s in stmts if "INSERT" in s.upper() and "magic_plan_window" in s)
window_insert = next(
s for s in stmts if "INSERT" in s.upper() and "magic_plan_window" in s
)
assert any(str(rid) in window_insert for rid in range(100, 114))

10
backend/app/db/models/magic_plan.py
View file

@ -3,7 +3,7 @@ from typing import Optional
from sqlmodel import Field, SQLModel
class MagicPlanPlan(SQLModel, table=True):
class MagicPlanPlanModel(SQLModel, table=True):
__tablename__ = "magic_plan_plan"
id: Optional[int] = Field(default=None, primary_key=True)
@ -13,7 +13,7 @@ class MagicPlanPlan(SQLModel, table=True):
postcode: Optional[str] = None
class MagicPlanFloor(SQLModel, table=True):
class MagicPlanFloorModel(SQLModel, table=True):
__tablename__ = "magic_plan_floor"
id: Optional[int] = Field(default=None, primary_key=True)
@ -21,7 +21,7 @@ class MagicPlanFloor(SQLModel, table=True):
level: Optional[int] = None
class MagicPlanRoom(SQLModel, table=True):
class MagicPlanRoomModel(SQLModel, table=True):
__tablename__ = "magic_plan_room"
id: Optional[int] = Field(default=None, primary_key=True)
@ -32,7 +32,7 @@ class MagicPlanRoom(SQLModel, table=True):
area_m2: Optional[float] = None
class MagicPlanWindow(SQLModel, table=True):
class MagicPlanWindowModel(SQLModel, table=True):
__tablename__ = "magic_plan_window"
id: Optional[int] = Field(default=None, primary_key=True)
@ -43,7 +43,7 @@ class MagicPlanWindow(SQLModel, table=True):
opening_type: Optional[str] = None
class MagicPlanDoor(SQLModel, table=True):
class MagicPlanDoorModel(SQLModel, table=True):
__tablename__ = "magic_plan_door"
id: Optional[int] = Field(default=None, primary_key=True)

134
backend/app/db/models/tests/test_magic_plan_models.py Normal file
View file

@ -0,0 +1,134 @@
from backend.app.db.models.magic_plan import (
MagicPlanDoorModel,
MagicPlanFloorModel,
MagicPlanPlanModel,
MagicPlanRoomModel,
MagicPlanWindowModel,
)
# --- MagicPlanPlan ---
def test_plan_table_name() -> None:
assert MagicPlanPlanModel.__tablename__ == "magic_plan_plan"
def test_plan_has_magic_plan_uid_column() -> None:
assert "magic_plan_uid" in MagicPlanPlanModel.__table__.columns
def test_plan_magic_plan_uid_is_unique() -> None:
col = MagicPlanPlanModel.__table__.columns["magic_plan_uid"]
assert (
any(
c.unique
for c in MagicPlanPlanModel.__table__.constraints
if hasattr(c, "columns")
and "magic_plan_uid" in [cc.name for cc in c.columns]
)
or col.unique
)
def test_plan_instantiation() -> None:
plan = MagicPlanPlanModel(
magic_plan_uid="uid-123", name="Test", address="1 High St", postcode="SW1A 1AA"
)
assert plan.magic_plan_uid == "uid-123"
assert plan.name == "Test"
assert plan.postcode == "SW1A 1AA"
# --- MagicPlanFloor ---
def test_floor_table_name() -> None:
assert MagicPlanFloorModel.__tablename__ == "magic_plan_floor"
def test_floor_fk_column_name() -> None:
assert "magic_plan_plan_id" in MagicPlanFloorModel.__table__.columns
def test_floor_has_level() -> None:
floor = MagicPlanFloorModel(magic_plan_plan_id=1, level=0)
assert floor.level == 0
# --- MagicPlanRoom ---
def test_room_table_name() -> None:
assert MagicPlanRoomModel.__tablename__ == "magic_plan_room"
def test_room_fk_column_name() -> None:
assert "magic_plan_floor_id" in MagicPlanRoomModel.__table__.columns
def test_room_has_measurement_columns() -> None:
cols = MagicPlanRoomModel.__table__.columns
assert "width_m" in cols
assert "length_m" in cols
assert "area_m2" in cols
def test_room_instantiation() -> None:
room = MagicPlanRoomModel(
magic_plan_floor_id=1, name="Kitchen", width_m=2.67, length_m=2.98, area_m2=7.95
)
assert room.name == "Kitchen"
assert room.width_m == 2.67
# --- MagicPlanWindow ---
def test_window_table_name() -> None:
assert MagicPlanWindowModel.__tablename__ == "magic_plan_window"
def test_window_fk_column_name() -> None:
assert "magic_plan_room_id" in MagicPlanWindowModel.__table__.columns
def test_window_has_measurement_columns() -> None:
cols = MagicPlanWindowModel.__table__.columns
assert "width_m" in cols
assert "height_m" in cols
assert "area_m2" in cols
assert "opening_type" in cols
def test_window_instantiation() -> None:
window = MagicPlanWindowModel(
magic_plan_room_id=1,
width_m=1.4,
height_m=1.2,
area_m2=1.68,
opening_type="casement",
)
assert window.opening_type == "casement"
# --- MagicPlanDoor ---
def test_door_table_name() -> None:
assert MagicPlanDoorModel.__tablename__ == "magic_plan_door"
def test_door_fk_column_name() -> None:
assert "magic_plan_room_id" in MagicPlanDoorModel.__table__.columns
def test_door_has_width_mm_and_type() -> None:
cols = MagicPlanDoorModel.__table__.columns
assert "width_mm" in cols
assert "type" in cols
def test_door_instantiation() -> None:
door = MagicPlanDoorModel(magic_plan_room_id=1, width_mm=0.79, type="hinged")
assert door.width_mm == 0.79
assert door.type == "hinged"

7
backend/magic_plan/address_matcher.py
View file

@ -1,10 +1,9 @@
import re
from typing import Optional
from datatypes.magicplan.api.response import PlanSummary
_UK_POSTCODE_RE = re.compile(
r"[A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2}", re.IGNORECASE
)
_UK_POSTCODE_RE = re.compile(r"[A-Z]{1,2}\d[A-Z\d]?\s*\d[A-Z]{2}", re.IGNORECASE)
def _extract_postcode(address: str) -> str | None:
@ -18,7 +17,7 @@ def _normalize_postcode(postcode: str) -> str:
return postcode.replace(" ", "").upper()
def find_matching_plan(plans: list[PlanSummary], address: str) -> PlanSummary | None:
def find_matching_plan(plans: list[PlanSummary], address: str) -> Optional[PlanSummary]:
postcode = _extract_postcode(address)
if postcode is None:
return None

10
backend/magic_plan/magic_plan_client.py
View file

@ -1,6 +1,6 @@
import requests
from datatypes.magicplan.api.response import MagicPlan, PlansListResponse
from datatypes.magicplan.api.response import MagicPlanPlan, PlansListResponse
_BASE_URL = "https://cloud.magicplan.app/api/v2"
@ -16,7 +16,9 @@ class MagicPlanClient:
r.raise_for_status()
return PlansListResponse.model_validate(r.json()["data"])
def get_plan(self, plan_id: str) -> MagicPlan:
r = self._session.get(f"{_BASE_URL}/plans/{plan_id}", params={"key": self._api_key})
def get_plan(self, plan_id: str) -> MagicPlanPlan:
r = self._session.get(
f"{_BASE_URL}/plans/{plan_id}", params={"key": self._api_key}
)
r.raise_for_status()
return MagicPlan.model_validate(r.json()["data"])
return MagicPlanPlan.model_validate(r.json()["data"])

15
backend/magic_plan/magic_plan_service.py
View file

@ -1,5 +1,10 @@
from typing import Optional
from datatypes.magicplan.api.response import (
MagicPlanPlan,
PlanSummary,
PlansListResponse,
)
from datatypes.magicplan.domain.mapper import map_plan
from datatypes.magicplan.domain.models import Plan
@ -20,14 +25,16 @@ class MagicPlanService:
if uprn is not None:
logger.info("MagicPlanService.run uprn=%s", uprn)
plans_response = self._client.get_plans()
matched = find_matching_plan(plans_response.plans, address)
plans_response: PlansListResponse = self._client.get_plans()
matched: Optional[PlanSummary] = find_matching_plan(
plans_response.plans, address
)
if matched is None:
raise ValueError(f"No MagicPlan found for address: {address!r}")
magic_plan = self._client.get_plan(matched.id)
plan = map_plan(magic_plan)
magic_plan: MagicPlanPlan = self._client.get_plan(matched.id)
plan: Plan = map_plan(magic_plan)
with db_session() as session:
save_plan(session, plan)

79
backend/magic_plan/tests/test_magic_plan_client.py
View file

@ -7,7 +7,7 @@ import pytest
import requests
from backend.magic_plan.magic_plan_client import MagicPlanClient
from datatypes.magicplan.api.response import MagicPlan, PlansListResponse
from datatypes.magicplan.api.response import MagicPlanPlan, PlansListResponse
FIXTURE_DIR = Path(__file__).parents[2] / "magic_plan"
BASE_URL = "https://cloud.magicplan.app/api/v2"
@ -20,7 +20,10 @@ def _load_fixture(name: str) -> dict[str, Any]:
def _make_client(mock_session: MagicMock) -> MagicPlanClient:
with patch("backend.magic_plan.magic_plan_client.requests.Session", return_value=mock_session):
with patch(
"backend.magic_plan.magic_plan_client.requests.Session",
return_value=mock_session,
):
return MagicPlanClient(customer_id=CUSTOMER_ID, api_key=API_KEY)
@ -47,10 +50,15 @@ def test_customer_header_set_on_session(mock_session: MagicMock) -> None:
# --- get_plans ---
def test_get_plans_calls_correct_url(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plans_calls_correct_url(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
plans_data = _load_fixture("magicplan_api_plans_response_example.json")["data"]
mock_session.get.return_value.json.return_value = {"message": "OK", "data": plans_data}
mock_session.get.return_value.json.return_value = {
"message": "OK",
"data": plans_data,
}
# Act
client.get_plans()
# Assert
@ -59,20 +67,30 @@ def test_get_plans_calls_correct_url(client: MagicPlanClient, mock_session: Magi
)
def test_get_plans_calls_raise_for_status(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plans_calls_raise_for_status(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
plans_data = _load_fixture("magicplan_api_plans_response_example.json")["data"]
mock_session.get.return_value.json.return_value = {"message": "OK", "data": plans_data}
mock_session.get.return_value.json.return_value = {
"message": "OK",
"data": plans_data,
}
# Act
client.get_plans()
# Assert
mock_session.get.return_value.raise_for_status.assert_called_once()
def test_get_plans_returns_plans_list_response(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plans_returns_plans_list_response(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
plans_data = _load_fixture("magicplan_api_plans_response_example.json")["data"]
mock_session.get.return_value.json.return_value = {"message": "OK", "data": plans_data}
mock_session.get.return_value.json.return_value = {
"message": "OK",
"data": plans_data,
}
# Act
result = client.get_plans()
# Assert
@ -80,9 +98,13 @@ def test_get_plans_returns_plans_list_response(client: MagicPlanClient, mock_ses
assert len(result.plans) == 1
def test_get_plans_propagates_http_error(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plans_propagates_http_error(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
mock_session.get.return_value.raise_for_status.side_effect = requests.HTTPError("404")
mock_session.get.return_value.raise_for_status.side_effect = requests.HTTPError(
"404"
)
# Act / Assert
with pytest.raises(requests.HTTPError):
client.get_plans()
@ -91,10 +113,15 @@ def test_get_plans_propagates_http_error(client: MagicPlanClient, mock_session:
# --- get_plan ---
def test_get_plan_calls_correct_url(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plan_calls_correct_url(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
plan_data = _load_fixture("magicplan_api_plan_response_example.json")["data"]
mock_session.get.return_value.json.return_value = {"message": "OK", "data": plan_data}
mock_session.get.return_value.json.return_value = {
"message": "OK",
"data": plan_data,
}
plan_id = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
# Act
client.get_plan(plan_id)
@ -104,30 +131,44 @@ def test_get_plan_calls_correct_url(client: MagicPlanClient, mock_session: Magic
)
def test_get_plan_calls_raise_for_status(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plan_calls_raise_for_status(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
plan_data = _load_fixture("magicplan_api_plan_response_example.json")["data"]
mock_session.get.return_value.json.return_value = {"message": "OK", "data": plan_data}
mock_session.get.return_value.json.return_value = {
"message": "OK",
"data": plan_data,
}
# Act
client.get_plan("a7285ed1-878d-47eb-8aa6-85ef9e187516")
# Assert
mock_session.get.return_value.raise_for_status.assert_called_once()
def test_get_plan_returns_magic_plan(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plan_returns_magic_plan(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
plan_data = _load_fixture("magicplan_api_plan_response_example.json")["data"]
mock_session.get.return_value.json.return_value = {"message": "OK", "data": plan_data}
mock_session.get.return_value.json.return_value = {
"message": "OK",
"data": plan_data,
}
# Act
result = client.get_plan("a7285ed1-878d-47eb-8aa6-85ef9e187516")
# Assert
assert isinstance(result, MagicPlan)
assert isinstance(result, MagicPlanPlan)
assert result.plan.id == "a7285ed1-878d-47eb-8aa6-85ef9e187516"
def test_get_plan_propagates_http_error(client: MagicPlanClient, mock_session: MagicMock) -> None:
def test_get_plan_propagates_http_error(
client: MagicPlanClient, mock_session: MagicMock
) -> None:
# Arrange
mock_session.get.return_value.raise_for_status.side_effect = requests.HTTPError("500")
mock_session.get.return_value.raise_for_status.side_effect = requests.HTTPError(
"500"
)
# Act / Assert
with pytest.raises(requests.HTTPError):
client.get_plan("some-id")

66
backend/magic_plan/tests/test_magic_plan_service.py
View file

@ -4,7 +4,7 @@ from unittest.mock import MagicMock, patch
import pytest
from datatypes.magicplan.api.response import MagicPlan, PlanSummary
from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary
from datatypes.magicplan.domain.mapper import map_plan
from datatypes.magicplan.domain.models import Plan
@ -17,20 +17,26 @@ PLAN_ID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
@pytest.fixture(scope="module")
def domain_plan() -> Plan:
data = json.loads((FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text())
return map_plan(MagicPlan.model_validate(data["data"]))
data = json.loads(
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
)
return map_plan(MagicPlanPlan.model_validate(data["data"]))
@pytest.fixture(scope="module")
def api_magic_plan() -> MagicPlan:
data = json.loads((FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text())
return MagicPlan.model_validate(data["data"])
def api_magic_plan() -> MagicPlanPlan:
data = json.loads(
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
)
return MagicPlanPlan.model_validate(data["data"])
@pytest.fixture(scope="module")
def plan_summary() -> PlanSummary:
data = json.loads((FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text())
return MagicPlan.model_validate(data["data"]).plan
data = json.loads(
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
)
return MagicPlanPlan.model_validate(data["data"]).plan
@pytest.fixture()
@ -59,7 +65,7 @@ def test_run_raises_when_no_plan_found(mock_client: MagicMock) -> None:
def test_run_fetches_plan_with_matched_id(
mock_client: MagicMock,
api_magic_plan: MagicPlan,
api_magic_plan: MagicPlanPlan,
plan_summary: PlanSummary,
domain_plan: Plan,
) -> None:
@ -67,9 +73,12 @@ def test_run_fetches_plan_with_matched_id(
mock_client.get_plans.return_value.plans = [plan_summary]
mock_client.get_plan.return_value = api_magic_plan
service = _make_service(mock_client)
with patch("backend.magic_plan.magic_plan_service.find_matching_plan", return_value=plan_summary), \
patch("backend.magic_plan.magic_plan_service.save_plan"), \
patch("backend.magic_plan.magic_plan_service.db_session"):
with patch(
"backend.magic_plan.magic_plan_service.find_matching_plan",
return_value=plan_summary,
), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
"backend.magic_plan.magic_plan_service.db_session"
):
service.run("2 Laburnum Way Bromley BR2 8BZ")
# Assert
mock_client.get_plan.assert_called_once_with(plan_summary.id)
@ -77,7 +86,7 @@ def test_run_fetches_plan_with_matched_id(
def test_run_returns_mapped_plan(
mock_client: MagicMock,
api_magic_plan: MagicPlan,
api_magic_plan: MagicPlanPlan,
plan_summary: PlanSummary,
domain_plan: Plan,
) -> None:
@ -85,9 +94,12 @@ def test_run_returns_mapped_plan(
mock_client.get_plans.return_value.plans = [plan_summary]
mock_client.get_plan.return_value = api_magic_plan
service = _make_service(mock_client)
with patch("backend.magic_plan.magic_plan_service.find_matching_plan", return_value=plan_summary), \
patch("backend.magic_plan.magic_plan_service.save_plan"), \
patch("backend.magic_plan.magic_plan_service.db_session"):
with patch(
"backend.magic_plan.magic_plan_service.find_matching_plan",
return_value=plan_summary,
), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
"backend.magic_plan.magic_plan_service.db_session"
):
result = service.run("2 Laburnum Way Bromley BR2 8BZ")
# Assert
assert isinstance(result, Plan)
@ -96,16 +108,19 @@ def test_run_returns_mapped_plan(
def test_run_calls_save_plan_with_mapped_plan(
mock_client: MagicMock,
api_magic_plan: MagicPlan,
api_magic_plan: MagicPlanPlan,
plan_summary: PlanSummary,
) -> None:
# Arrange
mock_client.get_plans.return_value.plans = [plan_summary]
mock_client.get_plan.return_value = api_magic_plan
service = _make_service(mock_client)
with patch("backend.magic_plan.magic_plan_service.find_matching_plan", return_value=plan_summary), \
patch("backend.magic_plan.magic_plan_service.save_plan") as mock_save, \
patch("backend.magic_plan.magic_plan_service.db_session"):
with patch(
"backend.magic_plan.magic_plan_service.find_matching_plan",
return_value=plan_summary,
), patch("backend.magic_plan.magic_plan_service.save_plan") as mock_save, patch(
"backend.magic_plan.magic_plan_service.db_session"
):
service.run("2 Laburnum Way Bromley BR2 8BZ")
# Assert — save_plan called with a Plan whose uid matches
call_args = mock_save.call_args
@ -115,14 +130,17 @@ def test_run_calls_save_plan_with_mapped_plan(
def test_run_accepts_uprn_without_error(
mock_client: MagicMock,
api_magic_plan: MagicPlan,
api_magic_plan: MagicPlanPlan,
plan_summary: PlanSummary,
) -> None:
# Arrange
mock_client.get_plans.return_value.plans = [plan_summary]
mock_client.get_plan.return_value = api_magic_plan
service = _make_service(mock_client)
with patch("backend.magic_plan.magic_plan_service.find_matching_plan", return_value=plan_summary), \
patch("backend.magic_plan.magic_plan_service.save_plan"), \
patch("backend.magic_plan.magic_plan_service.db_session"):
with patch(
"backend.magic_plan.magic_plan_service.find_matching_plan",
return_value=plan_summary,
), patch("backend.magic_plan.magic_plan_service.save_plan"), patch(
"backend.magic_plan.magic_plan_service.db_session"
):
service.run("2 Laburnum Way Bromley BR2 8BZ", uprn="100023336956")

3
datatypes/magicplan/api/response.py
View file

@ -2,7 +2,6 @@ from typing import Any, Optional, Union
from pydantic import BaseModel, ConfigDict, Field
_IGNORE = ConfigDict(extra="ignore")
_IGNORE_POPULATE = ConfigDict(extra="ignore", populate_by_name=True)
@ -287,7 +286,7 @@ class PlansListResponse(BaseModel):
plans: list[PlanSummary] = []
class MagicPlan(BaseModel):
class MagicPlanPlan(BaseModel):
model_config = _IGNORE
plan: PlanSummary
plan_detail: PlanDetail

34
datatypes/magicplan/api/tests/test_response.py
View file

@ -4,7 +4,7 @@ from typing import Any
import pytest
from datatypes.magicplan.api.response import MagicPlan, PlansListResponse
from datatypes.magicplan.api.response import MagicPlanPlan, PlansListResponse
FIXTURE_DIR = Path(__file__).parents[4] / "backend" / "magic_plan"
PLAN_ID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
@ -19,51 +19,51 @@ def raw_data() -> dict[str, Any]:
@pytest.fixture(scope="module")
def mp(raw_data: dict[str, Any]) -> MagicPlan:
return MagicPlan.model_validate(raw_data)
def mp(raw_data: dict[str, Any]) -> MagicPlanPlan:
return MagicPlanPlan.model_validate(raw_data)
def test_model_validate_does_not_raise(raw_data: dict[str, Any]):
# act
MagicPlan.model_validate(raw_data)
MagicPlanPlan.model_validate(raw_data)
def test_plan_id(mp: MagicPlan):
def test_plan_id(mp: MagicPlanPlan):
# assert
assert mp.plan.id == PLAN_ID
def test_url_3d_alias(mp: MagicPlan):
def test_url_3d_alias(mp: MagicPlanPlan):
# assert
assert mp.plan.url_3d is not None
assert mp.plan.url_3d.startswith("http")
def test_floor_count(mp: MagicPlan):
def test_floor_count(mp: MagicPlanPlan):
# assert
assert len(mp.plan_detail.plan.floors) == 2
def test_first_room_name(mp: MagicPlan):
def test_first_room_name(mp: MagicPlanPlan):
# assert
assert mp.plan_detail.plan.floors[0].rooms[0].name == "Kitchen"
def test_room_area_is_float(mp: MagicPlan):
def test_room_area_is_float(mp: MagicPlanPlan):
# arrange
room = mp.plan_detail.plan.floors[0].rooms[0]
# assert
assert isinstance(room.area, float)
def test_wall_item_symbol_id(mp: MagicPlan):
def test_wall_item_symbol_id(mp: MagicPlanPlan):
# arrange
room = mp.plan_detail.plan.floors[0].rooms[0]
# assert
assert room.wall_items[0].symbol.id != ""
def test_field_value_array(mp: MagicPlan):
def test_field_value_array(mp: MagicPlanPlan):
# arrange
room = mp.plan_detail.plan.floors[0].rooms[0]
array_field = next(f for f in room.displayable_fields if f.value.is_array)
@ -71,7 +71,7 @@ def test_field_value_array(mp: MagicPlan):
assert isinstance(array_field.value.value, list)
def test_field_value_scalar(mp: MagicPlan):
def test_field_value_scalar(mp: MagicPlanPlan):
# arrange
room = mp.plan_detail.plan.floors[0].rooms[0]
scalar_field = next(f for f in room.displayable_fields if not f.value.is_array)
@ -83,7 +83,7 @@ def test_extra_fields_ignored(raw_data: dict[str, Any]):
# arrange
data_with_extra = {**raw_data, "unknown_future_field": "whatever"}
# act
MagicPlan.model_validate(data_with_extra)
MagicPlanPlan.model_validate(data_with_extra)
# --- PlansListResponse ---
@ -102,7 +102,9 @@ def plans_response(plans_raw_data: dict[str, Any]) -> PlansListResponse:
return PlansListResponse.model_validate(plans_raw_data)
def test_plans_list_model_validate_does_not_raise(plans_raw_data: dict[str, Any]) -> None:
def test_plans_list_model_validate_does_not_raise(
plans_raw_data: dict[str, Any],
) -> None:
# act
PlansListResponse.model_validate(plans_raw_data)
@ -122,7 +124,9 @@ def test_plans_list_paging_page(plans_response: PlansListResponse) -> None:
assert plans_response.paging.page == 1
def test_plans_list_paging_next_page_is_false(plans_response: PlansListResponse) -> None:
def test_plans_list_paging_next_page_is_false(
plans_response: PlansListResponse,
) -> None:
# assert
assert plans_response.paging.next_page is False

10
datatypes/magicplan/domain/mapper.py
View file

@ -1,9 +1,11 @@
from typing import Optional
import datatypes.magicplan.api.response as api
from datatypes.magicplan.api.response import MagicPlan
from datatypes.magicplan.api.response import MagicPlanPlan
from datatypes.magicplan.domain.models import Plan, Floor, Room, Window, Door
def map_plan(mp: MagicPlan) -> Plan:
def map_plan(mp: MagicPlanPlan) -> Plan:
return Plan(
uid=mp.plan.id,
name=mp.plan.name,
@ -13,7 +15,7 @@ def map_plan(mp: MagicPlan) -> Plan:
)
def _map_address(addr: api.Address | None) -> str | None:
def _map_address(addr: Optional[api.Address]) -> Optional[str]:
if addr is None:
return None
street = " ".join(p for p in [addr.street_number, addr.street] if p) or None
@ -43,7 +45,7 @@ def _map_room(r: api.Room) -> Room:
)
def _parse_dimensions(dimensions: str | None) -> tuple[float, float]:
def _parse_dimensions(dimensions: Optional[str]) -> tuple[float, float]:
if not dimensions:
return 0.0, 0.0
parts = dimensions.split(" x ")

14
datatypes/magicplan/domain/tests/test_mapper.py
View file

@ -4,7 +4,7 @@ from typing import Any
import pytest
from datatypes.magicplan.api.response import MagicPlan
from datatypes.magicplan.api.response import MagicPlanPlan
from datatypes.magicplan.domain.mapper import map_plan
from datatypes.magicplan.domain.models import Plan
@ -22,12 +22,12 @@ def raw_data() -> dict[str, Any]:
@pytest.fixture(scope="module")
def mp(raw_data: dict[str, Any]) -> MagicPlan:
return MagicPlan.model_validate(raw_data)
def mp(raw_data: dict[str, Any]) -> MagicPlanPlan:
return MagicPlanPlan.model_validate(raw_data)
@pytest.fixture(scope="module")
def plan(mp: MagicPlan) -> Plan:
def plan(mp: MagicPlanPlan) -> Plan:
return map_plan(mp)
@ -119,7 +119,7 @@ def raw_data_2() -> dict[str, Any]:
@pytest.fixture(scope="module")
def plan2(raw_data_2: dict[str, Any]) -> Plan:
return map_plan(MagicPlan.model_validate(raw_data_2))
return map_plan(MagicPlanPlan.model_validate(raw_data_2))
def test_plan2_uid(plan2: Plan):
@ -201,7 +201,7 @@ def plan3() -> Plan:
payload = json.loads(
(FIXTURE_DIR / "magicplan_api_plan_response_example_3.json").read_text()
)
return map_plan(MagicPlan.model_validate(payload["data"]))
return map_plan(MagicPlanPlan.model_validate(payload["data"]))
def test_plan3_address_uses_street_number_and_omits_city(plan3: Plan):
@ -220,7 +220,7 @@ def plan4() -> Plan:
payload = json.loads(
(FIXTURE_DIR / "magicplan_api_plan_response_example_4.json").read_text()
)
return map_plan(MagicPlan.model_validate(payload["data"]))
return map_plan(MagicPlanPlan.model_validate(payload["data"]))
def test_plan4_address_uses_street_number_when_street_absent(plan4: Plan):