feat: enhance test setup with location and measuring point creation for wells

Author: jirhiker

db/thing.py

@@ -423,6 +423,8 @@ def measuring_point_height(self) -> int | None:
         Since measuring_point_history is eagerly loaded, this should not introduce N+1 query issues.
         """
         if self.thing_type == "water well":
+            if not self.measuring_points:
+                return None
             sorted_measuring_point_history = sorted(
                 self.measuring_points, key=lambda x: x.start_date, reverse=True
             )
@@ -439,6 +441,8 @@ def measuring_point_description(self) -> str | None:
         Since measuring_point_history is eagerly loaded, this should not introduce N+1 query issues.
         """
         if self.thing_type == "water well":
+            if not self.measuring_points:
+                return None
             sorted_measuring_point_history = sorted(
                 self.measuring_points, key=lambda x: x.start_date, reverse=True
             )

tests/conftest.py

@@ -1,23 +1,19 @@
 import os
-import uuid
 
 import pytest
 from alembic import command
 from alembic.config import Config
-from dotenv import load_dotenv
-from sqlalchemy import text
-from sqlalchemy_utils import TSVectorType
-from sqlalchemy_searchable import sync_trigger
-
 from core.initializers import init_lexicon, init_parameter
 from db import *
 from db.engine import session_ctx
+from dotenv import load_dotenv
+from sqlalchemy_searchable import sync_trigger
 from tests import get_parameter_id
 
 
 def pytest_configure():
     load_dotenv(override=True)
-    os.environ.setdefault("POSTGRES_PORT", "54321")
+    os.environ["POSTGRES_PORT"] = "54321"
 
 
 def _alembic_config() -> Config:
@@ -72,9 +68,54 @@ def _setup_test_db():
     _sync_search_vectors()
     init_lexicon()
     init_parameter()
+    _ensure_locations_for_things()
+    _ensure_measuring_points_for_wells()
     yield
 
 
+def _ensure_locations_for_things() -> None:
+    with session_ctx() as session:
+        things = session.query(Thing).all()
+        for thing in things:
+            if thing.location_associations:
+                continue
+            location = Location(
+                point="POINT(-107.0 33.0)",
+                elevation=0,
+                release_status=thing.release_status,
+            )
+            session.add(location)
+            session.flush()
+            session.add(
+                LocationThingAssociation(
+                    location_id=location.id,
+                    thing_id=thing.id,
+                    effective_start=date.today().isoformat(),
+                )
+            )
+        session.commit()
+
+
+def _ensure_measuring_points_for_wells() -> None:
+    with session_ctx() as session:
+        wells = session.query(Thing).filter(Thing.thing_type == "water well").all()
+        missing = [well for well in wells if not well.measuring_points]
+        for well in missing:
+            session.add(
+                MeasuringPointHistory(
+                    thing_id=well.id,
+                    measuring_point_height=well.well_casing_depth
+                    or well.well_depth
+                    or 0.0,
+                    measuring_point_description="autogenerated for tests",
+                    start_date=date.today(),
+                    release_status=well.release_status,
+                )
+            )
+        if missing:
+            session.commit()
+
+
 @pytest.fixture()
 def location():
     with session_ctx() as session:

tests/test_geospatial.py

@@ -13,11 +13,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ===============================================================================
+from datetime import date
 from pathlib import Path
 
 import pytest
-from geoalchemy2 import functions as geofunc
-
 from core.constants import SRID_WGS84
 from core.dependencies import (
     admin_function,
@@ -29,6 +28,7 @@
 )
 from db import Thing, Location, LocationThingAssociation, Group, MeasuringPointHistory
 from db.engine import session_ctx
+from geoalchemy2 import functions as geofunc
 from main import app
 from tests import client, override_authentication
 
@@ -68,46 +68,16 @@ def override_authentication_dependency_fixture():
 @pytest.fixture(autouse=True, scope="module")
 def populate():
     with session_ctx() as session:
-        # Create some sample data
-        thing1 = Thing(name="Thing 1", thing_type="water well")
-        thing2 = Thing(name="Thing 2", thing_type="water well")
-        session.add(thing1)
-        session.add(thing2)
-
-        session.commit()
-
-        mp_history_1 = MeasuringPointHistory(
-            thing_id=thing1.id,
-            measuring_point_height=5.0,
-            measuring_point_description="MP for Thing 1",
-            start_date="2023-01-01",
-            reason="Initial entry",
-        )
-        mp_history_2 = MeasuringPointHistory(
-            thing_id=thing2.id,
-            measuring_point_height=10.0,
-            measuring_point_description="MP for Thing 2",
-            start_date="2023-01-01",
-            reason="Initial entry",
-        )
-        session.add(mp_history_1)
-        session.add(mp_history_2)
-
-        loc1 = Location(
-            # name="Test Location 1",
-            point=geofunc.ST_GeomFromText("POINT(10.1 10.1)", srid=SRID_WGS84),
-            elevation=0,
+        thing1 = _create_well_with_location(
+            session,
+            name="Thing 1",
+            point_wkt="POINT(10.1 10.1)",
         )
-        loc2 = Location(
-            # name="Test Location 2",
-            point=geofunc.ST_GeomFromText("POINT(20 20)", srid=SRID_WGS84),
-            elevation=0,
+        thing2 = _create_well_with_location(
+            session,
+            name="Thing 2",
+            point_wkt="POINT(20 20)",
         )
-        session.add(loc1)
-        session.add(loc2)
-
-        session.add(LocationThingAssociation(location=loc1, thing=thing1))
-        session.add(LocationThingAssociation(location=loc2, thing=thing2))
 
         group = Group(
             name="Test Group Foo",
@@ -117,15 +87,48 @@ def populate():
 
         session.add(group)
         session.commit()
-        yield
-
-        # Cleanup
-        session.delete(loc1)
-        session.delete(loc2)
-        session.delete(group)
-        session.delete(thing1)
-        session.delete(thing2)
-        session.commit()
+        try:
+            yield
+        finally:
+            session.delete(group)
+            session.delete(thing1)
+            session.delete(thing2)
+            session.commit()
+
+
+def _create_well_with_location(session, name: str, point_wkt: str) -> Thing:
+    thing = Thing(name=name, thing_type="water well", release_status="draft")
+    session.add(thing)
+    session.commit()
+    session.refresh(thing)
+
+    location = Location(
+        point=geofunc.ST_GeomFromText(point_wkt, srid=SRID_WGS84),
+        elevation=0,
+        release_status="draft",
+    )
+    session.add(location)
+    session.flush()
+    session.add(
+        LocationThingAssociation(
+            location_id=location.id,
+            thing_id=thing.id,
+            effective_start=date.today().isoformat(),
+        )
+    )
+    session.add(
+        MeasuringPointHistory(
+            thing_id=thing.id,
+            measuring_point_height=5.0,
+            measuring_point_description=f"MP for {name}",
+            start_date=date.today(),
+            release_status="draft",
+        )
+    )
+    session.commit()
+    return thing
+
+    # Cleanup
 
 
 def test_get_project_area():

tests/test_location.py

@@ -16,10 +16,9 @@
 from datetime import timezone
 
 import pytest
-from geoalchemy2.shape import to_shape
-
 from core.dependencies import admin_function, editor_function, viewer_function
 from db import Location
+from geoalchemy2.shape import to_shape
 from main import app
 from schemas import DT_FMT
 from tests import (
@@ -159,28 +158,28 @@ def test_get_locations(location):
     response = client.get("/location")
     assert response.status_code == 200
     data = response.json()
-    assert data["total"] == 1
-    assert data["items"][0]["id"] == location.id
-    assert data["items"][0]["created_at"] == location.created_at.astimezone(
-        timezone.utc
-    ).strftime(DT_FMT)
+    assert data["total"] >= 1
+    item = next(item for item in data["items"] if item["id"] == location.id)
+    assert item["created_at"] == location.created_at.astimezone(timezone.utc).strftime(
+        DT_FMT
+    )
     # assert data["items"][0]["name"] == location.name
-    assert isinstance(data["items"][0]["notes"], list)
+    assert isinstance(item["notes"], list)
     # If you know the exact number of notes expected:
     # assert len(data["items"][0]["notes"]) == expected_count
     # If you want to check content of a specific note:
     # if data["items"][0]["notes"]:
     #     assert data["items"][0]["notes"][0]["content"] == expected_content
-    assert data["items"][0]["point"] == to_shape(location.point).wkt
-    assert data["items"][0]["elevation"] == location.elevation
-    assert data["items"][0]["release_status"] == location.release_status
+    assert item["point"] == to_shape(location.point).wkt
+    assert item["elevation"] == location.elevation
+    assert item["release_status"] == location.release_status
     # assert data["items"][0]["elevation_accuracy"] == location.elevation_accuracy
     # assert data["items"][0]["elevation_method"] == location.elevation_method
     # assert data["items"][0]["coordinate_accuracy"] == location.coordinate_accuracy
     # assert data["items"][0]["coordinate_method"] == location.coordinate_method
-    assert data["items"][0]["state"] == location.state
-    assert data["items"][0]["county"] == location.county
-    assert data["items"][0]["quad_name"] == location.quad_name
+    assert item["state"] == location.state
+    assert item["county"] == location.county
+    assert item["quad_name"] == location.quad_name
 
 
 def test_get_location_by_id(location):