test/python/api/test_api_connection.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for enhanced connection class for API functions.
   9 """
  10 from pathlib import Path
  11 import pytest
  12 import pytest_asyncio
  13
  14 import sqlalchemy as sa
  15
  16 from nominatim_api import NominatimAPIAsync
  17
  18 @pytest_asyncio.fixture
  19 async def apiobj(temp_db):
  20     """ Create an asynchronous SQLAlchemy engine for the test DB.
  21     """
  22     api = NominatimAPIAsync(Path('/invalid'), {})
  23     yield api
  24     await api.close()
  25
  26
  27 @pytest.mark.asyncio
  28 async def test_run_scalar(apiobj, table_factory):
  29     table_factory('foo', definition='that TEXT', content=(('a', ),))
  30
  31     async with apiobj.begin() as conn:
  32         assert await conn.scalar(sa.text('SELECT * FROM foo')) == 'a'
  33
  34
  35 @pytest.mark.asyncio
  36 async def test_run_execute(apiobj, table_factory):
  37     table_factory('foo', definition='that TEXT', content=(('a', ),))
  38
  39     async with apiobj.begin() as conn:
  40         result = await conn.execute(sa.text('SELECT * FROM foo'))
  41         assert result.fetchone()[0] == 'a'
  42
  43
  44 @pytest.mark.asyncio
  45 async def test_get_property_existing_cached(apiobj, table_factory):
  46     table_factory('nominatim_properties',
  47                   definition='property TEXT, value TEXT',
  48                   content=(('dbv', '96723'), ))
  49
  50     async with apiobj.begin() as conn:
  51         assert await conn.get_property('dbv') == '96723'
  52
  53         await conn.execute(sa.text('TRUNCATE nominatim_properties'))
  54
  55         assert await conn.get_property('dbv') == '96723'
  56
  57
  58 @pytest.mark.asyncio
  59 async def test_get_property_existing_uncached(apiobj, table_factory):
  60     table_factory('nominatim_properties',
  61                   definition='property TEXT, value TEXT',
  62                   content=(('dbv', '96723'), ))
  63
  64     async with apiobj.begin() as conn:
  65         assert await conn.get_property('dbv') == '96723'
  66
  67         await conn.execute(sa.text("UPDATE nominatim_properties SET value = '1'"))
  68
  69         assert await conn.get_property('dbv', cached=False) == '1'
  70
  71
  72 @pytest.mark.asyncio
  73 @pytest.mark.parametrize('param', ['foo', 'DB:server_version'])
  74 async def test_get_property_missing(apiobj, table_factory, param):
  75     table_factory('nominatim_properties',
  76                   definition='property TEXT, value TEXT')
  77
  78     async with apiobj.begin() as conn:
  79         with pytest.raises(ValueError):
  80             await conn.get_property(param)
  81
  82
  83 @pytest.mark.asyncio
  84 async def test_get_db_property_existing(apiobj):
  85     async with apiobj.begin() as conn:
  86         assert await conn.get_db_property('server_version') > 0
  87
  88
  89 @pytest.mark.asyncio
  90 async def test_get_db_property_existing(apiobj):
  91     async with apiobj.begin() as conn:
  92         with pytest.raises(ValueError):
  93             await conn.get_db_property('dfkgjd.rijg')