use pytest-postgresql in db tests instead of mocking and checking sql strings

backend/app/db/functions/tests/conftest.py

@@ -0,0 +1,41 @@
+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()

backend/app/db/functions/tests/test_magic_plan_functions.py

@@ -1,20 +1,25 @@
 import json
 from pathlib import Path
-from unittest.mock import MagicMock
 
 import pytest
-from sqlalchemy.dialects import postgresql
+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"
-PLAN_UID = "a7285ed1-878d-47eb-8aa6-85ef9e187516"
-
-# fixture 1: 2 floors, 14 rooms total, 13 windows, 27 doors
 
 
 @pytest.fixture(scope="module")
@@ -25,139 +30,66 @@ def domain_plan() -> Plan:
     return map_plan(MagicPlanPlan.model_validate(data["data"]))
 
 
-def _compiled(stmt: object) -> str:
-    return str(
-        stmt.compile(  # type: ignore[union-attr]
-            dialect=postgresql.dialect(),
-            compile_kwargs={"literal_binds": True},
-        )
-    )
+def _count(session: Session, model: type[SQLModel]) -> int:
+    return session.execute(select(func.count()).select_from(model)).scalar_one()
 
 
-@pytest.fixture()
-def mock_session() -> MagicMock:
-    session = MagicMock()
-
-    plan_result = MagicMock()
-    plan_result.scalar_one.return_value = 1
-
-    floor_result = MagicMock()
-    floor_result.scalars.return_value.all.return_value = [10, 20]
-
-    room_result = MagicMock()
-    room_result.scalars.return_value.all.return_value = list(range(100, 114))
-
-    session.execute.side_effect = [
-        plan_result,  # upsert plan
-        None,  # delete windows
-        None,  # delete doors
-        None,  # delete rooms
-        None,  # delete floors
-        floor_result,  # insert floors
-        room_result,  # insert rooms
-        None,  # insert windows
-        None,  # insert doors
-    ]
-    return session
-
-
-# --- save_plan orchestration ---
-
-
-def test_save_plan_does_not_raise(mock_session: MagicMock, domain_plan: Plan) -> None:
+def test_plan_row_present_after_save(db_session: Session, domain_plan: Plan) -> None:
     # Act
-    save_plan(mock_session, domain_plan)
-
-
-def test_save_plan_upserts_plan_table_first(
-    mock_session: MagicMock, domain_plan: Plan
-) -> None:
-    # Act
-    save_plan(mock_session, domain_plan)
+    save_plan(db_session, domain_plan)
     # Assert
-    first_stmt = mock_session.execute.call_args_list[0][0][0]
-    sql = _compiled(first_stmt)
-    assert "magic_plan_plan" in sql
-    assert "INSERT" in sql.upper()
+    assert _count(db_session, MagicPlanPlanModel) == 1
 
 
-def test_save_plan_upsert_contains_plan_uid(
-    mock_session: MagicMock, domain_plan: Plan
-) -> None:
+def test_floor_count_matches_domain(db_session: Session, domain_plan: Plan) -> None:
+    # Arrange
+    expected = len(domain_plan.floors)
     # Act
-    save_plan(mock_session, domain_plan)
+    save_plan(db_session, domain_plan)
     # Assert
-    first_stmt = mock_session.execute.call_args_list[0][0][0]
-    assert PLAN_UID in _compiled(first_stmt)
+    assert _count(db_session, MagicPlanFloorModel) == expected
 
 
-def test_save_plan_upsert_contains_plan_name(
-    mock_session: MagicMock, domain_plan: Plan
-) -> None:
+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(mock_session, domain_plan)
+    save_plan(db_session, domain_plan)
     # Assert
-    first_stmt = mock_session.execute.call_args_list[0][0][0]
-    assert domain_plan.name in _compiled(first_stmt)
+    assert _count(db_session, MagicPlanRoomModel) == expected
 
 
-def test_save_plan_deletes_floors_before_inserting(
-    mock_session: MagicMock, domain_plan: Plan
-) -> None:
+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(mock_session, domain_plan)
-    # Assert — find delete and insert stmts targeting magic_plan_floor
-    stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
-    floor_delete_idx = next(
-        i
-        for i, s in enumerate(stmts)
-        if "DELETE" in s.upper() and "magic_plan_floor" in s
-    )
-    floor_insert_idx = next(
-        i
-        for i, s in enumerate(stmts)
-        if "INSERT" in s.upper() and "magic_plan_floor" in s
-    )
-    assert floor_delete_idx < floor_insert_idx
+    save_plan(db_session, domain_plan)
+    # Assert
+    assert _count(db_session, MagicPlanWindowModel) == expected
 
 
-def test_save_plan_floor_insert_contains_all_levels(
-    mock_session: MagicMock, domain_plan: Plan
-) -> None:
+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(mock_session, domain_plan)
-    # Assert — each floor's level value appears in the INSERT
-    stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
-    floor_insert = next(
-        s for s in stmts if "INSERT" in s.upper() and "magic_plan_floor" in s
+    save_plan(db_session, domain_plan)
+    # 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)
+    save_plan(db_session, domain_plan)
+    # 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
     )
-    for floor in domain_plan.floors:
-        if floor.level is not None:
-            assert str(floor.level) in floor_insert
-
-
-def test_save_plan_room_insert_uses_all_floor_ids(
-    mock_session: MagicMock, domain_plan: Plan
-) -> None:
-    # Act
-    save_plan(mock_session, domain_plan)
-    # Assert — both mocked floor ids (10, 20) appear in the room INSERT
-    stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
-    room_insert = next(
-        s for s in stmts if "INSERT" in s.upper() and "magic_plan_room" in s
+    assert _count(db_session, MagicPlanWindowModel) == sum(
+        len(r.windows) for f in domain_plan.floors for r in f.rooms
     )
-    assert "10" in room_insert
-    assert "20" in room_insert
-
-
-def test_save_plan_windows_use_room_ids_from_insert(
-    mock_session: MagicMock, domain_plan: Plan
-) -> None:
-    # Act
-    save_plan(mock_session, domain_plan)
-    # Assert — window INSERT references one of the mocked room ids (100–113)
-    stmts = [_compiled(c[0][0]) for c in mock_session.execute.call_args_list]
-    window_insert = next(
-        s for s in stmts if "INSERT" in s.upper() and "magic_plan_window" in s
+    assert _count(db_session, MagicPlanDoorModel) == sum(
+        len(r.doors) for f in domain_plan.floors for r in f.rooms
     )
-    assert any(str(rid) in window_insert for rid in range(100, 114))

pytest.ini

@@ -3,6 +3,6 @@ pythonpath = .
 log_cli = true
 log_cli_level = INFO
 addopts = --cov-report term-missing --cov=etl/epc --cov=recommendations --cov=backend --cov=etl/epc_clean --cov=etl/spatial
-testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests backend/address2UPRN/tests backend/onboarders/tests backend/categorisation/tests backend/export/tests etl/hubspot/tests backend/hubspot_trigger_orchestrator/tests datatypes/epc/schema/tests datatypes/epc/surveys/tests datatypes/epc/domain/tests backend/ecmk_fetcher/tests/ backend/pashub_fetcher/tests backend/documents_parser/tests backend/magic_plan/tests datatypes/magicplan/api/tests datatypes/magicplan/domain/tests
+testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests backend/address2UPRN/tests backend/onboarders/tests backend/categorisation/tests backend/export/tests etl/hubspot/tests backend/hubspot_trigger_orchestrator/tests datatypes/epc/schema/tests datatypes/epc/surveys/tests datatypes/epc/domain/tests backend/ecmk_fetcher/tests/ backend/pashub_fetcher/tests backend/documents_parser/tests backend/magic_plan/tests datatypes/magicplan/api/tests datatypes/magicplan/domain/tests backend/app/db/functions/tests
 markers =
     integration: mark a test as an integration test