]> git.openstreetmap.org Git - nominatim.git/blob - test/python/api/test_api_connection.py
also force nominatim in build dir to venv python
[nominatim.git] / 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')