test/python/mocks.py

   1 """
   2 Custom mocks for testing.
   3 """
   4 import itertools
   5
   6 import psycopg2.extras
   7
   8 from nominatim.db import properties
   9
  10 class MockParamCapture:
  11     """ Mock that records the parameters with which a function was called
  12         as well as the number of calls.
  13     """
  14     def __init__(self, retval=0):
  15         self.called = 0
  16         self.return_value = retval
  17         self.last_args = None
  18         self.last_kwargs = None
  19
  20     def __call__(self, *args, **kwargs):
  21         self.called += 1
  22         self.last_args = args
  23         self.last_kwargs = kwargs
  24         return self.return_value
  25
  26
  27 class MockWordTable:
  28     """ A word table for testing.
  29     """
  30     def __init__(self, conn):
  31         self.conn = conn
  32         with conn.cursor() as cur:
  33             cur.execute("""CREATE TABLE word (word_id INTEGER,
  34                                               word_token text,
  35                                               word text,
  36                                               class text,
  37                                               type text,
  38                                               country_code varchar(2),
  39                                               search_name_count INTEGER,
  40                                               operator TEXT)""")
  41
  42         conn.commit()
  43
  44     def add_special(self, word_token, word, cls, typ, oper):
  45         with self.conn.cursor() as cur:
  46             cur.execute("""INSERT INTO word (word_token, word, class, type, operator)
  47                               VALUES (%s, %s, %s, %s, %s)
  48                         """, (word_token, word, cls, typ, oper))
  49         self.conn.commit()
  50
  51
  52     def add_country(self, country_code, word_token):
  53         with self.conn.cursor() as cur:
  54             cur.execute("INSERT INTO word (word_token, country_code) VALUES(%s, %s)",
  55                         (word_token, country_code))
  56         self.conn.commit()
  57
  58
  59     def add_postcode(self, word_token, postcode):
  60         with self.conn.cursor() as cur:
  61             cur.execute("""INSERT INTO word (word_token, word, class, type)
  62                               VALUES (%s, %s, 'place', 'postcode')
  63                         """, (word_token, postcode))
  64         self.conn.commit()
  65
  66
  67     def count(self):
  68         with self.conn.cursor() as cur:
  69             return cur.scalar("SELECT count(*) FROM word")
  70
  71
  72     def count_special(self):
  73         with self.conn.cursor() as cur:
  74             return cur.scalar("SELECT count(*) FROM word WHERE class != 'place'")
  75
  76
  77     def get_special(self):
  78         with self.conn.cursor() as cur:
  79             cur.execute("""SELECT word_token, word, class, type, operator
  80                            FROM word WHERE class != 'place'""")
  81             result = set((tuple(row) for row in cur))
  82             assert len(result) == cur.rowcount, "Word table has duplicates."
  83             return result
  84
  85
  86     def get_country(self):
  87         with self.conn.cursor() as cur:
  88             cur.execute("""SELECT country_code, word_token
  89                            FROM word WHERE country_code is not null""")
  90             result = set((tuple(row) for row in cur))
  91             assert len(result) == cur.rowcount, "Word table has duplicates."
  92             return result
  93
  94
  95     def get_postcodes(self):
  96         with self.conn.cursor() as cur:
  97             cur.execute("""SELECT word FROM word
  98                            WHERE class = 'place' and type = 'postcode'""")
  99             return set((row[0] for row in cur))
 100
 101     def get_partial_words(self):
 102         with self.conn.cursor() as cur:
 103             cur.execute("""SELECT word_token, search_name_count FROM word
 104                            WHERE class is null and country_code is null
 105                                  and not word_token like ' %'""")
 106             return set((tuple(row) for row in cur))
 107
 108
 109 class MockPlacexTable:
 110     """ A placex table for testing.
 111     """
 112     def __init__(self, conn):
 113         self.idseq = itertools.count(10000)
 114         self.conn = conn
 115         with conn.cursor() as cur:
 116             cur.execute("""CREATE TABLE placex (
 117                                place_id BIGINT,
 118                                parent_place_id BIGINT,
 119                                linked_place_id BIGINT,
 120                                importance FLOAT,
 121                                indexed_date TIMESTAMP,
 122                                geometry_sector INTEGER,
 123                                rank_address SMALLINT,
 124                                rank_search SMALLINT,
 125                                partition SMALLINT,
 126                                indexed_status SMALLINT,
 127                                osm_id int8,
 128                                osm_type char(1),
 129                                class text,
 130                                type text,
 131                                name hstore,
 132                                admin_level smallint,
 133                                address hstore,
 134                                extratags hstore,
 135                                geometry Geometry(Geometry,4326),
 136                                wikipedia TEXT,
 137                                country_code varchar(2),
 138                                housenumber TEXT,
 139                                postcode TEXT,
 140                                centroid GEOMETRY(Geometry, 4326))""")
 141             cur.execute("CREATE SEQUENCE IF NOT EXISTS seq_place")
 142         conn.commit()
 143
 144     def add(self, osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
 145             admin_level=None, address=None, extratags=None, geom='POINT(10 4)',
 146             country=None):
 147         with self.conn.cursor() as cur:
 148             psycopg2.extras.register_hstore(cur)
 149             cur.execute("""INSERT INTO placex (place_id, osm_type, osm_id, class,
 150                                                type, name, admin_level, address,
 151                                                extratags, geometry, country_code)
 152                             VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
 153                         (osm_type, osm_id or next(self.idseq), cls, typ, names,
 154                          admin_level, address, extratags, 'SRID=4326;' + geom,
 155                          country))
 156         self.conn.commit()
 157
 158
 159 class MockPropertyTable:
 160     """ A property table for testing.
 161     """
 162     def __init__(self, conn):
 163         self.conn = conn
 164
 165
 166     def set(self, name, value):
 167         """ Set a property in the table to the given value.
 168         """
 169         properties.set_property(self.conn, name, value)