mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
Merge c22ee3821b into 98297f803a
This commit is contained in:
Daniel Roth
GitHub
commit
1f8420434a
54 changed files with 1495 additions and 411270 deletions
|
|
@ -1,7 +1,7 @@
|
|||
import re
|
||||
from typing import Optional
|
||||
|
||||
from datatypes.magicplan.api.response import PlanSummary
|
||||
from domain.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)
|
||||
|
||||
47
applications/magic_plan/handler.py
Normal file
47
applications/magic_plan/handler.py
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
import os
|
||||
from typing import Any
|
||||
|
||||
import boto3
|
||||
|
||||
from infrastructure.magic_plan.config import MagicPlanConfig
|
||||
from infrastructure.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from infrastructure.s3.s3_client import S3Client
|
||||
from orchestration.magic_plan_orchestrator import MagicPlanOrchestrator
|
||||
from applications.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
from domain.magicplan.models import Plan
|
||||
from utilities.aws_lambda.subtask_handler import subtask_handler
|
||||
from utilities.logger import setup_logger
|
||||
|
||||
logger = setup_logger()
|
||||
|
||||
|
||||
@subtask_handler()
|
||||
def handler(body: dict[str, Any], context: Any) -> str:
|
||||
config = MagicPlanConfig.from_env(os.environ)
|
||||
payload = MagicPlanTriggerRequest.model_validate(body)
|
||||
client = MagicPlanClient(
|
||||
customer_id=config.customer_id,
|
||||
api_key=config.api_key,
|
||||
)
|
||||
|
||||
boto3_client: Any = boto3.client # type: ignore
|
||||
boto_s3: Any = boto3_client("s3")
|
||||
s3_client = S3Client(
|
||||
boto_s3_client=boto_s3, bucket="retrofit-energy-assessments-dev"
|
||||
)
|
||||
|
||||
# TODO: read s3_bucket from env var so staging/prod use the correct bucket
|
||||
plan: Plan = MagicPlanOrchestrator(client, s3_client).run(payload)
|
||||
logger.info("Saved MagicPlan plan uid=%s", plan.uid)
|
||||
return plan.uid
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
event = {
|
||||
"Records": [
|
||||
{
|
||||
"body": '{"address": "2 Laburnum Way Bromley BR2 8BZ", "hubspot_deal_id": "local-test-deal"}',
|
||||
}
|
||||
]
|
||||
}
|
||||
handler(event, None)
|
||||
17
applications/magic_plan/handler/Dockerfile
Normal file
17
applications/magic_plan/handler/Dockerfile
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
FROM public.ecr.aws/lambda/python:3.11
|
||||
|
||||
WORKDIR /var/task
|
||||
|
||||
COPY applications/magic_plan/handler/requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
COPY utilities/ utilities/
|
||||
COPY backend/ backend/
|
||||
COPY applications/ applications/
|
||||
COPY domain/ domain/
|
||||
COPY datatypes/ datatypes/
|
||||
COPY orchestration/ orchestration/
|
||||
COPY repositories/ repositories/
|
||||
COPY infrastructure/ infrastructure/
|
||||
|
||||
CMD ["applications.magic_plan.handler.handler"]
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
from datatypes.magicplan.api.response import PlanSummary
|
||||
from backend.magic_plan.address_matcher import find_matching_plan, _extract_postcode
|
||||
from domain.magicplan.api.response import PlanSummary
|
||||
from applications.magic_plan.address_matcher import find_matching_plan, _extract_postcode
|
||||
|
||||
|
||||
def _make_plan(
|
||||
|
|
@ -1,143 +0,0 @@
|
|||
from typing import Any, cast
|
||||
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlmodel import Session, col
|
||||
|
||||
from datatypes.magicplan.domain.models import Floor, Plan
|
||||
from backend.app.db.models.magic_plan import (
|
||||
MagicPlanDoorModel,
|
||||
MagicPlanFloorModel,
|
||||
MagicPlanPlanModel,
|
||||
MagicPlanRoomModel,
|
||||
MagicPlanWindowModel,
|
||||
)
|
||||
|
||||
|
||||
def save_plan(session: Session, plan: Plan, uploaded_file_id: int) -> None:
|
||||
plan_id: int = _upsert_plan(session, plan, uploaded_file_id)
|
||||
_delete_children(session, plan_id)
|
||||
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, uploaded_file_id: int) -> int:
|
||||
stmt = (
|
||||
pg_insert(MagicPlanPlanModel)
|
||||
.values(
|
||||
magic_plan_uid=plan.uid,
|
||||
name=plan.name,
|
||||
address=plan.address,
|
||||
postcode=plan.postcode,
|
||||
uploaded_file_id=uploaded_file_id,
|
||||
)
|
||||
.on_conflict_do_update(
|
||||
index_elements=["magic_plan_uid"],
|
||||
set_={
|
||||
"name": plan.name,
|
||||
"address": plan.address,
|
||||
"postcode": plan.postcode,
|
||||
"uploaded_file_id": uploaded_file_id,
|
||||
},
|
||||
)
|
||||
.returning(col(MagicPlanPlanModel.id))
|
||||
)
|
||||
row_id: int = session.execute(stmt).scalar_one()
|
||||
return row_id
|
||||
|
||||
|
||||
def _delete_children(session: Session, plan_id: int) -> None:
|
||||
floor_subq = (
|
||||
select(col(MagicPlanFloorModel.id))
|
||||
.where(col(MagicPlanFloorModel.magic_plan_plan_id) == plan_id)
|
||||
.scalar_subquery()
|
||||
)
|
||||
room_subq = (
|
||||
select(col(MagicPlanRoomModel.id))
|
||||
.where(col(MagicPlanRoomModel.magic_plan_floor_id).in_(floor_subq))
|
||||
.scalar_subquery()
|
||||
)
|
||||
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
|
||||
]
|
||||
result = session.execute(
|
||||
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]:
|
||||
rows: list[dict[str, Any]] = [
|
||||
{
|
||||
"magic_plan_floor_id": floor_id,
|
||||
"name": room.name,
|
||||
"width_m": room.width_m,
|
||||
"length_m": room.length_m,
|
||||
"area_m2": room.area_m2,
|
||||
}
|
||||
for floor, floor_id in zip(floors, floor_ids)
|
||||
for room in floor.rooms
|
||||
]
|
||||
result = session.execute(
|
||||
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:
|
||||
all_rooms = [room for floor in floors for room in floor.rooms]
|
||||
|
||||
window_rows: list[dict[str, Any]] = [
|
||||
{
|
||||
"magic_plan_room_id": room_id,
|
||||
"width_m": window.width_m,
|
||||
"height_m": window.height_m,
|
||||
"area_m2": window.area_m2,
|
||||
"opening_type": window.opening_type,
|
||||
}
|
||||
for room, room_id in zip(all_rooms, room_ids)
|
||||
for window in room.windows
|
||||
]
|
||||
door_rows: list[dict[str, Any]] = [
|
||||
{
|
||||
"magic_plan_room_id": room_id,
|
||||
"width_mm": door.width_mm,
|
||||
}
|
||||
for room, room_id in zip(all_rooms, room_ids)
|
||||
for door in room.doors
|
||||
]
|
||||
|
||||
if window_rows:
|
||||
session.execute(pg_insert(MagicPlanWindowModel).values(window_rows))
|
||||
if door_rows:
|
||||
session.execute(pg_insert(MagicPlanDoorModel).values(door_rows))
|
||||
|
|
@ -1,41 +0,0 @@
|
|||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlmodel import SQLModel
|
||||
|
||||
import backend.app.db.models.magic_plan # noqa: F401 — registers MagicPlan models with SQLModel.metadata
|
||||
|
||||
# TODO: promote to backend/app/db/conftest.py once a second DB-touching test directory appears under this tree
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def engine(postgresql):
|
||||
connection_string = (
|
||||
f"postgresql+psycopg://"
|
||||
f"{postgresql.info.user}:"
|
||||
f"{postgresql.info.password}@"
|
||||
f"{postgresql.info.host}:"
|
||||
f"{postgresql.info.port}/"
|
||||
f"{postgresql.info.dbname}"
|
||||
)
|
||||
|
||||
engine = create_engine(connection_string)
|
||||
SQLModel.metadata.create_all(engine)
|
||||
|
||||
yield engine
|
||||
|
||||
SQLModel.metadata.drop_all(engine)
|
||||
engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def db_session(engine):
|
||||
connection = engine.connect()
|
||||
transaction = connection.begin()
|
||||
session = sessionmaker(bind=connection)()
|
||||
|
||||
yield session
|
||||
|
||||
session.close()
|
||||
transaction.rollback()
|
||||
connection.close()
|
||||
|
|
@ -1,115 +0,0 @@
|
|||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlmodel import SQLModel
|
||||
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan
|
||||
from datatypes.magicplan.domain.mapper import map_plan
|
||||
from datatypes.magicplan.domain.models import Plan
|
||||
|
||||
from backend.app.db.functions.magic_plan_functions import save_plan
|
||||
from backend.app.db.models.magic_plan import (
|
||||
MagicPlanDoorModel,
|
||||
MagicPlanFloorModel,
|
||||
MagicPlanPlanModel,
|
||||
MagicPlanRoomModel,
|
||||
MagicPlanWindowModel,
|
||||
)
|
||||
|
||||
FIXTURE_DIR = Path(__file__).parents[4] / "magic_plan"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def domain_plan() -> Plan:
|
||||
data = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
|
||||
)
|
||||
return map_plan(MagicPlanPlan.model_validate(data["data"]))
|
||||
|
||||
|
||||
def _count(session: Session, model: type[SQLModel]) -> int:
|
||||
return session.execute(select(func.count()).select_from(model)).scalar_one()
|
||||
|
||||
|
||||
def test_plan_row_present_after_save(db_session: Session, domain_plan: Plan) -> None:
|
||||
# Act
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Assert
|
||||
assert _count(db_session, MagicPlanPlanModel) == 1
|
||||
|
||||
|
||||
def test_floor_count_matches_domain(db_session: Session, domain_plan: Plan) -> None:
|
||||
# Arrange
|
||||
expected = len(domain_plan.floors)
|
||||
# Act
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Assert
|
||||
assert _count(db_session, MagicPlanFloorModel) == expected
|
||||
|
||||
|
||||
def test_room_count_matches_domain(db_session: Session, domain_plan: Plan) -> None:
|
||||
# Arrange
|
||||
expected = sum(len(f.rooms) for f in domain_plan.floors)
|
||||
# Act
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Assert
|
||||
assert _count(db_session, MagicPlanRoomModel) == expected
|
||||
|
||||
|
||||
def test_window_count_matches_domain(db_session: Session, domain_plan: Plan) -> None:
|
||||
# Arrange
|
||||
expected = sum(len(r.windows) for f in domain_plan.floors for r in f.rooms)
|
||||
# Act
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Assert
|
||||
assert _count(db_session, MagicPlanWindowModel) == expected
|
||||
|
||||
|
||||
def test_door_count_matches_domain(db_session: Session, domain_plan: Plan) -> None:
|
||||
# Arrange
|
||||
expected = sum(len(r.doors) for f in domain_plan.floors for r in f.rooms)
|
||||
# Act
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Assert
|
||||
assert _count(db_session, MagicPlanDoorModel) == expected
|
||||
|
||||
|
||||
def test_save_plan_idempotent(db_session: Session, domain_plan: Plan) -> None:
|
||||
# Act — call twice within the same session
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Assert — same row counts as a single call
|
||||
assert _count(db_session, MagicPlanPlanModel) == 1
|
||||
assert _count(db_session, MagicPlanFloorModel) == len(domain_plan.floors)
|
||||
assert _count(db_session, MagicPlanRoomModel) == sum(
|
||||
len(f.rooms) for f in domain_plan.floors
|
||||
)
|
||||
assert _count(db_session, MagicPlanWindowModel) == sum(
|
||||
len(r.windows) for f in domain_plan.floors for r in f.rooms
|
||||
)
|
||||
assert _count(db_session, MagicPlanDoorModel) == sum(
|
||||
len(r.doors) for f in domain_plan.floors for r in f.rooms
|
||||
)
|
||||
|
||||
|
||||
def test_uploaded_file_id_stored_after_save(db_session: Session, domain_plan: Plan) -> None:
|
||||
# Act
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Assert
|
||||
row = db_session.execute(select(MagicPlanPlanModel)).scalar_one()
|
||||
assert row.uploaded_file_id == 1
|
||||
|
||||
|
||||
def test_save_plan_updates_uploaded_file_id_on_reingest(
|
||||
db_session: Session, domain_plan: Plan
|
||||
) -> None:
|
||||
# Arrange
|
||||
save_plan(db_session, domain_plan, 1)
|
||||
# Act
|
||||
save_plan(db_session, domain_plan, 2)
|
||||
# Assert
|
||||
row = db_session.execute(select(MagicPlanPlanModel)).scalar_one()
|
||||
assert row.uploaded_file_id == 2
|
||||
|
|
@ -1,53 +0,0 @@
|
|||
from typing import Optional
|
||||
|
||||
from sqlmodel import Field, SQLModel
|
||||
|
||||
|
||||
class MagicPlanPlanModel(SQLModel, table=True):
|
||||
__tablename__ = "magic_plan_plan"
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_uid: Optional[str] = Field(default=None, unique=True, index=True)
|
||||
name: Optional[str] = None
|
||||
address: Optional[str] = None
|
||||
postcode: Optional[str] = None
|
||||
uploaded_file_id: Optional[int] = Field(default=None)
|
||||
|
||||
|
||||
class MagicPlanFloorModel(SQLModel, table=True):
|
||||
__tablename__ = "magic_plan_floor"
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_plan_id: int = Field(foreign_key="magic_plan_plan.id")
|
||||
level: Optional[int] = None
|
||||
|
||||
|
||||
class MagicPlanRoomModel(SQLModel, table=True):
|
||||
__tablename__ = "magic_plan_room"
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_floor_id: int = Field(foreign_key="magic_plan_floor.id")
|
||||
name: Optional[str] = None
|
||||
width_m: Optional[float] = None
|
||||
length_m: Optional[float] = None
|
||||
area_m2: Optional[float] = None
|
||||
|
||||
|
||||
class MagicPlanWindowModel(SQLModel, table=True):
|
||||
__tablename__ = "magic_plan_window"
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_room_id: int = Field(foreign_key="magic_plan_room.id")
|
||||
width_m: Optional[float] = None
|
||||
height_m: Optional[float] = None
|
||||
area_m2: Optional[float] = None
|
||||
opening_type: Optional[str] = None
|
||||
|
||||
|
||||
class MagicPlanDoorModel(SQLModel, table=True):
|
||||
__tablename__ = "magic_plan_door"
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_room_id: int = Field(foreign_key="magic_plan_room.id")
|
||||
width_mm: Optional[float] = None
|
||||
type: Optional[str] = None
|
||||
|
|
@ -1,4 +1,8 @@
|
|||
from backend.app.db.models.magic_plan import (
|
||||
from typing import Any, cast
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from infrastructure.postgres.magic_plan_tables import (
|
||||
MagicPlanDoorModel,
|
||||
MagicPlanFloorModel,
|
||||
MagicPlanPlanModel,
|
||||
|
|
@ -6,6 +10,10 @@ from backend.app.db.models.magic_plan import (
|
|||
MagicPlanWindowModel,
|
||||
)
|
||||
|
||||
|
||||
def _table(model: type[Any]) -> sa.Table:
|
||||
return cast(sa.Table, getattr(model, "__table__"))
|
||||
|
||||
# --- MagicPlanPlan ---
|
||||
|
||||
|
||||
|
|
@ -14,20 +22,17 @@ def test_plan_table_name() -> None:
|
|||
|
||||
|
||||
def test_plan_has_magic_plan_uid_column() -> None:
|
||||
assert "magic_plan_uid" in MagicPlanPlanModel.__table__.columns
|
||||
assert "magic_plan_uid" in _table(MagicPlanPlanModel).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
|
||||
t = _table(MagicPlanPlanModel)
|
||||
col = t.columns["magic_plan_uid"]
|
||||
has_unique_constraint = any(
|
||||
isinstance(c, sa.UniqueConstraint) and "magic_plan_uid" in c.columns
|
||||
for c in t.constraints
|
||||
)
|
||||
assert has_unique_constraint or col.unique
|
||||
|
||||
|
||||
def test_plan_instantiation() -> None:
|
||||
|
|
@ -47,7 +52,7 @@ def test_floor_table_name() -> None:
|
|||
|
||||
|
||||
def test_floor_fk_column_name() -> None:
|
||||
assert "magic_plan_plan_id" in MagicPlanFloorModel.__table__.columns
|
||||
assert "magic_plan_plan_id" in _table(MagicPlanFloorModel).columns
|
||||
|
||||
|
||||
def test_floor_has_level() -> None:
|
||||
|
|
@ -63,11 +68,11 @@ def test_room_table_name() -> None:
|
|||
|
||||
|
||||
def test_room_fk_column_name() -> None:
|
||||
assert "magic_plan_floor_id" in MagicPlanRoomModel.__table__.columns
|
||||
assert "magic_plan_floor_id" in _table(MagicPlanRoomModel).columns
|
||||
|
||||
|
||||
def test_room_has_measurement_columns() -> None:
|
||||
cols = MagicPlanRoomModel.__table__.columns
|
||||
cols = _table(MagicPlanRoomModel).columns
|
||||
assert "width_m" in cols
|
||||
assert "length_m" in cols
|
||||
assert "area_m2" in cols
|
||||
|
|
@ -89,15 +94,14 @@ def test_window_table_name() -> None:
|
|||
|
||||
|
||||
def test_window_fk_column_name() -> None:
|
||||
assert "magic_plan_room_id" in MagicPlanWindowModel.__table__.columns
|
||||
assert "magic_plan_room_id" in _table(MagicPlanWindowModel).columns
|
||||
|
||||
|
||||
def test_window_has_measurement_columns() -> None:
|
||||
cols = MagicPlanWindowModel.__table__.columns
|
||||
cols = _table(MagicPlanWindowModel).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:
|
||||
|
|
@ -106,9 +110,8 @@ def test_window_instantiation() -> None:
|
|||
width_m=1.4,
|
||||
height_m=1.2,
|
||||
area_m2=1.68,
|
||||
opening_type="casement",
|
||||
)
|
||||
assert window.opening_type == "casement"
|
||||
assert window.width_m == 1.4
|
||||
|
||||
|
||||
# --- MagicPlanDoor ---
|
||||
|
|
@ -119,16 +122,16 @@ def test_door_table_name() -> None:
|
|||
|
||||
|
||||
def test_door_fk_column_name() -> None:
|
||||
assert "magic_plan_room_id" in MagicPlanDoorModel.__table__.columns
|
||||
assert "magic_plan_room_id" in _table(MagicPlanDoorModel).columns
|
||||
|
||||
|
||||
def test_door_has_width_mm_and_type() -> None:
|
||||
cols = MagicPlanDoorModel.__table__.columns
|
||||
cols = _table(MagicPlanDoorModel).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
|
||||
door = MagicPlanDoorModel(magic_plan_room_id=1, width_mm=790.0, type="hinged")
|
||||
assert door.width_mm == 790.0
|
||||
assert door.type == "hinged"
|
||||
|
|
|
|||
|
|
@ -1,39 +0,0 @@
|
|||
from typing import Any
|
||||
|
||||
from backend.app.config import get_settings
|
||||
from backend.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from backend.magic_plan.magic_plan_service import MagicPlanService
|
||||
from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
from datatypes.magicplan.domain.models import Plan
|
||||
from backend.app.db.models.tasks import SourceEnum
|
||||
from backend.utils.subtasks import task_handler
|
||||
from utils.logger import setup_logger
|
||||
|
||||
logger = setup_logger()
|
||||
|
||||
|
||||
@task_handler(task_source="magic_plan", source=SourceEnum.HUBSPOT_DEAL)
|
||||
def handler(body: dict[str, Any], context: Any) -> str:
|
||||
settings = get_settings()
|
||||
payload = MagicPlanTriggerRequest.model_validate(body)
|
||||
client = MagicPlanClient(
|
||||
customer_id=settings.MAGICPLAN_CUSTOMER_ID,
|
||||
api_key=settings.MAGICPLAN_API_KEY,
|
||||
)
|
||||
# TODO: read s3_bucket from env var so staging/prod use the correct bucket
|
||||
plan: Plan = MagicPlanService(
|
||||
client, s3_bucket="retrofit-energy-assessments-dev"
|
||||
).run(payload)
|
||||
logger.info("Saved MagicPlan plan uid=%s", plan.uid)
|
||||
return plan.uid
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
event = {
|
||||
"Records": [
|
||||
{
|
||||
"body": '{"address": "2 Laburnum Way Bromley BR2 8BZ", "hubspot_deal_id": "local-test-deal"}',
|
||||
}
|
||||
]
|
||||
}
|
||||
handler(event, None)
|
||||
|
|
@ -1,12 +0,0 @@
|
|||
FROM public.ecr.aws/lambda/python:3.11
|
||||
|
||||
WORKDIR /var/task
|
||||
|
||||
COPY backend/magic_plan/handler/requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
COPY utils/ utils/
|
||||
COPY backend/ backend/
|
||||
COPY datatypes/ datatypes/
|
||||
|
||||
CMD ["backend.magic_plan.handler.handler"]
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
|
|
@ -1,109 +0,0 @@
|
|||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from backend.magic_plan.handler import handler
|
||||
|
||||
ADDRESS = "2 Laburnum Way Bromley BR2 8BZ"
|
||||
PLAN_UID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
|
||||
|
||||
|
||||
def _make_settings(**overrides: str) -> MagicMock:
|
||||
settings = MagicMock()
|
||||
settings.MAGICPLAN_CUSTOMER_ID = overrides.get("customer_id", "cust-123")
|
||||
settings.MAGICPLAN_API_KEY = overrides.get("api_key", "key-abc")
|
||||
return settings
|
||||
|
||||
|
||||
def _call_handler(body: dict[str, Any]) -> Any:
|
||||
return handler.__wrapped__(body, None) # type: ignore[attr-defined]
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_plan() -> MagicMock:
|
||||
plan = MagicMock()
|
||||
plan.uid = PLAN_UID
|
||||
return plan
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_service(mock_plan: MagicMock) -> MagicMock:
|
||||
service = MagicMock()
|
||||
service.run.return_value = mock_plan
|
||||
return service
|
||||
|
||||
|
||||
# --- request validation ---
|
||||
|
||||
|
||||
def test_handler_raises_on_missing_address(mock_plan: MagicMock) -> None:
|
||||
# Arrange
|
||||
body: dict[str, Any] = {}
|
||||
with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings()), \
|
||||
patch("backend.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("backend.magic_plan.handler.MagicPlanService"):
|
||||
# Act / Assert
|
||||
with pytest.raises(ValidationError):
|
||||
_call_handler(body)
|
||||
|
||||
|
||||
# --- client construction ---
|
||||
|
||||
|
||||
def test_handler_constructs_client_from_settings(mock_service: MagicMock) -> None:
|
||||
# Arrange
|
||||
body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
|
||||
with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings(customer_id="cust-xyz", api_key="key-xyz")), \
|
||||
patch("backend.magic_plan.handler.MagicPlanClient") as MockClient, \
|
||||
patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):
|
||||
# Act
|
||||
_call_handler(body)
|
||||
# Assert
|
||||
MockClient.assert_called_once_with(customer_id="cust-xyz", api_key="key-xyz")
|
||||
|
||||
|
||||
# --- service orchestration ---
|
||||
|
||||
|
||||
def test_handler_calls_service_run_with_address(mock_service: MagicMock) -> None:
|
||||
# Arrange
|
||||
body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
|
||||
with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings()), \
|
||||
patch("backend.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):
|
||||
# Act
|
||||
_call_handler(body)
|
||||
# Assert
|
||||
mock_service.run.assert_called_once()
|
||||
request = mock_service.run.call_args.args[0]
|
||||
assert request.address == ADDRESS
|
||||
assert request.uprn is None
|
||||
|
||||
|
||||
def test_handler_passes_uprn_to_service(mock_service: MagicMock) -> None:
|
||||
# Arrange
|
||||
body = {"address": ADDRESS, "uprn": "100023336956", "hubspot_deal_id": "deal-123"}
|
||||
with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings()), \
|
||||
patch("backend.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):
|
||||
# Act
|
||||
_call_handler(body)
|
||||
# Assert
|
||||
mock_service.run.assert_called_once()
|
||||
request = mock_service.run.call_args.args[0]
|
||||
assert request.address == ADDRESS
|
||||
assert request.uprn == "100023336956"
|
||||
|
||||
|
||||
def test_handler_returns_plan_uid(mock_service: MagicMock) -> None:
|
||||
# Arrange
|
||||
body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
|
||||
with patch("backend.magic_plan.handler.get_settings", return_value=_make_settings()), \
|
||||
patch("backend.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("backend.magic_plan.handler.MagicPlanService", return_value=mock_service):
|
||||
# Act
|
||||
result = _call_handler(body)
|
||||
# Assert
|
||||
assert result == PLAN_UID
|
||||
|
|
@ -1,67 +0,0 @@
|
|||
from typing import Optional
|
||||
|
||||
import datatypes.magicplan.api.response as api
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan
|
||||
from datatypes.magicplan.domain.models import Plan, Floor, Room, Window, Door
|
||||
|
||||
|
||||
def map_plan(mp: MagicPlanPlan) -> Plan:
|
||||
return Plan(
|
||||
uid=mp.plan.id,
|
||||
name=mp.plan.name,
|
||||
address=_map_address(mp.plan.address),
|
||||
postcode=mp.plan.address.postal_code if mp.plan.address else None,
|
||||
floors=[_map_floor(f) for f in mp.plan_detail.plan.floors],
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
parts = [p for p in [street, addr.city, addr.country] if p]
|
||||
return ", ".join(parts) if parts else None
|
||||
|
||||
|
||||
def _map_floor(f: api.Floor) -> Floor:
|
||||
return Floor(
|
||||
level=f.level,
|
||||
name=f.name,
|
||||
rooms=[_map_room(r) for r in f.rooms],
|
||||
)
|
||||
|
||||
|
||||
def _map_room(r: api.Room) -> Room:
|
||||
width, length = _parse_dimensions(r.dimensions)
|
||||
return Room(
|
||||
name=r.name,
|
||||
width_m=width,
|
||||
length_m=length,
|
||||
area_m2=round(r.area, 2),
|
||||
windows=[
|
||||
_map_window(wi) for wi in r.wall_items if wi.symbol.id.startswith("window")
|
||||
],
|
||||
doors=[_map_door(wi) for wi in r.wall_items if wi.symbol.id.startswith("door")],
|
||||
)
|
||||
|
||||
|
||||
def _parse_dimensions(dimensions: Optional[str]) -> tuple[float, float]:
|
||||
if not dimensions:
|
||||
return 0.0, 0.0
|
||||
parts = dimensions.split(" x ")
|
||||
width = round(float(parts[0].split(" m")[0]), 2)
|
||||
length = round(float(parts[1].split(" m")[0]), 2)
|
||||
return width, length
|
||||
|
||||
|
||||
def _map_window(wi: api.WallItem) -> Window:
|
||||
return Window(
|
||||
width_m=round(wi.size.x, 2),
|
||||
height_m=round(wi.size.z, 2),
|
||||
area_m2=round(wi.size.x * wi.size.z, 2),
|
||||
opening_type=wi.symbol.id.removeprefix("window"),
|
||||
)
|
||||
|
||||
|
||||
def _map_door(wi: api.WallItem) -> Door:
|
||||
return Door(width_mm=round(wi.size.x, 2))
|
||||
|
|
@ -1,227 +0,0 @@
|
|||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan
|
||||
from datatypes.magicplan.domain.mapper import map_plan
|
||||
from datatypes.magicplan.domain.models import Plan
|
||||
|
||||
FIXTURE_DIR = Path(__file__).parents[4] / "backend" / "magic_plan"
|
||||
PLAN_ID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
|
||||
PLAN_ID_2 = "9f9889ff-793e-4e9a-a6f0-e22f5b0f5365"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def raw_data() -> dict[str, Any]:
|
||||
payload = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
|
||||
)
|
||||
return payload["data"]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def mp(raw_data: dict[str, Any]) -> MagicPlanPlan:
|
||||
return MagicPlanPlan.model_validate(raw_data)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def plan(mp: MagicPlanPlan) -> Plan:
|
||||
return map_plan(mp)
|
||||
|
||||
|
||||
def test_plan_uid(plan: Plan):
|
||||
assert plan.uid == PLAN_ID
|
||||
|
||||
|
||||
def test_floor_count(plan: Plan):
|
||||
assert len(plan.floors) == 2
|
||||
|
||||
|
||||
def test_first_room_name(plan: Plan):
|
||||
assert plan.floors[0].rooms[0].name == "Kitchen"
|
||||
|
||||
|
||||
def test_room_dimensions_are_floats(plan: Plan):
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert isinstance(room.width_m, float)
|
||||
assert isinstance(room.length_m, float)
|
||||
assert isinstance(room.area_m2, float)
|
||||
|
||||
|
||||
def test_room_area_rounded_to_2dp(plan: Plan):
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert room.area_m2 == 7.95
|
||||
|
||||
|
||||
def test_room_dimensions_parsed_from_string(plan: Plan):
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert room.width_m == pytest.approx(2.67)
|
||||
assert room.length_m == pytest.approx(2.98)
|
||||
|
||||
|
||||
def test_kitchen_has_windows(plan: Plan):
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert len(room.windows) >= 1
|
||||
|
||||
|
||||
def test_window_fields_are_floats(plan: Plan):
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
assert isinstance(window.width_m, float)
|
||||
assert isinstance(window.height_m, float)
|
||||
assert isinstance(window.area_m2, float)
|
||||
|
||||
|
||||
def test_window_opening_type_prefix_stripped(plan: Plan):
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
assert not window.opening_type.startswith("window")
|
||||
assert window.opening_type == "casement"
|
||||
|
||||
|
||||
def test_window_area_is_width_times_height(plan: Plan):
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
assert window.area_m2 == pytest.approx(window.width_m * window.height_m, rel=1e-2)
|
||||
|
||||
|
||||
def test_window_dimensions_rounded_to_2dp(plan: Plan):
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
assert window.width_m == 1.40
|
||||
assert window.height_m == 1.20
|
||||
assert window.area_m2 == 1.68
|
||||
|
||||
|
||||
def test_door_width_rounded_to_2dp(plan: Plan):
|
||||
door = plan.floors[0].rooms[0].doors[0]
|
||||
assert door.width_mm == 0.79
|
||||
|
||||
|
||||
def test_kitchen_has_doors(plan: Plan):
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert len(room.doors) >= 1
|
||||
|
||||
|
||||
def test_door_width_is_float(plan: Plan):
|
||||
door = plan.floors[0].rooms[0].doors[0]
|
||||
assert isinstance(door.width_mm, float)
|
||||
|
||||
|
||||
# --- Fixture 2: magicplan_api_plan_response_example_2.json ---
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def raw_data_2() -> dict[str, Any]:
|
||||
payload = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example_2.json").read_text()
|
||||
)
|
||||
return payload["data"]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def plan2(raw_data_2: dict[str, Any]) -> Plan:
|
||||
return map_plan(MagicPlanPlan.model_validate(raw_data_2))
|
||||
|
||||
|
||||
def test_plan2_uid(plan2: Plan):
|
||||
assert plan2.uid == PLAN_ID_2
|
||||
|
||||
|
||||
def test_plan2_floor_count(plan2: Plan):
|
||||
assert len(plan2.floors) == 3
|
||||
|
||||
|
||||
def test_plan2_first_room_name(plan2: Plan):
|
||||
assert plan2.floors[0].rooms[0].name == "Toilet"
|
||||
|
||||
|
||||
def test_plan2_room_area_rounded_to_2dp(plan2: Plan):
|
||||
room = plan2.floors[0].rooms[0]
|
||||
assert room.area_m2 == 0.96
|
||||
|
||||
|
||||
def test_plan2_room_dimensions_parsed_from_string(plan2: Plan):
|
||||
room = plan2.floors[0].rooms[0]
|
||||
assert room.width_m == pytest.approx(1.12)
|
||||
assert room.length_m == pytest.approx(0.86)
|
||||
|
||||
|
||||
def test_plan2_room_with_no_windows(plan2: Plan):
|
||||
hall = plan2.floors[0].rooms[1]
|
||||
assert hall.name == "Hall"
|
||||
assert hall.windows == []
|
||||
|
||||
|
||||
def test_plan2_window_dimensions_rounded_to_2dp(plan2: Plan):
|
||||
window = plan2.floors[0].rooms[0].windows[0]
|
||||
assert window.width_m == 0.39
|
||||
assert window.height_m == 0.67
|
||||
assert window.area_m2 == 0.26
|
||||
|
||||
|
||||
def test_plan2_window_opening_type_casement(plan2: Plan):
|
||||
window = plan2.floors[0].rooms[0].windows[0]
|
||||
assert window.opening_type == "casement"
|
||||
|
||||
|
||||
def test_plan2_window_opening_type_hung(plan2: Plan):
|
||||
bathroom1 = plan2.floors[1].rooms[1]
|
||||
assert bathroom1.name == "Bathroom 1"
|
||||
assert bathroom1.windows[0].opening_type == "hung"
|
||||
|
||||
|
||||
def test_plan2_door_width_rounded_to_2dp(plan2: Plan):
|
||||
door = plan2.floors[0].rooms[0].doors[0]
|
||||
assert door.width_mm == 0.71
|
||||
|
||||
|
||||
# --- Address and postcode fields ---
|
||||
|
||||
|
||||
def test_plan_postcode(plan: Plan):
|
||||
assert plan.postcode == "BR2 8BZ"
|
||||
|
||||
|
||||
def test_plan_address(plan: Plan):
|
||||
assert plan.address == "2 Laburnum Way, Bromley, GB"
|
||||
|
||||
|
||||
def test_plan2_postcode(plan2: Plan):
|
||||
assert plan2.postcode == "BR1 3LP"
|
||||
|
||||
|
||||
def test_plan2_address(plan2: Plan):
|
||||
assert plan2.address == "11 Station Road, Bromley, GB"
|
||||
|
||||
|
||||
# --- Fixture 3: street_number set, city absent ---
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def plan3() -> Plan:
|
||||
payload = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example_3.json").read_text()
|
||||
)
|
||||
return map_plan(MagicPlanPlan.model_validate(payload["data"]))
|
||||
|
||||
|
||||
def test_plan3_address_uses_street_number_and_omits_city(plan3: Plan):
|
||||
assert plan3.address == "2 Laburnum Way, GB"
|
||||
|
||||
|
||||
def test_plan3_postcode(plan3: Plan):
|
||||
assert plan3.postcode == "BR2 8BZ"
|
||||
|
||||
|
||||
# --- Fixture 4: street_number set, street absent ---
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def plan4() -> Plan:
|
||||
payload = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example_4.json").read_text()
|
||||
)
|
||||
return map_plan(MagicPlanPlan.model_validate(payload["data"]))
|
||||
|
||||
|
||||
def test_plan4_address_uses_street_number_when_street_absent(plan4: Plan):
|
||||
assert plan4.address == "2, Bromley, GB"
|
||||
|
|
@ -37,10 +37,10 @@ module "lambda" {
|
|||
LOG_LEVEL = "info"
|
||||
MAGICPLAN_CUSTOMER_ID = var.magicplan_customer_id
|
||||
MAGICPLAN_API_KEY = var.magicplan_api_key
|
||||
DB_USERNAME = local.db_credentials.db_assessment_model_username
|
||||
DB_PASSWORD = local.db_credentials.db_assessment_model_password
|
||||
DB_HOST = var.db_host
|
||||
DB_NAME = var.db_name
|
||||
DB_PORT = var.db_port
|
||||
POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username
|
||||
POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password
|
||||
POSTGRES_HOST = var.db_host
|
||||
POSTGRES_DATABASE = var.db_name
|
||||
POSTGRES_PORT = var.db_port
|
||||
}
|
||||
}
|
||||
|
|
|
|||
128
domain/magicplan/mapper.py
Normal file
128
domain/magicplan/mapper.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
from typing import Optional
|
||||
|
||||
import domain.magicplan.api.response as api
|
||||
from domain.magicplan.api.response import MagicPlanPlan
|
||||
from domain.magicplan.models import (
|
||||
Door,
|
||||
DoorVentilation,
|
||||
Floor,
|
||||
Plan,
|
||||
Room,
|
||||
Window,
|
||||
WindowVentilation,
|
||||
)
|
||||
|
||||
|
||||
def map_plan(mp: MagicPlanPlan) -> Plan:
|
||||
return Plan(
|
||||
uid=mp.plan.id,
|
||||
name=mp.plan.name,
|
||||
address=map_address(mp.plan.address),
|
||||
postcode=mp.plan.address.postal_code if mp.plan.address else None,
|
||||
floors=[_map_floor(f) for f in mp.plan_detail.plan.floors],
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
parts = [p for p in [street, addr.city, addr.country] if p]
|
||||
return ", ".join(parts) if parts else None
|
||||
|
||||
|
||||
def _map_floor(f: api.Floor) -> Floor:
|
||||
return Floor(
|
||||
level=f.level,
|
||||
name=f.name,
|
||||
rooms=[_map_room(r) for r in f.rooms],
|
||||
)
|
||||
|
||||
|
||||
def _map_room(r: api.Room) -> Room:
|
||||
width, length = _parse_dimensions(r.dimensions)
|
||||
return Room(
|
||||
name=r.name,
|
||||
width_m=width,
|
||||
length_m=length,
|
||||
area_m2=round(r.area, 2),
|
||||
windows=[
|
||||
_map_window(wi)
|
||||
for wi in r.wall_items
|
||||
if wi.symbol.id.startswith("window") or wi.symbol.id == "doorglass"
|
||||
],
|
||||
doors=[
|
||||
_map_door(wi)
|
||||
for wi in r.wall_items
|
||||
if wi.symbol.id.startswith("door") and wi.symbol.id != "doorglass"
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def _parse_dimensions(dimensions: Optional[str]) -> tuple[float, float]:
|
||||
if not dimensions:
|
||||
return 0.0, 0.0
|
||||
parts = dimensions.split(" x ")
|
||||
width = round(float(parts[0].split(" m")[0]), 2)
|
||||
length = round(float(parts[1].split(" m")[0]), 2)
|
||||
return width, length
|
||||
|
||||
|
||||
def _map_window(wi: api.WallItem) -> Window:
|
||||
return Window(
|
||||
width_m=round(wi.size.x, 2),
|
||||
height_m=round(wi.size.z, 2),
|
||||
area_m2=round(wi.size.x * wi.size.z, 2),
|
||||
ventilation=_map_window_ventilation(wi.custom_displayable_fields),
|
||||
)
|
||||
|
||||
|
||||
def _map_window_ventilation(
|
||||
fields: list[api.SurveyField],
|
||||
) -> Optional[WindowVentilation]:
|
||||
if not fields:
|
||||
return None
|
||||
by_label = {f.label: f for f in fields}
|
||||
|
||||
def _str(label: str) -> Optional[str]:
|
||||
f = by_label.get(label)
|
||||
if f is None or not f.value.has_value:
|
||||
return None
|
||||
v = f.value.value
|
||||
return v[0] if isinstance(v, list) else v
|
||||
|
||||
def _int(label: str) -> Optional[int]:
|
||||
raw = _str(label)
|
||||
return int(raw) if raw is not None else None
|
||||
|
||||
return WindowVentilation(
|
||||
opening_type=_str("Opening Type"),
|
||||
num_openings=_int("Number of Openings (In Same Window)"),
|
||||
pct_openable=_int("% of Window Openable"),
|
||||
trickle_vent_area_mm2=_int(
|
||||
"Trickle Vent Effective Area (mm2) (No Code Then Width x Height)"
|
||||
),
|
||||
num_trickle_vents=_int("No. of Trickle Vents"),
|
||||
)
|
||||
|
||||
|
||||
def _map_door(wi: api.WallItem) -> Door:
|
||||
return Door(
|
||||
width_mm=round(wi.size.x * 1000, 0),
|
||||
height_mm=round(wi.size.z * 1000, 0),
|
||||
ventilation=_map_door_ventilation(wi.custom_displayable_fields),
|
||||
)
|
||||
|
||||
|
||||
def _map_door_ventilation(
|
||||
fields: list[api.SurveyField],
|
||||
) -> Optional[DoorVentilation]:
|
||||
if not fields:
|
||||
return None
|
||||
by_label = {f.label: f for f in fields}
|
||||
f = by_label.get("Door Undercut (mm)")
|
||||
if f is None or not f.value.has_value:
|
||||
return None
|
||||
raw = f.value.value
|
||||
undercut = float(raw[0] if isinstance(raw, list) else raw)
|
||||
return DoorVentilation(undercut_mm=undercut)
|
||||
|
|
@ -1,4 +1,19 @@
|
|||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class WindowVentilation:
|
||||
opening_type: Optional[str] = None
|
||||
num_openings: Optional[int] = None
|
||||
pct_openable: Optional[int] = None
|
||||
trickle_vent_area_mm2: Optional[int] = None
|
||||
num_trickle_vents: Optional[int] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class DoorVentilation:
|
||||
undercut_mm: Optional[float] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -6,12 +21,14 @@ class Window:
|
|||
width_m: float
|
||||
height_m: float
|
||||
area_m2: float
|
||||
opening_type: str
|
||||
ventilation: Optional[WindowVentilation] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Door:
|
||||
width_mm: float # TODO: should this be m or mm?
|
||||
width_mm: float
|
||||
height_mm: float
|
||||
ventilation: Optional[DoorVentilation] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
15
infrastructure/magic_plan/config.py
Normal file
15
infrastructure/magic_plan/config.py
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
from dataclasses import dataclass
|
||||
from typing import Mapping
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MagicPlanConfig:
|
||||
customer_id: str
|
||||
api_key: str
|
||||
|
||||
@classmethod
|
||||
def from_env(cls, env: Mapping[str, str]) -> "MagicPlanConfig":
|
||||
return cls(
|
||||
customer_id=env["MAGICPLAN_CUSTOMER_ID"],
|
||||
api_key=env["MAGICPLAN_API_KEY"],
|
||||
)
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
import requests
|
||||
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary, PlansListResponse
|
||||
from domain.magicplan.api.response import MagicPlanPlan, PlanSummary, PlansListResponse
|
||||
|
||||
_BASE_URL = "https://cloud.magicplan.app/api/v2"
|
||||
|
||||
144
infrastructure/postgres/magic_plan_tables.py
Normal file
144
infrastructure/postgres/magic_plan_tables.py
Normal file
|
|
@ -0,0 +1,144 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import ClassVar, Optional
|
||||
|
||||
from sqlmodel import Field, SQLModel
|
||||
|
||||
from domain.magicplan.models import (
|
||||
Door,
|
||||
DoorVentilation,
|
||||
Floor,
|
||||
Plan,
|
||||
Room,
|
||||
Window,
|
||||
WindowVentilation,
|
||||
)
|
||||
|
||||
|
||||
class MagicPlanPlanModel(SQLModel, table=True):
|
||||
__tablename__: ClassVar[str] = "magic_plan_plan" # pyright: ignore[reportIncompatibleVariableOverride]
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_uid: Optional[str] = Field(default=None, unique=True, index=True)
|
||||
name: Optional[str] = None
|
||||
address: Optional[str] = None
|
||||
postcode: Optional[str] = None
|
||||
uploaded_file_id: Optional[int] = Field(default=None)
|
||||
|
||||
@classmethod
|
||||
def from_domain(cls, plan: Plan, uploaded_file_id: int) -> "MagicPlanPlanModel":
|
||||
return cls(
|
||||
magic_plan_uid=plan.uid,
|
||||
name=plan.name,
|
||||
address=plan.address,
|
||||
postcode=plan.postcode,
|
||||
uploaded_file_id=uploaded_file_id,
|
||||
)
|
||||
|
||||
|
||||
class MagicPlanFloorModel(SQLModel, table=True):
|
||||
__tablename__: ClassVar[str] = "magic_plan_floor" # pyright: ignore[reportIncompatibleVariableOverride]
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_plan_id: int = Field(foreign_key="magic_plan_plan.id")
|
||||
level: Optional[int] = None
|
||||
|
||||
@classmethod
|
||||
def from_domain(cls, floor: Floor, plan_id: int) -> "MagicPlanFloorModel":
|
||||
return cls(magic_plan_plan_id=plan_id, level=floor.level)
|
||||
|
||||
|
||||
class MagicPlanRoomModel(SQLModel, table=True):
|
||||
__tablename__: ClassVar[str] = "magic_plan_room" # pyright: ignore[reportIncompatibleVariableOverride]
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_floor_id: int = Field(foreign_key="magic_plan_floor.id")
|
||||
name: Optional[str] = None
|
||||
width_m: Optional[float] = None
|
||||
length_m: Optional[float] = None
|
||||
area_m2: Optional[float] = None
|
||||
|
||||
@classmethod
|
||||
def from_domain(cls, room: Room, floor_id: int) -> "MagicPlanRoomModel":
|
||||
return cls(
|
||||
magic_plan_floor_id=floor_id,
|
||||
name=room.name,
|
||||
width_m=room.width_m,
|
||||
length_m=room.length_m,
|
||||
area_m2=room.area_m2,
|
||||
)
|
||||
|
||||
|
||||
class MagicPlanWindowModel(SQLModel, table=True):
|
||||
__tablename__: ClassVar[str] = "magic_plan_window" # pyright: ignore[reportIncompatibleVariableOverride]
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_room_id: int = Field(foreign_key="magic_plan_room.id")
|
||||
width_m: Optional[float] = None
|
||||
height_m: Optional[float] = None
|
||||
area_m2: Optional[float] = None
|
||||
|
||||
@classmethod
|
||||
def from_domain(cls, window: Window, room_id: int) -> "MagicPlanWindowModel":
|
||||
return cls(
|
||||
magic_plan_room_id=room_id,
|
||||
width_m=window.width_m,
|
||||
height_m=window.height_m,
|
||||
area_m2=window.area_m2,
|
||||
)
|
||||
|
||||
|
||||
class MagicPlanDoorModel(SQLModel, table=True):
|
||||
__tablename__: ClassVar[str] = "magic_plan_door" # pyright: ignore[reportIncompatibleVariableOverride]
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_room_id: int = Field(foreign_key="magic_plan_room.id")
|
||||
width_mm: Optional[float] = None
|
||||
height_mm: Optional[float] = None
|
||||
type: Optional[str] = None
|
||||
|
||||
@classmethod
|
||||
def from_domain(cls, door: Door, room_id: int) -> "MagicPlanDoorModel":
|
||||
return cls(magic_plan_room_id=room_id, width_mm=door.width_mm, height_mm=door.height_mm)
|
||||
|
||||
|
||||
class MagicPlanWindowVentilationModel(SQLModel, table=True):
|
||||
__tablename__: ClassVar[str] = "magic_plan_window_ventilation" # pyright: ignore[reportIncompatibleVariableOverride]
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_window_id: int = Field(foreign_key="magic_plan_window.id")
|
||||
opening_type: Optional[str] = None
|
||||
num_openings: Optional[int] = None
|
||||
pct_openable: Optional[int] = None
|
||||
trickle_vent_area_mm2: Optional[int] = None
|
||||
num_trickle_vents: Optional[int] = None
|
||||
|
||||
@classmethod
|
||||
def from_domain(
|
||||
cls, ventilation: WindowVentilation, window_id: int
|
||||
) -> "MagicPlanWindowVentilationModel":
|
||||
return cls(
|
||||
magic_plan_window_id=window_id,
|
||||
opening_type=ventilation.opening_type,
|
||||
num_openings=ventilation.num_openings,
|
||||
pct_openable=ventilation.pct_openable,
|
||||
trickle_vent_area_mm2=ventilation.trickle_vent_area_mm2,
|
||||
num_trickle_vents=ventilation.num_trickle_vents,
|
||||
)
|
||||
|
||||
|
||||
class MagicPlanDoorVentilationModel(SQLModel, table=True):
|
||||
__tablename__: ClassVar[str] = "magic_plan_door_ventilation" # pyright: ignore[reportIncompatibleVariableOverride]
|
||||
|
||||
id: Optional[int] = Field(default=None, primary_key=True)
|
||||
magic_plan_door_id: int = Field(foreign_key="magic_plan_door.id")
|
||||
undercut_mm: Optional[float] = None
|
||||
|
||||
@classmethod
|
||||
def from_domain(
|
||||
cls, ventilation: DoorVentilation, door_id: int
|
||||
) -> "MagicPlanDoorVentilationModel":
|
||||
return cls(
|
||||
magic_plan_door_id=door_id,
|
||||
undercut_mm=ventilation.undercut_mm,
|
||||
)
|
||||
|
|
@ -1,32 +1,39 @@
|
|||
import gzip
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
import os
|
||||
from typing import Optional, cast
|
||||
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary
|
||||
from datatypes.magicplan.domain.mapper import map_plan
|
||||
from datatypes.magicplan.domain.models import Plan
|
||||
from domain.magicplan.api.response import MagicPlanPlan, PlanSummary
|
||||
from domain.magicplan.mapper import map_plan
|
||||
from domain.magicplan.models import Plan
|
||||
|
||||
from backend.app.db.connection import db_session
|
||||
from backend.app.db.functions.magic_plan_functions import save_plan
|
||||
from backend.app.db.models.uploaded_file import (
|
||||
FileSourceEnum,
|
||||
FileTypeEnum,
|
||||
UploadedFile,
|
||||
)
|
||||
from backend.magic_plan.address_matcher import find_matching_plan
|
||||
from backend.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
from utils.logger import setup_logger
|
||||
from utils.s3 import save_data_to_s3
|
||||
from applications.magic_plan.address_matcher import find_matching_plan
|
||||
from infrastructure.postgres.config import PostgresConfig
|
||||
from infrastructure.postgres.engine import make_engine, make_session
|
||||
from infrastructure.s3.s3_client import S3Client
|
||||
from repositories.magic_plan.magic_plan_postgres_repository import (
|
||||
MagicPlanPostgresRepository,
|
||||
)
|
||||
from infrastructure.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from applications.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
from utilities.logger import setup_logger
|
||||
|
||||
logger = setup_logger()
|
||||
|
||||
|
||||
class MagicPlanService:
|
||||
def __init__(self, client: MagicPlanClient, s3_bucket: str) -> None:
|
||||
self._client = client
|
||||
self._s3_bucket = s3_bucket
|
||||
class MagicPlanOrchestrator:
|
||||
def __init__(
|
||||
self, magic_plan_api_client: MagicPlanClient, s3_client: S3Client
|
||||
) -> None:
|
||||
self._api_client = magic_plan_api_client
|
||||
# self._s3_bucket = s3_bucket
|
||||
self._s3_client = s3_client
|
||||
|
||||
def run(self, request: MagicPlanTriggerRequest) -> Plan:
|
||||
address = request.address
|
||||
|
|
@ -35,13 +42,13 @@ class MagicPlanService:
|
|||
if uprn is not None:
|
||||
logger.info("MagicPlanService.run uprn=%s", uprn)
|
||||
|
||||
plans: list[PlanSummary] = self._client.get_plans()
|
||||
plans: list[PlanSummary] = self._api_client.get_plans()
|
||||
matched: Optional[PlanSummary] = find_matching_plan(plans, address)
|
||||
|
||||
if matched is None:
|
||||
raise ValueError(f"No MagicPlan found for address: {address!r}")
|
||||
|
||||
raw_bytes: bytes = self._client.get_plan_raw(matched.id)
|
||||
raw_bytes: bytes = self._api_client.get_plan_raw(matched.id)
|
||||
magic_plan: MagicPlanPlan = MagicPlanPlan.model_validate(
|
||||
json.loads(raw_bytes)["data"]
|
||||
)
|
||||
|
|
@ -54,10 +61,14 @@ class MagicPlanService:
|
|||
hubspot_deal_id=request.hubspot_deal_id,
|
||||
)
|
||||
|
||||
with db_session() as session:
|
||||
session.add(uploaded_file)
|
||||
session.flush()
|
||||
save_plan(session, plan, cast(int, uploaded_file.id))
|
||||
engine = make_engine(PostgresConfig.from_env(os.environ))
|
||||
session = make_session(engine)
|
||||
|
||||
session.add(uploaded_file)
|
||||
session.flush()
|
||||
MagicPlanPostgresRepository(session).save(
|
||||
plan, cast(int, uploaded_file.id)
|
||||
) # TODO: refactor to use postgres Unit of Work
|
||||
|
||||
return plan
|
||||
|
||||
|
|
@ -73,9 +84,11 @@ class MagicPlanService:
|
|||
s3_key = f"documents/uprn/{uprn}/magic_plan_{plan_id}.json.gz"
|
||||
else:
|
||||
s3_key = f"documents/hubspot_deal_id/{hubspot_deal_id}/magic_plan_{plan_id}.json.gz"
|
||||
save_data_to_s3(compressed, self._s3_bucket, s3_key)
|
||||
|
||||
self._s3_client.put_object(s3_key, compressed)
|
||||
|
||||
return UploadedFile(
|
||||
s3_file_bucket=self._s3_bucket,
|
||||
s3_file_bucket=self._s3_client.bucket,
|
||||
s3_file_key=s3_key,
|
||||
s3_upload_timestamp=datetime.now(timezone.utc),
|
||||
uprn=int(uprn) if uprn is not None else None,
|
||||
193
repositories/magic_plan/magic_plan_postgres_repository.py
Normal file
193
repositories/magic_plan/magic_plan_postgres_repository.py
Normal file
|
|
@ -0,0 +1,193 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import Any, cast
|
||||
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlmodel import Session, col
|
||||
|
||||
from domain.magicplan.models import Floor, Plan
|
||||
from infrastructure.postgres.magic_plan_tables import (
|
||||
MagicPlanDoorModel,
|
||||
MagicPlanDoorVentilationModel,
|
||||
MagicPlanFloorModel,
|
||||
MagicPlanPlanModel,
|
||||
MagicPlanRoomModel,
|
||||
MagicPlanWindowModel,
|
||||
MagicPlanWindowVentilationModel,
|
||||
)
|
||||
from repositories.magic_plan.magic_plan_repository import MagicPlanRepository
|
||||
|
||||
|
||||
class MagicPlanPostgresRepository(MagicPlanRepository):
|
||||
def __init__(self, session: Session) -> None:
|
||||
self._session = session
|
||||
|
||||
def save(self, plan: Plan, uploaded_file_id: int) -> None:
|
||||
plan_id = self._upsert_plan(plan, uploaded_file_id)
|
||||
self._delete_children(plan_id)
|
||||
floor_ids = self._insert_floors(plan.floors, plan_id)
|
||||
room_ids = self._insert_rooms(plan.floors, floor_ids)
|
||||
window_ids, door_ids = self._insert_windows_and_doors(plan.floors, room_ids)
|
||||
self._insert_ventilation(plan.floors, window_ids, door_ids)
|
||||
|
||||
def _upsert_plan(self, plan: Plan, uploaded_file_id: int) -> int:
|
||||
row_data: dict[str, Any] = MagicPlanPlanModel.from_domain(
|
||||
plan, uploaded_file_id
|
||||
).model_dump(exclude={"id"})
|
||||
stmt = (
|
||||
pg_insert(MagicPlanPlanModel)
|
||||
.values(**row_data)
|
||||
.on_conflict_do_update(
|
||||
index_elements=["magic_plan_uid"],
|
||||
set_={k: v for k, v in row_data.items() if k != "magic_plan_uid"},
|
||||
)
|
||||
.returning(col(MagicPlanPlanModel.id))
|
||||
)
|
||||
return cast(
|
||||
int, self._session.execute(stmt).scalar_one()
|
||||
) # pyright: ignore[reportDeprecated]
|
||||
|
||||
def _delete_children(self, plan_id: int) -> None:
|
||||
floor_subq = (
|
||||
select(col(MagicPlanFloorModel.id))
|
||||
.where(col(MagicPlanFloorModel.magic_plan_plan_id) == plan_id)
|
||||
.scalar_subquery()
|
||||
)
|
||||
room_subq = (
|
||||
select(col(MagicPlanRoomModel.id))
|
||||
.where(col(MagicPlanRoomModel.magic_plan_floor_id).in_(floor_subq))
|
||||
.scalar_subquery()
|
||||
)
|
||||
window_subq = (
|
||||
select(col(MagicPlanWindowModel.id))
|
||||
.where(col(MagicPlanWindowModel.magic_plan_room_id).in_(room_subq))
|
||||
.scalar_subquery()
|
||||
)
|
||||
door_subq = (
|
||||
select(col(MagicPlanDoorModel.id))
|
||||
.where(col(MagicPlanDoorModel.magic_plan_room_id).in_(room_subq))
|
||||
.scalar_subquery()
|
||||
)
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
delete(MagicPlanWindowVentilationModel).where(
|
||||
col(MagicPlanWindowVentilationModel.magic_plan_window_id).in_(
|
||||
window_subq
|
||||
)
|
||||
)
|
||||
)
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
delete(MagicPlanDoorVentilationModel).where(
|
||||
col(MagicPlanDoorVentilationModel.magic_plan_door_id).in_(door_subq)
|
||||
)
|
||||
)
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
delete(MagicPlanWindowModel).where(
|
||||
col(MagicPlanWindowModel.magic_plan_room_id).in_(room_subq)
|
||||
)
|
||||
)
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
delete(MagicPlanDoorModel).where(
|
||||
col(MagicPlanDoorModel.magic_plan_room_id).in_(room_subq)
|
||||
)
|
||||
)
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
delete(MagicPlanRoomModel).where(
|
||||
col(MagicPlanRoomModel.magic_plan_floor_id).in_(floor_subq)
|
||||
)
|
||||
)
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
delete(MagicPlanFloorModel).where(
|
||||
col(MagicPlanFloorModel.magic_plan_plan_id) == plan_id
|
||||
)
|
||||
)
|
||||
|
||||
def _insert_floors(self, floors: list[Floor], plan_id: int) -> list[int]:
|
||||
rows: list[dict[str, Any]] = [
|
||||
MagicPlanFloorModel.from_domain(floor, plan_id).model_dump(exclude={"id"})
|
||||
for floor in floors
|
||||
]
|
||||
result = self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
pg_insert(MagicPlanFloorModel)
|
||||
.values(rows)
|
||||
.returning(col(MagicPlanFloorModel.id))
|
||||
)
|
||||
return cast(list[int], list(result.scalars().all()))
|
||||
|
||||
def _insert_rooms(self, floors: list[Floor], floor_ids: list[int]) -> list[int]:
|
||||
rows: list[dict[str, Any]] = [
|
||||
MagicPlanRoomModel.from_domain(room, floor_id).model_dump(exclude={"id"})
|
||||
for floor, floor_id in zip(floors, floor_ids)
|
||||
for room in floor.rooms
|
||||
]
|
||||
result = self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
pg_insert(MagicPlanRoomModel)
|
||||
.values(rows)
|
||||
.returning(col(MagicPlanRoomModel.id))
|
||||
)
|
||||
return cast(list[int], list(result.scalars().all()))
|
||||
|
||||
def _insert_windows_and_doors(
|
||||
self, floors: list[Floor], room_ids: list[int]
|
||||
) -> tuple[list[int], list[int]]:
|
||||
all_rooms = [room for floor in floors for room in floor.rooms]
|
||||
window_rows: list[dict[str, Any]] = [
|
||||
MagicPlanWindowModel.from_domain(window, room_id).model_dump(exclude={"id"})
|
||||
for room, room_id in zip(all_rooms, room_ids)
|
||||
for window in room.windows
|
||||
]
|
||||
door_rows: list[dict[str, Any]] = [
|
||||
MagicPlanDoorModel.from_domain(door, room_id).model_dump(exclude={"id"})
|
||||
for room, room_id in zip(all_rooms, room_ids)
|
||||
for door in room.doors
|
||||
]
|
||||
window_ids: list[int] = []
|
||||
door_ids: list[int] = []
|
||||
if window_rows:
|
||||
result = self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
pg_insert(MagicPlanWindowModel)
|
||||
.values(window_rows)
|
||||
.returning(col(MagicPlanWindowModel.id))
|
||||
)
|
||||
window_ids = cast(list[int], list(result.scalars().all()))
|
||||
if door_rows:
|
||||
result = self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
pg_insert(MagicPlanDoorModel)
|
||||
.values(door_rows)
|
||||
.returning(col(MagicPlanDoorModel.id))
|
||||
)
|
||||
door_ids = cast(list[int], list(result.scalars().all()))
|
||||
return window_ids, door_ids
|
||||
|
||||
def _insert_ventilation(
|
||||
self,
|
||||
floors: list[Floor],
|
||||
window_ids: list[int],
|
||||
door_ids: list[int],
|
||||
) -> None:
|
||||
all_rooms = [room for floor in floors for room in floor.rooms]
|
||||
all_windows = [w for room in all_rooms for w in room.windows]
|
||||
all_doors = [d for room in all_rooms for d in room.doors]
|
||||
|
||||
window_vent_rows: list[dict[str, Any]] = [
|
||||
MagicPlanWindowVentilationModel.from_domain(w.ventilation, wid).model_dump(
|
||||
exclude={"id"}
|
||||
)
|
||||
for w, wid in zip(all_windows, window_ids)
|
||||
if w.ventilation is not None
|
||||
]
|
||||
door_vent_rows: list[dict[str, Any]] = [
|
||||
MagicPlanDoorVentilationModel.from_domain(d.ventilation, did).model_dump(
|
||||
exclude={"id"}
|
||||
)
|
||||
for d, did in zip(all_doors, door_ids)
|
||||
if d.ventilation is not None
|
||||
]
|
||||
if window_vent_rows:
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
pg_insert(MagicPlanWindowVentilationModel).values(window_vent_rows)
|
||||
)
|
||||
if door_vent_rows:
|
||||
self._session.execute( # pyright: ignore[reportDeprecated]
|
||||
pg_insert(MagicPlanDoorVentilationModel).values(door_vent_rows)
|
||||
)
|
||||
18
repositories/magic_plan/magic_plan_repository.py
Normal file
18
repositories/magic_plan/magic_plan_repository.py
Normal file
|
|
@ -0,0 +1,18 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from domain.magicplan.models import Plan
|
||||
|
||||
|
||||
class MagicPlanRepository(ABC):
|
||||
"""Persists a MagicPlan aggregate.
|
||||
|
||||
Implementations are expected to write child rows using bulk inserts
|
||||
(one round-trip per table) — the codebase targets Postgres exclusively.
|
||||
Upsert semantics on magic_plan_uid; child rows are deleted and re-inserted
|
||||
on each save (idempotent re-run behaviour per ADR-0012).
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def save(self, plan: Plan, uploaded_file_id: int) -> None: ...
|
||||
|
|
@ -1,51 +1,182 @@
|
|||
UPRN,Address,Postcode
|
||||
U1035052,"1 Sudbury Crescent, Bromley",BR1 4PY
|
||||
U1027449,"11 Station Road, Bromley",BR1 3LP
|
||||
U1021310,"126 Faringdon Avenue, Bromley",BR2 8BU
|
||||
U1010811,"13 Gilbert Road, Bromley",BR1 3QP
|
||||
U1024017,"13 Manor Way, Bromley",BR2 8ES
|
||||
U1042232,"154 Southover, Bromley",BR1 4RZ
|
||||
U1009369,"17 Minster Road, Bromley",BR1 4DY
|
||||
U1022305,"18a Lansdowne Road, Bromley",BR1 3LZ
|
||||
U1033165,"2 Laburnum Way, Bromley",BR2 8BZ
|
||||
U1035326,"2 Whitebeam Avenue, Bromley",BR2 8DL
|
||||
U1037872,"20 Sudbury Crescent, Bromley",BR1 4PZ
|
||||
U1007432,"21 Detling Road, Bromley",BR1 4SH
|
||||
U1005123,"24 Bonville Road, Bromley",BR1 4QA
|
||||
U1034810,"24 Newbury Road, Bromley",BR2 0QW
|
||||
U1020351,"27 Laburnum Way, Bromley",BR2 8BY
|
||||
U1009511,"27 Newbury Road, Bromley",BR2 0QN
|
||||
U1034985,"272 Southborough Lane, Bromley",BR2 8AS
|
||||
U1037954,"28 Treewall Gardens, Bromley",BR1 5BT
|
||||
U1038103,"29 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1013358,"3 Bird In Hand Lane, Bromley",BR1 2NA
|
||||
U1024709,"3 Parkfield Way, Bromley",BR2 8AE
|
||||
U1031058,"303 Keedonwood Road, Bromley",BR1 4QR
|
||||
U1014077,"32 Aylesbury Road, Bromley",BR2 0QP
|
||||
U1019564,"32 Brook Lane, Bromley",BR1 4PU
|
||||
U1020237,"33 Hornbeam Way, Bromley",BR2 8DB
|
||||
U1027392,"26 Silverdale Road, Oprington",BR5 2LT
|
||||
U1003906,"54 Barnsdale Crescent, Oprington",BR5 2AX
|
||||
U1034479,"90 Mead Way, Bromley",BR2 9EU
|
||||
U1005549,"79 Lower Gravel Road, Bromley",BR2 8LP
|
||||
U1016743,"16 Princes Plain, Bromley",BR2 8LE
|
||||
U1041937,"75 Turpington Lane, Bromley",BR2 8JD
|
||||
U1034805,"38 Narrow Way, Bromley",BR2 8JB
|
||||
U1041933,"31 Turpington Lane, Bromley",BR2 8JA
|
||||
U1037833,"3 Stiles Close, Bromley",BR2 8EQ
|
||||
U1042734,"86 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1042575,"90 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1033177,"30 Larch Way, Bromley",BR2 8DU
|
||||
U1027989,"5 Larch Way, Bromley",BR2 8DT
|
||||
U1012309,"13 Almond Close, Bromley",BR2 8DS
|
||||
U1022525,"30 Lovelace Avenue, Bromley",BR2 8DQ
|
||||
U1047613,"13 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1027549,"14 Thorn Close, Bromley",BR2 8DH
|
||||
U1022726,"31 Hornbeam Way, Bromley",BR2 8DB
|
||||
U1021308,"70 Faringdon Avenue, Bromley",BR2 8BU
|
||||
U1026958,"77 Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1007553,"115d Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1032132,"115f Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1014627,"81 Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1009607,"20 Parkfield Way, Bromley",BR2 8AF
|
||||
U1027838,"15 Holmcroft Way, Bromley",BR2 8AD
|
||||
U1052376,"3 Shoreham Way, Bromley",BR2 7PU
|
||||
U1015499,"25 Boughton Avenue, Bromley",BR2 7PL
|
||||
U1005741,"2 Malling Way, Bromley",BR2 7PJ
|
||||
U1019906,"26 Eastry Avenue, Bromley",BR2 7PF
|
||||
U1021769,"49 Eastry Avenue, Bromley",BR2 7PE
|
||||
U1031121,"32 Laburnum Way, Bromley",BR2 8BZ
|
||||
U1022281,"35 Kingsdown Way, Bromley",BR2 7PT
|
||||
U1035115,"30 Thorn Close, Bromley",BR2 8DH
|
||||
U1027493,"35 Sudbury Crescent, Bromley",BR1 4PY
|
||||
U1042298,"39 Sudbury Crescent, Bromley",BR1 4PY
|
||||
U1016746,"68 Princes Plain, Bromley",BR2 8LE
|
||||
U1005168,"38 Holbrook Way, Bromley",BR2 8EE
|
||||
U1036446,"14 Meath Close, Oprington",BR5 2HF
|
||||
U1010452,"5 Canbury Path, Oprington",BR5 2EU
|
||||
U1019785,"50 Cray Valley Road, Oprington",BR5 2EZ
|
||||
U1024065,"64 Marion Crescent, Oprington",BR5 2HD
|
||||
U1042248,"16 Stanley Way, Oprington",BR5 2HE
|
||||
U1029229,"2 Meath Close, Oprington",BR5 2HF
|
||||
U1037768,"13 Silverdale Road, Oprington",BR5 2LU
|
||||
U1014589,"71 Empress Drive, Chislehurst",BR7 5BQ
|
||||
U1024698,"4 Palace View, Bromley",BR1 3EL
|
||||
U1052186,"4 Ravensleigh Gardens, Bromley",BR1 5SN
|
||||
U1052536,"12 Thorn Close, Bromley",BR2 8DH
|
||||
U1022018,"12 Hazel Walk, Bromley",BR2 8DF
|
||||
U1007728,"2 Hazel Walk, Bromley",BR2 8DF
|
||||
U1002456,"54 Birch Row, Bromley",BR2 8DA
|
||||
U1020349,"21 Laburnum Way, Bromley",BR2 8BY
|
||||
U1032129,"78 Faringdon Avenue, Bromley",BR2 8BU
|
||||
U1032130,"86 Faringdon Avenue, Bromley",BR2 8BU
|
||||
U1021824,"115g Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1021827,"121 Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1026960,"105 Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1010578,"6 Cranbrook Close, Bromley",BR2 7QA
|
||||
U1024709,"3 Parkfield Way, Bromley",BR2 8AE
|
||||
U1024580,"24 Parkfield Way, Bromley",BR2 8AF
|
||||
U1011190,"11 Shoreham Way, Bromley",BR2 7PU
|
||||
U1011191,"32 Shoreham Way, Bromley",BR2 7PU
|
||||
U1021535,"4 Chilham Way, Bromley",BR2 7PR
|
||||
U1007556,"11 Farleigh Avenue, Bromley",BR2 7PP
|
||||
U1010255,"7 Boughton Avenue, Bromley",BR2 7PL
|
||||
U1034810,"24 Newbury Road, Bromley",BR2 0QW
|
||||
U1009032,"10 Malling Way, Bromley",BR2 7PJ
|
||||
U1004686,"55 Baston Road, Bromley",BR2 7BD
|
||||
U1032061,"30 Eastry Avenue, Bromley",BR2 7PF
|
||||
U1010582,"13 Cranworth Cottages, Keston",BR2 6DB
|
||||
U1009511,"27 Newbury Road, Bromley",BR2 0QN
|
||||
U1037954,"28 Treewall Gardens, Bromley",BR1 5BT
|
||||
U1014793,"59 Headcorn Road, Bromley",BR1 4SQ
|
||||
U1005123,"24 Bonville Road, Bromley",BR1 4QA
|
||||
U1037872,"20 Sudbury Crescent, Bromley",BR1 4PZ
|
||||
U1022305,"18a Lansdowne Road, Bromley",BR1 3LZ
|
||||
U1035052,"1 Sudbury Crescent, Bromley",BR1 4PY
|
||||
U1042298,"39 Sudbury Crescent, Bromley",BR1 4PY
|
||||
U1019564,"32 Brook Lane, Bromley",BR1 4PU
|
||||
U1024511,"81 Nightingale Lane, Bromley",BR1 2SA
|
||||
U1032133,"119 Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1032134,"125 Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1032131,"93 Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1021825,"117a Faringdon Avenue, Bromley",BR2 8BT
|
||||
U1010107,"42 Birch Row, Bromley",BR2 8DA
|
||||
U1016880,"25 Almond Way, Bromley",BR2 8DR
|
||||
U1038107,"152 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1052455,"10 Stiles Close, Bromley",BR2 8EQ
|
||||
U1028328,"76 Magpie Hall Lane, Bromley",BR2 8ER
|
||||
U1020064,"28 Green Way, Bromley",BR2 8EY
|
||||
U1041934,"51 Turpington Lane, Bromley",BR2 8JA
|
||||
U1005696,"158 Magpie Hall Lane, Bromley",BR2 8JG
|
||||
U1030892,"140 Poverest Road, Oprington",BR5 1RH
|
||||
U1011072,"45 Rookery Gardens, Oprington",BR5 4BA
|
||||
U1031058,"303 Keedonwood Road, Bromley",BR1 4QR
|
||||
U1052429,"76 Southover, Bromley",BR1 4RY
|
||||
U1042153,"4 Scotts Road, Bromley",BR1 3QD
|
||||
U1037814,"42 Stanley Road, Bromley",BR2 9JH
|
||||
U1014078,"43 Aylesbury Road, Bromley",BR2 0QR
|
||||
U1007701,"46 Harwood Avenue, Bromley",BR1 3DU
|
||||
U1036758,"46 Newbury Road, Bromley",BR2 0QW
|
||||
U1025820,"46 Princes Plain, Bromley",BR2 8LE
|
||||
U1022991,"5 Link Way, Bromley",BR2 8JH
|
||||
U1024484,"55 Mounthurst Road, Bromley",BR2 7PG
|
||||
U1014793,"59 Headcorn Road, Bromley",BR1 4SQ
|
||||
U1008158,"71 Lower Gravel Road, Bromley",BR2 8LP
|
||||
U1032062,"46 Eastry Avenue, Bromley",BR2 7PF
|
||||
U1016742,"4 Princes Plain, Bromley",BR2 8LE
|
||||
U1038106,"84 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1013831,"33 Ash Row, Bromley",BR2 8DZ
|
||||
U1005742,"18 Malling Way, Bromley",BR2 7PJ
|
||||
U1042572,"37 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1005167,"30 Holbrook Way, Bromley",BR2 8EE
|
||||
U1024581,"30 Parkfield Way, Bromley",BR2 8AF
|
||||
U1024050,"10 Marden Avenue, Bromley",BR2 7PX
|
||||
U1020328,"31 Kingsdown Way, Bromley",BR2 7PT
|
||||
U1020327,"5 Kingsdown Way, Bromley",BR2 7PT
|
||||
U1005131,"10 Boughton Avenue, Bromley",BR2 7PL
|
||||
U1023415,"27 Malling Way, Bromley",BR2 7PJ
|
||||
U1052580,"25 Trentham Drive, Oprington",BR5 2EP
|
||||
U1026447,"50 Princes Plain, Bromley",BR2 8LE
|
||||
U1007432,"21 Detling Road, Bromley",BR1 4SH
|
||||
U1053388,"52 Whitebeam Avenue, Bromley",BR2 8DL
|
||||
U1022307,"20 Larch Way, Bromley",BR2 8DU
|
||||
U1027743,"17 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1022533,"81 Lower Gravel Road, Bromley",BR2 8LP
|
||||
U1014630,"118 Faringdon Avenue, Bromley",BR2 8BU
|
||||
U1030897,"60 Princes Plain, Bromley",BR2 8LE
|
||||
U1022931,"15 Lennard Road, Bromley",BR2 8LN
|
||||
U1042735,"108 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1013830,"31 Ash Row, Bromley",BR2 8DZ
|
||||
U1020368,"11 Larch Way, Bromley",BR2 8DT
|
||||
U1020369,"13 Larch Way, Bromley",BR2 8DT
|
||||
U1042576,"112 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1027830,"26 Holbrook Way, Bromley",BR2 8EE
|
||||
U1016744,"18 Princes Plain, Bromley",BR2 8LE
|
||||
U1041772,"13 Stanley Road, Bromley",BR2 9JE
|
||||
U1042573,"47 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1008151,"61 Lovelace Avenue, Bromley",BR2 8EA
|
||||
U1034985,"272 Southborough Lane, Bromley",BR2 8AS
|
||||
U1007947,"8 Laburnum Way, Bromley",BR2 8BZ
|
||||
U1027744,"33 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1038102,"14 Whitebeam Avenue, Bromley",BR2 8DL
|
||||
U1027745,"148 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1020208,"16 Holbrook Way, Bromley",BR2 8EE
|
||||
U1023463,"30 MANOR WAY, Bromley",BR2 8ES
|
||||
U1032647,"32 Narrow Way, Bromley",BR2 8JB
|
||||
U1033406,"67 Lower Gravel Road, Bromley",BR2 8LP
|
||||
U1000649,"42 Barnsdale Crescent, Oprington",BR5 2AX
|
||||
U1036332,"50 Marion Crescent, Oprington",BR5 2HD
|
||||
U1038103,"29 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1042574,"66 Whitebeam Avenue, Bromley",BR2 8DL
|
||||
U1042256,"2 Stiles Close, Bromley",BR2 8EQ
|
||||
U1034017,"48 Princes Plain, Bromley",BR2 8LE
|
||||
U1041435,"64 Princes Plain, Bromley",BR2 8LE
|
||||
U1053386,"27 Whitebeam Avenue, Bromley",BR2 8DJ
|
||||
U1038104,"32 Whitebeam Avenue, Bromley",BR2 8DL
|
||||
U1019567,"8 Broom Close, Bromley",BR2 8EU
|
||||
U1023539,"2 Marsham Close, Chislehurst",BR7 6JD
|
||||
U1037465,"6 Princes Plain, Bromley",BR2 8LE
|
||||
U1009202,"63 Mead Way, Bromley",BR2 9ER
|
||||
U1021353,"66 George Lane, Bromley",BR2 7LQ
|
||||
U1042733,"68 Whitebeam Avenue, Bromley",BR2 8DL
|
||||
U1030962,"7 Ravensleigh Gardens, Bromley",BR1 5SN
|
||||
U1031294,"70 London Lane, Bromley",BR1 4HE
|
||||
U1004442,"7 Birch Row, Bromley",BR2 8BX
|
||||
U1005006,"44 Birch Row, Bromley",BR2 8DA
|
||||
U1035331,"104 Whitebeam Avenue, Bromley",BR2 8DW
|
||||
U1024017,"13 Manor Way, Bromley",BR2 8ES
|
||||
U1014078,"43 Aylesbury Road, Bromley",BR2 0QR
|
||||
U1052186,"4 Ravensleigh Gardens, Bromley",BR1 5SN
|
||||
U1037450,"70 Pontefract Road, Bromley",BR1 4RB
|
||||
U1014589,"71 Empress Drive, Chislehurst",BR7 5BQ
|
||||
U1052429,"76 Southover, Bromley",BR1 4RY
|
||||
U1031294,"70 London Lane, Bromley",BR1 4HE
|
||||
U1014077,"32 Aylesbury Road, Bromley",BR2 0QP
|
||||
U1030962,"7 Ravensleigh Gardens, Bromley",BR1 5SN
|
||||
U1042232,"154 Southover, Bromley",BR1 4RZ
|
||||
U1024484,"55 Mounthurst Road, Bromley",BR2 7PG
|
||||
U1007701,"46 Harwood Avenue, Bromley",BR1 3DU
|
||||
U1020199,"78 Hillside Road, Bromley",BR2 0ST
|
||||
U1024511,"81 Nightingale Lane, Bromley",BR1 2SA
|
||||
U1036758,"46 Newbury Road, Bromley",BR2 0QW
|
||||
U1009369,"17 Minster Road, Bromley",BR1 4DY
|
||||
U1009194,"84 Mays Hill Road, Bromley",BR2 0HT
|
||||
U1013358,"3 Bird In Hand Lane, Bromley",BR1 2NA
|
||||
U1009202,"63 Mead Way, Bromley",BR2 9ER
|
||||
U1022991,"5 Link Way, Bromley",BR2 8JH
|
||||
U1025820,"46 Princes Plain, Bromley",BR2 8LE
|
||||
U1020237,"33 Hornbeam Way, Bromley",BR2 8DB
|
||||
U1021310,"126 Faringdon Avenue, Bromley",BR2 8BU
|
||||
U1021353,"66 George Lane, Bromley",BR2 7LQ
|
||||
U1033165,"2 Laburnum Way, Bromley",BR2 8BZ
|
||||
U1010811,"13 Gilbert Road, Bromley",BR1 3QP
|
||||
U1027449,"11 Station Road, Bromley",BR1 3LP
|
||||
U1035326,"2 Whitebeam Avenue, Bromley",BR2 8DL
|
||||
U1020351,"27 Laburnum Way, Bromley",BR2 8BY
|
||||
|
|
|
|||
|
91
tests/applications/magic_plan/test_magic_plan_handler.py
Normal file
91
tests/applications/magic_plan/test_magic_plan_handler.py
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from applications.magic_plan.handler import handler
|
||||
|
||||
ADDRESS = "2 Laburnum Way Bromley BR2 8BZ"
|
||||
PLAN_UID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
|
||||
|
||||
_ENV = {"MAGICPLAN_CUSTOMER_ID": "cust-123", "MAGICPLAN_API_KEY": "key-abc"}
|
||||
|
||||
|
||||
def _call_handler(body: dict[str, Any]) -> Any:
|
||||
return handler.__wrapped__(body, None) # type: ignore[attr-defined]
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_plan() -> MagicMock:
|
||||
plan = MagicMock()
|
||||
plan.uid = PLAN_UID
|
||||
return plan
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mock_orchestrator(mock_plan: MagicMock) -> MagicMock:
|
||||
orchestrator = MagicMock()
|
||||
orchestrator.run.return_value = mock_plan
|
||||
return orchestrator
|
||||
|
||||
|
||||
# --- request validation ---
|
||||
|
||||
|
||||
def test_handler_raises_on_missing_address(mock_plan: MagicMock) -> None:
|
||||
body: dict[str, Any] = {}
|
||||
with patch("applications.magic_plan.handler.os.environ", _ENV), \
|
||||
patch("applications.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("applications.magic_plan.handler.MagicPlanOrchestrator"):
|
||||
with pytest.raises(ValidationError):
|
||||
_call_handler(body)
|
||||
|
||||
|
||||
# --- client construction ---
|
||||
|
||||
|
||||
def test_handler_constructs_client_from_env(mock_orchestrator: MagicMock) -> None:
|
||||
body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
|
||||
env = {"MAGICPLAN_CUSTOMER_ID": "cust-xyz", "MAGICPLAN_API_KEY": "key-xyz"}
|
||||
with patch("applications.magic_plan.handler.os.environ", env), \
|
||||
patch("applications.magic_plan.handler.MagicPlanClient") as MockClient, \
|
||||
patch("applications.magic_plan.handler.MagicPlanOrchestrator", return_value=mock_orchestrator):
|
||||
_call_handler(body)
|
||||
MockClient.assert_called_once_with(customer_id="cust-xyz", api_key="key-xyz")
|
||||
|
||||
|
||||
# --- orchestrator orchestration ---
|
||||
|
||||
|
||||
def test_handler_calls_orchestrator_run_with_address(mock_orchestrator: MagicMock) -> None:
|
||||
body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
|
||||
with patch("applications.magic_plan.handler.os.environ", _ENV), \
|
||||
patch("applications.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("applications.magic_plan.handler.MagicPlanOrchestrator", return_value=mock_orchestrator):
|
||||
_call_handler(body)
|
||||
mock_orchestrator.run.assert_called_once()
|
||||
request = mock_orchestrator.run.call_args.args[0]
|
||||
assert request.address == ADDRESS
|
||||
assert request.uprn is None
|
||||
|
||||
|
||||
def test_handler_passes_uprn_to_orchestrator(mock_orchestrator: MagicMock) -> None:
|
||||
body = {"address": ADDRESS, "uprn": "100023336956", "hubspot_deal_id": "deal-123"}
|
||||
with patch("applications.magic_plan.handler.os.environ", _ENV), \
|
||||
patch("applications.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("applications.magic_plan.handler.MagicPlanOrchestrator", return_value=mock_orchestrator):
|
||||
_call_handler(body)
|
||||
mock_orchestrator.run.assert_called_once()
|
||||
request = mock_orchestrator.run.call_args.args[0]
|
||||
assert request.address == ADDRESS
|
||||
assert request.uprn == "100023336956"
|
||||
|
||||
|
||||
def test_handler_returns_plan_uid(mock_orchestrator: MagicMock) -> None:
|
||||
body = {"address": ADDRESS, "hubspot_deal_id": "deal-123"}
|
||||
with patch("applications.magic_plan.handler.os.environ", _ENV), \
|
||||
patch("applications.magic_plan.handler.MagicPlanClient"), \
|
||||
patch("applications.magic_plan.handler.MagicPlanOrchestrator", return_value=mock_orchestrator):
|
||||
result = _call_handler(body)
|
||||
assert result == PLAN_UID
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
from applications.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
|
||||
|
||||
def test_valid_payload_with_address_only() -> None:
|
||||
231
tests/domain/magicplan/test_mapper.py
Normal file
231
tests/domain/magicplan/test_mapper.py
Normal file
|
|
@ -0,0 +1,231 @@
|
|||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
import domain.magicplan.api.response as api
|
||||
from domain.magicplan.api.response import MagicPlanPlan, Symbol, Vec3, WallItem
|
||||
from domain.magicplan.mapper import (
|
||||
_map_window, # pyright: ignore[reportPrivateUsage]
|
||||
map_address,
|
||||
map_plan,
|
||||
)
|
||||
from domain.magicplan.models import Plan
|
||||
|
||||
FIXTURE_DIR = Path(__file__).parents[3] / "tests" / "magic_plan"
|
||||
PLAN_ID = "72efd2e0-b2b9-48cd-b82e-41f5b3166c9a"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def raw_data() -> dict[str, Any]:
|
||||
payload = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response.json").read_text()
|
||||
)
|
||||
return payload["data"]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def mp(raw_data: dict[str, Any]) -> MagicPlanPlan:
|
||||
return MagicPlanPlan.model_validate(raw_data)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def plan(mp: MagicPlanPlan) -> Plan:
|
||||
return map_plan(mp)
|
||||
|
||||
|
||||
# --- Plan-level ---
|
||||
|
||||
|
||||
def test_plan_uid(plan: Plan) -> None:
|
||||
assert plan.uid == PLAN_ID
|
||||
|
||||
|
||||
def test_plan_postcode(plan: Plan) -> None:
|
||||
assert plan.postcode == "BR2 8DU"
|
||||
|
||||
|
||||
def test_plan_address(plan: Plan) -> None:
|
||||
assert plan.address == "20 Larch Way, Bromley, GB"
|
||||
|
||||
|
||||
def test_floor_count(plan: Plan) -> None:
|
||||
assert len(plan.floors) == 2
|
||||
|
||||
|
||||
# --- Room dimensions ---
|
||||
|
||||
|
||||
def test_first_room_name(plan: Plan) -> None:
|
||||
assert plan.floors[0].rooms[0].name == "Kitchen"
|
||||
|
||||
|
||||
def test_room_dimensions_are_floats(plan: Plan) -> None:
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert isinstance(room.width_m, float)
|
||||
assert isinstance(room.length_m, float)
|
||||
assert isinstance(room.area_m2, float)
|
||||
|
||||
|
||||
def test_room_area_rounded_to_2dp(plan: Plan) -> None:
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert room.area_m2 == 9.39
|
||||
|
||||
|
||||
def test_room_dimensions_parsed_from_string(plan: Plan) -> None:
|
||||
room = plan.floors[0].rooms[0]
|
||||
assert room.width_m == 3.19
|
||||
assert room.length_m == 2.94
|
||||
|
||||
|
||||
# --- Windows ---
|
||||
|
||||
|
||||
def test_kitchen_has_windows(plan: Plan) -> None:
|
||||
assert len(plan.floors[0].rooms[0].windows) >= 1
|
||||
|
||||
|
||||
def test_window_fields_are_floats(plan: Plan) -> None:
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
assert isinstance(window.width_m, float)
|
||||
assert isinstance(window.height_m, float)
|
||||
assert isinstance(window.area_m2, float)
|
||||
|
||||
|
||||
def test_window_area_is_width_times_height(plan: Plan) -> None:
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
assert window.area_m2 == round(window.width_m * window.height_m, 2)
|
||||
|
||||
|
||||
def test_window_dimensions_rounded_to_2dp(plan: Plan) -> None:
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
assert window.width_m == 0.90
|
||||
assert window.height_m == 1.00
|
||||
assert window.area_m2 == 0.90
|
||||
|
||||
|
||||
def test_hallway_has_no_windows(plan: Plan) -> None:
|
||||
hallway = plan.floors[0].rooms[3]
|
||||
assert hallway.name == "Hallway"
|
||||
assert hallway.windows == []
|
||||
|
||||
|
||||
# --- Doors ---
|
||||
|
||||
|
||||
def test_kitchen_has_doors(plan: Plan) -> None:
|
||||
assert len(plan.floors[0].rooms[0].doors) >= 1
|
||||
|
||||
|
||||
def test_door_width_is_float(plan: Plan) -> None:
|
||||
door = plan.floors[0].rooms[0].doors[0]
|
||||
assert isinstance(door.width_mm, float)
|
||||
|
||||
|
||||
def test_door_width_rounded_to_2dp(plan: Plan) -> None:
|
||||
door = plan.floors[0].rooms[0].doors[0]
|
||||
assert door.width_mm == 800.0
|
||||
|
||||
|
||||
def test_door_height_is_correct(plan: Plan) -> None:
|
||||
# Kitchen doorhinged has size.z = 2.04 m = 2040 mm
|
||||
door = plan.floors[0].rooms[0].doors[0]
|
||||
assert door.height_mm == 2040.0
|
||||
|
||||
|
||||
# --- Ventilation ---
|
||||
|
||||
|
||||
def test_window_with_no_custom_fields_has_no_ventilation() -> None:
|
||||
wi = WallItem(
|
||||
uid="test",
|
||||
symbol=Symbol(id="windowcasement", name="Casement Window", valid=True),
|
||||
size=Vec3(x=1.0, y=0.0, z=1.2),
|
||||
position=Vec3(x=0.0, y=0.0, z=0.0),
|
||||
rotation=Vec3(x=0.0, y=0.0, z=0.0),
|
||||
)
|
||||
|
||||
window = _map_window(wi)
|
||||
|
||||
assert window.ventilation is None
|
||||
|
||||
|
||||
def test_mapped_window_has_no_opening_type() -> None:
|
||||
wi = WallItem(
|
||||
uid="test",
|
||||
symbol=Symbol(id="windowcasement", name="Casement Window", valid=True),
|
||||
size=Vec3(x=1.0, y=0.0, z=1.2),
|
||||
position=Vec3(x=0.0, y=0.0, z=0.0),
|
||||
rotation=Vec3(x=0.0, y=0.0, z=0.0),
|
||||
)
|
||||
|
||||
window = _map_window(wi)
|
||||
|
||||
assert not hasattr(window, "opening_type")
|
||||
|
||||
|
||||
def test_kitchen_window_has_ventilation(plan: Plan) -> None:
|
||||
window = plan.floors[0].rooms[0].windows[0]
|
||||
|
||||
assert window.ventilation is not None
|
||||
assert window.ventilation.trickle_vent_area_mm2 == 1700
|
||||
|
||||
|
||||
def test_toilet_door_has_ventilation_undercut(plan: Plan) -> None:
|
||||
toilet_doors = plan.floors[0].rooms[2].doors
|
||||
hinged = next(d for d in toilet_doors if d.ventilation is not None)
|
||||
|
||||
assert hinged.ventilation is not None
|
||||
assert hinged.ventilation.undercut_mm == 70.0
|
||||
|
||||
|
||||
def test_doorglass_is_classified_as_window(plan: Plan) -> None:
|
||||
room = plan.floors[0].rooms[1]
|
||||
|
||||
assert any(
|
||||
w.ventilation is not None and w.ventilation.opening_type == "External.Door"
|
||||
for w in room.windows
|
||||
)
|
||||
|
||||
|
||||
def test_glass_door_ventilation_opening_type(plan: Plan) -> None:
|
||||
room = plan.floors[0].rooms[1]
|
||||
glass = next(
|
||||
w for w in room.windows
|
||||
if w.ventilation is not None and w.ventilation.opening_type == "External.Door"
|
||||
)
|
||||
|
||||
assert glass.ventilation is not None
|
||||
assert glass.ventilation.opening_type == "External.Door"
|
||||
|
||||
|
||||
# --- Address unit tests ---
|
||||
|
||||
|
||||
def test_map_address_with_street_and_number() -> None:
|
||||
addr = api.Address(street_number="2", street="Laburnum Way", city="Bromley", country="GB")
|
||||
|
||||
assert map_address(addr) == "2 Laburnum Way, Bromley, GB"
|
||||
|
||||
|
||||
def test_map_address_with_street_number_only() -> None:
|
||||
addr = api.Address(street_number="2", city="Bromley", country="GB")
|
||||
|
||||
assert map_address(addr) == "2, Bromley, GB"
|
||||
|
||||
|
||||
def test_map_address_with_street_only() -> None:
|
||||
addr = api.Address(street="Laburnum Way", city="Bromley", country="GB")
|
||||
|
||||
assert map_address(addr) == "Laburnum Way, Bromley, GB"
|
||||
|
||||
|
||||
def test_map_address_city_absent_is_omitted() -> None:
|
||||
addr = api.Address(street_number="2", street="Laburnum Way", country="GB")
|
||||
|
||||
assert map_address(addr) == "2 Laburnum Way, GB"
|
||||
|
||||
|
||||
def test_map_address_none_returns_none() -> None:
|
||||
assert map_address(None) is None
|
||||
|
|
@ -4,16 +4,16 @@ from typing import Any
|
|||
|
||||
import pytest
|
||||
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan, PlansListResponse
|
||||
from domain.magicplan.api.response import MagicPlanPlan, PlansListResponse
|
||||
|
||||
FIXTURE_DIR = Path(__file__).parents[4] / "backend" / "magic_plan"
|
||||
PLAN_ID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
|
||||
FIXTURE_DIR = Path(__file__).parents[3] / "tests" / "magic_plan"
|
||||
PLAN_ID = "72efd2e0-b2b9-48cd-b82e-41f5b3166c9a"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def raw_data() -> dict[str, Any]:
|
||||
payload = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response.json").read_text()
|
||||
)
|
||||
return payload["data"]
|
||||
|
||||
|
|
@ -23,66 +23,55 @@ 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
|
||||
def test_model_validate_does_not_raise(raw_data: dict[str, Any]) -> None:
|
||||
MagicPlanPlan.model_validate(raw_data)
|
||||
|
||||
|
||||
def test_plan_id(mp: MagicPlanPlan):
|
||||
# assert
|
||||
def test_plan_id(mp: MagicPlanPlan) -> None:
|
||||
assert mp.plan.id == PLAN_ID
|
||||
|
||||
|
||||
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: MagicPlanPlan):
|
||||
# assert
|
||||
def test_floor_count(mp: MagicPlanPlan) -> None:
|
||||
assert len(mp.plan_detail.plan.floors) == 2
|
||||
|
||||
|
||||
def test_first_room_name(mp: MagicPlanPlan):
|
||||
# assert
|
||||
def test_first_room_name(mp: MagicPlanPlan) -> None:
|
||||
assert mp.plan_detail.plan.floors[0].rooms[0].name == "Kitchen"
|
||||
|
||||
|
||||
def test_room_area_is_float(mp: MagicPlanPlan):
|
||||
# arrange
|
||||
def test_room_area_is_float(mp: MagicPlanPlan) -> None:
|
||||
room = mp.plan_detail.plan.floors[0].rooms[0]
|
||||
# assert
|
||||
assert isinstance(room.area, float)
|
||||
|
||||
|
||||
def test_wall_item_symbol_id(mp: MagicPlanPlan):
|
||||
# arrange
|
||||
def test_wall_item_symbol_id(mp: MagicPlanPlan) -> None:
|
||||
room = mp.plan_detail.plan.floors[0].rooms[0]
|
||||
# assert
|
||||
assert room.wall_items[0].symbol.id != ""
|
||||
|
||||
|
||||
def test_field_value_array(mp: MagicPlanPlan):
|
||||
# arrange
|
||||
def test_custom_displayable_fields_parsed(mp: MagicPlanPlan) -> None:
|
||||
# Kitchen windowcasement carries custom_displayable_fields in the new fixture.
|
||||
room = mp.plan_detail.plan.floors[0].rooms[0]
|
||||
array_field = next(f for f in room.displayable_fields if f.value.is_array)
|
||||
# assert
|
||||
wi = next(w for w in room.wall_items if w.symbol.id == "windowcasement")
|
||||
assert len(wi.custom_displayable_fields) > 0
|
||||
|
||||
|
||||
def test_field_value_array(mp: MagicPlanPlan) -> None:
|
||||
room = mp.plan_detail.plan.floors[0].rooms[0]
|
||||
wi = next(w for w in room.wall_items if w.symbol.id == "windowcasement")
|
||||
array_field = next(f for f in wi.custom_displayable_fields if f.value.is_array)
|
||||
assert isinstance(array_field.value.value, list)
|
||||
|
||||
|
||||
def test_field_value_scalar(mp: MagicPlanPlan):
|
||||
# arrange
|
||||
def test_field_value_scalar(mp: MagicPlanPlan) -> None:
|
||||
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)
|
||||
# assert
|
||||
wi = next(w for w in room.wall_items if w.symbol.id == "windowcasement")
|
||||
scalar_field = next(f for f in wi.custom_displayable_fields if not f.value.is_array)
|
||||
assert isinstance(scalar_field.value.value, str)
|
||||
|
||||
|
||||
def test_extra_fields_ignored(raw_data: dict[str, Any]):
|
||||
# arrange
|
||||
def test_extra_fields_ignored(raw_data: dict[str, Any]) -> None:
|
||||
data_with_extra = {**raw_data, "unknown_future_field": "whatever"}
|
||||
# act
|
||||
MagicPlanPlan.model_validate(data_with_extra)
|
||||
|
||||
|
||||
|
|
@ -105,39 +94,31 @@ def plans_response(plans_raw_data: dict[str, Any]) -> PlansListResponse:
|
|||
def test_plans_list_model_validate_does_not_raise(
|
||||
plans_raw_data: dict[str, Any],
|
||||
) -> None:
|
||||
# act
|
||||
PlansListResponse.model_validate(plans_raw_data)
|
||||
|
||||
|
||||
def test_plans_list_count(plans_response: PlansListResponse) -> None:
|
||||
# assert
|
||||
assert len(plans_response.plans) == 1
|
||||
|
||||
|
||||
def test_plans_list_first_plan_id(plans_response: PlansListResponse) -> None:
|
||||
# assert
|
||||
assert plans_response.plans[0].id == "9f9889ff-793e-4e9a-a6f0-e22f5b0f5365"
|
||||
|
||||
|
||||
def test_plans_list_paging_page(plans_response: PlansListResponse) -> None:
|
||||
# assert
|
||||
assert plans_response.paging.page == 1
|
||||
|
||||
|
||||
def test_plans_list_paging_next_page_is_false(
|
||||
plans_response: PlansListResponse,
|
||||
) -> None:
|
||||
# assert
|
||||
assert plans_response.paging.next_page is False
|
||||
|
||||
|
||||
def test_plans_list_paging_count(plans_response: PlansListResponse) -> None:
|
||||
# assert
|
||||
assert plans_response.paging.count == 1
|
||||
|
||||
|
||||
def test_plans_list_unknown_keys_ignored(plans_raw_data: dict[str, Any]) -> None:
|
||||
# arrange
|
||||
data_with_extra = {**plans_raw_data, "unknown_future_field": "whatever"}
|
||||
# act
|
||||
PlansListResponse.model_validate(data_with_extra)
|
||||
|
|
@ -6,8 +6,8 @@ from unittest.mock import MagicMock, patch
|
|||
import pytest
|
||||
import requests
|
||||
|
||||
from backend.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary
|
||||
from infrastructure.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from domain.magicplan.api.response import MagicPlanPlan, PlanSummary
|
||||
|
||||
FIXTURE_DIR = Path(__file__).parents[2] / "magic_plan"
|
||||
BASE_URL = "https://cloud.magicplan.app/api/v2"
|
||||
|
|
@ -22,7 +22,7 @@ def _load_fixture(name: str) -> dict[str, Any]:
|
|||
def _make_client(mock_session: MagicMock) -> MagicPlanClient:
|
||||
mock_session.headers = {}
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_client.requests.Session",
|
||||
"infrastructure.magic_plan.magic_plan_client.requests.Session",
|
||||
return_value=mock_session,
|
||||
):
|
||||
return MagicPlanClient(customer_id=CUSTOMER_ID, api_key=API_KEY)
|
||||
|
|
@ -129,11 +129,17 @@ def test_get_plans_multi_page_fetches_all_pages(
|
|||
page2_plan = {**page1_plan, "id": "page-2-plan-id"}
|
||||
page1_response = MagicMock()
|
||||
page1_response.json.return_value = {
|
||||
"data": {"paging": {"page": 1, "next_page": True, "count": 2}, "plans": [page1_plan]}
|
||||
"data": {
|
||||
"paging": {"page": 1, "next_page": True, "count": 2},
|
||||
"plans": [page1_plan],
|
||||
}
|
||||
}
|
||||
page2_response = MagicMock()
|
||||
page2_response.json.return_value = {
|
||||
"data": {"paging": {"page": 2, "next_page": False, "count": 2}, "plans": [page2_plan]}
|
||||
"data": {
|
||||
"paging": {"page": 2, "next_page": False, "count": 2},
|
||||
"plans": [page2_plan],
|
||||
}
|
||||
}
|
||||
mock_session.get.side_effect = [page1_response, page2_response]
|
||||
# Act
|
||||
|
|
@ -154,7 +160,7 @@ 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"]
|
||||
plan_data = _load_fixture("magicplan_api_plan_response.json")["data"]
|
||||
mock_session.get.return_value.json.return_value = {
|
||||
"message": "OK",
|
||||
"data": plan_data,
|
||||
|
|
@ -170,7 +176,7 @@ 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"]
|
||||
plan_data = _load_fixture("magicplan_api_plan_response.json")["data"]
|
||||
mock_session.get.return_value.json.return_value = {
|
||||
"message": "OK",
|
||||
"data": plan_data,
|
||||
|
|
@ -185,7 +191,7 @@ 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"]
|
||||
plan_data = _load_fixture("magicplan_api_plan_response.json")["data"]
|
||||
mock_session.get.return_value.json.return_value = {
|
||||
"message": "OK",
|
||||
"data": plan_data,
|
||||
|
|
@ -194,7 +200,7 @@ def test_get_plan_returns_magic_plan(
|
|||
result = client.get_plan("a7285ed1-878d-47eb-8aa6-85ef9e187516")
|
||||
# Assert
|
||||
assert isinstance(result, MagicPlanPlan)
|
||||
assert result.plan.id == "a7285ed1-878d-47eb-8aa6-85ef9e187516"
|
||||
assert result.plan.id == "72efd2e0-b2b9-48cd-b82e-41f5b3166c9a"
|
||||
|
||||
|
||||
def test_get_plan_propagates_http_error(
|
||||
21
tests/infrastructure/magic_plan/test_magic_plan_config.py
Normal file
21
tests/infrastructure/magic_plan/test_magic_plan_config.py
Normal file
|
|
@ -0,0 +1,21 @@
|
|||
import pytest
|
||||
|
||||
from infrastructure.magic_plan.config import MagicPlanConfig
|
||||
|
||||
_ENV = {"MAGICPLAN_CUSTOMER_ID": "cust-123", "MAGICPLAN_API_KEY": "key-abc"}
|
||||
|
||||
|
||||
def test_from_env_constructs_config() -> None:
|
||||
config = MagicPlanConfig.from_env(_ENV)
|
||||
assert config.customer_id == "cust-123"
|
||||
assert config.api_key == "key-abc"
|
||||
|
||||
|
||||
def test_from_env_raises_on_missing_customer_id() -> None:
|
||||
with pytest.raises(KeyError):
|
||||
MagicPlanConfig.from_env({"MAGICPLAN_API_KEY": "key-abc"})
|
||||
|
||||
|
||||
def test_from_env_raises_on_missing_api_key() -> None:
|
||||
with pytest.raises(KeyError):
|
||||
MagicPlanConfig.from_env({"MAGICPLAN_CUSTOMER_ID": "cust-123"})
|
||||
1
tests/magic_plan/magicplan_api_plan_response.json
Normal file
1
tests/magic_plan/magicplan_api_plan_response.json
Normal file
File diff suppressed because one or more lines are too long
39
tests/magic_plan/magicplan_api_plans_response_example.json
Normal file
39
tests/magic_plan/magicplan_api_plans_response_example.json
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
{
|
||||
"data": {
|
||||
"paging": {
|
||||
"page": 1,
|
||||
"next_page": false,
|
||||
"count": 1
|
||||
},
|
||||
"plans": [
|
||||
{
|
||||
"id": "9f9889ff-793e-4e9a-a6f0-e22f5b0f5365",
|
||||
"project_id": "269422e7-45b6-4582-b124-405053dcd967",
|
||||
"name": "11, Br1 3lp",
|
||||
"address": {
|
||||
"street": "11 Station Road",
|
||||
"street_number": null,
|
||||
"postal_code": "BR1 3LP",
|
||||
"city": "Bromley",
|
||||
"country": "GB",
|
||||
"longitude": 0.01593668,
|
||||
"latitude": 51.40901033
|
||||
},
|
||||
"creation_date": "2026-04-28T09:35:44+00:00",
|
||||
"update_date": "2026-05-05T12:53:36+00:00",
|
||||
"thumbnail_url": "https://s3.amazonaws.com/prod.plans.sensopia.com/9f9889ff-793e-4e9a-a6f0-e22f5b0f5365/plan.thumb",
|
||||
"public_url": "https://cloud.magicplan.app/plan/9f9889ff-793e-4e9a-a6f0-e22f5b0f5365",
|
||||
"cloud_url": "https://cloud.magicplan.app/projects/9f9889ff-793e-4e9a-a6f0-e22f5b0f5365",
|
||||
"3d_url": "https://3d.magicplan.app/#embed/?key=MmFkZDJjNGRmYWRjM2Y5ZDAwMjEyZGRlY2I3NmJjOWFjOWRmMDdkNzIxZTViZDdhNTgxZDBiYWE1YTYzZTJmY%2FJNEogVfW%2FZwVfY25qc24oCKnfVxiF%2FupeeA7vwS8FECF0L9E7DUFE%2ByzEYzYaoVc%2FbtsZ%2FqZOSPopiR4OqD3zbCziU0QTydELS32cnSFOT",
|
||||
"workgroup_id": "677d01685458a",
|
||||
"team_id": null,
|
||||
"created_by": {
|
||||
"id": "b19771e9-1aad-45a5-9a41-f01a835172ea",
|
||||
"firstname": null,
|
||||
"lastname": null,
|
||||
"email": "archie.ratcliff@domna.homes"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
|
@ -1,48 +1,44 @@
|
|||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from unittest.mock import ANY, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary
|
||||
from datatypes.magicplan.domain.mapper import map_plan
|
||||
from datatypes.magicplan.domain.models import Plan
|
||||
from domain.magicplan.api.response import MagicPlanPlan, PlanSummary
|
||||
from domain.magicplan.mapper import map_plan
|
||||
from domain.magicplan.models import Plan
|
||||
|
||||
from backend.app.db.models.uploaded_file import (
|
||||
FileSourceEnum,
|
||||
FileTypeEnum,
|
||||
UploadedFile,
|
||||
)
|
||||
from backend.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from backend.magic_plan.magic_plan_service import MagicPlanService
|
||||
from backend.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
from infrastructure.magic_plan.magic_plan_client import MagicPlanClient
|
||||
from infrastructure.s3.s3_client import S3Client
|
||||
from orchestration.magic_plan_orchestrator import MagicPlanOrchestrator
|
||||
from applications.magic_plan.magic_plan_trigger_request import MagicPlanTriggerRequest
|
||||
|
||||
FIXTURE_DIR = Path(__file__).parents[2] / "magic_plan"
|
||||
PLAN_ID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
|
||||
PLAN_ID = "72efd2e0-b2b9-48cd-b82e-41f5b3166c9a"
|
||||
S3_BUCKET = "test-bucket"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def domain_plan() -> Plan:
|
||||
data = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
|
||||
)
|
||||
data = json.loads((FIXTURE_DIR / "magicplan_api_plan_response.json").read_text())
|
||||
return map_plan(MagicPlanPlan.model_validate(data["data"]))
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def api_magic_plan() -> MagicPlanPlan:
|
||||
data = json.loads(
|
||||
(FIXTURE_DIR / "magicplan_api_plan_response_example.json").read_text()
|
||||
)
|
||||
data = json.loads((FIXTURE_DIR / "magicplan_api_plan_response.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()
|
||||
)
|
||||
data = json.loads((FIXTURE_DIR / "magicplan_api_plan_response.json").read_text())
|
||||
return MagicPlanPlan.model_validate(data["data"]).plan
|
||||
|
||||
|
||||
|
|
@ -50,25 +46,43 @@ def plan_summary() -> PlanSummary:
|
|||
def mock_client() -> MagicMock:
|
||||
client = MagicMock(spec=MagicPlanClient)
|
||||
client.get_plan_raw.return_value = (
|
||||
FIXTURE_DIR / "magicplan_api_plan_response_example.json"
|
||||
FIXTURE_DIR / "magicplan_api_plan_response.json"
|
||||
).read_bytes()
|
||||
return client
|
||||
|
||||
|
||||
def _make_service(mock_client: MagicMock) -> MagicPlanService:
|
||||
return MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
|
||||
def _make_s3_client(bucket: str = S3_BUCKET) -> MagicMock:
|
||||
s3 = MagicMock(spec=S3Client)
|
||||
s3.bucket = bucket
|
||||
return s3
|
||||
|
||||
|
||||
def _make_service(
|
||||
mock_client: MagicMock, mock_s3: Optional[MagicMock] = None
|
||||
) -> MagicPlanOrchestrator:
|
||||
if mock_s3 is None:
|
||||
mock_s3 = _make_s3_client()
|
||||
return MagicPlanOrchestrator(magic_plan_api_client=mock_client, s3_client=mock_s3)
|
||||
|
||||
|
||||
def _make_request(
|
||||
address: str = "2 Laburnum Way Bromley BR2 8BZ",
|
||||
hubspot_deal_id: str = "deal-123",
|
||||
uprn: str | None = None,
|
||||
uprn: Optional[str] = None,
|
||||
) -> MagicPlanTriggerRequest:
|
||||
return MagicPlanTriggerRequest(
|
||||
address=address, hubspot_deal_id=hubspot_deal_id, uprn=uprn
|
||||
)
|
||||
|
||||
|
||||
def _patch_db() -> tuple[patch, patch, patch]: # type: ignore[type-arg]
|
||||
return (
|
||||
patch("orchestration.magic_plan_orchestrator.PostgresConfig"),
|
||||
patch("orchestration.magic_plan_orchestrator.make_engine"),
|
||||
patch("orchestration.magic_plan_orchestrator.make_session"),
|
||||
)
|
||||
|
||||
|
||||
# --- no match ---
|
||||
|
||||
|
||||
|
|
@ -94,14 +108,13 @@ def test_run_fetches_plan_with_matched_id(
|
|||
mock_client.get_plans.return_value = [plan_summary]
|
||||
mock_client.get_plan.return_value = api_magic_plan
|
||||
service = _make_service(mock_client)
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
), patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
):
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository"
|
||||
), p_config, p_engine, p_session:
|
||||
service.run(_make_request())
|
||||
# Assert
|
||||
mock_client.get_plan_raw.assert_called_once_with(plan_summary.id)
|
||||
|
|
@ -117,21 +130,20 @@ def test_run_returns_mapped_plan(
|
|||
mock_client.get_plans.return_value = [plan_summary]
|
||||
mock_client.get_plan.return_value = api_magic_plan
|
||||
service = _make_service(mock_client)
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
), patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
):
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository"
|
||||
), p_config, p_engine, p_session:
|
||||
result = service.run(_make_request())
|
||||
# Assert
|
||||
assert isinstance(result, Plan)
|
||||
assert result.uid == PLAN_ID
|
||||
|
||||
|
||||
def test_run_calls_save_plan_with_mapped_plan(
|
||||
def test_run_calls_save_with_mapped_plan(
|
||||
mock_client: MagicMock,
|
||||
api_magic_plan: MagicPlanPlan,
|
||||
plan_summary: PlanSummary,
|
||||
|
|
@ -140,18 +152,18 @@ def test_run_calls_save_plan_with_mapped_plan(
|
|||
mock_client.get_plans.return_value = [plan_summary]
|
||||
mock_client.get_plan.return_value = api_magic_plan
|
||||
service = _make_service(mock_client)
|
||||
mock_repo = MagicMock()
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
), patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
):
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository",
|
||||
return_value=mock_repo,
|
||||
), p_config, p_engine, p_session:
|
||||
service.run(_make_request())
|
||||
# Assert — save_plan called with a Plan whose uid matches
|
||||
call_args = mock_save.call_args
|
||||
saved_plan: Plan = call_args[0][1]
|
||||
# Assert — save called with a Plan whose uid matches
|
||||
saved_plan: Plan = mock_repo.save.call_args[0][0]
|
||||
assert saved_plan.uid == PLAN_ID
|
||||
|
||||
|
||||
|
|
@ -164,14 +176,13 @@ def test_run_accepts_uprn_without_error(
|
|||
mock_client.get_plans.return_value = [plan_summary]
|
||||
mock_client.get_plan.return_value = api_magic_plan
|
||||
service = _make_service(mock_client)
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
), patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
):
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository"
|
||||
), p_config, p_engine, p_session:
|
||||
service.run(_make_request(uprn="100023336956"))
|
||||
|
||||
|
||||
|
|
@ -180,56 +191,51 @@ def test_run_accepts_uprn_without_error(
|
|||
|
||||
def test_run_uploads_to_s3_with_uprn_key(
|
||||
mock_client: MagicMock,
|
||||
api_magic_plan: MagicPlanPlan,
|
||||
plan_summary: PlanSummary,
|
||||
) -> None:
|
||||
# Arrange
|
||||
mock_client.get_plans.return_value = [plan_summary]
|
||||
mock_s3 = _make_s3_client()
|
||||
request = _make_request(uprn="100023336956")
|
||||
service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
|
||||
service = _make_service(mock_client, mock_s3)
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
), patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
) as mock_s3:
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository"
|
||||
), p_config, p_engine, p_session:
|
||||
# Act
|
||||
service.run(request)
|
||||
# Assert
|
||||
mock_s3.assert_called_once_with(
|
||||
ANY,
|
||||
S3_BUCKET,
|
||||
mock_s3.put_object.assert_called_once_with(
|
||||
f"documents/uprn/100023336956/magic_plan_{plan_summary.id}.json.gz",
|
||||
ANY,
|
||||
)
|
||||
|
||||
|
||||
def test_run_uploads_to_s3_with_deal_id_key_when_uprn_absent(
|
||||
mock_client: MagicMock,
|
||||
api_magic_plan: MagicPlanPlan,
|
||||
plan_summary: PlanSummary,
|
||||
) -> None:
|
||||
# Arrange
|
||||
mock_client.get_plans.return_value = [plan_summary]
|
||||
mock_client.get_plan.return_value = api_magic_plan
|
||||
mock_s3 = _make_s3_client()
|
||||
request = _make_request(hubspot_deal_id="deal-456", uprn=None)
|
||||
service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
|
||||
service = _make_service(mock_client, mock_s3)
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
), patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
) as mock_s3:
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository"
|
||||
), p_config, p_engine, p_session:
|
||||
# Act
|
||||
service.run(request)
|
||||
# Assert
|
||||
mock_s3.assert_called_once_with(
|
||||
ANY,
|
||||
S3_BUCKET,
|
||||
mock_s3.put_object.assert_called_once_with(
|
||||
f"documents/hubspot_deal_id/deal-456/magic_plan_{plan_summary.id}.json.gz",
|
||||
ANY,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -238,24 +244,22 @@ def test_run_uploads_to_s3_with_deal_id_key_when_uprn_absent(
|
|||
|
||||
def test_run_creates_uploaded_file_record(
|
||||
mock_client: MagicMock,
|
||||
api_magic_plan: MagicPlanPlan,
|
||||
plan_summary: PlanSummary,
|
||||
) -> None:
|
||||
# Arrange
|
||||
mock_client.get_plans.return_value = [plan_summary]
|
||||
mock_client.get_plan.return_value = api_magic_plan
|
||||
mock_s3 = _make_s3_client()
|
||||
request = _make_request(hubspot_deal_id="deal-789", uprn="100023336956")
|
||||
service = MagicPlanService(client=mock_client, s3_bucket=S3_BUCKET)
|
||||
service = _make_service(mock_client, mock_s3)
|
||||
mock_session = MagicMock()
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
) as mock_db, patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
):
|
||||
mock_db.return_value.__enter__.return_value = mock_session
|
||||
), patch(
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository"
|
||||
), p_config, p_engine, p_session as mock_make_session:
|
||||
mock_make_session.return_value = mock_session
|
||||
# Act
|
||||
service.run(request)
|
||||
# Assert
|
||||
|
|
@ -267,13 +271,16 @@ def test_run_creates_uploaded_file_record(
|
|||
assert uploaded_file.file_source == FileSourceEnum.MAGIC_PLAN.value
|
||||
assert uploaded_file.file_type == FileTypeEnum.MAGIC_PLAN_JSON.value
|
||||
assert uploaded_file.s3_file_bucket == S3_BUCKET
|
||||
assert uploaded_file.s3_file_key == f"documents/uprn/100023336956/magic_plan_{plan_summary.id}.json.gz"
|
||||
assert (
|
||||
uploaded_file.s3_file_key
|
||||
== f"documents/uprn/100023336956/magic_plan_{plan_summary.id}.json.gz"
|
||||
)
|
||||
assert uploaded_file.s3_upload_timestamp is not None
|
||||
assert uploaded_file.uprn == 100023336956
|
||||
assert uploaded_file.hubspot_deal_id == "deal-789"
|
||||
|
||||
|
||||
def test_run_passes_flushed_uploaded_file_id_to_save_plan(
|
||||
def test_run_passes_flushed_uploaded_file_id_to_save(
|
||||
mock_client: MagicMock,
|
||||
plan_summary: PlanSummary,
|
||||
) -> None:
|
||||
|
|
@ -281,7 +288,7 @@ def test_run_passes_flushed_uploaded_file_id_to_save_plan(
|
|||
mock_client.get_plans.return_value = [plan_summary]
|
||||
service = _make_service(mock_client)
|
||||
mock_session = MagicMock()
|
||||
added_objects: list = []
|
||||
added_objects: list[object] = []
|
||||
|
||||
mock_session.add.side_effect = added_objects.append
|
||||
|
||||
|
|
@ -291,18 +298,19 @@ def test_run_passes_flushed_uploaded_file_id_to_save_plan(
|
|||
obj.id = 42
|
||||
|
||||
mock_session.flush.side_effect = simulate_flush
|
||||
mock_repo = MagicMock()
|
||||
p_config, p_engine, p_session = _patch_db()
|
||||
|
||||
with patch(
|
||||
"backend.magic_plan.magic_plan_service.find_matching_plan",
|
||||
"orchestration.magic_plan_orchestrator.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"
|
||||
) as mock_db, patch(
|
||||
"backend.magic_plan.magic_plan_service.save_data_to_s3"
|
||||
):
|
||||
mock_db.return_value.__enter__.return_value = mock_session
|
||||
), patch(
|
||||
"orchestration.magic_plan_orchestrator.MagicPlanPostgresRepository",
|
||||
return_value=mock_repo,
|
||||
), p_config, p_engine, p_session as mock_make_session:
|
||||
mock_make_session.return_value = mock_session
|
||||
# Act
|
||||
service.run(_make_request())
|
||||
|
||||
# Assert
|
||||
assert mock_save.call_args[0][2] == 42
|
||||
# Assert — save called with the flushed uploaded_file_id
|
||||
assert mock_repo.save.call_args[0][1] == 42
|
||||
0
tests/repositories/magic_plan/__init__.py
Normal file
0
tests/repositories/magic_plan/__init__.py
Normal file
|
|
@ -0,0 +1,113 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy import Engine
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from domain.magicplan.models import (
|
||||
Door,
|
||||
DoorVentilation,
|
||||
Floor,
|
||||
Plan,
|
||||
Room,
|
||||
Window,
|
||||
WindowVentilation,
|
||||
)
|
||||
from infrastructure.postgres.magic_plan_tables import (
|
||||
MagicPlanDoorModel,
|
||||
MagicPlanDoorVentilationModel,
|
||||
MagicPlanFloorModel,
|
||||
MagicPlanPlanModel,
|
||||
MagicPlanRoomModel,
|
||||
MagicPlanWindowModel,
|
||||
MagicPlanWindowVentilationModel,
|
||||
)
|
||||
from repositories.magic_plan.magic_plan_postgres_repository import (
|
||||
MagicPlanPostgresRepository,
|
||||
)
|
||||
|
||||
|
||||
def _plan() -> Plan:
|
||||
window = Window(
|
||||
width_m=1.2,
|
||||
height_m=1.5,
|
||||
area_m2=1.8,
|
||||
ventilation=WindowVentilation(
|
||||
opening_type="30.Hinged.Pivot.Window",
|
||||
num_openings=2,
|
||||
pct_openable=70,
|
||||
trickle_vent_area_mm2=1700,
|
||||
num_trickle_vents=2,
|
||||
),
|
||||
)
|
||||
door = Door(width_mm=762.0, height_mm=2040.0, ventilation=DoorVentilation(undercut_mm=70.0))
|
||||
room = Room(
|
||||
name="Living Room",
|
||||
width_m=4.0,
|
||||
length_m=5.0,
|
||||
area_m2=20.0,
|
||||
windows=[window],
|
||||
doors=[door],
|
||||
)
|
||||
floor = Floor(level=0, name="Ground Floor", rooms=[room])
|
||||
return Plan(
|
||||
uid="test-uid-abc123",
|
||||
name="Test Plan",
|
||||
address="1 Test Street",
|
||||
postcode="TE1 1ST",
|
||||
floors=[floor],
|
||||
)
|
||||
|
||||
|
||||
def test_save_writes_all_rows(db_engine: Engine) -> None:
|
||||
# Arrange
|
||||
plan = _plan()
|
||||
|
||||
# Act
|
||||
with Session(db_engine) as session:
|
||||
MagicPlanPostgresRepository(session).save(plan, uploaded_file_id=1)
|
||||
session.commit()
|
||||
|
||||
# Assert — one row written to every table in the aggregate
|
||||
with Session(db_engine) as session:
|
||||
assert len(session.exec(select(MagicPlanPlanModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanFloorModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanRoomModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanWindowModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanDoorModel)).all()) == 1
|
||||
|
||||
|
||||
def test_save_writes_ventilation_rows(db_engine: Engine) -> None:
|
||||
# Arrange — plan with one window (with ventilation) and one door (with ventilation)
|
||||
plan = _plan()
|
||||
|
||||
# Act
|
||||
with Session(db_engine) as session:
|
||||
MagicPlanPostgresRepository(session).save(plan, uploaded_file_id=1)
|
||||
session.commit()
|
||||
|
||||
# Assert
|
||||
with Session(db_engine) as session:
|
||||
assert len(session.exec(select(MagicPlanWindowVentilationModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanDoorVentilationModel)).all()) == 1
|
||||
|
||||
|
||||
def test_save_is_idempotent(db_engine: Engine) -> None:
|
||||
# Arrange
|
||||
plan = _plan()
|
||||
|
||||
# Act — save twice with the same plan uid; second must overwrite, not append
|
||||
with Session(db_engine) as session:
|
||||
repo = MagicPlanPostgresRepository(session)
|
||||
repo.save(plan, uploaded_file_id=1)
|
||||
repo.save(plan, uploaded_file_id=1)
|
||||
session.commit()
|
||||
|
||||
# Assert — row counts do not double
|
||||
with Session(db_engine) as session:
|
||||
assert len(session.exec(select(MagicPlanPlanModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanFloorModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanRoomModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanWindowModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanDoorModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanWindowVentilationModel)).all()) == 1
|
||||
assert len(session.exec(select(MagicPlanDoorVentilationModel)).all()) == 1
|
||||
41
utilities/logger.py
Normal file
41
utilities/logger.py
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
import logging
|
||||
from os import PathLike
|
||||
from typing import Optional, Union
|
||||
|
||||
|
||||
def setup_logger(
|
||||
log_file: Optional[Union[str, PathLike[str]]] = None,
|
||||
level: int = logging.INFO,
|
||||
overwrite_handler: bool = False,
|
||||
) -> logging.Logger:
|
||||
# Create a logger and set the logging level
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(level)
|
||||
|
||||
# if logger already has handlers, just return it
|
||||
if logger.hasHandlers() and not overwrite_handler:
|
||||
return logger
|
||||
|
||||
# Define the log message format
|
||||
log_format = "%(asctime)s [%(levelname)s] %(message)s"
|
||||
date_format = "%Y-%m-%d %H:%M:%S"
|
||||
formatter = logging.Formatter(log_format, datefmt=date_format)
|
||||
|
||||
# Create a file handler and set the file path and format
|
||||
if log_file:
|
||||
file_handler = logging.FileHandler(log_file)
|
||||
file_handler.setLevel(level)
|
||||
file_handler.setFormatter(formatter)
|
||||
logger.addHandler(file_handler)
|
||||
|
||||
# Create a console handler and set the format
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setLevel(level)
|
||||
|
||||
# Set the formatter for the handlers
|
||||
console_handler.setFormatter(formatter)
|
||||
|
||||
# Add the handlers to the logger
|
||||
logger.addHandler(console_handler)
|
||||
|
||||
return logger
|
||||
Loading…
Add table
Reference in a new issue