test/bdd/steps/table_compare.py

   1 """
   2 Functions to facilitate accessing and comparing the content of DB tables.
   3 """
   4 import re
   5
   6 ID_REGEX = re.compile(r"(?P<typ>[NRW])(?P<oid>\d+)(:(?P<cls>\w+))?")
   7
   8 class NominatimID:
   9     """ Splits a unique identifier for places into its components.
  10         As place_ids cannot be used for testing, we use a unique
  11         identifier instead that is of the form <osmtype><osmid>[:<class>].
  12     """
  13
  14     def __init__(self, oid):
  15         self.typ = self.oid = self.cls = None
  16
  17         if oid is not None:
  18             m = ID_REGEX.fullmatch(oid)
  19             assert m is not None, \
  20                    "ID '{}' not of form <osmtype><osmid>[:<class>]".format(oid)
  21
  22             self.typ = m.group('typ')
  23             self.oid = m.group('oid')
  24             self.cls = m.group('cls')
  25
  26     def __str__(self):
  27         if self.cls is None:
  28             return self.typ + self.oid
  29
  30         return '{self.typ}{self.oid}:{self.cls}'.format(self=self)
  31
  32     def query_osm_id(self, cur, query):
  33         """ Run a query on cursor `cur` using osm ID, type and class. The
  34             `query` string must contain exactly one placeholder '{}' where
  35             the 'where' query should go.
  36         """
  37         where = 'osm_type = %s and osm_id = %s'
  38         params = [self.typ, self. oid]
  39
  40         if self.cls is not None:
  41             where += ' and class = %s'
  42             params.append(self.cls)
  43
  44         return cur.execute(query.format(where), params)
  45
  46     def query_place_id(self, cur, query):
  47         """ Run a query on cursor `cur` using the place ID. The `query` string
  48             must contain exactly one placeholder '%s' where the 'where' query
  49             should go.
  50         """
  51         pid = self.get_place_id(cur)
  52         return cur.execute(query, (pid, ))
  53
  54     def get_place_id(self, cur):
  55         """ Look up the place id for the ID. Throws an assertion if the ID
  56             is not unique.
  57         """
  58         self.query_osm_id(cur, "SELECT place_id FROM placex WHERE {}")
  59         assert cur.rowcount == 1, \
  60                "Place ID {!s} not unique. Found {} entries.".format(self, cur.rowcount)
  61
  62         return cur.fetchone()[0]