add tests for new arm and export Python functions

nominatim/clicmd/admin.py

@@ -89,19 +89,22 @@ class AdminFuncs:
 
         api = napi.NominatimAPI(args.project_dir)
 
-        if args.target != 'reverse':
-            for _ in range(1000):
-                api.reverse((random.uniform(-90, 90), random.uniform(-180, 180)),
-                            address_details=True)
+        try:
+            if args.target != 'reverse':
+                for _ in range(1000):
+                    api.reverse((random.uniform(-90, 90), random.uniform(-180, 180)),
+                                address_details=True)
 
-        if args.target != 'search':
-            from ..tokenizer import factory as tokenizer_factory
+            if args.target != 'search':
+                from ..tokenizer import factory as tokenizer_factory
 
-            tokenizer = tokenizer_factory.get_tokenizer_for_db(args.config)
-            with connect(args.config.get_libpq_dsn()) as conn:
-                words = tokenizer.most_frequent_words(conn, 1000)
+                tokenizer = tokenizer_factory.get_tokenizer_for_db(args.config)
+                with connect(args.config.get_libpq_dsn()) as conn:
+                    words = tokenizer.most_frequent_words(conn, 1000)
 
-            for word in words:
-                api.search(word)
+                for word in words:
+                    api.search(word)
+        finally:
+            api.close()
 
         return 0

nominatim/clicmd/export.py

@@ -52,6 +52,8 @@ RANK_TO_OUTPUT_MAP = {
 class QueryExport:
     """\
     Export places as CSV file from the database.
+
+
     """
 
     def add_args(self, parser: argparse.ArgumentParser) -> None:
@@ -63,7 +65,8 @@ class QueryExport:
         group.add_argument('--output-format',
                            default='street;suburb;city;county;state;country',
                            help=("Semicolon-separated list of address types "
-                                 "(see --output-type)."))
+                                 "(see --output-type). Additionally accepts:"
+                                 "placeid,postcode"))
         group.add_argument('--language',
                            help=("Preferred language for output "
                                  "(use local name, if omitted)"))
@@ -91,46 +94,49 @@ async def export(args: NominatimArgs) -> int:
 
     api = napi.NominatimAPIAsync(args.project_dir)
 
-    output_range = RANK_RANGE_MAP[args.output_type]
+    try:
+        output_range = RANK_RANGE_MAP[args.output_type]
 
-    writer = init_csv_writer(args.output_format)
+        writer = init_csv_writer(args.output_format)
 
-    async with api.begin() as conn, api.begin() as detail_conn:
-        t = conn.t.placex
+        async with api.begin() as conn, api.begin() as detail_conn:
+            t = conn.t.placex
 
-        sql = sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
-                    t.c.class_, t.c.type, t.c.admin_level,
-                    t.c.address, t.c.extratags,
-                    t.c.housenumber, t.c.postcode, t.c.country_code,
-                    t.c.importance, t.c.wikipedia, t.c.indexed_date,
-                    t.c.rank_address, t.c.rank_search,
-                    t.c.centroid)\
-                 .where(t.c.linked_place_id == None)\
-                 .where(t.c.rank_address.between(*output_range))
+            sql = sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
+                        t.c.class_, t.c.type, t.c.admin_level,
+                        t.c.address, t.c.extratags,
+                        t.c.housenumber, t.c.postcode, t.c.country_code,
+                        t.c.importance, t.c.wikipedia, t.c.indexed_date,
+                        t.c.rank_address, t.c.rank_search,
+                        t.c.centroid)\
+                     .where(t.c.linked_place_id == None)\
+                     .where(t.c.rank_address.between(*output_range))
 
-        parent_place_id = await get_parent_id(conn, args.node, args.way, args.relation)
-        if parent_place_id:
-            taddr = conn.t.addressline
+            parent_place_id = await get_parent_id(conn, args.node, args.way, args.relation)
+            if parent_place_id:
+                taddr = conn.t.addressline
 
-            sql = sql.join(taddr, taddr.c.place_id == t.c.place_id)\
-                     .where(taddr.c.address_place_id == parent_place_id)\
-                     .where(taddr.c.isaddress)
+                sql = sql.join(taddr, taddr.c.place_id == t.c.place_id)\
+                         .where(taddr.c.address_place_id == parent_place_id)\
+                         .where(taddr.c.isaddress)
 
-        if args.restrict_to_country:
-            sql = sql.where(t.c.country_code == args.restrict_to_country.lower())
+            if args.restrict_to_country:
+                sql = sql.where(t.c.country_code == args.restrict_to_country.lower())
 
-        results = []
-        for row in await conn.execute(sql):
-            result = create_from_placex_row(row, ReverseResult)
-            if result is not None:
-                results.append(result)
+            results = []
+            for row in await conn.execute(sql):
+                result = create_from_placex_row(row, ReverseResult)
+                if result is not None:
+                    results.append(result)
 
-            if len(results) == 1000:
+                if len(results) == 1000:
+                    await dump_results(detail_conn, results, writer, args.language)
+                    results = []
+
+            if results:
                 await dump_results(detail_conn, results, writer, args.language)
-                results = []
-
-        if results:
-            await dump_results(detail_conn, results, writer, args.language)
+    finally:
+        await api.close()
 
     return 0
 
@@ -159,9 +165,11 @@ async def dump_results(conn: napi.SearchConnection,
 
         result.localize(locale)
         for line in (result.address_rows or []):
-            if line.isaddress and line.local_name\
-               and line.rank_address in RANK_TO_OUTPUT_MAP:
-                data[RANK_TO_OUTPUT_MAP[line.rank_address]] = line.local_name
+            if line.isaddress and line.local_name:
+                if line.category[1] == 'postcode':
+                    data['postcode'] = line.local_name
+                elif line.rank_address in RANK_TO_OUTPUT_MAP:
+                    data[RANK_TO_OUTPUT_MAP[line.rank_address]] = line.local_name
 
         writer.writerow(data)
 

test/python/api/test_export.py

@@ -0,0 +1,72 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2023 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Tests for export CLI function.
+"""
+import pytest
+
+import nominatim.cli
+
+@pytest.fixture
+def run_export(tmp_path, capsys):
+    def _exec(args):
+        assert 0 == nominatim.cli.nominatim(module_dir='MODULE NOT AVAILABLE',
+                                            osm2pgsql_path='OSM2PGSQL NOT AVAILABLE',
+                                            cli_args=['export', '--project-dir', str(tmp_path)]
+                                                     + args)
+        return capsys.readouterr().out.split('\r\n')
+
+    return _exec
+
+
+@pytest.fixture(autouse=True)
+def setup_database_with_context(apiobj):
+    apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+                     class_='highway', type='residential',  name='Street',
+                     country_code='pl', postcode='55674',
+                     rank_search=27, rank_address=26)
+    apiobj.add_address_placex(332, fromarea=False, isaddress=False,
+                              distance=0.0034,
+                              place_id=1000, osm_type='N', osm_id=3333,
+                              class_='place', type='suburb', name='Smallplace',
+                              country_code='pl', admin_level=13,
+                              rank_search=24, rank_address=23)
+    apiobj.add_address_placex(332, fromarea=True, isaddress=True,
+                              place_id=1001, osm_type='N', osm_id=3334,
+                              class_='place', type='city', name='Bigplace',
+                              country_code='pl',
+                              rank_search=17, rank_address=16)
+
+
+def test_export_default(run_export):
+    csv = run_export([])
+
+    assert csv == ['street,suburb,city,county,state,country', 'Street,,Bigplace,,,', '']
+
+
+def test_export_output_type(run_export):
+    csv = run_export(['--output-type', 'city'])
+
+    assert csv == ['street,suburb,city,county,state,country', ',,Bigplace,,,', '']
+
+
+def test_export_output_format(run_export):
+    csv = run_export(['--output-format', 'placeid;street;nothing;postcode'])
+
+    assert csv == ['placeid,street,nothing,postcode', '332,Street,,55674', '']
+
+
+def test_export_restrict_to_node_good(run_export):
+    csv = run_export(['--restrict-to-osm-node', '3334'])
+
+    assert csv == ['street,suburb,city,county,state,country', 'Street,,Bigplace,,,', '']
+
+
+def test_export_restrict_to_node_not_address(run_export):
+    csv = run_export(['--restrict-to-osm-node', '3333'])
+
+    assert csv == ['street,suburb,city,county,state,country', '']

test/python/api/test_warm.py

@@ -0,0 +1,33 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2023 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Tests for warm-up CLI function.
+"""
+import pytest
+
+import nominatim.cli
+
+@pytest.fixture(autouse=True)
+def setup_database_with_context(apiobj, table_factory):
+    table_factory('word',
+                  definition='word_id INT, word_token TEXT, type TEXT, word TEXT, info JSONB',
+                  content=[(55, 'test', 'W', 'test', None),
+                           (2, 'test', 'w', 'test', None)])
+
+    apiobj.add_data('properties',
+                    [{'property': 'tokenizer', 'value': 'icu'},
+                     {'property': 'tokenizer_import_normalisation', 'value': ':: lower();'},
+                     {'property': 'tokenizer_import_transliteration', 'value': "'1' > '/1/'; 'ä' > 'ä '"},
+                    ])
+
+
+@pytest.mark.parametrize('args', [['--search-only'], ['--reverse-only']])
+def test_warm_all(tmp_path, args):
+    assert 0 == nominatim.cli.nominatim(module_dir='MODULE NOT AVAILABLE',
+                                        osm2pgsql_path='OSM2PGSQL NOT AVAILABLE',
+                                        cli_args=['admin', '--project-dir', str(tmp_path),
+                                                  '--warm'] + args)

test/python/cli/conftest.py

@@ -46,7 +46,7 @@ class DummyTokenizer:
 
 
 @pytest.fixture
-def cli_call(src_dir):
+def cli_call():
     """ Call the nominatim main function with the correct paths set.
         Returns a function that can be called with the desired CLI arguments.
     """