return set((tuple(row) for row in self))
+ def table_exists(self, table):
+ """ Check that a table with the given name exists in the database.
+ """
+ num = self.scalar("""SELECT count(*) FROM pg_tables
+ WHERE tablename = %s""", (table, ))
+ return num == 1
+
+
@pytest.fixture
def temp_db(monkeypatch):
""" Create an empty database for the test. The database name is also
def temp_db_conn(temp_db):
""" Connection to the test database.
"""
- conn = connection.connect('dbname=' + temp_db)
- yield conn
- conn.close()
+ with connection.connect('dbname=' + temp_db) as conn:
+ yield conn
@pytest.fixture
def def_config():
return Configuration(None, SRC_DIR.resolve() / 'settings')
+@pytest.fixture
+def src_dir():
+ return SRC_DIR.resolve()
@pytest.fixture
def status_table(temp_db_conn):
temp_db_conn.commit()
-
+@pytest.fixture
+def osm2pgsql_options(temp_db):
+ return dict(osm2pgsql='echo',
+ osm2pgsql_cache=10,
+ osm2pgsql_style='style.file',
+ threads=1,
+ dsn='dbname=' + temp_db,
+ flatnode_file='',
+ tablespaces=dict(slim_data='', slim_index='',
+ main_data='', main_index=''))