test/python/mocks.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Custom mocks for testing.
   9 """
  10 import itertools
  11
  12
  13 class MockPlacexTable:
  14     """ A placex table for testing.
  15     """
  16     def __init__(self, conn):
  17         self.idseq = itertools.count(10000)
  18         self.conn = conn
  19         with conn.cursor() as cur:
  20             cur.execute("""CREATE TABLE placex (
  21                                place_id BIGINT,
  22                                parent_place_id BIGINT,
  23                                linked_place_id BIGINT,
  24                                importance FLOAT,
  25                                indexed_date TIMESTAMP,
  26                                geometry_sector INTEGER,
  27                                rank_address SMALLINT,
  28                                rank_search SMALLINT,
  29                                partition SMALLINT,
  30                                indexed_status SMALLINT,
  31                                osm_id int8,
  32                                osm_type char(1),
  33                                class text,
  34                                type text,
  35                                name hstore,
  36                                admin_level smallint,
  37                                address hstore,
  38                                extratags hstore,
  39                                token_info jsonb,
  40                                geometry Geometry(Geometry,4326),
  41                                wikipedia TEXT,
  42                                country_code varchar(2),
  43                                housenumber TEXT,
  44                                postcode TEXT,
  45                                centroid GEOMETRY(Geometry, 4326))""")
  46             cur.execute("CREATE SEQUENCE IF NOT EXISTS seq_place")
  47         conn.commit()
  48
  49     def add(self, osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
  50             admin_level=None, address=None, extratags=None, geom='POINT(10 4)',
  51             country=None, housenumber=None, rank_search=30, centroid=None):
  52         with self.conn.cursor() as cur:
  53             cur.execute("""INSERT INTO placex (place_id, osm_type, osm_id, class,
  54                                                type, name, admin_level, address,
  55                                                housenumber, rank_search,
  56                                                extratags, centroid, geometry, country_code)
  57                             VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s,
  58                                    %s, %s, %s, %s, %s, %s, %s)
  59                             RETURNING place_id""",
  60                         (osm_type, osm_id or next(self.idseq), cls, typ, names,
  61                          admin_level, address, housenumber, rank_search,
  62                          extratags, centroid, 'SRID=4326;' + geom,
  63                          country))
  64             place_id = cur.fetchone()[0]
  65         self.conn.commit()
  66         return place_id