3 from pathlib import Path
9 SRC_DIR = Path(__file__) / '..' / '..' / '..'
11 # always test against the source
12 sys.path.insert(0, str(SRC_DIR.resolve()))
14 from nominatim.config import Configuration
15 from nominatim.db import connection
17 class _TestingCursor(psycopg2.extras.DictCursor):
18 """ Extension to the DictCursor class that provides execution
19 short-cuts that simplify writing assertions.
22 def scalar(self, sql, params=None):
23 """ Execute a query with a single return value and return this value.
24 Raises an assertion when not exactly one row is returned.
26 self.execute(sql, params)
27 assert self.rowcount == 1
28 return self.fetchone()[0]
30 def row_set(self, sql, params=None):
31 """ Execute a query and return the result as a set of tuples.
33 self.execute(sql, params)
34 if self.rowcount == 1:
35 return set(tuple(self.fetchone()))
37 return set((tuple(row) for row in self))
40 def temp_db(monkeypatch):
41 """ Create an empty database for the test. The database name is also
42 exported into NOMINATIM_DATABASE_DSN.
44 name = 'test_nominatim_python_unittest'
45 conn = psycopg2.connect(database='postgres')
47 conn.set_isolation_level(0)
48 with conn.cursor() as cur:
49 cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
50 cur.execute('CREATE DATABASE {}'.format(name))
54 monkeypatch.setenv('NOMINATIM_DATABASE_DSN' , 'dbname=' + name)
58 conn = psycopg2.connect(database='postgres')
60 conn.set_isolation_level(0)
61 with conn.cursor() as cur:
62 cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
67 def temp_db_with_extensions(temp_db):
68 conn = psycopg2.connect(database=temp_db)
69 with conn.cursor() as cur:
70 cur.execute('CREATE EXTENSION hstore; CREATE EXTENSION postgis;')
77 def temp_db_conn(temp_db):
78 """ Connection to the test database.
80 conn = connection.connect('dbname=' + temp_db)
86 def temp_db_cursor(temp_db):
87 """ Connection and cursor towards the test database. The connection will
88 be in auto-commit mode.
90 conn = psycopg2.connect('dbname=' + temp_db)
91 conn.set_isolation_level(0)
92 with conn.cursor(cursor_factory=_TestingCursor) as cur:
99 return Configuration(None, SRC_DIR.resolve() / 'settings')
103 def status_table(temp_db_conn):
104 """ Create an empty version of the status table.
106 with temp_db_conn.cursor() as cur:
107 cur.execute("""CREATE TABLE import_status (
108 lastimportdate timestamp with time zone NOT NULL,
112 temp_db_conn.commit()
116 def place_table(temp_db_with_extensions, temp_db_conn):
117 """ Create an empty version of the place table.
119 with temp_db_conn.cursor() as cur:
120 cur.execute("""CREATE TABLE place (
121 osm_id int8 NOT NULL,
122 osm_type char(1) NOT NULL,
126 admin_level smallint,
129 geometry Geometry(Geometry,4326) NOT NULL)""")
130 temp_db_conn.commit()
134 def place_row(place_table, temp_db_cursor):
135 """ A factory for rows in the place table. The table is created as a
136 prerequisite to the fixture.
138 idseq = itertools.count(1001)
139 def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
140 admin_level=None, address=None, extratags=None, geom=None):
141 temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
142 (osm_id or next(idseq), osm_type, cls, typ, names,
143 admin_level, address, extratags,
144 geom or 'SRID=4326;POINT(0 0 )'))