adapt unit tests to use separate interpolation table

test/python/conftest.py

@@ -192,9 +192,9 @@ def place_table(temp_db_with_extensions, table_factory):
                      type text NOT NULL,
                      name hstore,
                      admin_level smallint,
-                     address hstore,
+                     address HSTORE,
-                     extratags hstore,
+                     extratags HSTORE,
-                     geometry Geometry(Geometry,4326) NOT NULL""")
+                     geometry GEOMETRY(Geometry,4326) NOT NULL""")
 
 
 @pytest.fixture
@@ -223,9 +223,9 @@ def place_postcode_table(temp_db_with_extensions, table_factory):
                   """osm_type char(1) NOT NULL,
                      osm_id bigint NOT NULL,
                      postcode text NOT NULL,
-                     country_code text,
+                     country_code TEXT,
-                     centroid Geometry(Point, 4326) NOT NULL,
+                     centroid GEOMETRY(Point, 4326) NOT NULL,
-                     geometry Geometry(Geometry, 4326)""")
+                     geometry GEOMETRY(Geometry, 4326)""")
 
 
 @pytest.fixture
@@ -246,6 +246,36 @@ def place_postcode_row(place_postcode_table, temp_db_cursor):
     return _insert
 
 
+@pytest.fixture
+def place_interpolation_table(temp_db_with_extensions, table_factory):
+    """ Create an empty version of the place_interpolation table.
+    """
+    table_factory('place_interpolation',
+                  """osm_type char(1) NOT NULL,
+                     osm_id bigint NOT NULL,
+                     type TEXT,
+                     address HSTORE,
+                     nodes BIGINT[],
+                     geometry GEOMETRY(Geometry, 4326)""")
+
+
+@pytest.fixture
+def place_interpolation_row(place_interpolation_table, temp_db_cursor):
+    """ A factory for rows in the place_interpolation table. The table is created as a
+        prerequisite to the fixture.
+    """
+    idseq = itertools.count(30001)
+
+    def _insert(osm_type='N', osm_id=None, typ='odd', address=None,
+                nodes=None, geom='LINESTRING(0.1 0.21, 0.1 0.2)'):
+        params = {'osm_type': osm_type, 'osm_id': osm_id or next(idseq),
+                  'type': typ, 'address': address, 'nodes': nodes,
+                  'geometry': _with_srid(geom)}
+        temp_db_cursor.insert_row('place_interpolation', **params)
+
+    return _insert
+
+
 @pytest.fixture
 def placex_table(temp_db_with_extensions, temp_db_conn, load_sql, place_table):
     """ Create an empty version of the placex table.
@@ -289,10 +319,11 @@ def osmline_table(temp_db_with_extensions, load_sql):
 def osmline_row(osmline_table, temp_db_cursor):
     idseq = itertools.count(20001)
 
-    def _add(osm_id=None, geom='LINESTRING(12.0 11.0, 12.003 11.0)'):
+    def _add(osm_type='W', osm_id=None, geom='LINESTRING(12.0 11.0, 12.003 11.0)'):
         return temp_db_cursor.insert_row(
             'location_property_osmline',
             place_id=pysql.SQL("nextval('seq_place')"),
+            osm_type=osm_type,
             osm_id=osm_id or next(idseq),
             geometry_sector=pysql.Literal(20),
             partition=pysql.Literal(0),

test/python/tools/test_admin.py

@@ -76,8 +76,8 @@ def test_analyse_indexing_with_osm_id(project_env, placex_row):
 class TestAdminCleanDeleted:
 
     @pytest.fixture(autouse=True)
-    def setup_polygon_delete(self, project_env, table_factory, place_table, placex_row,
+    def setup_polygon_delete(self, project_env, table_factory, place_interpolation_table,
-                             osmline_table, temp_db_cursor, load_sql):
+                             placex_row, osmline_table, temp_db_cursor, load_sql):
         """ Set up place_force_delete function and related tables
         """
         self.project_env = project_env
@@ -106,16 +106,24 @@ class TestAdminCleanDeleted:
                       class TEXT NOT NULL,
                       type TEXT NOT NULL,
                       deferred BOOLEAN""")
+        table_factory('place_interpolation_to_be_deleted',
+                      """osm_id BIGINT,
+                      osm_type CHAR(1)""")
         table_factory('import_polygon_error', """osm_id BIGINT,
                       osm_type CHAR(1),
                       class TEXT NOT NULL,
                       type TEXT NOT NULL""")
-        temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION place_delete()
+        temp_db_cursor.execute("""
-                               RETURNS TRIGGER AS $$
+            CREATE OR REPLACE FUNCTION place_delete() RETURNS TRIGGER AS $$
-                               BEGIN RETURN NULL; END;
+                BEGIN RETURN NULL; END;
-                               $$ LANGUAGE plpgsql;""")
+            $$ LANGUAGE plpgsql;
-        temp_db_cursor.execute("""CREATE TRIGGER place_before_delete BEFORE DELETE ON place
+
-                               FOR EACH ROW EXECUTE PROCEDURE place_delete();""")
+            CREATE TRIGGER place_before_delete BEFORE DELETE ON place
+            FOR EACH ROW EXECUTE PROCEDURE place_delete();
+
+            CREATE TRIGGER place_interpolation_before_delete BEFORE DELETE ON place_interpolation
+            FOR EACH ROW EXECUTE PROCEDURE place_delete();
+        """)
         load_sql('functions/utils.sql')
 
     def test_admin_clean_deleted_no_records(self):

test/python/tools/test_database_import.py

@@ -165,12 +165,12 @@ def test_truncate_database_tables(temp_db_conn, temp_db_cursor, table_factory, w
 
 @pytest.mark.parametrize("threads", (1, 5))
 @pytest.mark.asyncio
-async def test_load_data(dsn, place_row, placex_table, osmline_table,
+async def test_load_data(dsn, place_row, place_interpolation_row, placex_table, osmline_table,
                          temp_db_cursor, threads):
     for oid in range(100, 130):
         place_row(osm_id=oid)
-    place_row(osm_type='W', osm_id=342, cls='place', typ='houses',
+    place_interpolation_row(osm_type='W', osm_id=342, typ='odd',
-              geom='LINESTRING(0 0, 10 10)')
+                            geom='LINESTRING(0 0, 10 10)')
 
     temp_db_cursor.execute("""
         CREATE OR REPLACE FUNCTION placex_insert() RETURNS TRIGGER AS $$
@@ -238,7 +238,7 @@ class TestSetupSQL:
         assert not reverse == temp_db_cursor.table_exists('search_name')
 
     def test_create_table_triggers(self, temp_db_conn, placex_table, osmline_table,
-                                   postcode_table, load_sql):
+                                   place_interpolation_table, postcode_table, load_sql):
         load_sql('functions.sql')
 
         database_import.create_table_triggers(temp_db_conn, self.config)