Stabilize shared-db API tests

Author: jirhiker

tests/test_location.py

@@ -161,32 +161,33 @@ 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), None)
+    assert item is not None
+    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 "nma_location_notes" in data["items"][0]
-    assert data["items"][0]["nma_location_notes"] == location.nma_location_notes
-    assert "nma_data_reliability" in data["items"][0]
-    assert data["items"][0]["nma_data_reliability"] == location.nma_data_reliability
+    assert item["point"] == to_shape(location.point).wkt
+    assert item["elevation"] == location.elevation
+    assert item["release_status"] == location.release_status
+    assert "nma_location_notes" in item
+    assert item["nma_location_notes"] == location.nma_location_notes
+    assert "nma_data_reliability" in item
+    assert item["nma_data_reliability"] == location.nma_data_reliability
     # 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):

tests/test_sample.py

@@ -285,7 +285,11 @@ def test_get_samples(water_chemistry_sample, groundwater_level_sample):
     response = client.get("/sample")
     assert response.status_code == 200
     data = response.json()
-    assert len(data["items"]) == 2
+    assert len(data["items"]) >= 2
+
+    item_ids = {item["id"] for item in data["items"]}
+    assert water_chemistry_sample.id in item_ids
+    assert groundwater_level_sample.id in item_ids
 
     for item in data["items"]:
         assert "id" in item
@@ -312,7 +316,7 @@ def test_get_samples_by_thing_id(
     response = client.get(f"/sample?thing_id={water_well_thing.id}")
     assert response.status_code == 200
     data = response.json()
-    assert data["total"] == 2
+    assert data["total"] >= 2
 
     data_ids = [d["id"] for d in data["items"]]
     sorted_data_ids = sorted(data_ids)

tests/test_sensor.py

@@ -175,20 +175,21 @@ def test_get_sensors(sensor):
     response = client.get("/sensor")
     assert response.status_code == 200
     data = response.json()
-    assert data["total"] == 1
-    assert data["items"][0]["id"] == sensor.id
-    assert data["items"][0]["created_at"] == sensor.created_at.astimezone(
-        timezone.utc
-    ).strftime(DT_FMT)
-    assert data["items"][0]["release_status"] == sensor.release_status
-    assert data["items"][0]["name"] == sensor.name
-    assert data["items"][0]["sensor_type"] == sensor.sensor_type
-    assert data["items"][0]["model"] == sensor.model
-    assert data["items"][0]["serial_no"] == sensor.serial_no
-    assert data["items"][0]["pcn_number"] == sensor.pcn_number
-    assert data["items"][0]["owner_agency"] == sensor.owner_agency
-    assert data["items"][0]["sensor_status"] == sensor.sensor_status
-    assert data["items"][0]["notes"] == sensor.notes
+    assert data["total"] >= 1
+    item = next((item for item in data["items"] if item["id"] == sensor.id), None)
+    assert item is not None
+    assert item["created_at"] == sensor.created_at.astimezone(timezone.utc).strftime(
+        DT_FMT
+    )
+    assert item["release_status"] == sensor.release_status
+    assert item["name"] == sensor.name
+    assert item["sensor_type"] == sensor.sensor_type
+    assert item["model"] == sensor.model
+    assert item["serial_no"] == sensor.serial_no
+    assert item["pcn_number"] == sensor.pcn_number
+    assert item["owner_agency"] == sensor.owner_agency
+    assert item["sensor_status"] == sensor.sensor_status
+    assert item["notes"] == sensor.notes
 
 
 def test_get_sensors_by_thing_id(
@@ -219,20 +220,21 @@ def test_get_sensors_by_parameter_id(sensor, groundwater_level_observation):
     response = client.get(f"/sensor?parameter_id={_groundwater_level_parameter_id()}")
     assert response.status_code == 200
     data = response.json()
-    assert data["total"] == 1
-    assert data["items"][0]["id"] == sensor.id
-    assert data["items"][0]["created_at"] == sensor.created_at.astimezone(
-        timezone.utc
-    ).strftime(DT_FMT)
-    assert data["items"][0]["release_status"] == sensor.release_status
-    assert data["items"][0]["name"] == sensor.name
-    assert data["items"][0]["sensor_type"] == sensor.sensor_type
-    assert data["items"][0]["model"] == sensor.model
-    assert data["items"][0]["serial_no"] == sensor.serial_no
-    assert data["items"][0]["pcn_number"] == sensor.pcn_number
-    assert data["items"][0]["owner_agency"] == sensor.owner_agency
-    assert data["items"][0]["sensor_status"] == sensor.sensor_status
-    assert data["items"][0]["notes"] == sensor.notes
+    assert data["total"] >= 1
+    item = next((item for item in data["items"] if item["id"] == sensor.id), None)
+    assert item is not None
+    assert item["created_at"] == sensor.created_at.astimezone(timezone.utc).strftime(
+        DT_FMT
+    )
+    assert item["release_status"] == sensor.release_status
+    assert item["name"] == sensor.name
+    assert item["sensor_type"] == sensor.sensor_type
+    assert item["model"] == sensor.model
+    assert item["serial_no"] == sensor.serial_no
+    assert item["pcn_number"] == sensor.pcn_number
+    assert item["owner_agency"] == sensor.owner_agency
+    assert item["sensor_status"] == sensor.sensor_status
+    assert item["notes"] == sensor.notes
 
 
 def test_get_sensor_by_id(sensor):

tests/test_thing.py

@@ -731,19 +731,19 @@ def test_get_thing_id_links(thing_id_link):
     response = client.get("/thing/id-link")
     assert response.status_code == 200
     data = response.json()
-    assert data["total"] == 1
-    assert data["items"][0]["id"] == thing_id_link.id
-    assert data["items"][0]["created_at"] == thing_id_link.created_at.astimezone(
+    assert data["total"] >= 1
+    item = next(
+        (item for item in data["items"] if item["id"] == thing_id_link.id), None
+    )
+    assert item is not None
+    assert item["created_at"] == thing_id_link.created_at.astimezone(
         timezone.utc
     ).strftime(DT_FMT)
-    assert data["items"][0]["release_status"] == thing_id_link.release_status
-    assert data["items"][0]["thing_id"] == thing_id_link.thing_id
-    assert data["items"][0]["relation"] == thing_id_link.relation
-    assert data["items"][0]["alternate_id"] == thing_id_link.alternate_id
-    assert (
-        data["items"][0]["alternate_organization"]
-        == thing_id_link.alternate_organization
-    )
+    assert item["release_status"] == thing_id_link.release_status
+    assert item["thing_id"] == thing_id_link.thing_id
+    assert item["relation"] == thing_id_link.relation
+    assert item["alternate_id"] == thing_id_link.alternate_id
+    assert item["alternate_organization"] == thing_id_link.alternate_organization
 
 
 def test_get_thing_id_link_by_id(thing_id_link):
@@ -797,11 +797,11 @@ def test_get_things(water_well_thing, spring_thing, location):
     response = client.get("/thing")
     assert response.status_code == 200
 
-    expected_location = LocationResponse.model_validate(location).model_dump()
-    # created_at is already serialized to UTC format by UTCAwareDatetime
-
     data = response.json()
-    assert data["total"] == 2
+    assert data["total"] >= 2
+    item_ids = {item["id"] for item in data["items"]}
+    assert water_well_thing.id in item_ids
+    assert spring_thing.id in item_ids
 
 
 @pytest.mark.skip("Needs to be updated per changes made from feature files")

tests/test_well_inventory.py

@@ -15,6 +15,7 @@
 from cli.service_adapter import well_inventory_csv
 from core.constants import SRID_UTM_ZONE_13N, SRID_WGS84
 from db import (
+    Base,
     Location,
     LocationThingAssociation,
     Thing,
@@ -29,6 +30,24 @@
 from shapely import Point
 
 
+def _reset_well_inventory_tables() -> None:
+    with session_ctx() as session:
+        for table in reversed(Base.metadata.sorted_tables):
+            if table.name in ("alembic_version", "parameter"):
+                continue
+            if table.name.startswith("lexicon"):
+                continue
+            session.execute(table.delete())
+        session.commit()
+
+
+@pytest.fixture(autouse=True)
+def isolate_well_inventory_tables():
+    _reset_well_inventory_tables()
+    yield
+    _reset_well_inventory_tables()
+
+
 def test_well_inventory_db_contents():
     """
     Test that the well inventory upload creates the correct database contents.
@@ -417,17 +436,6 @@ def test_well_inventory_db_contents():
                 else:
                     assert participant.participant.name == file_content["field_staff_2"]
 
-        # CLEAN UP THE DATABASE AFTER TESTING
-        session.query(FieldEventParticipant).delete()
-        session.query(FieldActivity).delete()
-        session.query(FieldEvent).delete()
-        session.query(ThingContactAssociation).delete()
-        session.query(LocationThingAssociation).delete()
-        session.query(Contact).delete()
-        session.query(Location).delete()
-        session.query(Thing).delete()
-        session.commit()
-
 
 # =============================================================================
 # Error Handling Tests - Cover API error paths
@@ -582,7 +590,7 @@ def test_upload_missing_contact_type(self):
             result = well_inventory_csv(file_path)
             assert result.exit_code == 1
 
-    def test_upload_missing_contact_type(self):
+    def test_upload_missing_contact_role(self):
         """Upload fails when contact is provided without role."""
         file_path = Path("tests/features/data/well-inventory-missing-contact-role.csv")
         if file_path.exists():
@@ -966,15 +974,5 @@ def test_upload_valid_with_comma_in_quotes(self):
             # Should succeed - commas in quoted fields are valid CSV
             assert result.exit_code in (0, 1)  # 1 if other validation fails
 
-            # Clean up if records were created
-            if result.exit_code == 0:
-                with session_ctx() as session:
-                    session.query(Thing).delete()
-                    session.query(Location).delete()
-                    session.query(Contact).delete()
-                    session.query(FieldEvent).delete()
-                    session.query(FieldActivity).delete()
-                    session.commit()
-
 
 # ============= EOF =============================================