""" A collection of integration tests. The tests are organised as follows: The global module setup creates a test database which is used by all tests in this test suite. Each TestCase class represents one import/update run of osm2pgsql. The import is run in the class setup method of the class. The tests themself then check the contents of the resulting database. Each test executes exactly one SQL statement, so that always all checks are run on each import regardless of the result of the other checks. The tests are organised in groups, each of which is implemented in its own class (called *Tests). Each TestCase defines the arguments with which to call osm2pgsql and then inherits from all test groups it wants executed. """ import logging from os import path as op import os import subprocess import sys import psycopg2 import unittest # Base configuration for the tests. May be overwritten by the importer. CONFIG = { 'executable' : './osm2pgsql', 'test_database' : 'osm2pgsql-test', 'test_data_path' : 'tests/data', 'default_data_path' : '.', 'have_lua' : True, 'have_tablespace' : True } ####################################################################### # # Global setup and teardown. def setUpModule(): """ Create a database that is used for all tests. """ loglevel = logging.ERROR for arg in sys.argv: if arg[0] == '-' and arg[1:] == (len(arg) - 1) * 'v': loglevel -= (len(arg) - 1) * 10 logging.basicConfig(level=loglevel if loglevel > 0 else 1) logging.info("Setting up test database") dbname = CONFIG['test_database'] with psycopg2.connect("dbname='template1'") as conn: conn.autocommit = True with conn.cursor() as cur: cur.execute('DROP DATABASE IF EXISTS "{}"'.format(dbname)) cur.execute("CREATE DATABASE \"{}\" WITH ENCODING 'UTF8'" .format(dbname)) with psycopg2.connect("dbname='{}'".format(dbname)) as conn: conn.autocommit = True with conn.cursor() as cur: # Check if there is a dataspace, we will skip tests otherwise. if CONFIG['have_tablespace']: cur.execute("""SELECT spcname FROM pg_tablespace WHERE spcname = 'tablespacetest'""") CONFIG['have_tablespace'] = cur.fetchone() cur.execute('CREATE EXTENSION postgis') cur.execute('CREATE EXTENSION hstore') def tearDownModule(): """ Destroy the global database. """ logging.info("Cleaning up test database") with psycopg2.connect("dbname='template1'") as conn: with conn.cursor() as cur: conn.autocommit = True cur.execute('DROP DATABASE IF EXISTS "{}"' .format(CONFIG['test_database'])) ######################################################################## # # Runners # # These provide the boiler plate for importing and updating the database. class BaseRunner(object): """ The base class for all runners. Provides the setup function that imports and optionally updates the database. It also implements convenience functions for the database checks. """ conn = None extra_params = None use_gazetteer = False use_lua_tagtransform = False import_file = None update_file = None schema = None @classmethod def setUpClass(cls): try: # In case one is left over from a previous test os.remove('flat.nodes') except (FileNotFoundError): pass if cls.use_lua_tagtransform and not CONFIG['have_lua']: cls.skipTest(None, "No Lua configured.") if 'tablespacetest' in cls.extra_params and not CONFIG['have_lua']: cls.skipTest(None, "Test tablespace 'tablespacetest' not configured.") with psycopg2.connect("dbname='{}'".format(CONFIG['test_database'])) as conn: with conn.cursor() as cur: for t in ('nodes', 'ways', 'rels', 'point', 'line', 'roads', 'polygon'): cur.execute("DROP TABLE IF EXISTS planet_osm_" + t) cur.execute("DROP SCHEMA IF EXISTS osm CASCADE") if cls.schema: cur.execute("CREATE SCHEMA " + cls.schema) if cls.import_file: cls.run_import(cls.get_def_params() + cls.extra_params, op.join(CONFIG['test_data_path'], cls.import_file)) if cls.update_file: cls.run_import(['-a'] + cls.get_def_params() + cls.extra_params, op.join(CONFIG['test_data_path'], cls.update_file)) BaseRunner.conn = psycopg2.connect("dbname='{}'".format(CONFIG['test_database'])) @classmethod def tearDownClass(cls): if BaseRunner.conn: BaseRunner.conn.close() BaseRunner.conn = None if cls.schema: with psycopg2.connect("dbname='{}'".format(CONFIG['test_database'])) as conn: conn.cursor().execute("DROP SCHEMA IF EXISTS {} CASCADE".format(cls.schema)) try: os.remove('flat.nodes') except (FileNotFoundError): pass @classmethod def get_def_params(cls): if cls.use_gazetteer: params = ['-S', op.join(CONFIG['test_data_path'], 'gazetteer-test.style'), '-O', 'gazetteer'] else: params = ['-S', op.join(CONFIG['default_data_path'], 'default.style')] if cls.use_lua_tagtransform: params += ['--tag-transform-script', op.join(CONFIG['default_data_path'], 'style.lua')] return params @classmethod def run_import(cls, params, filename): cmdline = [CONFIG['executable']] if not '-d' in params and not '--database' in params: cmdline.extend(('-d', CONFIG['test_database'])) cmdline.extend(params) cmdline.append(op.join(CONFIG['test_data_path'], filename)) logging.info("Executing command: {}".format(' '.join(cmdline))) proc = subprocess.Popen(cmdline, stdout=subprocess.PIPE, stderr=subprocess.PIPE) (outp, err) = proc.communicate() if proc.returncode == 0: logging.debug(err.decode('utf-8')) else: logging.warning(err.decode('utf-8')) raise RuntimeError("Import failed.") def assert_count(self, count, table, where=None): if self.schema: table = self.schema + '.' + table sql = 'SELECT count(*) FROM ' + table if where: sql += ' WHERE ' + where self.assert_sql(count, sql) def assert_sql(self, expect, sql): with BaseRunner.conn.cursor() as cur: cur.execute(sql) self.assertEqual(cur.rowcount, 1) self.assertEqual(expect, cur.fetchone()[0]) class BaseImportRunner(BaseRunner): import_file = 'liechtenstein-2013-08-03.osm.pbf' update = False class BaseUpdateRunner(BaseRunner): import_file = 'liechtenstein-2013-08-03.osm.pbf' update_file = '000466354.osc.gz' update = True class MultipolygonImportRunner(BaseRunner): import_file = 'test_multipolygon.osm' update = False class MultipolygonUpdateRunner(BaseRunner): import_file = 'test_multipolygon.osm' update_file = 'test_multipolygon_diff.osc' update = True class BaseUpdateRunnerWithOutputSchema(BaseUpdateRunner): schema = 'osm' class DependencyUpdateRunner(BaseRunner): import_file = 'test_forward_dependencies.opl' update_file = 'test_forward_dependencies_diff.opl' ######################################################################## # # Test groups # # Classes with test that check the content of the database. class PgsqlBaseTests(object): """ Tests the basic counts of a Liechtenstein import with the pgsql output. """ def is_full_hstore(self): return (('--hstore' in self.extra_params or '-k' in self.extra_params or '-j' in self.extra_params) and '--hstore-match-only'not in self.extra_params) def test_point_count(self): if self.is_full_hstore(): count = 1475 if self.update else 1360 else: count = 1457 if self.update else 1342 self.assert_count(count, 'planet_osm_point') def test_line_count(self): if self.is_full_hstore(): count = 3297 if self.update else 3254 elif '-l' in self.extra_params: count = 3274 if self.update else 3229 else: count = 3274 if self.update else 3231 self.assert_count(count, 'planet_osm_line') def test_road_count(self): if '-l' in self.extra_params: count = 380 if self.update else 374 else: count = 380 if self.update else 375 self.assert_count(count, 'planet_osm_roads') def test_polygon_count(self): if self.is_full_hstore(): count = 4278 if self.update else 4131 elif self.use_lua_tagtransform: # There is a difference in handling mixture of tags on ways and # relations with Lua tagtransform, where the correct behaviour is # non-obvious. count = 4283 if self.update else 4136 else: count = 4277 if self.update else 4130 self.assert_count(count, 'planet_osm_polygon') class PgsqlMercGeomTests(object): """ Tests that mercator-transformed geometries are plausible. """ def test_basic_polygon_area(self): area = 1272140688 if self.use_lua_tagtransform else 1247245186 self.assert_sql(area, """SELECT round(sum(cast(ST_Area(way) as numeric)),0) FROM planet_osm_polygon""") def test_basic_line_length(self): self.assert_sql(4211350, 'SELECT round(sum(ST_Length(way))) FROM planet_osm_line') def test_basic_road_length(self): self.assert_sql(2032023, 'SELECT round(sum(ST_Length(way))) FROM planet_osm_roads') class DroppedTableTests(object): """ Tests for the absence of middle tables.""" def test_absence_nodes_table(self): self.assert_count(0, 'pg_tables', where="tablename = 'planet_osm_nodes'") def test_absence_way_table(self): self.assert_count(0, 'pg_tables', where="tablename = 'planet_osm_ways'") def test_absence_rel_table(self): self.assert_count(0, 'pg_tables', where="tablename = 'planet_osm_rels'") class ExtraTagTests(object): """ Tests for presence of extra meta tags. """ def test_metatags_point(self): self.assert_count(1360, 'planet_osm_point', """tags ? 'osm_user' AND tags ? 'osm_version' AND tags ? 'osm_uid' AND tags ? 'osm_changeset'""") def test_metatags_line(self): self.assert_count(3254, 'planet_osm_line', """tags ? 'osm_user' AND tags ? 'osm_version' AND tags ? 'osm_uid' AND tags ? 'osm_changeset'""") def test_metatags_polygon(self): self.assert_count(4131, 'planet_osm_polygon', """tags ? 'osm_user' AND tags ? 'osm_version' AND tags ? 'osm_uid' AND tags ? 'osm_changeset'""") class HstoreTagTests(object): """ Tests for entries in hstore columns in --hstore-all mode. """ def test_number_hstore_tags_points(self): self.assert_sql(4352 if self.update else 4228, """SELECT sum(array_length(akeys(tags),1)) FROM planet_osm_point""") def test_number_hstore_tags_roads(self): self.assert_sql(2336 if self.update else 2317, """SELECT sum(array_length(akeys(tags),1)) FROM planet_osm_roads""") def test_number_hstore_tags_lines(self): self.assert_sql(10505 if self.update else 10387, """SELECT sum(array_length(akeys(tags),1)) FROM planet_osm_line""") def test_number_hstore_tags_polygons(self): self.assert_sql(9832 if self.update else 9538, """SELECT sum(array_length(akeys(tags),1)) FROM planet_osm_polygon""") class GazetteerTests(object): """ Tests for gazetteer output using Liechtenstein file. """ def test_place_count(self): self.assert_count(2877 if self.update else 2836, 'place') def test_place_node_count(self): self.assert_count(764 if self.update else 759, 'place', where="osm_type = 'N'") def test_place_way_count(self): self.assert_count(2095 if self.update else 2059, 'place', where="osm_type = 'W'") def test_place_rel_count(self): self.assert_count(18 if self.update else 18, 'place', where="osm_type = 'R'") def test_place_housenumber_count(self): self.assert_count(199, 'place', where="address ? 'housenumber'") def test_place_address_count(self): self.assert_count(319, 'place', where='address is not null') class MultipolygonTests(object): """ Tests for the special multipolygon input. """ def assert_area(self, area, osm_id, k, v, name = None): sql = """SELECT round(ST_Area(way)) FROM planet_osm_polygon WHERE osm_id = {} AND "{}" = '{}'""".format(osm_id, k, v) if name is not None: sql += " AND name = '{}'".format(name) self.assert_sql(area, sql) def assert_length(self, length, osm_id, k, v, name): sql = """SELECT round(ST_Length(way)) FROM planet_osm_line WHERE osm_id = {} AND "{}" = '{}' AND name = '{}'""" self.assert_sql(length, sql.format(osm_id, k, v, name)) def assert_inner_rings(self, num, osm_id, k, v, name): sql = """SELECT round(ST_NumInteriorRing(way)) FROM planet_osm_polygon WHERE osm_id = {} AND "{}" = '{}' AND name = '{}'""" self.assert_sql(num, sql.format(osm_id, k, v, name)) def assert_sum_area(self, area, osm_id, k, v, name=None): sql = """SELECT round(sum(ST_Area(way))) FROM planet_osm_polygon WHERE osm_id = {} AND "{}" = '{}'""" if name is not None: sql += " AND name = '{}'".format(name) self.assert_sql(area, sql.format(osm_id, k, v, name)) def assert_num_rows(self, area, osm_id, k, v, name = None): sql = """SELECT count(*) FROM planet_osm_polygon WHERE osm_id = {} AND "{}" = '{}'""" if name is not None: sql += " AND name = '{}'".format(name) self.assert_sql(area, sql.format(osm_id, k, v, name)) def assert_num_geoms(self, area, osm_id, k, v, name): sql = """SELECT ST_NumGeometries(way) FROM planet_osm_polygon WHERE osm_id = {} AND "{}" = '{}' AND name = '{}'""" self.assert_sql(area, sql.format(osm_id, k, v, name)) def test_tags_from_relation(self): self.assert_area(13949 if self.update else 12895, -1, 'landuse', 'residential', 'Name_rel') def test_named_inner_inner_way(self): self.assert_area(3144, 4, 'landuse', 'farmland', 'Name_way3') def test_named_inner_outer_way(self): self.assert_area(12894, -8, 'landuse', 'residential', 'Name_rel2') def test_named_inner_outer_way_2(self): self.assert_area(3144, 5, 'landuse', 'farmland', 'Name_way4') def test_nonarea_inner_outer(self): self.assert_area(12894, -14, 'landuse', 'residential', 'Name_way5') def test_nonarea_inner_inner(self): self.assert_length(228, 6, 'highway', 'residential', 'Name_way6') def test_multipolygon_no_holes(self): self.assert_area(11529, -11, 'landuse', 'residential', 'Name_rel6') def test_multipolygon_two_holes_area(self): self.assert_area(9286, -3, 'landuse', 'residential', 'Name_rel11') def test_multipolygon_two_holes_rings(self): self.assert_inner_rings(2, -3, 'landuse', 'residential', 'Name_rel11') def test_multipolygon_two_outer(self): self.assert_sum_area(17581, -13, 'landuse', 'farmland', 'Name_rel9') def test_multipolygon_two_outer_num(self): if '-G' in self.extra_params: self.assert_num_geoms(2, -13, 'landuse', 'farmland', 'Name_rel9') else: self.assert_num_rows(2, -13, 'landuse', 'farmland', 'Name_rel9') def test_nested_outer_ways_area(self): self.assert_sum_area(16169, -7, 'landuse', 'farmland', 'Name_rel15') def test_nested_outer_ways_num(self): if '-G' in self.extra_params: self.assert_num_geoms(2, -7, 'landuse', 'farmland', 'Name_rel15') else: self.assert_num_rows(2, -7, 'landuse', 'farmland', 'Name_rel15') def test_no_copy_tags_from_outer(self): self.assert_area(18501, -24, 'natural', 'water') def test_outer_way_is_present(self): self.assert_area(24859, 83, 'landuse', 'farmland') def test_diff_move_point_inner_way(self): self.assert_area(13949 if self.update else 12895, -1, 'landuse', 'residential', 'Name_rel') def test_diff_remove_relation(self): self.assert_count(0 if self.update else 1, 'planet_osm_polygon', where='osm_id = -25') def test_tags_on_inner_and_outer(self): self.assert_count(1, 'planet_osm_polygon', where='osm_id = 113') def test_tags_on_inner_and_outer(self): self.assert_count(1, 'planet_osm_polygon', where='osm_id = 118') def test_tags_on_outer_present(self): self.assert_count(1, 'planet_osm_polygon', where='osm_id = 114') def test_tags_on_outer_absence_relation(self): self.assert_num_rows(0, -33, 'natural', 'water') def test_tags_on_relation_includes_relatoin(self): self.assert_sum_area(29155 if self.update else 68494, -29, 'natural', 'water') def test_tags_relation_outer_1_absence(self): self.assert_count(0, 'planet_osm_polygon', where='osm_id = 109') def test_tags_relation_outer_2_absence(self): self.assert_count(0, 'planet_osm_polygon', where='osm_id = 104') def test_tags_relation_outer_1_presence(self): self.assert_count(1, 'planet_osm_polygon', where='osm_id = 107') def test_tags_relation_outer_2_presence(self): self.assert_count(1, 'planet_osm_polygon', where='osm_id = 102') def test_tags_relation_outer_2_presence_area(self): self.assert_area(12994, 102, 'natural', 'water') def test_tags_on_relation_and_way_relation_presence(self): self.assert_sum_area(10377, -39, 'landuse', 'forest') def test_tags_on_relation_and_way_way_presence(self): self.assert_count(1, 'planet_osm_polygon', where='osm_id = 138') def test_different_tags_on_relation_and_way_relation_presence(self): self.assert_sum_area(12397, -40, 'landuse', 'forest') def test_different_tags_on_relation_and_way_way_presence(self): self.assert_count(1, 'planet_osm_polygon', where='osm_id = 140') def test_planet_osm_nodes(self): self.assert_count(1, 'pg_tables', where="tablename = 'planet_osm_nodes'") ######################################################################## # # Test runs # # Each class represents one import/update run. The class setup function # in the runner trigger a single import for each class. Then the tests # themselves check the state of the database after the import. class TestPgsqlImportNonSlim(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests, DroppedTableTests): extra_params = [] class TestPgsqlImportNonSlimLatLon(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, DroppedTableTests): extra_params = ['-l'] class TestPgsqlImportSlimDrop(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests, DroppedTableTests): extra_params = ['--slim', '--drop'] class TestPgsqlImportNonSlimLua(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests, DroppedTableTests): extra_params = [] use_lua_tagtransform = True class TextPgsqlImportSlimHstoreDrop(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, DroppedTableTests): extra_params = ['--slim', '--hstore', '--hstore-add-index', '--drop'] class TestPgsqlImportSlim(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim'] class TestPgsqlImportSlimParallel(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '--number-processes', '16'] class TestPgsqlImportSlimSmallCache(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '--number-processes', '8', '-C1', '--cache-strategy=dense'] class TestPgsqlImportSlimNoCache(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '-C0'] class TestPgsqlImportSlimHstoreMatchOnly(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '-k', '--hstore-match-only'] class TestPgsqlImportSlimHstoreNameColumn(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '-z', 'name:'] class TestPgsqlImportSlimHstore(BaseImportRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-k'] class TestPgsqlImportSlimHstoreAll(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, HstoreTagTests): extra_params = ['--slim', '-j'] class TestPgsqlImportSlimHstoreIndex(BaseImportRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--hstore', '--hstore-add-index'] class TestPgsqlImportSlimHstoreExtra(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '-k', '-x', '--hstore-match-only'] class TestPgsqlImportSlimHstoreAllExtra(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, ExtraTagTests): extra_params = ['--slim', '-j', '-x'] class TestPgsqlImportTablespaceMainData(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '--tablespace-main-data', 'tablespacetest'] class TestPgsqlImportTablespaceMainIndex(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '--tablespace-main-index', 'tablespacetest'] class TestPgsqlImportTablespaceSlimData(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '--tablespace-slim-data', 'tablespacetest'] class TestPgsqlImportTablespaceSlimIndex(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim', '--tablespace-slim-index', 'tablespacetest'] class TestPgsqlImportSlimLua(BaseImportRunner, unittest.TestCase, PgsqlBaseTests, PgsqlMercGeomTests): extra_params = ['--slim'] use_lua_tagtransform = True # Standard output updates class TestPgsqlUpdate(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim'] class TestPgsqlUpdateParallel(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--number-processes', '15'] class TestPgsqlUpdateSmallCache(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--number-processes', '8', '-C1', '--cache-strategy=dense'] class TestPgsqlUpdateNoCache(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-C0'] class TestPgsqlUpdateHstoreMatchOnly(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-k', '--hstore-match-only'] class TestPgsqlUpdateHstoreNameColumn(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-z', 'name:'] class TestPgsqlUpdateHstore(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-k'] class TestPgsqlUpdateHstoreAll(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests, HstoreTagTests): extra_params = ['--slim', '-j'] class TestPgsqlUpdateHstoreIndex(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--hstore', '--hstore-add-index'] class TestPgsqlUpdateHstoreExtra(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-k', '-x', '--hstore-match-only'] class TestPgsqlUpdateHstoreAllExtra(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-j', '-x'] class TestPgsqlUpdateTablespaceMainData(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--tablespace-main-data', 'tablespacetest'] class TestPgsqlUpdateTablespaceMainIndex(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--tablespace-main-index', 'tablespacetest'] class TestPgsqlUpdateTablespaceSlimData(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--tablespace-slim-data', 'tablespacetest'] class TestPgsqlUpdateTablespaceSlimIndex(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--tablespace-slim-index', 'tablespacetest'] class TestPgsqlUpdateLua(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim'] use_lua_tagtransform = True # Gazetteer output class TestGazetteerImport(BaseImportRunner, unittest.TestCase, GazetteerTests): extra_params = ['--slim'] use_gazetteer = True class TestGazetteerUpdate(BaseUpdateRunner, unittest.TestCase, GazetteerTests): extra_params = ['--slim'] use_gazetteer = True # Multipolygon import tests class TestMPImportSlim(MultipolygonImportRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim'] class TestMPImportSlimMultiGeom(MultipolygonImportRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-G'] class TestMPImportSlimHstore(MultipolygonImportRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-k'] class TestMPImportSlimHstoreMatch(MultipolygonImportRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-k', '--hstore-match-only'] class TestMPImportSlimHstoreExtra(MultipolygonImportRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-x', '-k', '--hstore-match-only'] class TestMPImportSlimLua(MultipolygonImportRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim'] use_lua_tagtransform = True class TestMPImportSlimLuaHstore(MultipolygonImportRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-k'] use_lua_tagtransform = True # Multipolygon update tests class TestMPUpdateSlim(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim'] class TestMPUpdateSlimMultiGeom(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-G'] class TestMPUpdateSlimHstore(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-k'] class TestMPUpdateSlimHstoreMatch(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-k', '--hstore-match-only'] class TestMPUpdateSlimHstoreExtra(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-x', '-k', '--hstore-match-only'] class TestMPUpdateSlimLua(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim'] use_lua_tagtransform = True class TestMPUpdateSlimLuaHstore(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-k'] use_lua_tagtransform = True class TestMPUpdateSlimWithFlatNodes(MultipolygonUpdateRunner, unittest.TestCase, MultipolygonTests): extra_params = ['--slim', '-F', 'flat.nodes'] def test_planet_osm_nodes(self): self.assert_count(0, 'pg_tables', where="tablename = 'planet_osm_nodes'") class TestMPUpdateWithForwardDependencies(DependencyUpdateRunner, unittest.TestCase): extra_params = ['--latlong', '--slim', '--with-forward-dependencies=true'] def test_count(self): self.assert_count(1, 'planet_osm_point') self.assert_count(1, 'planet_osm_line') self.assert_count(0, 'planet_osm_line', 'abs(ST_X(ST_StartPoint(way)) - 3.0) < 0.0001') self.assert_count(1, 'planet_osm_line', 'abs(ST_X(ST_StartPoint(way)) - 3.1) < 0.0001') self.assert_count(1, 'planet_osm_roads') self.assert_count(1, 'planet_osm_polygon') class TestMPUpdateWithoutForwardDependencies(DependencyUpdateRunner, unittest.TestCase): extra_params = ['--latlong', '--slim', '--with-forward-dependencies=false'] def test_count(self): self.assert_count(1, 'planet_osm_point') self.assert_count(1, 'planet_osm_line') self.assert_count(1, 'planet_osm_line', 'abs(ST_X(ST_StartPoint(way)) - 3.0) < 0.0001') self.assert_count(0, 'planet_osm_line', 'abs(ST_X(ST_StartPoint(way)) - 3.1) < 0.0001') self.assert_count(1, 'planet_osm_roads') self.assert_count(2, 'planet_osm_polygon') # Database access tests class TestDBAccessNorm(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-d', CONFIG['test_database']] class TestDBAccessNormLong(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--database', CONFIG['test_database']] class TestDBAccessConninfo(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-d', 'dbname=' + CONFIG['test_database']] class TestDBAccessConninfoLong(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--database', 'dbname=' + CONFIG['test_database']] class TestDBAccessURIPostgresql(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-d', 'postgresql:///' + CONFIG['test_database']] class TestDBAccessURIPostgres(BaseUpdateRunner, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '-d', 'postgres:///' + CONFIG['test_database']] # Schema tests class TestDBOutputSchema(BaseUpdateRunnerWithOutputSchema, unittest.TestCase, PgsqlBaseTests): extra_params = ['--slim', '--output-pgsql-schema=osm']