#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2023 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Helper fixtures for API call tests.
import pytest
import time
-from nominatim.api import NominatimAPI
+import nominatim.api as napi
+
+class APITester:
+
+ def __init__(self):
+ self.api = napi.NominatimAPI(Path('/invalid'), {})
+ self.async_to_sync(self.api._async_api.setup_database())
+
+
+ def async_to_sync(self, func):
+ """ Run an asynchronous function until completion using the
+ internal loop of the API.
+ """
+ return self.api._loop.run_until_complete(func)
+
+
+ def add_data(self, table, data):
+ """ Insert data into the given table.
+ """
+ sql = getattr(self.api._async_api._tables, table).insert()
+ self.async_to_sync(self.exec_async(sql, data))
+
+
+ async def exec_async(self, sql, *args, **kwargs):
+ async with self.api._async_api.begin() as conn:
+ return await conn.execute(sql, *args, **kwargs)
+
+
+ async def create_tables(self):
+ async with self.api._async_api._engine.begin() as conn:
+ await conn.run_sync(self.api._async_api._tables.meta.create_all)
+
@pytest.fixture
-def apiobj(temp_db):
+def apiobj(temp_db_with_extensions):
""" Create an asynchronous SQLAlchemy engine for the test DB.
"""
- api = NominatimAPI(Path('/invalid'), {})
- yield api
- api.close()
+ testapi = APITester()
+ testapi.async_to_sync(testapi.create_tables())
+ yield testapi
+ testapi.api.close()
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2023 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the status API call.
import pytest
from nominatim.version import NOMINATIM_VERSION, NominatimVersion
-from nominatim.api import NominatimAPI
+import nominatim.api as napi
-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()
+def test_status_no_extra_info(apiobj):
+ result = apiobj.api.status()
assert result.status == 0
assert result.message == 'OK'
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'), ))
+def test_status_full(apiobj):
+ import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0, tzinfo=dt.timezone.utc)
+ apiobj.add_data('import_status',
+ [{'lastimportdate': import_date}])
+ apiobj.add_data('properties',
+ [{'property': 'database_version', 'value': '99.5.4-2'}])
- result = apiobj.status()
+ result = apiobj.api.status()
assert result.status == 0
assert result.message == 'OK'
assert result.software_version == NOMINATIM_VERSION
assert result.database_version == NominatimVersion(99, 5, 4, 2)
- assert result.data_updated == dt.datetime(2022, 12, 7, 14, 14, 46, 0, tzinfo=dt.timezone.utc)
+ assert result.data_updated == import_date
def test_status_database_not_found(monkeypatch):
monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'dbname=rgjdfkgjedkrgdfkngdfkg')
- api = NominatimAPI(Path('/invalid'), {})
+ api = napi.NominatimAPI(Path('/invalid'), {})
result = api.status()