test/python/tools/test_admin.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for maintenance and analysis functions.
   9 """
  10 import pytest
  11
  12 from nominatim.errors import UsageError
  13 from nominatim.tools import admin
  14 from nominatim.tokenizer import factory
  15
  16 @pytest.fixture(autouse=True)
  17 def create_placex_table(project_env, tokenizer_mock, temp_db_cursor, placex_table):
  18     """ All tests in this module require the placex table to be set up.
  19     """
  20     temp_db_cursor.execute("DROP TYPE IF EXISTS prepare_update_info CASCADE")
  21     temp_db_cursor.execute("""CREATE TYPE prepare_update_info AS (
  22                              name HSTORE,
  23                              address HSTORE,
  24                              rank_address SMALLINT,
  25                              country_code TEXT,
  26                              class TEXT,
  27                              type TEXT,
  28                              linked_place_id BIGINT
  29                            )""")
  30     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION placex_indexing_prepare(p placex,
  31                                                      OUT result prepare_update_info)
  32                            AS $$
  33                            BEGIN
  34                              result.address := p.address;
  35                              result.name := p.name;
  36                              result.class := p.class;
  37                              result.type := p.type;
  38                              result.country_code := p.country_code;
  39                              result.rank_address := p.rank_address;
  40                            END;
  41                            $$ LANGUAGE plpgsql STABLE;
  42                         """)
  43     factory.create_tokenizer(project_env)
  44
  45
  46 def test_analyse_indexing_no_objects(project_env):
  47     with pytest.raises(UsageError):
  48         admin.analyse_indexing(project_env)
  49
  50
  51 @pytest.mark.parametrize("oid", ['1234', 'N123a', 'X123'])
  52 def test_analyse_indexing_bad_osmid(project_env, oid):
  53     with pytest.raises(UsageError):
  54         admin.analyse_indexing(project_env, osm_id=oid)
  55
  56
  57 def test_analyse_indexing_unknown_osmid(project_env):
  58     with pytest.raises(UsageError):
  59         admin.analyse_indexing(project_env, osm_id='W12345674')
  60
  61
  62 def test_analyse_indexing_with_place_id(project_env, temp_db_cursor):
  63     temp_db_cursor.execute("INSERT INTO placex (place_id) VALUES(12345)")
  64
  65     admin.analyse_indexing(project_env, place_id=12345)
  66
  67
  68 def test_analyse_indexing_with_osm_id(project_env, temp_db_cursor):
  69     temp_db_cursor.execute("""INSERT INTO placex (place_id, osm_type, osm_id)
  70                               VALUES(9988, 'N', 10000)""")
  71
  72     admin.analyse_indexing(project_env, osm_id='N10000')
  73
  74
  75 class TestAdminCleanDeleted:
  76
  77     @pytest.fixture(autouse=True)
  78     def setup_polygon_delete(self, project_env, table_factory, temp_db_cursor):
  79         """ Set up import_polygon_delete table and simplified place_force_delete function
  80         """
  81         self.project_env = project_env
  82         self.temp_db_cursor = temp_db_cursor
  83         table_factory('import_polygon_delete',
  84                       """osm_id BIGINT,
  85                       osm_type CHAR(1),
  86                       class TEXT NOT NULL,
  87                       type TEXT NOT NULL""",
  88                       ((100, 'N', 'boundary', 'administrative'),
  89                       (145, 'N', 'boundary', 'administrative'),
  90                       (175, 'R', 'landcover', 'grass')))
  91         table_factory('place_to_be_deleted',
  92                       """osm_id BIGINT,
  93                       osm_type CHAR(1),
  94                       class TEXT NOT NULL,
  95                       type TEXT NOT NULL,
  96                       deferred BOOLEAN""")
  97         temp_db_cursor.execute("""INSERT INTO placex (place_id, osm_id, osm_type, class, type, indexed_date, indexed_status)
  98                               VALUES(1, 100, 'N', 'boundary', 'administrative', current_date - INTERVAL '1 month', 1),
  99                                (2, 145, 'N', 'boundary', 'administrative', current_date - INTERVAL '1 month', 1),
 100                                (3, 175, 'R', 'landcover', 'grass', current_date - INTERVAL '1 month', 1)""")
 101         temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION flush_deleted_places()
 102                                RETURNS INTEGER
 103                                AS $$
 104                                BEGIN
 105                                 UPDATE placex p SET indexed_status = 100 FROM place_to_be_deleted d
 106                                 WHERE p.osm_type = d.osm_type
 107                                 AND p.osm_id = d.osm_id
 108                                 AND p.class = d.class
 109                                 AND p.type = d.type
 110                                 AND NOT deferred;
 111                                TRUNCATE TABLE place_to_be_deleted;
 112                                 RETURN NULL;
 113                                END;
 114                                $$
 115                                LANGUAGE plpgsql;
 116                                """)
 117         temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION place_force_delete(placeid BIGINT)
 118                                RETURNS BOOLEAN
 119                                AS $$
 120                                DECLARE
 121                                 osmid BIGINT;
 122                                 osmtype character(1);
 123                                 pclass text;
 124                                 ptype text;
 125                                BEGIN
 126                                 SELECT osm_type, osm_id, class, type FROM placex WHERE place_id = placeid INTO osmtype, osmid, pclass, ptype;
 127                                 DELETE FROM import_polygon_delete WHERE osm_type = osmtype AND osm_id = osmid AND class = pclass AND type = ptype;
 128                                 INSERT INTO place_to_be_deleted (osm_type, osm_id, class, type, deferred)
 129                                         VALUES(osmtype, osmid, pclass, ptype, false);
 130                                 PERFORM flush_deleted_places();
 131                                 RETURN TRUE;
 132                                END;
 133                                $$
 134                                LANGUAGE plpgsql;
 135                                """)
 136
 137
 138     def test_admin_clean_deleted_no_records(self):
 139         admin.clean_deleted_relations(self.project_env, age='1 year')
 140         assert self.temp_db_cursor.row_set('SELECT osm_id, osm_type, class, type, indexed_status FROM placex') == {(100, 'N', 'boundary', 'administrative', 1),
 141                                                                                                                    (145, 'N', 'boundary', 'administrative', 1),
 142                                                                                                                    (175, 'R', 'landcover', 'grass', 1)}
 143         assert self.temp_db_cursor.table_rows('import_polygon_delete') == 3
 144
 145
 146     def test_admin_clean_deleted_no_age(self):
 147         with pytest.raises(UsageError):
 148             admin.clean_deleted_relations(self.project_env)
 149
 150
 151     @pytest.mark.parametrize('test_age', ['T week', '1 welk', 'P1E'])
 152     def test_admin_clean_deleted_bad_age(self, test_age):
 153         with pytest.raises(UsageError):
 154             admin.clean_deleted_relations(self.project_env, age = test_age)
 155
 156
 157     @pytest.mark.parametrize('test_age', ['1 week', 'P3D', '5 hours'])
 158     def test_admin_clean_deleted(self, test_age):
 159         admin.clean_deleted_relations(self.project_env, age = test_age)
 160         assert self.temp_db_cursor.row_set('SELECT osm_id, osm_type, class, type, indexed_status FROM placex') == {(100, 'N', 'boundary', 'administrative', 100),
 161                                                                                                                    (145, 'N', 'boundary', 'administrative', 100),
 162                                                                                                                    (175, 'R', 'landcover', 'grass', 100)}
 163         assert self.temp_db_cursor.table_rows('import_polygon_delete') == 0