X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/ab71f17c47ca056c88412b896d78cf013b30948e..7205491b8495e48c62b28373d1746e77d475582b:/test/bdd/steps/nominatim_environment.py?ds=sidebyside diff --git a/test/bdd/steps/nominatim_environment.py b/test/bdd/steps/nominatim_environment.py index 6b83c2e4..1fc6f887 100644 --- a/test/bdd/steps/nominatim_environment.py +++ b/test/bdd/steps/nominatim_environment.py @@ -5,6 +5,7 @@ # Copyright (C) 2022 by the Nominatim developer community. # For a full list of authors see the git log. from pathlib import Path +import importlib import sys import tempfile @@ -15,7 +16,7 @@ sys.path.insert(1, str((Path(__file__) / '..' / '..' / '..' / '..').resolve())) from nominatim import cli from nominatim.config import Configuration -from nominatim.db.connection import _Connection +from nominatim.db.connection import Connection from nominatim.tools import refresh from nominatim.tokenizer import factory as tokenizer_factory from steps.utils import run_script @@ -36,18 +37,25 @@ class NominatimEnvironment: self.api_test_db = config['API_TEST_DB'] self.api_test_file = config['API_TEST_FILE'] self.tokenizer = config['TOKENIZER'] + self.import_style = config['STYLE'] self.server_module_path = config['SERVER_MODULE_PATH'] self.reuse_template = not config['REMOVE_TEMPLATE'] self.keep_scenario_db = config['KEEP_TEST_DB'] self.code_coverage_path = config['PHPCOV'] self.code_coverage_id = 1 - self.default_config = Configuration(None, self.src_dir / 'settings').get_os_env() + self.default_config = Configuration(None).get_os_env() self.test_env = None self.template_db_done = False self.api_db_done = False self.website_dir = None + self.api_engine = None + if config['API_ENGINE'] != 'php': + if not hasattr(self, f"create_api_request_func_{config['API_ENGINE']}"): + raise RuntimeError(f"Unknown API engine '{config['API_ENGINE']}'") + self.api_engine = getattr(self, f"create_api_request_func_{config['API_ENGINE']}")() + def connect_database(self, dbname): """ Return a connection to the database with the given name. Uses configured host, user and port. @@ -61,7 +69,7 @@ class NominatimEnvironment: dbargs['user'] = self.db_user if self.db_pass: dbargs['password'] = self.db_pass - conn = psycopg2.connect(connection_factory=_Connection, **dbargs) + conn = psycopg2.connect(connection_factory=Connection, **dbargs) return conn def next_code_coverage_file(self): @@ -78,7 +86,10 @@ class NominatimEnvironment: be picked up by dotenv and creates a project directory with the appropriate website scripts. """ - dsn = 'pgsql:dbname={}'.format(dbname) + if dbname.startswith('sqlite:'): + dsn = 'sqlite:dbname={}'.format(dbname[7:]) + else: + dsn = 'pgsql:dbname={}'.format(dbname) if self.db_host: dsn += ';host=' + self.db_host if self.db_port: @@ -88,11 +99,6 @@ class NominatimEnvironment: if self.db_pass: dsn += ';password=' + self.db_pass - if self.website_dir is not None \ - and self.test_env is not None \ - and dsn == self.test_env['NOMINATIM_DATABASE_DSN']: - return # environment already set uo - self.test_env = dict(self.default_config) self.test_env['NOMINATIM_DATABASE_DSN'] = dsn self.test_env['NOMINATIM_LANGUAGES'] = 'en,de,fr,ja' @@ -104,9 +110,10 @@ class NominatimEnvironment: self.test_env['NOMINATIM_CONFIGDIR'] = str((self.src_dir / 'settings').resolve()) self.test_env['NOMINATIM_DATABASE_MODULE_SRC_PATH'] = str((self.build_dir / 'module').resolve()) self.test_env['NOMINATIM_OSM2PGSQL_BINARY'] = str((self.build_dir / 'osm2pgsql' / 'osm2pgsql').resolve()) - self.test_env['NOMINATIM_NOMINATIM_TOOL'] = str((self.build_dir / 'nominatim').resolve()) if self.tokenizer is not None: self.test_env['NOMINATIM_TOKENIZER'] = self.tokenizer + if self.import_style is not None: + self.test_env['NOMINATIM_IMPORT_STYLE'] = self.import_style if self.server_module_path: self.test_env['NOMINATIM_DATABASE_MODULE_PATH'] = self.server_module_path @@ -128,13 +135,9 @@ class NominatimEnvironment: def get_test_config(self): - cfg = Configuration(Path(self.website_dir.name), self.src_dir / 'settings', - environ=self.test_env) + cfg = Configuration(Path(self.website_dir.name), environ=self.test_env) cfg.set_libdirs(module=self.build_dir / 'module', - osm2pgsql=self.build_dir / 'osm2pgsql' / 'osm2pgsql', - php=self.src_dir / 'lib-php', - sql=self.src_dir / 'lib-sql', - data=self.src_dir / 'data') + osm2pgsql=self.build_dir / 'osm2pgsql' / 'osm2pgsql') return cfg def get_libpq_dsn(self): @@ -197,23 +200,28 @@ class NominatimEnvironment: """ self.write_nominatim_config(self.api_test_db) + if self.api_test_db.startswith('sqlite:'): + return + if not self.api_db_done: self.api_db_done = True if not self._reuse_or_drop_db(self.api_test_db): - testdata = Path('__file__') / '..' / '..' / 'testdb' - self.test_env['NOMINATIM_WIKIPEDIA_DATA_PATH'] = str(testdata.resolve()) + testdata = (Path(__file__) / '..' / '..' / '..' / 'testdb').resolve() + self.test_env['NOMINATIM_WIKIPEDIA_DATA_PATH'] = str(testdata) + simp_file = Path(self.website_dir.name) / 'secondary_importance.sql.gz' + simp_file.symlink_to(testdata / 'secondary_importance.sql.gz') try: self.run_nominatim('import', '--osm-file', str(self.api_test_file)) - self.run_nominatim('add-data', '--tiger-data', str((testdata / 'tiger').resolve())) + self.run_nominatim('add-data', '--tiger-data', str(testdata / 'tiger')) self.run_nominatim('freeze') if self.tokenizer == 'legacy': - phrase_file = str((testdata / 'specialphrases_testdb.sql').resolve()) + phrase_file = str(testdata / 'specialphrases_testdb.sql') run_script(['psql', '-d', self.api_test_db, '-f', phrase_file]) else: - csv_path = str((testdata / 'full_en_phrases_test.csv').resolve()) + csv_path = str(testdata / 'full_en_phrases_test.csv') self.run_nominatim('special-phrases', '--import-from-csv', csv_path) except: self.db_drop_database(self.api_test_db) @@ -262,8 +270,8 @@ class NominatimEnvironment: self.db_drop_database(self.test_db) def _reuse_or_drop_db(self, name): - """ Check for the existance of the given DB. If reuse is enabled, - then the function checks for existance and returns True if the + """ Check for the existence of the given DB. If reuse is enabled, + then the function checks for existnce and returns True if the database is already there. Otherwise an existing database is dropped and always false returned. """ @@ -302,12 +310,7 @@ class NominatimEnvironment: cli.nominatim(module_dir='', osm2pgsql_path=str(self.build_dir / 'osm2pgsql' / 'osm2pgsql'), - phplib_dir=str(self.src_dir / 'lib-php'), - sqllib_dir=str(self.src_dir / 'lib-sql'), - data_dir=str(self.src_dir / 'data'), - config_dir=str(self.src_dir / 'settings'), cli_args=cmdline, - phpcgi_path='', environ=self.test_env) @@ -332,3 +335,40 @@ class NominatimEnvironment: WHERE class='place' and type='houses' and osm_type='W' and ST_GeometryType(geometry) = 'ST_LineString'""") + + + def create_api_request_func_starlette(self): + import nominatim.server.starlette.server + from asgi_lifespan import LifespanManager + import httpx + + async def _request(endpoint, params, project_dir, environ, http_headers): + app = nominatim.server.starlette.server.get_application(project_dir, environ) + + async with LifespanManager(app): + async with httpx.AsyncClient(app=app, base_url="http://nominatim.test") as client: + response = await client.get(f"/{endpoint}", params=params, + headers=http_headers) + + return response.text, response.status_code + + return _request + + + def create_api_request_func_falcon(self): + import nominatim.server.falcon.server + import falcon.testing + + async def _request(endpoint, params, project_dir, environ, http_headers): + app = nominatim.server.falcon.server.get_application(project_dir, environ) + + async with falcon.testing.ASGIConductor(app) as conductor: + response = await conductor.get(f"/{endpoint}", params=params, + headers=http_headers) + + return response.text, response.status_code + + return _request + + +