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
  13 import sqlalchemy as sa
  14
  15
  16 @pytest.mark.asyncio
  17 async def test_run_scalar(api, table_factory):
  18     table_factory('foo', definition='that TEXT', content=(('a', ),))
  19
  20     async with api.begin() as conn:
  21         assert await conn.scalar(sa.text('SELECT * FROM foo')) == 'a'
  22
  23
  24 @pytest.mark.asyncio
  25 async def test_run_execute(api, table_factory):
  26     table_factory('foo', definition='that TEXT', content=(('a', ),))
  27
  28     async with api.begin() as conn:
  29         result = await conn.execute(sa.text('SELECT * FROM foo'))
  30         assert result.fetchone()[0] == 'a'
  31
  32
  33 @pytest.mark.asyncio
  34 async def test_get_property_existing_cached(api, table_factory):
  35     table_factory('nominatim_properties',
  36                   definition='property TEXT, value TEXT',
  37                   content=(('dbv', '96723'), ))
  38
  39     async with api.begin() as conn:
  40         assert await conn.get_property('dbv') == '96723'
  41
  42         await conn.execute(sa.text('TRUNCATE nominatim_properties'))
  43
  44         assert await conn.get_property('dbv') == '96723'
  45
  46
  47 @pytest.mark.asyncio
  48 async def test_get_property_existing_uncached(api, table_factory):
  49     table_factory('nominatim_properties',
  50                   definition='property TEXT, value TEXT',
  51                   content=(('dbv', '96723'), ))
  52
  53     async with api.begin() as conn:
  54         assert await conn.get_property('dbv') == '96723'
  55
  56         await conn.execute(sa.text("UPDATE nominatim_properties SET value = '1'"))
  57
  58         assert await conn.get_property('dbv', cached=False) == '1'
  59
  60
  61 @pytest.mark.asyncio
  62 @pytest.mark.parametrize('param', ['foo', 'DB:server_version'])
  63 async def test_get_property_missing(api, table_factory, param):
  64     table_factory('nominatim_properties',
  65                   definition='property TEXT, value TEXT')
  66
  67     async with api.begin() as conn:
  68         with pytest.raises(ValueError):
  69             await conn.get_property(param)
  70
  71
  72 @pytest.mark.asyncio
  73 async def test_get_db_property_existing(api):
  74     async with api.begin() as conn:
  75         assert await conn.get_db_property('server_version') > 0
  76
  77
  78 @pytest.mark.asyncio
  79 async def test_get_db_property_existing(api):
  80     async with api.begin() as conn:
  81         with pytest.raises(ValueError):
  82             await conn.get_db_property('dfkgjd.rijg')