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