test: more use of table_factory

This commit is contained in:
Sarah Hoffmann
2021-05-19 17:37:03 +02:00
parent af52eed0dd
commit 01f5a9ff84

View File

@@ -135,43 +135,37 @@ def property_table(table_factory):
@pytest.fixture @pytest.fixture
def status_table(temp_db_conn): def status_table(table_factory):
""" Create an empty version of the status table and """ Create an empty version of the status table and
the status logging table. the status logging table.
""" """
with temp_db_conn.cursor() as cur: table_factory('import_status',
cur.execute("""CREATE TABLE import_status ( """lastimportdate timestamp with time zone NOT NULL,
lastimportdate timestamp with time zone NOT NULL, sequence_id integer,
sequence_id integer, indexed boolean""")
indexed boolean table_factory('import_osmosis_log',
)""") """batchend timestamp,
cur.execute("""CREATE TABLE import_osmosis_log ( batchseq integer,
batchend timestamp, batchsize bigint,
batchseq integer, starttime timestamp,
batchsize bigint, endtime timestamp,
starttime timestamp, event text""")
endtime timestamp,
event text
)""")
temp_db_conn.commit()
@pytest.fixture @pytest.fixture
def place_table(temp_db_with_extensions, temp_db_conn): def place_table(temp_db_with_extensions, table_factory):
""" Create an empty version of the place table. """ Create an empty version of the place table.
""" """
with temp_db_conn.cursor() as cur: table_factory('place',
cur.execute("""CREATE TABLE place ( """osm_id int8 NOT NULL,
osm_id int8 NOT NULL, osm_type char(1) NOT NULL,
osm_type char(1) NOT NULL, class text NOT NULL,
class text NOT NULL, type text NOT NULL,
type text NOT NULL, name hstore,
name hstore, admin_level smallint,
admin_level smallint, address hstore,
address hstore, extratags hstore,
extratags hstore, geometry Geometry(Geometry,4326) NOT NULL""")
geometry Geometry(Geometry,4326) NOT NULL)""")
temp_db_conn.commit()
@pytest.fixture @pytest.fixture
@@ -197,24 +191,22 @@ def placex_table(temp_db_with_extensions, temp_db_conn):
@pytest.fixture @pytest.fixture
def osmline_table(temp_db_with_extensions, temp_db_conn): def osmline_table(temp_db_with_extensions, table_factory):
with temp_db_conn.cursor() as cur: table_factory('location_property_osmline',
cur.execute("""CREATE TABLE location_property_osmline ( """place_id BIGINT,
place_id BIGINT, osm_id BIGINT,
osm_id BIGINT, parent_place_id BIGINT,
parent_place_id BIGINT, geometry_sector INTEGER,
geometry_sector INTEGER, indexed_date TIMESTAMP,
indexed_date TIMESTAMP, startnumber INTEGER,
startnumber INTEGER, endnumber INTEGER,
endnumber INTEGER, partition SMALLINT,
partition SMALLINT, indexed_status SMALLINT,
indexed_status SMALLINT, linegeo GEOMETRY,
linegeo GEOMETRY, interpolationtype TEXT,
interpolationtype TEXT, address HSTORE,
address HSTORE, postcode TEXT,
postcode TEXT, country_code VARCHAR(2)""")
country_code VARCHAR(2))""")
temp_db_conn.commit()
@pytest.fixture @pytest.fixture