future=True)
+ async def close(self) -> None:
+ """ Close all active connections to the database. The NominatimAPIAsync
+ object remains usable after closing. If a new API functions is
+ called, new connections are created.
+ """
+ await self.engine.dispose()
+
+
async def status(self) -> StatusResult:
""" Return the status of the database.
"""
self.async_api = NominatimAPIAsync(project_dir, environ)
+ def close(self) -> None:
+ """ Close all active connections to the database. The NominatimAPIAsync
+ object remains usable after closing. If a new API functions is
+ called, new connections are created.
+ """
+ asyncio.get_event_loop().run_until_complete(self.async_api.close())
+
+
def status(self) -> StatusResult:
""" Return the status of the database.
"""
async with engine.begin() as conn:
status.data_updated = await _get_database_date(conn)
status.database_version = await _get_database_version(conn)
- except asyncpg.PostgresError as err:
- return StatusResult(700, str(err))
+ except asyncpg.PostgresError:
+ return StatusResult(700, 'No database')
return status
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Helper fixtures for API call tests.
+"""
+from pathlib import Path
+import pytest
+import time
+
+from nominatim.api import NominatimAPI
+
+@pytest.fixture
+def apiobj(temp_db):
+ """ Create an asynchronous SQLAlchemy engine for the test DB.
+ """
+ api = NominatimAPI(Path('/invalid'), {})
+ yield api
+ api.close()
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Tests for the status API call.
+"""
+from pathlib import Path
+import datetime as dt
+import pytest
+
+from nominatim.version import version_str
+from nominatim.api import NominatimAPI
+
+def test_status_no_extra_info(apiobj, table_factory):
+ table_factory('import_status',
+ definition="lastimportdate timestamp with time zone NOT NULL")
+ table_factory('nominatim_properties',
+ definition='property TEXT, value TEXT')
+
+ result = apiobj.status()
+
+ assert result.status == 0
+ assert result.message == 'OK'
+ assert result.software_version == version_str()
+ assert result.database_version is None
+ assert result.data_updated is None
+
+
+def test_status_full(apiobj, table_factory):
+ table_factory('import_status',
+ definition="lastimportdate timestamp with time zone NOT NULL",
+ content=(('2022-12-07 15:14:46+01',),))
+ table_factory('nominatim_properties',
+ definition='property TEXT, value TEXT',
+ content=(('database_version', '99.5.4-2'), ))
+
+ result = apiobj.status()
+
+ assert result.status == 0
+ assert result.message == 'OK'
+ assert result.software_version == version_str()
+ assert result.database_version == '99.5.4-2'
+ assert result.data_updated == dt.datetime(2022, 12, 7, 14, 14, 46, 0, tzinfo=dt.timezone.utc)
+
+
+def test_status_database_not_found(monkeypatch):
+ monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'dbname=rgjdfkgjedkrgdfkngdfkg')
+
+ api = NominatimAPI(Path('/invalid'), {})
+
+ result = api.status()
+
+ assert result.status == 700
+ assert result.message == 'No database'
+ assert result.software_version == version_str()
+ assert result.database_version is None
+ assert result.data_updated is None
correct functionionality. They use a lot of monkeypatching to avoid executing
the actual functions.
"""
+import importlib
import pytest
import nominatim.indexer.indexer
assert mock.called == 1
-def test_cli_serve_command(cli_call, mock_func_factory):
+def test_cli_serve_php(cli_call, mock_func_factory):
func = mock_func_factory(nominatim.cli, 'run_php_server')
cli_call('serve') == 0
assert func.called == 1
+def test_cli_serve_sanic(cli_call, mock_func_factory):
+ mod = pytest.importorskip("sanic")
+ func = mock_func_factory(mod.Sanic, "run")
+
+ cli_call('serve', '--engine', 'sanic') == 0
+
+ assert func.called == 1
+
+
+def test_cli_serve_starlette_custom_server(cli_call, mock_func_factory):
+ pytest.importorskip("starlette")
+ mod = pytest.importorskip("uvicorn")
+ func = mock_func_factory(mod, "run")
+
+ cli_call('serve', '--engine', 'starlette', '--server', 'foobar:4545') == 0
+
+ assert func.called == 1
+ assert func.last_kwargs['host'] == 'foobar'
+ assert func.last_kwargs['port'] == 4545
+
+
+def test_cli_serve_starlette_custom_server_bad_port(cli_call, mock_func_factory):
+ pytest.importorskip("starlette")
+ mod = pytest.importorskip("uvicorn")
+ func = mock_func_factory(mod, "run")
+
+ cli_call('serve', '--engine', 'starlette', '--server', 'foobar:45:45') == 1
+
+
+@pytest.mark.parametrize("engine", ['falcon', 'starlette'])
+def test_cli_serve_uvicorn_based(cli_call, engine, mock_func_factory):
+ pytest.importorskip(engine)
+ mod = pytest.importorskip("uvicorn")
+ func = mock_func_factory(mod, "run")
+
+ cli_call('serve', '--engine', engine) == 0
+
+ assert func.called == 1
+ assert func.last_kwargs['host'] == '127.0.0.1'
+ assert func.last_kwargs['port'] == 8088
+
def test_cli_export_command(cli_call, mock_run_legacy):
assert cli_call('export', '--output-all-postcodes') == 0
"""
Tests for API access commands of command-line interface wrapper.
"""
+import json
import pytest
import nominatim.clicmd.api
+import nominatim.api
+from nominatim.apicmd.status import StatusResult
@pytest.mark.parametrize("endpoint", (('search', 'reverse', 'lookup', 'details', 'status')))
('details', '--node', '1'),
('details', '--way', '1'),
('details', '--relation', '1'),
- ('details', '--place_id', '10001'),
- ('status',)])
-class TestCliApiCall:
+ ('details', '--place_id', '10001')])
+class TestCliApiCallPhp:
@pytest.fixture(autouse=True)
def setup_cli_call(self, params, cli_call, mock_func_factory, tmp_path):
assert self.run_nominatim() == 1
+class TestCliStatusCall:
+
+ @pytest.fixture(autouse=True)
+ def setup_status_mock(self, monkeypatch):
+ monkeypatch.setattr(nominatim.api.NominatimAPI, 'status',
+ lambda self: StatusResult(200, 'OK'))
+
+
+ def test_status_simple(self, cli_call, tmp_path):
+ result = cli_call('status', '--project-dir', str(tmp_path))
+
+ assert result == 0
+
+
+ def test_status_json_format(self, cli_call, tmp_path, capsys):
+ result = cli_call('status', '--project-dir', str(tmp_path),
+ '--format', 'json')
+
+ assert result == 0
+
+ json.loads(capsys.readouterr().out)
+
+
QUERY_PARAMS = {
'search': ('--query', 'somewhere'),
'reverse': ('--lat', '20', '--lon', '30'),
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Tests for formatting results for the V1 API.
+"""
+import datetime as dt
+import pytest
+
+import nominatim.result_formatter.v1 as format_module
+from nominatim.apicmd.status import StatusResult
+from nominatim.version import version_str
+
+STATUS_FORMATS = {'text', 'json'}
+
+class TestStatusResultFormat:
+
+
+ @pytest.fixture(autouse=True)
+ def make_formatter(self):
+ self.formatter = format_module.create(StatusResult)
+
+
+ def test_format_list(self):
+ assert set(self.formatter.list_formats()) == STATUS_FORMATS
+
+
+ @pytest.mark.parametrize('fmt', list(STATUS_FORMATS))
+ def test_supported(self, fmt):
+ assert self.formatter.supports_format(fmt)
+
+
+ def test_unsupported(self):
+ assert not self.formatter.supports_format('gagaga')
+
+
+ def test_format_text(self):
+ assert self.formatter.format(StatusResult(0, 'message here'), 'text') == 'message here'
+
+
+ def test_format_json_minimal(self):
+ status = StatusResult(700, 'Bad format.')
+
+ result = self.formatter.format(status, 'json')
+
+ assert result == '{"status": 700, "message": "Bad format.", "software_version": "%s"}' % (version_str())
+
+
+ def test_format_json_full(self):
+ status = StatusResult(0, 'OK')
+ status.data_updated = dt.datetime(2010, 2, 7, 20, 20, 3, 0, tzinfo=dt.timezone.utc)
+ status.database_version = '5.6'
+
+ result = self.formatter.format(status, 'json')
+
+ assert result == '{"status": 0, "message": "OK", "data_updated": "2010-02-07T20:20:03+00:00", "software_version": "%s", "database_version": "5.6"}' % (version_str())
projects / nominatim.git / commitdiff