test/python/mocks.py

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