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:
"""
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
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.
"""
cli.nominatim(module_dir='',
osm2pgsql_path=str(self.build_dir / 'osm2pgsql' / 'osm2pgsql'),
cli_args=cmdline,
- phpcgi_path='',
environ=self.test_env)
from asgi_lifespan import LifespanManager
import httpx
- async def _request(endpoint, params, project_dir, environ):
+ 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)
-
- return response.text, response.status_code
-
- return _request
-
-
- def create_api_request_func_sanic(self):
- import nominatim.server.sanic.server
-
- async def _request(endpoint, params, project_dir, environ):
- app = nominatim.server.sanic.server.get_application(project_dir, environ)
-
- _, response = await app.asgi_client.get(f"/{endpoint}", params=params)
+ response = await client.get(f"/{endpoint}", params=params,
+ headers=http_headers)
return response.text, response.status_code
import nominatim.server.falcon.server
import falcon.testing
- async def _request(endpoint, params, project_dir, environ):
+ 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)
+ response = await conductor.get(f"/{endpoint}", params=params,
+ headers=http_headers)
return response.text, response.status_code