Add trigger to prevent overlapping land use areas

database/migrations/env.py

@@ -18,7 +18,9 @@
     generate_new_lifecycle_status_triggers,
     generate_update_lifecycle_status_triggers,
     generate_validate_polygon_geometry_triggers,
+    trg_prevent_land_use_area_overlaps,
     trg_validate_line_geometry,
+    trgfunc_prevent_land_use_area_overlaps,
     trgfunc_validate_line_geometry,
 )
 
@@ -60,6 +62,8 @@
     + validate_polygon_geometry_trgfuncs
     + [trg_validate_line_geometry]
     + [trgfunc_validate_line_geometry]
+    + [trgfunc_prevent_land_use_area_overlaps]
+    + [trg_prevent_land_use_area_overlaps]
 )
 
 register_entities(entities=imported_functions, entity_types=[PGTrigger, PGFunction])

database/migrations/versions/2024_11_27_1633-9c28ba93d9be_add_land_use_overlap_prevention_trigger.py

@@ -0,0 +1,63 @@
+"""add land use overlap prevention trigger
+
+Revision ID: 9c28ba93d9be
+Revises: 84142a5b86b6
+Create Date: 2024-11-27 16:33:53.394187
+
+"""
+from typing import Sequence, Union
+
+import geoalchemy2
+import sqlalchemy as sa
+from alembic import op
+from alembic_utils.pg_function import PGFunction
+from alembic_utils.pg_trigger import PGTrigger
+from sqlalchemy import text as sql_text
+
+# revision identifiers, used by Alembic.
+revision: str = "9c28ba93d9be"
+down_revision: Union[str, None] = "84142a5b86b6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    hame_trgfunc_land_use_area_prevent_overlap = PGFunction(
+        schema="hame",
+        signature="trgfunc_land_use_area_prevent_overlap()",
+        definition="RETURNS TRIGGER AS $$\n    DECLARE\n        overlapping_id UUID;\n    BEGIN\n        SELECT id INTO overlapping_id\n        FROM hame.land_use_area\n        WHERE\n            plan_id = NEW.plan_id\n            AND ST_Overlaps(geom, NEW.geom)\n        ;\n        IF overlapping_id IS NOT NULL THEN\n            RAISE EXCEPTION 'Geometries overlap\\: % - %',\n                NEW.id, overlapping_id\n                USING HINT = 'Two land use areas cannot overlap';\n        END IF;\n        RETURN NEW;\n    END;\n    $$ language 'plpgsql'",
+    )
+    op.create_entity(hame_trgfunc_land_use_area_prevent_overlap)
+
+    hame_land_use_area_trg_land_use_area_prevent_overlap = PGTrigger(
+        schema="hame",
+        signature="trg_land_use_area_prevent_overlap",
+        on_entity="hame.land_use_area",
+        is_constraint=False,
+        definition="BEFORE INSERT OR UPDATE ON land_use_area\n        FOR EACH ROW\n        EXECUTE FUNCTION hame.trgfunc_land_use_area_prevent_overlap()",
+    )
+    op.create_entity(hame_land_use_area_trg_land_use_area_prevent_overlap)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    hame_land_use_area_trg_land_use_area_prevent_overlap = PGTrigger(
+        schema="hame",
+        signature="trg_land_use_area_prevent_overlap",
+        on_entity="hame.land_use_area",
+        is_constraint=False,
+        definition="BEFORE INSERT OR UPDATE ON land_use_area\n        FOR EACH ROW\n        EXECUTE FUNCTION hame.trgfunc_land_use_area_prevent_overlap()",
+    )
+    op.drop_entity(hame_land_use_area_trg_land_use_area_prevent_overlap)
+
+    hame_trgfunc_land_use_area_prevent_overlap = PGFunction(
+        schema="hame",
+        signature="trgfunc_land_use_area_prevent_overlap()",
+        definition="RETURNS TRIGGER AS $$\n    DECLARE\n        overlapping_id UUID;\n    BEGIN\n        SELECT id INTO overlapping_id\n        FROM hame.land_use_area\n        WHERE\n            plan_id = NEW.plan_id\n            AND ST_Overlaps(geom, NEW.geom)\n        ;\n        IF overlapping_id IS NOT NULL THEN\n            RAISE EXCEPTION 'Geometries overlap\\: % - %',\n                NEW.id, overlapping_id\n                USING HINT = 'Two land use areas cannot overlap';\n        END IF;\n        RETURN NEW;\n    END;\n    $$ language 'plpgsql'",
+    )
+    op.drop_entity(hame_trgfunc_land_use_area_prevent_overlap)
+
+    # ### end Alembic commands ###

database/test/conftest.py

@@ -463,6 +463,12 @@ def session(connection_string):
     yield session()
 
 
+@pytest.fixture
+def rollback_after(session: Session):
+    yield
+    session.rollback()
+
+
 @pytest.fixture()
 def code_instance(session):
     instance = codes.LifeCycleStatus(value="test", status="LOCAL")

database/test/test_triggers.py

@@ -4,6 +4,7 @@
 import models
 import pytest
 from geoalchemy2.shape import from_shape
+from shapely import transform
 from shapely.geometry import MultiLineString, MultiPoint, MultiPolygon, shape
 from sqlalchemy.exc import InternalError
 from sqlalchemy.orm import Session
@@ -620,41 +621,71 @@ def test_validate_line_geometry(
     session.rollback()
 
 
-def test_intersecting_other_area_geometries_trigger(
+def test_overlapping_land_use_area_geometries_trigger(
     session: Session,
-    empty_value_plan_regulation_instance: models.PlanRegulation,
+    plan_instance: models.Plan,
     code_instance: codes.LifeCycleStatus,
     type_of_underground_instance: codes.TypeOfUnderground,
-    plan_regulation_group_instance: models.PlanRegulationGroup,
+    rollback_after,
 ):
-    paakayttotarkoitus_additional_information_instance = (
-        codes.TypeOfAdditionalInformation(value="paakayttotarkoitus", status="LOCAL")
-    )
-    session.add(paakayttotarkoitus_additional_information_instance)
-    empty_value_plan_regulation_instance.intended_use = (
-        paakayttotarkoitus_additional_information_instance
+    square = MultiPolygon([(((0, 0), (0, 1), (1, 1), (1, 0), (0, 0)),)])
+    overlapping_square = transform(square, lambda x: x + 0.5)
+
+    land_use_area_instance = models.LandUseArea(
+        plan=plan_instance,
+        name="area 1",
+        geom=from_shape(square),
+        lifecycle_status=code_instance,
+        type_of_underground=type_of_underground_instance,
     )
+    session.add(land_use_area_instance)
     session.flush()
 
-    another_area_instance = models.OtherArea(
-        geom=from_shape(
-            MultiPolygon([(((1.0, 2.0), (2.0, 2.0), (2.0, 1.0), (1.0, 1.0)),)])
-        ),
+    # Create a new land_use_area that overlaps land_use_area_instance
+    new_land_use_area_instance = models.LandUseArea(
+        plan=plan_instance,
+        name="area 2",
+        geom=from_shape(overlapping_square),
         lifecycle_status=code_instance,
         type_of_underground=type_of_underground_instance,
-        plan_regulation_group=plan_regulation_group_instance,
     )
-    session.add(another_area_instance)
-    # Create a new other_area_instance that intersects another_area_instance
-    new_other_area_instance = models.OtherArea(
-        geom=from_shape(
-            MultiPolygon([(((1.0, 2.0), (2.0, 2.0), (2.0, 1.0), (1.0, 1.0)),)])
-        ),
+
+    with pytest.raises(InternalError) as excinfo:
+        session.add(new_land_use_area_instance)
+        session.flush()
+    assert "Geometries overlap" in str(excinfo.value.orig.pgerror)
+
+
+def test_adjacent_land_use_areas_should_be_fine(
+    session: Session,
+    plan_instance: models.Plan,
+    code_instance: codes.LifeCycleStatus,
+    type_of_underground_instance: codes.TypeOfUnderground,
+    rollback_after,
+):
+    square = MultiPolygon([(((0, 0), (0, 1), (1, 1), (1, 0), (0, 0)),)])
+    adjacent_square = transform(square, lambda vertex: vertex + [1, 0])
+
+    land_use_area_instance = models.LandUseArea(
+        plan=plan_instance,
+        name="area 1",
+        geom=from_shape(square),
         lifecycle_status=code_instance,
         type_of_underground=type_of_underground_instance,
-        plan_regulation_group=plan_regulation_group_instance,
     )
-    session.add(new_other_area_instance)
-    with pytest.raises(InternalError):
-        session.flush()
-    session.rollback()
+    session.add(land_use_area_instance)
+    session.flush()
+
+    # Create a new land_use_area that is adjacent to land_use_area_instance
+    new_land_use_area_instance = models.LandUseArea(
+        plan=plan_instance,
+        name="area 2",
+        geom=from_shape(adjacent_square),
+        lifecycle_status=code_instance,
+        type_of_underground=type_of_underground_instance,
+    )
+
+    session.add(new_land_use_area_instance)
+    session.flush()
+
+    assert True  # No exception was raised

database/triggers.py

@@ -508,3 +508,39 @@ def generate_validate_polygon_geometry_triggers():
     FOR EACH ROW
     EXECUTE FUNCTION hame.trgfunc_line_validate_geometry()""",
 )
+
+trgfunc_prevent_land_use_area_overlaps = PGFunction(
+    schema="hame",
+    signature="trgfunc_land_use_area_prevent_overlap()",
+    definition="""
+    RETURNS TRIGGER AS $$
+    DECLARE
+        overlapping_id UUID;
+    BEGIN
+        SELECT id INTO overlapping_id
+        FROM hame.land_use_area
+        WHERE
+            plan_id = NEW.plan_id
+            AND ST_Overlaps(geom, NEW.geom)
+        ;
+        IF overlapping_id IS NOT NULL THEN
+            RAISE EXCEPTION 'Geometries overlap: % - %',
+                NEW.id, overlapping_id
+                USING HINT = 'Two land use areas cannot overlap';
+        END IF;
+        RETURN NEW;
+    END;
+    $$ language 'plpgsql'
+    """,
+)
+
+trg_prevent_land_use_area_overlaps = PGTrigger(
+    schema="hame",
+    signature="trg_land_use_area_prevent_overlap",
+    on_entity="hame.land_use_area",
+    definition="""
+        BEFORE INSERT OR UPDATE ON land_use_area
+        FOR EACH ROW
+        EXECUTE FUNCTION hame.trgfunc_land_use_area_prevent_overlap()
+    """,
+)